diff --git a/volatility3/framework/plugins/linux/check_idt.py b/volatility3/framework/plugins/linux/check_idt.py index dbdb0e9be..e199d98d3 100644 --- a/volatility3/framework/plugins/linux/check_idt.py +++ b/volatility3/framework/plugins/linux/check_idt.py @@ -10,7 +10,6 @@ from volatility3.framework import interfaces, renderers, symbols from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints from volatility3.framework.symbols import linux -from volatility3.plugins.linux import lsmod vollog = logging.getLogger(__name__) @@ -34,14 +33,16 @@ class Check_idt(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + 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) ), - requirements.PluginRequirement( - name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0) - ), ] @staticmethod @@ -82,10 +83,10 @@ class Check_idt(interfaces.plugins.PluginInterface): vmlinux = self.context.modules[self.config["kernel"]] - modules = lsmod.Lsmod.list_modules(self.context, vmlinux.name) - - handlers = linux.LinuxUtilities.generate_kernel_handler_info( - self.context, vmlinux.name, modules + known_modules = linux_utilities_modules.Modules.run_modules_scanners( + context=self.context, + kernel_module_name=self.config["kernel"], + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, ) idt_table_size = 256 @@ -134,19 +135,24 @@ class Check_idt(interfaces.plugins.PluginInterface): module_name = renderers.NotAvailableValue() symbol_name = renderers.NotAvailableValue() else: - module_name, symbol_name = ( - linux_utilities_modules.Modules.lookup_module_address( - self.context, vmlinux.name, handlers, idt_addr + module_info, symbol_name = ( + linux_utilities_modules.Modules.module_lookup_by_address( + self.context, vmlinux.name, known_modules, idt_addr ) ) + if module_info: + module_name = module_info.name + else: + module_name = renderers.NotAvailableValue() + yield ( 0, [ format_hints.Hex(i), format_hints.Hex(idt_addr), module_name, - symbol_name, + symbol_name or renderers.NotAvailableValue(), ], ) diff --git a/volatility3/framework/plugins/linux/check_modules.py b/volatility3/framework/plugins/linux/check_modules.py index 76f75ec50..44cb568e6 100644 --- a/volatility3/framework/plugins/linux/check_modules.py +++ b/volatility3/framework/plugins/linux/check_modules.py @@ -33,7 +33,7 @@ class Check_modules(plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + version=(3, 0, 0), ), ] @@ -41,7 +41,7 @@ class Check_modules(plugins.PluginInterface): @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.get_kset_modules, removal_date="2025-09-25", - replacement_version=(2, 0, 0), + replacement_version=(3, 0, 0), ) def get_kset_modules( cls, context: interfaces.context.ContextInterface, vmlinux_name: str diff --git a/volatility3/framework/plugins/linux/hidden_modules.py b/volatility3/framework/plugins/linux/hidden_modules.py index 9891bf138..985d4cfcb 100644 --- a/volatility3/framework/plugins/linux/hidden_modules.py +++ b/volatility3/framework/plugins/linux/hidden_modules.py @@ -10,7 +10,6 @@ from volatility3.framework import renderers, interfaces, exceptions, deprecation from volatility3.framework.constants import architectures from volatility3.framework.renderers import format_hints from volatility3.framework.configuration import requirements -from volatility3.plugins.linux import lsmod vollog = logging.getLogger(__name__) @@ -29,13 +28,10 @@ class Hidden_modules(interfaces.plugins.PluginInterface): description="Linux kernel", architectures=architectures.LINUX_ARCHS, ), - requirements.PluginRequirement( - name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0) - ), requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + version=(3, 0, 0), ), ] @@ -43,7 +39,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface): @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.get_modules_memory_boundaries, removal_date="2025-09-25", - replacement_version=(2, 0, 0), + replacement_version=(3, 0, 0), ) def get_modules_memory_boundaries( context: interfaces.context.ContextInterface, @@ -56,7 +52,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface): @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.get_module_address_alignment, removal_date="2025-09-25", - replacement_version=(2, 0, 0), + replacement_version=(3, 0, 0), ) @classmethod def _get_module_address_alignment( @@ -84,7 +80,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface): @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.get_hidden_modules, removal_date="2025-09-25", - replacement_version=(2, 0, 0), + replacement_version=(3, 0, 0), ) @classmethod def get_hidden_modules( @@ -124,7 +120,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface): @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.validate_alignment_patterns, removal_date="2025-09-25", - replacement_version=(2, 0, 0), + replacement_version=(3, 0, 0), ) def _validate_alignment_patterns( addresses: Iterable[int], @@ -163,7 +159,9 @@ class Hidden_modules(interfaces.plugins.PluginInterface): known_module_addresses = { vmlinux_layer.canonicalize(module.vol.offset) - for module in lsmod.Lsmod.list_modules(context, vmlinux_module_name) + for module in linux_utilities_modules.Modules.list_modules( + context, vmlinux_module_name + ) } return known_module_addresses diff --git a/volatility3/framework/plugins/linux/keyboard_notifiers.py b/volatility3/framework/plugins/linux/keyboard_notifiers.py index 8fd2846c1..215704350 100644 --- a/volatility3/framework/plugins/linux/keyboard_notifiers.py +++ b/volatility3/framework/plugins/linux/keyboard_notifiers.py @@ -9,7 +9,6 @@ from volatility3.framework import interfaces, renderers, exceptions from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints from volatility3.framework.symbols import linux -from volatility3.plugins.linux import lsmod vollog = logging.getLogger(__name__) @@ -30,10 +29,12 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + version=(3, 0, 0), ), - requirements.PluginRequirement( - name="lsmod", plugin=lsmod.Lsmod, version=(2, 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) @@ -43,12 +44,6 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface): def _generator(self): vmlinux = self.context.modules[self.config["kernel"]] - modules = lsmod.Lsmod.list_modules(self.context, vmlinux.name) - - handlers = linux.LinuxUtilities.generate_kernel_handler_info( - self.context, vmlinux.name, modules - ) - try: knl_addr = vmlinux.object_from_symbol("keyboard_notifier_list") except exceptions.SymbolError: @@ -65,6 +60,12 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface): vollog.error("The head of the keyboard notifier list is paged out.") return + known_modules = linux_utilities_modules.Modules.run_modules_scanners( + context=self.context, + kernel_module_name=self.config["kernel"], + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, + ) + knl = vmlinux.object( object_type="atomic_notifier_head", offset=knl_addr.vol.offset, @@ -76,13 +77,25 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface): ): call_addr = call_back.notifier_call - module_name, symbol_name = ( - linux_utilities_modules.Modules.lookup_module_address( - self.context, vmlinux.name, handlers, call_addr + module_info, symbol_name = ( + linux_utilities_modules.Modules.module_lookup_by_address( + self.context, vmlinux.name, known_modules, call_addr ) ) - yield (0, [format_hints.Hex(call_addr), module_name, symbol_name]) + if module_info: + module_name = module_info.name + else: + module_name = renderers.NotAvailableValue() + + yield ( + 0, + [ + format_hints.Hex(call_addr), + module_name, + symbol_name or renderers.NotAvailableValue(), + ], + ) def run(self): return renderers.TreeGrid( diff --git a/volatility3/framework/plugins/linux/kthreads.py b/volatility3/framework/plugins/linux/kthreads.py index 60d24f06e..06e94b221 100644 --- a/volatility3/framework/plugins/linux/kthreads.py +++ b/volatility3/framework/plugins/linux/kthreads.py @@ -5,14 +5,14 @@ import logging from typing import List import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules -from volatility3.framework import constants, exceptions, interfaces, renderers +from volatility3.framework import exceptions, interfaces, renderers from volatility3.framework.configuration import requirements from volatility3.framework.interfaces import plugins from volatility3.framework.renderers import format_hints from volatility3.framework.symbols import linux from volatility3.framework.constants import architectures from volatility3.framework.objects import utility -from volatility3.plugins.linux import pslist, lsmod +from volatility3.plugins.linux import pslist vollog = logging.getLogger(__name__) @@ -34,7 +34,12 @@ class Kthreads(plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + 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) @@ -42,28 +47,24 @@ class Kthreads(plugins.PluginInterface): requirements.PluginRequirement( name="pslist", plugin=pslist.PsList, version=(4, 0, 0) ), - requirements.PluginRequirement( - name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0) - ), ] def _generator(self): vmlinux = self.context.modules[self.config["kernel"]] - modules = lsmod.Lsmod.list_modules(self.context, vmlinux.name) - handlers = linux.LinuxUtilities.generate_kernel_handler_info( - self.context, vmlinux.name, modules - ) - - kthread_type = vmlinux.get_type( - vmlinux.symbol_table_name + constants.BANG + "kthread" - ) + kthread_type = vmlinux.get_type("kthread") if not kthread_type.has_member("threadfn"): raise exceptions.VolatilityException( "Unsupported kthread implementation. This plugin only works with kernels >= 5.8" ) + known_modules = linux_utilities_modules.Modules.run_modules_scanners( + context=self.context, + kernel_module_name=self.config["kernel"], + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, + ) + for task in pslist.PsList.list_tasks( self.context, vmlinux.name, include_threads=True ): @@ -86,9 +87,7 @@ class Kthreads(plugins.PluginInterface): if not (threadfn and threadfn.is_readable()): continue - task_name = utility.array_to_string(task.comm) - - thread_name = task_name + thread_name = utility.array_to_string(task.comm) # kernels >= 5.17 in d6986ce24fc00b0638bd29efe8fb7ba7619ed2aa full_name was added to kthread if kthread.has_member("full_name"): @@ -101,18 +100,23 @@ class Kthreads(plugins.PluginInterface): f"full_name pointer for thread at {kthread.vol.offset:#x} is paged out." ) - module_name, symbol_name = ( - linux_utilities_modules.Modules.lookup_module_address( - self.context, vmlinux.name, handlers, threadfn + module_info, symbol_name = ( + linux_utilities_modules.Modules.module_lookup_by_address( + self.context, vmlinux.name, known_modules, threadfn ) ) + if module_info: + module_name = module_info.name + else: + module_name = renderers.NotAvailableValue() + fields = [ task.pid, thread_name, format_hints.Hex(threadfn), module_name, - symbol_name, + symbol_name or renderers.NotAvailableValue(), ] yield 0, fields diff --git a/volatility3/framework/plugins/linux/lsmod.py b/volatility3/framework/plugins/linux/lsmod.py index b4f881801..466bfa0b4 100644 --- a/volatility3/framework/plugins/linux/lsmod.py +++ b/volatility3/framework/plugins/linux/lsmod.py @@ -33,14 +33,14 @@ class Lsmod(plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + version=(3, 0, 0), ), ] @classmethod @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.list_modules, - replacement_version=(2, 0, 0), + replacement_version=(3, 0, 0), removal_date="2025-09-25", ) def list_modules( diff --git a/volatility3/framework/plugins/linux/modxview.py b/volatility3/framework/plugins/linux/modxview.py index f6a6f7727..ed21acfd1 100644 --- a/volatility3/framework/plugins/linux/modxview.py +++ b/volatility3/framework/plugins/linux/modxview.py @@ -34,7 +34,22 @@ spot modules presence and taints.""" requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + version=(3, 0, 0), + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherer_lsmod", + component=linux_utilities_modules.ModuleGathererLsmod, + version=(1, 0, 0), + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherer_sysfs", + component=linux_utilities_modules.ModuleGathererSysFs, + version=(1, 0, 0), + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherer_scanner", + component=linux_utilities_modules.ModuleGathererScanner, + version=(1, 0, 0), ), requirements.VersionRequirement( name="linux-tainting", component=tainting.Tainting, version=(1, 0, 0) @@ -50,7 +65,7 @@ spot modules presence and taints.""" @classmethod @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.flatten_run_modules_results, - replacement_version=(2, 0, 0), + replacement_version=(3, 0, 0), removal_date="2025-09-25", ) def flatten_run_modules_results( @@ -73,7 +88,7 @@ spot modules presence and taints.""" @classmethod @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.run_modules_scanners, - replacement_version=(2, 0, 0), + replacement_version=(3, 0, 0), removal_date="2025-09-25", ) def run_modules_scanners( @@ -89,35 +104,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_gatherers = [ + linux_utilities_modules.ModuleGathererLsmod, + linux_utilities_modules.ModuleGathererSysFs, + linux_utilities_modules.ModuleGathererScanner, + ] run_results = linux_utilities_modules.Modules.run_modules_scanners( - self.context, kernel_name, flatten=False + context=self.context, + kernel_module_name=self.config["kernel"], + 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 ["lsmod", "check_modules", "hidden_modules"]: + 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.name]: # 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.name) else: - aggregated_modules[mod_info.offset] = [plugin_name] + aggregated_modules[mod_info.offset] = [gatherer.name] - 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, ) @@ -125,7 +147,7 @@ spot modules presence and taints.""" taints = ",".join( tainting.Tainting.get_taints_parsed( self.context, - kernel_name, + self.config["kernel"], module.taints, True, ) @@ -136,9 +158,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.name in gatherers, + linux_utilities_modules.ModuleGathererSysFs.name in gatherers, + linux_utilities_modules.ModuleGathererScanner.name in gatherers, taints or NotAvailableValue(), ), ) @@ -149,7 +171,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/netfilter.py b/volatility3/framework/plugins/linux/netfilter.py index 9c0055a54..e8a33be61 100644 --- a/volatility3/framework/plugins/linux/netfilter.py +++ b/volatility3/framework/plugins/linux/netfilter.py @@ -726,7 +726,7 @@ class Netfilter(interfaces.plugins.PluginInterface): _version = (1, 1, 1) - _required_linux_utilities_modules_version = (2, 0, 0) + _required_linux_utilities_modules_version = (3, 0, 0) _required_linuxutils_version = (2, 1, 0) _required_lsmod_version = (2, 0, 0) _required_linuxnet_version = (1, 0, 0) diff --git a/volatility3/framework/plugins/linux/tracing/ftrace.py b/volatility3/framework/plugins/linux/tracing/ftrace.py index 17766cc74..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 @@ -79,7 +79,12 @@ class CheckFtrace(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + 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", @@ -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 @@ -223,7 +225,9 @@ class CheckFtrace(interfaces.plugins.PluginInterface): return known_modules = linux_utilities_modules.Modules.run_modules_scanners( - self.context, kernel_name, run_hidden_modules=True + context=self.context, + kernel_module_name=self.config["kernel"], + 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 fe6d11af9..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 @@ -52,7 +52,12 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + version=(3, 0, 0), + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), ), ] @@ -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]]: @@ -229,7 +234,9 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): return known_modules = linux_utilities_modules.Modules.run_modules_scanners( - self.context, kernel_name, run_hidden_modules=False + context=self.context, + kernel_module_name=self.config["kernel"], + 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 281d46eda..7d30b84ee 100644 --- a/volatility3/framework/plugins/linux/tty_check.py +++ b/volatility3/framework/plugins/linux/tty_check.py @@ -12,7 +12,6 @@ from volatility3.framework.interfaces import plugins from volatility3.framework.objects import utility from volatility3.framework.renderers import format_hints from volatility3.framework.symbols import linux -from volatility3.plugins.linux import lsmod vollog = logging.getLogger(__name__) @@ -33,10 +32,12 @@ class tty_check(plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(2, 0, 0), + version=(3, 0, 0), ), - requirements.PluginRequirement( - name="lsmod", plugin=lsmod.Lsmod, version=(2, 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) @@ -46,12 +47,6 @@ class tty_check(plugins.PluginInterface): def _generator(self): vmlinux = self.context.modules[self.config["kernel"]] - modules = lsmod.Lsmod.list_modules(self.context, vmlinux.name) - - handlers = linux.LinuxUtilities.generate_kernel_handler_info( - self.context, vmlinux.name, modules - ) - try: tty_drivers = vmlinux.object_from_symbol("tty_drivers").cast("list_head") except exceptions.SymbolError: @@ -64,6 +59,12 @@ class tty_check(plugins.PluginInterface): "This means you are either analyzing an unsupported kernel version or that your symbol table is corrupt." ) + known_modules = linux_utilities_modules.Modules.run_modules_scanners( + context=self.context, + kernel_module_name=self.config["kernel"], + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, + ) + for tty in tty_drivers.to_list( vmlinux.symbol_table_name + constants.BANG + "tty_driver", "tty_drivers" ): @@ -87,13 +88,23 @@ class tty_check(plugins.PluginInterface): except exceptions.InvalidAddressException: continue - module_name, symbol_name = ( - linux_utilities_modules.Modules.lookup_module_address( - self.context, vmlinux.name, handlers, recv_buf + module_info, symbol_name = ( + linux_utilities_modules.Modules.module_lookup_by_address( + self.context, vmlinux.name, known_modules, recv_buf ) ) - yield (0, (name, format_hints.Hex(recv_buf), module_name, symbol_name)) + if module_info: + module_name = module_info.name + else: + module_name = renderers.NotAvailableValue() + + yield 0, ( + name, + format_hints.Hex(recv_buf), + module_name, + symbol_name or renderers.NotAvailableValue(), + ) def run(self): return renderers.TreeGrid( diff --git a/volatility3/framework/symbols/linux/utilities/modules.py b/volatility3/framework/symbols/linux/utilities/modules.py index ef20ed044..8f1b5b67d 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,37 +22,63 @@ from volatility3.framework import ( exceptions, objects, ) + +from volatility3.framework.configuration import requirements from volatility3.framework.objects import utility from volatility3.framework.symbols.linux import extensions vollog = logging.getLogger(__name__) -class Modules(interfaces.configuration.VersionableInterface): - """Kernel modules related utilities.""" +class ModuleInfo(NamedTuple): + """ + Used to track the name and boundary of a kernel module + """ - _version = (2, 0, 0) + 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) - class ModuleInfo(NamedTuple): + gatherer_return_type = Generator[Union[ModuleInfo, "extensions.module"], None, None] + + # Must be set to a unique, descriptive name of the gathering technique or data structure source + name = None + + @classmethod + @abstractmethod + def gather_modules( + cls, context: interfaces.context.ContextInterface, kernel_module_name: str + ) -> gatherer_return_type: """ - Used to track the name and boundary of a kernel module + This method must return a generator (yield) of each `gatherer_return_type` found from its source """ - offset: int - name: str - start: int - end: int - @staticmethod +class Modules(interfaces.configuration.VersionableInterface): + """Kernel modules related utilities.""" + + _version = (3, 0, 0) + _required_framework_version = (2, 0, 0) + + framework.require_interface_version(*_required_framework_version) + + @classmethod def module_lookup_by_address( + cls, context: interfaces.context.ContextInterface, kernel_module_name: str, modules: Iterable[ModuleInfo], target_address: int, - run_hidden_modules: bool = True, ) -> Optional[Tuple[ModuleInfo, Optional[str]]]: """ Determine if a target address lies in a module memory space. @@ -189,98 +227,80 @@ 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, - ) -> 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 run_modules_scanners( cls, context: interfaces.context.ContextInterface, - kernel_name: str, - run_hidden_modules: bool = True, + kernel_module_name: str, + caller_wanted_gatherers: List[ModuleGathererInterface], flatten: bool = True, ) -> Dict[str, List[ModuleInfo]]: """Run module scanning plugins and aggregate the results. It is designed to not operate any inter-plugin results triage. - Args: - run_hidden_modules: specify if the hidden_modules plugin should be run - Returns: - Dictionary mapping each plugin to its corresponding result - """ + Rules for `caller_wanted_gatherers`: + If `ModuleGatherers.all_gathers_identifier` is specified then every source will be populated - kernel = context.modules[kernel_name] + If empty or an invalid gatherer is specified then a ValueError is thrown + + All gatherer names must be unique + Args: + called_wanted_sources: The list of sources to gather modules. + flatten: Whether to de-duplicate modules across gatherers + Returns: + Dictionary mapping each gatherer to its corresponding result + """ + if not caller_wanted_gatherers: + raise ValueError( + "`caller_wanted_gatherers` must have at least one gatherer." + ) + + if not isinstance(caller_wanted_gatherers, Iterable): + raise ValueError("`caller_wanted_gatherers` must be iterable") + + seen_names = set() + + for gatherer in caller_wanted_gatherers: + if not issubclass(gatherer, ModuleGathererInterface): + raise ValueError( + f"Invalid gatherer sent through `caller_wanted_gatherers`: {gatherer}" + ) + + if not gatherer.name: + raise ValueError( + f"{gatherer} does not have a valid name attribute, which is required. It must be a non-zero length string." + ) + + if gatherer.name in seen_names: + raise ValueError( + f"{gatherer} has a name {gatherer.name} which has already been processed. Names must be unique." + ) + + seen_names.add(gatherer.name) + + kernel = context.modules[kernel_module_name] address_mask = context.layers[kernel.layer_name].address_mask - run_results = {} + run_results: Dict[ModuleGathererInterface, List[ModuleInfo]] = {} - # the kernel module boundaries - run_results["kernel"] = [cls.get_kernel_module_info(context, kernel_name)] + # Walk each source gathering modules + for gatherer in caller_wanted_gatherers: + run_results[gatherer.name] = [] - # lsmod - run_results["lsmod"] = [] + # process each module coming from back the current source + for module in gatherer.gather_modules(context, kernel_module_name): - for module in cls.list_modules(context, kernel_name): - modinfo = cls.get_module_info_for_module(address_mask, module) - if modinfo: - run_results["lsmod"].append(modinfo) + # the kernel sends back a ModuleInfo directly + if isinstance(module, ModuleInfo): + modinfo = module + else: + modinfo = cls.get_module_info_for_module(address_mask, module) - # check_modules - run_results["check_modules"] = [] - - sysfs_modules: dict = cls.get_kset_modules(context, kernel_name) - - for m_offset in sysfs_modules.values(): - module = kernel.object(object_type="module", offset=m_offset, absolute=True) - modinfo = cls.get_module_info_for_module(address_mask, module) - if modinfo: - run_results["check_modules"].append(modinfo) - - # hidden_modules - if run_hidden_modules: - known_modules_addresses = set( - context.layers[kernel.layer_name].canonicalize(modinfo.start) - for modinfo in run_results["kernel"] - + run_results["lsmod"] - + run_results["check_modules"] - ) - modules_memory_boundaries = cls.get_modules_memory_boundaries( - context, kernel_name - ) - run_results["hidden_modules"] = [] - - for module in cls.get_hidden_modules( - context, - kernel_name, - known_modules_addresses, - modules_memory_boundaries, - ): - modinfo = cls.get_module_info_for_module(address_mask, module) if modinfo: - run_results["hidden_modules"].append(modinfo) + run_results[gatherer.name].append(modinfo) if flatten: return cls.flatten_run_modules_results(run_results) @@ -338,7 +358,7 @@ class Modules(interfaces.configuration.VersionableInterface): Returns: List of ModuleInfo objects """ - uniq_modules: List[Modules.ModuleInfo] = [] + uniq_modules: List[ModuleInfo] = [] seen_addresses: int = set() @@ -534,3 +554,133 @@ 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 + """ + + _version = (1, 0, 0) + + name = "Lsmod" + + @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 + """ + + _version = (1, 0, 0) + + name = "SysFs" + + @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 + """ + + _version = (1, 0, 0) + + name = "Scanner" + + @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 + """ + + _version = (1, 0, 0) + + name = "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, + interfaces.configuration.ConfigurableInterface, +): + _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, + ] + + @classmethod + def get_requirements(cls): + reqs = [] + + # for now, all versions are 1, this will be broken out if/when that changes + for gatherer in ModuleGatherers.all_gatherers_identifier: + reqs.append( + requirements.VersionRequirement( + name=gatherer.name.replace(" ", ""), + component=gatherer, + version=(1, 0, 0), + ) + ) + + return reqs