diff --git a/volatility3/framework/plugins/linux/check_idt.py b/volatility3/framework/plugins/linux/check_idt.py index 119531836..e199d98d3 100644 --- a/volatility3/framework/plugins/linux/check_idt.py +++ b/volatility3/framework/plugins/linux/check_idt.py @@ -35,6 +35,11 @@ class Check_idt(interfaces.plugins.PluginInterface): component=linux_utilities_modules.Modules, version=(3, 0, 0), ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), + ), requirements.VersionRequirement( name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0) ), @@ -81,7 +86,7 @@ class Check_idt(interfaces.plugins.PluginInterface): known_modules = linux_utilities_modules.Modules.run_modules_scanners( context=self.context, kernel_module_name=self.config["kernel"], - caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier, + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, ) idt_table_size = 256 diff --git a/volatility3/framework/plugins/linux/keyboard_notifiers.py b/volatility3/framework/plugins/linux/keyboard_notifiers.py index b4f9dd3ca..215704350 100644 --- a/volatility3/framework/plugins/linux/keyboard_notifiers.py +++ b/volatility3/framework/plugins/linux/keyboard_notifiers.py @@ -31,6 +31,11 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface): component=linux_utilities_modules.Modules, version=(3, 0, 0), ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), + ), requirements.VersionRequirement( name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0) ), @@ -58,7 +63,7 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface): known_modules = linux_utilities_modules.Modules.run_modules_scanners( context=self.context, kernel_module_name=self.config["kernel"], - caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier, + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, ) knl = vmlinux.object( diff --git a/volatility3/framework/plugins/linux/kthreads.py b/volatility3/framework/plugins/linux/kthreads.py index 7d6abcecc..06e94b221 100644 --- a/volatility3/framework/plugins/linux/kthreads.py +++ b/volatility3/framework/plugins/linux/kthreads.py @@ -36,6 +36,11 @@ class Kthreads(plugins.PluginInterface): component=linux_utilities_modules.Modules, version=(3, 0, 0), ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), + ), requirements.VersionRequirement( name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0) ), @@ -57,7 +62,7 @@ class Kthreads(plugins.PluginInterface): known_modules = linux_utilities_modules.Modules.run_modules_scanners( context=self.context, kernel_module_name=self.config["kernel"], - caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier, + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, ) for task in pslist.PsList.list_tasks( diff --git a/volatility3/framework/plugins/linux/modxview.py b/volatility3/framework/plugins/linux/modxview.py index e43672974..b04eb74ca 100644 --- a/volatility3/framework/plugins/linux/modxview.py +++ b/volatility3/framework/plugins/linux/modxview.py @@ -36,6 +36,11 @@ spot modules presence and taints.""" component=linux_utilities_modules.Modules, version=(3, 0, 0), ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), + ), requirements.VersionRequirement( name="linux-tainting", component=tainting.Tainting, version=(1, 0, 0) ), @@ -89,44 +94,42 @@ spot modules presence and taints.""" ) def _generator(self): - kernel_name = self.config["kernel"] + kernel = self.context.modules[self.config["kernel"]] - kernel = self.context.modules[kernel_name] - - wanted_sources = [ - linux_utilities_modules.Modules.source_lsmod_identifier, - linux_utilities_modules.Modules.source_sysfs_identifier, - linux_utilities_modules.Modules.source_hidden_identifier, + wanted_gatherers = [ + linux_utilities_modules.ModuleGathererLsmod, + linux_utilities_modules.ModuleGathererSysFs, + linux_utilities_modules.ModuleGathererScanner, ] run_results = linux_utilities_modules.Modules.run_modules_scanners( context=self.context, kernel_module_name=self.config["kernel"], - caller_wanted_sources=wanted_sources, + caller_wanted_gatherers=wanted_gatherers, flatten=False, ) aggregated_modules = {} # We want to be explicit on the plugins results we are interested in - for plugin_name in wanted_sources: + for gatherer in wanted_gatherers: # Iterate over each recovered module - for mod_info in run_results[plugin_name]: + for mod_info in run_results[gatherer]: # Use offsets as unique keys, whether a module # appears in many plugin runs or not if aggregated_modules.get(mod_info.offset, None) is not None: # Append the plugin to the list of originating plugins - aggregated_modules[mod_info.offset].append(plugin_name) + aggregated_modules[mod_info.offset].append(gatherer) else: - aggregated_modules[mod_info.offset] = [plugin_name] + aggregated_modules[mod_info.offset] = [gatherer] - for module_offset, originating_plugins in aggregated_modules.items(): - # Tainting parsing capabilities applied to the module + for module_offset, gatherers in aggregated_modules.items(): module = kernel.object("module", offset=module_offset, absolute=True) + # Tainting parsing capabilities applied to the module if self.config.get("plain_taints"): taints = tainting.Tainting.get_taints_as_plain_string( self.context, - kernel_name, + self.config["kernel"], module.taints, True, ) @@ -134,7 +137,7 @@ spot modules presence and taints.""" taints = ",".join( tainting.Tainting.get_taints_parsed( self.context, - kernel_name, + self.config["kernel"], module.taints, True, ) @@ -145,9 +148,9 @@ spot modules presence and taints.""" ( module.get_name() or NotAvailableValue(), format_hints.Hex(module_offset), - "lsmod" in originating_plugins, - "check_modules" in originating_plugins, - "hidden_modules" in originating_plugins, + linux_utilities_modules.ModuleGathererLsmod in gatherers, + linux_utilities_modules.ModuleGathererSysFs in gatherers, + linux_utilities_modules.ModuleGathererScanner in gatherers, taints or NotAvailableValue(), ), ) @@ -158,7 +161,7 @@ spot modules presence and taints.""" ("Address", format_hints.Hex), ("In procfs", bool), ("In sysfs", bool), - ("Hidden", bool), + ("In scan", bool), ("Taints", str), ] diff --git a/volatility3/framework/plugins/linux/tracing/ftrace.py b/volatility3/framework/plugins/linux/tracing/ftrace.py index 146230f02..c5e4f9ef8 100644 --- a/volatility3/framework/plugins/linux/tracing/ftrace.py +++ b/volatility3/framework/plugins/linux/tracing/ftrace.py @@ -5,7 +5,7 @@ # Public researches: https://i.blackhat.com/USA21/Wednesday-Handouts/us-21-Fixing-A-Memory-Forensics-Blind-Spot-Linux-Kernel-Tracing-wp.pdf import logging -from typing import Dict, List, Generator +from typing import List, Generator from enum import Enum from dataclasses import dataclass @@ -65,7 +65,7 @@ class CheckFtrace(interfaces.plugins.PluginInterface): Investigate the ftrace infrastructure to uncover kernel attached callbacks, which can be leveraged to hook kernel functions and modify their behaviour.""" - _version = (3, 0, 0) + _version = (4, 0, 0) _required_framework_version = (2, 19, 0) @classmethod @@ -81,6 +81,11 @@ class CheckFtrace(interfaces.plugins.PluginInterface): component=linux_utilities_modules.Modules, version=(3, 0, 0), ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), + ), requirements.BooleanRequirement( name="show_ftrace_flags", description="Show ftrace flags associated with an ftrace_ops struct", @@ -127,9 +132,8 @@ class CheckFtrace(interfaces.plugins.PluginInterface): cls, context: interfaces.context.ContextInterface, kernel_module_name: str, - known_modules: Dict[str, List[linux_utilities_modules.Modules.ModuleInfo]], + known_modules: List[linux_utilities_modules.ModuleInfo], ftrace_ops: interfaces.objects.ObjectInterface, - run_hidden_modules: bool = True, ) -> Generator[ParsedFtraceOps, None, None]: """Parse an ftrace_ops struct to highlight ftrace kernel hooking. Iterates over embedded ftrace_func_entry entries, which point to hooked memory areas. @@ -137,8 +141,6 @@ class CheckFtrace(interfaces.plugins.PluginInterface): Args: known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through run_modules_scanners(). ftrace_ops: The ftrace_ops struct to parse - run_hidden_modules: Whether to run the hidden_modules plugin or not. Note: it won't be run, even if specified, \ - if the "hidden_modules" key is present in known_modules. Yields: An iterable of ParsedFtraceOps dataclasses, containing a selection of useful fields (callback, hook, module) related to an ftrace_ops struct @@ -225,7 +227,7 @@ class CheckFtrace(interfaces.plugins.PluginInterface): known_modules = linux_utilities_modules.Modules.run_modules_scanners( context=self.context, kernel_module_name=self.config["kernel"], - caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier, + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, ) for ftrace_ops in self.iterate_ftrace_ops_list(self.context, kernel_name): diff --git a/volatility3/framework/plugins/linux/tracing/tracepoints.py b/volatility3/framework/plugins/linux/tracing/tracepoints.py index 83903b19b..9d4a4a2e3 100644 --- a/volatility3/framework/plugins/linux/tracing/tracepoints.py +++ b/volatility3/framework/plugins/linux/tracing/tracepoints.py @@ -5,7 +5,7 @@ # Public researches: https://i.blackhat.com/USA21/Wednesday-Handouts/us-21-Fixing-A-Memory-Forensics-Blind-Spot-Linux-Kernel-Tracing-wp.pdf import logging -from typing import Dict, Iterable, List, Optional +from typing import Iterable, List, Optional from dataclasses import dataclass import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules @@ -38,7 +38,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): Investigate the tracepoints subsystem to uncover kernel attached probes, which can be leveraged to hook kernel functions and modify their behaviour.""" - _version = (1, 0, 0) + _version = (2, 0, 0) _required_framework_version = (2, 19, 0) @classmethod @@ -54,6 +54,11 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): component=linux_utilities_modules.Modules, version=(3, 0, 0), ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), + ), ] @classmethod @@ -96,7 +101,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): cls, context: interfaces.context.ContextInterface, kernel_module_name: str, - known_modules: Dict[str, List[linux_utilities_modules.Modules.ModuleInfo]], + known_modules: List[linux_utilities_modules.ModuleInfo], tracepoint: interfaces.objects.ObjectInterface, run_hidden_modules: bool = True, ) -> Optional[Iterable[ParsedTracepointFunc]]: @@ -231,7 +236,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): known_modules = linux_utilities_modules.Modules.run_modules_scanners( context=self.context, kernel_module_name=self.config["kernel"], - caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier, + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, ) tracepoints = self.iterate_tracepoints_array(self.context, kernel_name) diff --git a/volatility3/framework/plugins/linux/tty_check.py b/volatility3/framework/plugins/linux/tty_check.py index f4b581756..7d30b84ee 100644 --- a/volatility3/framework/plugins/linux/tty_check.py +++ b/volatility3/framework/plugins/linux/tty_check.py @@ -34,6 +34,11 @@ class tty_check(plugins.PluginInterface): component=linux_utilities_modules.Modules, version=(3, 0, 0), ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), + ), requirements.VersionRequirement( name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0) ), @@ -57,7 +62,7 @@ class tty_check(plugins.PluginInterface): known_modules = linux_utilities_modules.Modules.run_modules_scanners( context=self.context, kernel_module_name=self.config["kernel"], - caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier, + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, ) for tty in tty_drivers.to_list( diff --git a/volatility3/framework/symbols/linux/utilities/modules.py b/volatility3/framework/symbols/linux/utilities/modules.py index dac52dce3..a930bef20 100644 --- a/volatility3/framework/symbols/linux/utilities/modules.py +++ b/volatility3/framework/symbols/linux/utilities/modules.py @@ -1,6 +1,18 @@ import logging import warnings -from typing import Iterable, Iterator, List, Optional, Tuple, NamedTuple, Dict, Set +from typing import ( + Iterable, + Iterator, + List, + Optional, + Tuple, + NamedTuple, + Dict, + Set, + Generator, + Union, +) +from abc import ABCMeta, abstractmethod from volatility3 import framework from volatility3.framework import ( @@ -10,12 +22,44 @@ from volatility3.framework import ( exceptions, objects, ) + from volatility3.framework.objects import utility from volatility3.framework.symbols.linux import extensions vollog = logging.getLogger(__name__) +class ModuleInfo(NamedTuple): + """ + Used to track the name and boundary of a kernel module + """ + + offset: int + name: str + start: int + end: int + + +class ModuleGathererInterface( + interfaces.configuration.VersionableInterface, metaclass=ABCMeta +): + _version = (1, 0, 0) + _required_framework_version = (2, 0, 0) + + framework.require_interface_version(*_required_framework_version) + + gatherer_return_type = Generator[Union[ModuleInfo, "extensions.module"], None, None] + + @classmethod + @abstractmethod + def gather_modules( + cls, context: interfaces.context.ContextInterface, kernel_module_name: str + ) -> gatherer_return_type: + """ + This method must return a generator (yield) of each `gatherer_return_type` found from its source + """ + + class Modules(interfaces.configuration.VersionableInterface): """Kernel modules related utilities.""" @@ -24,31 +68,6 @@ class Modules(interfaces.configuration.VersionableInterface): framework.require_interface_version(*_required_framework_version) - # Valid sources of kernel modules to send to `run_module_scanners` - source_kernel_identifier = "kernel" - source_lsmod_identifier = "lsmod" - source_sysfs_identifier = "check_modules" - source_hidden_identifier = "hidden_modules" - - # With few exceptions, rootkit checking plugins want all sources - # This provides a stable identifier as new sources are added over time - all_sources_identifier = [ - source_kernel_identifier, - source_lsmod_identifier, - source_sysfs_identifier, - source_hidden_identifier, - ] - - class ModuleInfo(NamedTuple): - """ - Used to track the name and boundary of a kernel module - """ - - offset: int - name: str - start: int - end: int - @classmethod def module_lookup_by_address( cls, @@ -204,138 +223,41 @@ class Modules(interfaces.configuration.VersionableInterface): end = start + module.get_core_size() - return Modules.ModuleInfo(module.vol.offset, mod_name, start, end) - - @staticmethod - def get_kernel_module_info( - context: interfaces.context.ContextInterface, - kernel_module_name: str, - ) -> Iterator[ModuleInfo]: - """ - Returns a ModuleInfo instance that encodes the kernel - This is required to map function pointers to the kerenl executable - """ - kernel = context.modules[kernel_module_name] - - address_mask = context.layers[kernel.layer_name].address_mask - - start_addr = kernel.object_from_symbol("_text") - start_addr = start_addr.vol.offset & address_mask - - end_addr = kernel.object_from_symbol("_etext") - end_addr = end_addr.vol.offset & address_mask - - return [ - Modules.ModuleInfo( - start_addr, constants.linux.KERNEL_NAME, start_addr, end_addr - ) - ] + return ModuleInfo(module.vol.offset, mod_name, start, end) @classmethod - def _get_hidden_modules_results( - cls, - context: str, - kernel_module_name: str, - run_results: Dict[str, List[ModuleInfo]], - ): - known_modules_addresses = set() - - kernel = context.modules[kernel_module_name] - - # Walk each sources' results - for results in run_results.values(): - for modinfo in results: - address = context.layers[kernel.layer_name].canonicalize(modinfo.start) - known_modules_addresses.add(address) - - modules_memory_boundaries = cls.get_modules_memory_boundaries( - context, kernel_module_name - ) - - hidden_results = [] - - address_mask = context.layers[kernel.layer_name].address_mask - - for module in cls.get_hidden_modules( - context, - kernel_module_name, - known_modules_addresses, - modules_memory_boundaries, - ): - modinfo = cls.get_module_info_for_module(address_mask, module) - if modinfo: - hidden_results.append(modinfo) - - return hidden_results - - @classmethod - def _get_list_modules( - cls, context: interfaces.context.ContextInterface, kernel_module_name: str - ) -> List[ModuleInfo]: + def _validate_gatherers(cls, caller_wanted_gatherers) -> List[str]: """ - Gather `module` instances from lsmod + Called by `run_modules_scanners` to validate the caller supplied gatherers list + An exception is thrown if an empty gatherers list is given or a list containing an invalid source """ - yield from cls.list_modules(context, kernel_module_name) - - @classmethod - def _get_sysfs_modules( - cls, context: interfaces.context.ContextInterface, kernel_module_name: str - ) -> List[ModuleInfo]: - """ - Gather the `module` instances from sysfs - """ - kernel = context.modules[kernel_module_name] - - sysfs_modules: dict = cls.get_kset_modules(context, kernel_module_name) - - for m_offset in sysfs_modules.values(): - yield kernel.object(object_type="module", offset=m_offset, absolute=True) - - @classmethod - def _validated_sources(cls, caller_wanted_sources) -> List[str]: - """ - Called by `run_modules_scanners` to validate the caller supplied sources list - An exception is thrown if an empty source list is given or a list containing an invalid source - """ - if not caller_wanted_sources: - raise ValueError("`caller_wanted_sources` must have at least one source.") - - if ( - len(caller_wanted_sources) == 1 - and caller_wanted_sources[0] == Modules.source_hidden_identifier - ): + if not caller_wanted_gatherers: raise ValueError( - f"{Modules.source_hidden_identifier} cannot be the only source or there is nothing to compare against." + "`caller_wanted_gatherers` must have at least one gatherer." ) - wanted_sources = [] - - for source in caller_wanted_sources: - if source not in Modules.all_sources_identifier: + for gatherer in caller_wanted_gatherers: + if gatherer not in ModuleGatherers.all_gatherers_identifier: raise ValueError( - f"Invalid source sent through `caller_wanted_sources`: {source}" + f"Invalid gatherer sent through `caller_wanted_gatherers`: {gatherer}" ) - wanted_sources.append(source) - - return wanted_sources - @classmethod def run_modules_scanners( cls, context: interfaces.context.ContextInterface, kernel_module_name: str, - caller_wanted_sources: List[str], + caller_wanted_gatherers: List[ModuleGathererInterface], flatten: bool = True, - ) -> Dict[str, List[ModuleInfo]]: + ) -> Dict[ModuleGathererInterface, List[ModuleInfo]]: """Run module scanning plugins and aggregate the results. It is designed to not operate any inter-plugin results triage. Rules for `caller_wanted_sources`: - If `Modules.all_sources_identifier` is specified then every source will be populated + If `ModuleGathers.all_gathers_identifier` is specified then every source will be populated - If `Modules.source_hidden_identifier` is in the list, then at least one other sources must be + If `ModuleGathers.Scanner` is in the list, then at least one other sources must be specified so a comparison will be populated If empty or an invalid source is specified then a ValueError is thrown @@ -346,52 +268,30 @@ class Modules(interfaces.configuration.VersionableInterface): Returns: Dictionary mapping each plugin to its corresponding result """ - - module_gatherers = { - Modules.source_kernel_identifier: cls.get_kernel_module_info, - Modules.source_lsmod_identifier: cls._get_list_modules, - Modules.source_sysfs_identifier: cls._get_sysfs_modules, - } + # Throws ValueError if invalid gatherers sent in + Modules._validate_gatherers(caller_wanted_gatherers) kernel = context.modules[kernel_module_name] address_mask = context.layers[kernel.layer_name].address_mask - wanted_sources = Modules._validated_sources(caller_wanted_sources) + run_results: Dict[ModuleGathererInterface, List[ModuleInfo]] = {} - run_results = {} - - run_hidden_modules = False - - # Special case hidden modules since it gathers modules on its own - if Modules.source_hidden_identifier in wanted_sources: - run_hidden_modules = True - wanted_sources.remove(Modules.source_hidden_identifier) - - # Walk each source, gathering modules - for wanted_source in wanted_sources: - run_results[wanted_source] = [] - - gatherer = module_gatherers[wanted_source] + # Walk each source gathering modules + for gatherer in caller_wanted_gatherers: + run_results[gatherer] = [] # process each module coming from back the current source - for module in gatherer(context, kernel_module_name): + for module in gatherer.gather_modules(context, kernel_module_name): + # the kernel sends back a ModuleInfo directly - if wanted_source == Modules.source_kernel_identifier: + if gatherer == ModuleGathererKernel: modinfo = module else: modinfo = cls.get_module_info_for_module(address_mask, module) if modinfo: - run_results[wanted_source].append(modinfo) - - # run hidden modules against the other sources - if run_hidden_modules: - run_results[Modules.source_hidden_identifier] = ( - cls._get_hidden_modules_results( - context, kernel_module_name, run_results - ) - ) + run_results[gatherer].append(modinfo) if flatten: return cls.flatten_run_modules_results(run_results) @@ -449,7 +349,7 @@ class Modules(interfaces.configuration.VersionableInterface): Returns: List of ModuleInfo objects """ - uniq_modules: List[Modules.ModuleInfo] = [] + uniq_modules: List[ModuleInfo] = [] seen_addresses: int = set() @@ -645,3 +545,98 @@ class Modules(interfaces.configuration.VersionableInterface): True if all the addresses meet the alignment """ return all(addr % address_alignment == 0 for addr in addresses) + + +class ModuleGathererLsmod(ModuleGathererInterface): + """ + Gathers modules from the main kernel list + """ + + @classmethod + def gather_modules( + cls, context: interfaces.context.ContextInterface, kernel_module_name: str + ) -> ModuleGathererInterface.gatherer_return_type: + yield from Modules.list_modules(context, kernel_module_name) + + +class ModuleGathererSysFs(ModuleGathererInterface): + """ + Gathers modules from the sysfs /sys/modules objects + """ + + @classmethod + def gather_modules( + cls, context: interfaces.context.ContextInterface, kernel_module_name: str + ) -> ModuleGathererInterface.gatherer_return_type: + kernel = context.modules[kernel_module_name] + + sysfs_modules: dict = Modules.get_kset_modules(context, kernel_module_name) + + for m_offset in sysfs_modules.values(): + yield kernel.object(object_type="module", offset=m_offset, absolute=True) + + +class ModuleGathererScanner(ModuleGathererInterface): + """ + Gathers modules by scanning memory + """ + + @classmethod + def gather_modules( + cls, context: interfaces.context.ContextInterface, kernel_module_name: str + ) -> ModuleGathererInterface.gatherer_return_type: + modules_memory_boundaries = Modules.get_modules_memory_boundaries( + context, kernel_module_name + ) + + # Send in an empty list to not filter on any modules + yield from Modules.get_hidden_modules( + context=context, + vmlinux_module_name=kernel_module_name, + known_module_addresses=[], + modules_memory_boundaries=modules_memory_boundaries, + ) + + +class ModuleGathererKernel(ModuleGathererInterface): + """ + Creates a ModuleInfo instance for the kernel so that plugins + can determine when function pointers reference the kernel + """ + + @classmethod + def gather_modules( + cls, context: interfaces.context.ContextInterface, kernel_module_name: str + ) -> ModuleGathererInterface.gatherer_return_type: + """ + Returns a ModuleInfo instance that encodes the kernel + This is required to map function pointers to the kerenl executable + """ + kernel = context.modules[kernel_module_name] + + address_mask = context.layers[kernel.layer_name].address_mask + + start_addr = kernel.object_from_symbol("_text") + start_addr = start_addr.vol.offset & address_mask + + end_addr = kernel.object_from_symbol("_etext") + end_addr = end_addr.vol.offset & address_mask + + yield ModuleInfo(start_addr, constants.linux.KERNEL_NAME, start_addr, end_addr) + + +class ModuleGatherers(interfaces.configuration.VersionableInterface): + _version = (1, 0, 0) + _required_framework_version = (2, 0, 0) + + framework.require_interface_version(*_required_framework_version) + + # Valid sources of cores kernel module gatherers to send to `run_module_scanners` + # With few exceptions, rootkit checking plugins want all sources + # This provides a stable identifier as new sources are added over time + all_gatherers_identifier = [ + ModuleGathererLsmod, + ModuleGathererSysFs, + ModuleGathererScanner, + ModuleGathererKernel, + ]