diff --git a/volatility3/framework/__init__.py b/volatility3/framework/__init__.py index 60acb9465..0bbdefa43 100644 --- a/volatility3/framework/__init__.py +++ b/volatility3/framework/__init__.py @@ -11,12 +11,9 @@ import inspect import logging import os import traceback -import functools -import warnings -from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, TypeVar +from typing import Any, Dict, Generator, List, Optional, Tuple, Type, TypeVar -from volatility3.framework import constants, exceptions, interfaces -from volatility3.framework.configuration import requirements +from volatility3.framework import constants, interfaces if ( sys.version_info.major != constants.REQUIRED_PYTHON_VERSION[0] @@ -66,61 +63,6 @@ def require_interface_version(*args) -> None: ) -class Deprecation: - """Deprecation related methods.""" - - @staticmethod - def deprecated_method( - replacement: Callable, - replacement_version: Tuple[int, int, int] = None, - additional_information: str = "", - ): - """A decorator for marking functions as deprecated. - - Args: - replacement: The replacement function overriding the deprecated API, in the form of a Callable (typically a method) - replacement_version: The "replacement" base class version that the deprecated method expects before proxying to it. This implies that "replacement" is a method from a class that inherits from VersionableInterface. - additional_information: Information appended at the end of the deprecation message - """ - - def decorator(deprecated_func): - @functools.wraps(deprecated_func) - def wrapper(*args, **kwargs): - nonlocal replacement, replacement_version, additional_information - # Prevent version mismatches between deprecated (proxy) methods and the ones they proxy - if ( - replacement_version is not None - and callable(replacement) - and hasattr(replacement, "__self__") - ): - replacement_base_class = replacement.__self__ - - # Verify that the base class inherits from VersionableInterface - if inspect.isclass(replacement_base_class) and issubclass( - replacement_base_class, - interfaces.configuration.VersionableInterface, - ): - # SemVer check - if not requirements.VersionRequirement.matches_required( - replacement_version, replacement_base_class.version - ): - raise exceptions.VersionMismatchException( - deprecated_func, - replacement_base_class, - replacement_version, - "This is a bug, the deprecated call needs to be removed and the caller needs to update their code to use the new method.", - ) - - deprecation_msg = f"Method \"{deprecated_func.__module__ + '.' + deprecated_func.__qualname__}\" is deprecated, use \"{replacement.__module__ + '.' + replacement.__qualname__}\" instead. {additional_information}" - warnings.warn(deprecation_msg, FutureWarning) - # Return the wrapped function with its original arguments - return deprecated_func(*args, **kwargs) - - return wrapper - - return decorator - - class NonInheritable: def __init__(self, value: Any, cls: Type) -> None: self.default_value = value diff --git a/volatility3/framework/deprecation.py b/volatility3/framework/deprecation.py new file mode 100644 index 000000000..e2e91a0eb --- /dev/null +++ b/volatility3/framework/deprecation.py @@ -0,0 +1,90 @@ +# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# + +# This file contains the Deprecation class used to deprecate methods in an orderly manner + +import warnings +import functools +import inspect + +from typing import Callable, Tuple + +from volatility3.framework import interfaces, exceptions +from volatility3.framework.configuration import requirements + + +def method_being_removed(message: str, removal_date: str): + """A decorator for marking functions as being removed in the future and without a replacement. + Callers to this function should explicitly list the API paths that should be used instead. + + Args: + message: A message added to the standard deprecation warning. Should include the replacement API paths + removal_date: A YYYY-MM-DD formatted date of when the function will be removed from the framework + """ + + def decorator(deprecated_func): + @functools.wraps(deprecated_func) + def wrapper(*args, **kwargs): + warnings.warn( + f"This API ({deprecated_func.__module__}.{deprecated_func.__qualname__}) will be removed in the first release after {removal_date}. {message}", + FutureWarning, + ) + return deprecated_func(*args, **kwargs) + + return wrapper + + return decorator + + +def deprecated_method( + replacement: Callable, + removal_date: str, + replacement_version: Tuple[int, int, int] = None, + additional_information: str = "", +): + """A decorator for marking functions as deprecated. + + Args: + replacement: The replacement function overriding the deprecated API, in the form of a Callable (typically a method) + removal_date: A YYYY-MM-DD formatted date of when the function will be removed from the framework + replacement_version: The "replacement" base class version that the deprecated method expects before proxying to it. This implies that "replacement" is a method from a class that inherits from VersionableInterface. + additional_information: Information appended at the end of the deprecation message + """ + + def decorator(deprecated_func): + @functools.wraps(deprecated_func) + def wrapper(*args, **kwargs): + nonlocal replacement, replacement_version, additional_information + # Prevent version mismatches between deprecated (proxy) methods and the ones they proxy + if ( + replacement_version is not None + and callable(replacement) + and hasattr(replacement, "__self__") + ): + replacement_base_class = replacement.__self__ + + # Verify that the base class inherits from VersionableInterface + if inspect.isclass(replacement_base_class) and issubclass( + replacement_base_class, + interfaces.configuration.VersionableInterface, + ): + # SemVer check + if not requirements.VersionRequirement.matches_required( + replacement_version, replacement_base_class.version + ): + raise exceptions.VersionMismatchException( + deprecated_func, + replacement_base_class, + replacement_version, + "This is a bug, the deprecated call needs to be removed and the caller needs to update their code to use the new method.", + ) + + deprecation_msg = f"Method \"{deprecated_func.__module__ + '.' + deprecated_func.__qualname__}\" is deprecated and will be removed in the first release after {removal_date}, use \"{replacement.__module__ + '.' + replacement.__qualname__}\" instead. {additional_information}" + warnings.warn(deprecation_msg, FutureWarning) + # Return the wrapped function with its original arguments + return deprecated_func(*args, **kwargs) + + return wrapper + + return decorator diff --git a/volatility3/framework/plugins/linux/check_idt.py b/volatility3/framework/plugins/linux/check_idt.py index 653ecc081..dbdb0e9be 100644 --- a/volatility3/framework/plugins/linux/check_idt.py +++ b/volatility3/framework/plugins/linux/check_idt.py @@ -34,7 +34,7 @@ class Check_idt(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(1, 0, 0), + version=(2, 0, 0), ), requirements.VersionRequirement( name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0) diff --git a/volatility3/framework/plugins/linux/check_modules.py b/volatility3/framework/plugins/linux/check_modules.py index 23da29680..76f75ec50 100644 --- a/volatility3/framework/plugins/linux/check_modules.py +++ b/volatility3/framework/plugins/linux/check_modules.py @@ -3,14 +3,15 @@ # import logging -from typing import List +from typing import List, Dict -from volatility3.framework import interfaces, renderers, exceptions, constants +import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules +from volatility3.framework import interfaces, renderers, deprecation from volatility3.framework.configuration import requirements from volatility3.framework.interfaces import plugins from volatility3.framework.objects import utility from volatility3.framework.renderers import format_hints -from volatility3.plugins.linux import lsmod +from volatility3.framework.symbols.linux import extensions vollog = logging.getLogger(__name__) @@ -18,7 +19,7 @@ vollog = logging.getLogger(__name__) class Check_modules(plugins.PluginInterface): """Compares module list to sysfs info, if available""" - _version = (1, 0, 0) + _version = (2, 0, 0) _required_framework_version = (2, 0, 0) @classmethod @@ -29,58 +30,34 @@ class Check_modules(plugins.PluginInterface): description="Linux kernel", architectures=["Intel32", "Intel64"], ), - 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), ), ] @classmethod + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.get_kset_modules, + removal_date="2025-09-25", + replacement_version=(2, 0, 0), + ) def get_kset_modules( cls, context: interfaces.context.ContextInterface, vmlinux_name: str - ): - vmlinux = context.modules[vmlinux_name] - - try: - module_kset = vmlinux.object_from_symbol("module_kset") - except exceptions.SymbolError: - module_kset = None - - if not module_kset: - raise TypeError( - "This plugin requires the module_kset structure. This structure is not present in the supplied symbol table. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupt." - ) - - ret = {} - - kobj_off = vmlinux.get_type("module_kobject").relative_child_offset("kobj") - - for kobj in module_kset.list.to_list( - vmlinux.symbol_table_name + constants.BANG + "kobject", "entry" - ): - mod_kobj = vmlinux.object( - object_type="module_kobject", - offset=kobj.vol.offset - kobj_off, - absolute=True, - ) - - mod = mod_kobj.mod - - try: - name = utility.pointer_to_string(kobj.name, 32) - except exceptions.InvalidAddressException: - continue - - if kobj.name and kobj.reference_count() > 2: - ret[name] = mod - - return ret + ) -> Dict[str, extensions.module]: + return linux_utilities_modules.Modules.get_kset_modules(context, vmlinux_name) def _generator(self): - kset_modules = self.get_kset_modules(self.context, self.config["kernel"]) + kset_modules = linux_utilities_modules.Modules.get_kset_modules( + self.context, self.config["kernel"] + ) lsmod_modules = set( str(utility.array_to_string(modules.name)) - for modules in lsmod.Lsmod.list_modules(self.context, self.config["kernel"]) + for modules in linux_utilities_modules.Modules.list_modules( + self.context, self.config["kernel"] + ) ) for mod_name in set(kset_modules.keys()).difference(lsmod_modules): diff --git a/volatility3/framework/plugins/linux/hidden_modules.py b/volatility3/framework/plugins/linux/hidden_modules.py index e1ba40926..9891bf138 100644 --- a/volatility3/framework/plugins/linux/hidden_modules.py +++ b/volatility3/framework/plugins/linux/hidden_modules.py @@ -3,7 +3,10 @@ # import logging from typing import List, Set, Tuple, Iterable -from volatility3.framework import renderers, interfaces, exceptions, objects +from volatility3.framework.symbols.linux.utilities import ( + modules as linux_utilities_modules, +) +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 @@ -16,7 +19,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface): """Carves memory to find hidden kernel modules""" _required_framework_version = (2, 10, 0) - _version = (1, 0, 1) + _version = (2, 0, 0) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -29,46 +32,32 @@ class Hidden_modules(interfaces.plugins.PluginInterface): 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), + ), ] - @classmethod + @staticmethod + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.get_modules_memory_boundaries, + removal_date="2025-09-25", + replacement_version=(2, 0, 0), + ) def get_modules_memory_boundaries( - cls, context: interfaces.context.ContextInterface, vmlinux_module_name: str, - ) -> Tuple[int]: - """Determine the boundaries of the module allocation area - - Args: - context: The context to retrieve required elements (layers, symbol tables) from - vmlinux_module_name: The name of the kernel module on which to operate - - Returns: - A tuple containing the minimum and maximum addresses for the module allocation area. - """ - vmlinux = context.modules[vmlinux_module_name] - if vmlinux.has_symbol("mod_tree"): - # Kernel >= 5.19 58d208de3e8d87dbe196caf0b57cc58c7a3836ca - mod_tree = vmlinux.object_from_symbol("mod_tree") - modules_addr_min = mod_tree.addr_min - modules_addr_max = mod_tree.addr_max - elif vmlinux.has_symbol("module_addr_min"): - # 2.6.27 <= kernel < 5.19 3a642e99babe0617febb6f402e1e063479f489db - modules_addr_min = vmlinux.object_from_symbol("module_addr_min") - modules_addr_max = vmlinux.object_from_symbol("module_addr_max") - - if isinstance(modules_addr_min, objects.Void): - raise exceptions.VolatilityException( - "Your ISF symbols lack type information. You may need to update the" - "ISF using the latest version of dwarf2json" - ) - else: - raise exceptions.VolatilityException( - "Cannot find the module memory allocation area. Unsupported kernel" - ) - - return modules_addr_min, modules_addr_max + ) -> Tuple[int, int]: + return linux_utilities_modules.Modules.get_modules_memory_boundaries( + context, vmlinux_module_name + ) + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.get_module_address_alignment, + removal_date="2025-09-25", + replacement_version=(2, 0, 0), + ) @classmethod def _get_module_address_alignment( cls, @@ -88,27 +77,15 @@ class Hidden_modules(interfaces.plugins.PluginInterface): Returns: The struct module alignment """ - # FIXME: When dwarf2json/ISF supports type alignments. Read it directly from the type metadata - # Additionally, while 'context' and 'vmlinux_module_name' are currently unused, they will be - # essential for retrieving type metadata in the future. - return 64 - - @staticmethod - def _validate_alignment_patterns( - addresses: Iterable[int], - address_alignment: int, - ) -> bool: - """Check if the memory addresses meet our alignments patterns - - Args: - addresses: Iterable with the address values - address_alignment: Number of bytes for alignment validation - - Returns: - True if all the addresses meet the alignment - """ - return all(addr % address_alignment == 0 for addr in addresses) + return linux_utilities_modules.get_module_address_alignment( + context, vmlinux_module_name + ) + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.get_hidden_modules, + removal_date="2025-09-25", + replacement_version=(2, 0, 0), + ) @classmethod def get_hidden_modules( cls, @@ -139,54 +116,32 @@ class Hidden_modules(interfaces.plugins.PluginInterface): Yields: module objects """ - vmlinux = context.modules[vmlinux_module_name] - vmlinux_layer = context.layers[vmlinux.layer_name] - - module_addr_min, module_addr_max = modules_memory_boundaries - module_address_alignment = cls._get_module_address_alignment( - context, vmlinux_module_name + return linux_utilities_modules.get_hidden_modules( + vmlinux_module_name, known_module_addresses, modules_memory_boundaries ) - if not cls._validate_alignment_patterns( - known_module_addresses, module_address_alignment - ): - vollog.warning( - f"Module addresses aren't aligned to {module_address_alignment} bytes. " - "Switching to 1 byte aligment scan method." - ) - module_address_alignment = 1 - mkobj_offset = vmlinux.get_type("module").relative_child_offset("mkobj") - mod_offset = vmlinux.get_type("module_kobject").relative_child_offset("mod") - offset_to_mkobj_mod = mkobj_offset + mod_offset - mod_member_template = vmlinux.get_type("module_kobject").child_template("mod") - mod_size = mod_member_template.size - mod_member_data_format = mod_member_template.data_format + @staticmethod + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.validate_alignment_patterns, + removal_date="2025-09-25", + replacement_version=(2, 0, 0), + ) + def _validate_alignment_patterns( + addresses: Iterable[int], + address_alignment: int, + ) -> bool: + """Check if the memory addresses meet our alignments patterns - for module_addr in range( - module_addr_min, module_addr_max, module_address_alignment - ): - if module_addr in known_module_addresses: - continue + Args: + addresses: Iterable with the address values + address_alignment: Number of bytes for alignment validation - try: - # This is just a pre-filter. Module readability and consistency are verified in module.is_valid() - self_referential_bytes = vmlinux_layer.read( - module_addr + offset_to_mkobj_mod, mod_size - ) - self_referential = objects.convert_data_to_value( - self_referential_bytes, int, mod_member_data_format - ) - if self_referential != module_addr: - continue - except ( - exceptions.PagedInvalidAddressException, - exceptions.InvalidAddressException, - ): - continue - - module = vmlinux.object("module", offset=module_addr, absolute=True) - if module and module.is_valid(): - yield module + Returns: + True if all the addresses meet the alignment + """ + return linux_utilities_modules.validate_alignment_patterns( + addresses, address_alignment + ) @classmethod def get_lsmod_module_addresses( @@ -217,10 +172,13 @@ class Hidden_modules(interfaces.plugins.PluginInterface): known_module_addresses = self.get_lsmod_module_addresses( self.context, vmlinux_module_name ) - modules_memory_boundaries = self.get_modules_memory_boundaries( - self.context, vmlinux_module_name + modules_memory_boundaries = ( + linux_utilities_modules.Modules.get_modules_memory_boundaries( + self.context, vmlinux_module_name + ) ) - for module in self.get_hidden_modules( + + for module in linux_utilities_modules.Modules.get_hidden_modules( self.context, vmlinux_module_name, known_module_addresses, diff --git a/volatility3/framework/plugins/linux/keyboard_notifiers.py b/volatility3/framework/plugins/linux/keyboard_notifiers.py index 726280cbe..8fd2846c1 100644 --- a/volatility3/framework/plugins/linux/keyboard_notifiers.py +++ b/volatility3/framework/plugins/linux/keyboard_notifiers.py @@ -30,7 +30,7 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(1, 0, 0), + version=(2, 0, 0), ), requirements.PluginRequirement( name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0) diff --git a/volatility3/framework/plugins/linux/kthreads.py b/volatility3/framework/plugins/linux/kthreads.py index 674eae1e5..60d24f06e 100644 --- a/volatility3/framework/plugins/linux/kthreads.py +++ b/volatility3/framework/plugins/linux/kthreads.py @@ -34,7 +34,7 @@ class Kthreads(plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(1, 0, 0), + version=(2, 0, 0), ), requirements.VersionRequirement( name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0) diff --git a/volatility3/framework/plugins/linux/lsmod.py b/volatility3/framework/plugins/linux/lsmod.py index e9a2a7137..b4f881801 100644 --- a/volatility3/framework/plugins/linux/lsmod.py +++ b/volatility3/framework/plugins/linux/lsmod.py @@ -6,7 +6,8 @@ import logging from typing import List, Iterable -from volatility3.framework import exceptions, renderers, constants, interfaces +import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules +from volatility3.framework import exceptions, renderers, interfaces, deprecation from volatility3.framework.configuration import requirements from volatility3.framework.interfaces import plugins from volatility3.framework.objects import utility @@ -29,35 +30,31 @@ class Lsmod(plugins.PluginInterface): description="Linux kernel", architectures=["Intel32", "Intel64"], ), + requirements.VersionRequirement( + name="linux_utilities_modules", + component=linux_utilities_modules.Modules, + version=(2, 0, 0), + ), ] @classmethod + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.list_modules, + replacement_version=(2, 0, 0), + removal_date="2025-09-25", + ) def list_modules( cls, context: interfaces.context.ContextInterface, vmlinux_module_name: str ) -> Iterable[interfaces.objects.ObjectInterface]: - """Lists all the modules in the primary layer. - - Args: - context: The context to retrieve required elements (layers, symbol tables) from - layer_name: The name of the layer on which to operate - vmlinux_symbols: The name of the table containing the kernel symbols - - Yields: - The modules present in the `layer_name` layer's modules list - - This function will throw a SymbolError exception if kernel module support is not enabled. - """ - vmlinux = context.modules[vmlinux_module_name] - - modules = vmlinux.object_from_symbol(symbol_name="modules").cast("list_head") - - table_name = modules.vol.type_name.split(constants.BANG)[0] - - yield from modules.to_list(table_name + constants.BANG + "module", "list") + return linux_utilities_modules.Modules.list_modules( + context, vmlinux_module_name + ) def _generator(self): try: - for module in self.list_modules(self.context, self.config["kernel"]): + for module in linux_utilities_modules.Modules.list_modules( + self.context, self.config["kernel"] + ): mod_size = module.get_init_size() + module.get_core_size() mod_name = utility.array_to_string(module.name) @@ -65,7 +62,7 @@ class Lsmod(plugins.PluginInterface): yield 0, (format_hints.Hex(module.vol.offset), mod_name, mod_size) except exceptions.SymbolError: - vollog.debug( + vollog.warning( "The required symbol 'module' is not present in symbol table. Please check that kernel modules are enabled for the system under analysis." ) diff --git a/volatility3/framework/plugins/linux/modxview.py b/volatility3/framework/plugins/linux/modxview.py index 0dd503829..f6a6f7727 100644 --- a/volatility3/framework/plugins/linux/modxview.py +++ b/volatility3/framework/plugins/linux/modxview.py @@ -3,8 +3,10 @@ # import logging from typing import List, Dict, Iterator -from volatility3.plugins.linux import lsmod, check_modules, hidden_modules -from volatility3.framework import interfaces + +import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules + +from volatility3.framework import interfaces, deprecation from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue from volatility3.framework.symbols.linux import extensions @@ -29,22 +31,14 @@ spot modules presence and taints.""" description="Linux kernel", architectures=architectures.LINUX_ARCHS, ), + requirements.VersionRequirement( + name="linux_utilities_modules", + component=linux_utilities_modules.Modules, + version=(2, 0, 0), + ), requirements.VersionRequirement( name="linux-tainting", component=tainting.Tainting, version=(1, 0, 0) ), - requirements.PluginRequirement( - name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0) - ), - requirements.PluginRequirement( - name="check_modules", - plugin=check_modules.Check_modules, - version=(1, 0, 0), - ), - requirements.PluginRequirement( - name="hidden_modules", - plugin=hidden_modules.Hidden_modules, - version=(1, 0, 0), - ), requirements.BooleanRequirement( name="plain_taints", description="Display the plain taints string for each module.", @@ -54,6 +48,11 @@ spot modules presence and taints.""" ] @classmethod + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.flatten_run_modules_results, + replacement_version=(2, 0, 0), + removal_date="2025-09-25", + ) def flatten_run_modules_results( cls, run_results: Dict[str, List[extensions.module]], deduplicate: bool = True ) -> Iterator[extensions.module]: @@ -67,15 +66,16 @@ spot modules presence and taints.""" Returns: Iterator of modules objects """ - seen_addresses = set() - for modules in run_results.values(): - for module in modules: - if deduplicate and module.vol.offset in seen_addresses: - continue - seen_addresses.add(module.vol.offset) - yield module + return linux_utilities_modules.Modules.flatten_run_modules_results( + run_results, deduplicate + ) @classmethod + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.run_modules_scanners, + replacement_version=(2, 0, 0), + removal_date="2025-09-25", + ) def run_modules_scanners( cls, context: interfaces.context.ContextInterface, @@ -83,67 +83,37 @@ spot modules presence and taints.""" run_hidden_modules: bool = True, ) -> Dict[str, List[extensions.module]]: """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 - """ - - kernel = context.modules[kernel_name] - run_results = {} - # lsmod - run_results["lsmod"] = list(lsmod.Lsmod.list_modules(context, kernel_name)) - # check_modules - sysfs_modules: dict = check_modules.Check_modules.get_kset_modules( - context, kernel_name + to not operate any inter-plugin results triage.""" + return linux_utilities_modules.Modules.run_modules_scanners( + context, kernel_name, run_hidden_modules ) - ## Convert get_kset_modules() offsets back to module objects - run_results["check_modules"] = [ - kernel.object(object_type="module", offset=m_offset, absolute=True) - for m_offset in sysfs_modules.values() - ] - # hidden_modules - if run_hidden_modules: - known_modules_addresses = set( - context.layers[kernel.layer_name].canonicalize(module.vol.offset) - for module in run_results["lsmod"] + run_results["check_modules"] - ) - modules_memory_boundaries = ( - hidden_modules.Hidden_modules.get_modules_memory_boundaries( - context, kernel_name - ) - ) - run_results["hidden_modules"] = list( - hidden_modules.Hidden_modules.get_hidden_modules( - context, - kernel_name, - known_modules_addresses, - modules_memory_boundaries, - ) - ) - - return run_results def _generator(self): kernel_name = self.config["kernel"] - run_results = self.run_modules_scanners(self.context, kernel_name) + + kernel = self.context.modules[kernel_name] + + run_results = linux_utilities_modules.Modules.run_modules_scanners( + self.context, kernel_name, 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"]: # Iterate over each recovered module - for module in run_results[plugin_name]: + for mod_info in run_results[plugin_name]: # Use offsets as unique keys, whether a module # appears in many plugin runs or not - if aggregated_modules.get(module.vol.offset, None) is not None: + if aggregated_modules.get(mod_info.offset, None) is not None: # Append the plugin to the list of originating plugins - aggregated_modules[module.vol.offset][1].append(plugin_name) + aggregated_modules[mod_info.offset].append(plugin_name) else: - aggregated_modules[module.vol.offset] = (module, [plugin_name]) + aggregated_modules[mod_info.offset] = [plugin_name] - for module_offset, (module, originating_plugins) in aggregated_modules.items(): + for module_offset, originating_plugins in aggregated_modules.items(): # Tainting parsing capabilities applied to the module + module = kernel.object("module", offset=module_offset, absolute=True) + if self.config.get("plain_taints"): taints = tainting.Tainting.get_taints_as_plain_string( self.context, diff --git a/volatility3/framework/plugins/linux/netfilter.py b/volatility3/framework/plugins/linux/netfilter.py index 9c8c9feb0..9c0055a54 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 = (1, 0, 0) + _required_linux_utilities_modules_version = (2, 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 59168d5bc..17766cc74 100644 --- a/volatility3/framework/plugins/linux/tracing/ftrace.py +++ b/volatility3/framework/plugins/linux/tracing/ftrace.py @@ -10,11 +10,9 @@ from enum import Enum from dataclasses import dataclass import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules -from volatility3.plugins.linux import hidden_modules, modxview from volatility3.framework import constants, exceptions, interfaces from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue -from volatility3.framework.symbols.linux import extensions from volatility3.framework.constants import architectures vollog = logging.getLogger(__name__) @@ -67,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 = (2, 0, 0) + _version = (3, 0, 0) _required_framework_version = (2, 19, 0) @classmethod @@ -81,15 +79,7 @@ class CheckFtrace(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(1, 1, 0), - ), - requirements.PluginRequirement( - name="modxview", plugin=modxview.Modxview, version=(1, 0, 0) - ), - requirements.PluginRequirement( - name="hidden_modules", - plugin=hidden_modules.Hidden_modules, - version=(1, 0, 0), + version=(2, 0, 0), ), requirements.BooleanRequirement( name="show_ftrace_flags", @@ -136,8 +126,8 @@ class CheckFtrace(interfaces.plugins.PluginInterface): def parse_ftrace_ops( cls, context: interfaces.context.ContextInterface, - kernel_name: str, - known_modules: Dict[str, List[extensions.module]], + kernel_module_name: str, + known_modules: Dict[str, List[linux_utilities_modules.Modules.ModuleInfo]], ftrace_ops: interfaces.objects.ObjectInterface, run_hidden_modules: bool = True, ) -> Generator[ParsedFtraceOps, None, None]: @@ -145,70 +135,33 @@ class CheckFtrace(interfaces.plugins.PluginInterface): Iterates over embedded ftrace_func_entry entries, which point to hooked memory areas. Args: - known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through modxview.run_modules_scanners(). + 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. + 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 """ - kernel = context.modules[kernel_name] - kernel_layer = context.layers[kernel.layer_name] + kernel = context.modules[kernel_module_name] callback = ftrace_ops.func - callback_symbol = module_address = module_name = None - # Try to lookup within the known modules if the callback address fits - module = linux_utilities_modules.Modules.module_lookup_by_address( - context, - kernel.layer_name, - modxview.Modxview.flatten_run_modules_results(known_modules), - callback, - ) - # Run hidden_modules plugin if a callback origin couldn't be determined (only done once, results are re-used afterwards) - if ( - module is None - and run_hidden_modules - and "hidden_modules" not in known_modules - ): - vollog.info( - "A callback module origin could not be determined. hidden_modules plugin will be run to detect additional modules.", - ) - known_modules_addresses = set( - kernel_layer.canonicalize(module.vol.offset) - for module in modxview.Modxview.flatten_run_modules_results( - known_modules - ) - ) - modules_memory_boundaries = ( - hidden_modules.Hidden_modules.get_modules_memory_boundaries( - context, kernel_name - ) - ) - known_modules["hidden_modules"] = list( - hidden_modules.Hidden_modules.get_hidden_modules( - context, - kernel_name, - known_modules_addresses, - modules_memory_boundaries, - ) - ) - # Lookup the updated list to see if hidden_modules was able - # to find the missing module - module = linux_utilities_modules.Modules.module_lookup_by_address( + mod_info, callback_symbol = ( + linux_utilities_modules.Modules.module_lookup_by_address( context, - kernel.layer_name, - modxview.Modxview.flatten_run_modules_results(known_modules), + kernel_module_name, + known_modules, callback, ) + ) - # Fetch more information about the module - if module is not None: - module_address = module.vol.offset - module_name = module.get_name() - callback_symbol = module.get_symbol_by_address(callback) + if mod_info: + module_address = mod_info.start + module_name = mod_info.name else: - vollog.warning( + callback_symbol = module_address = module_name = None + + vollog.debug( f"Could not determine ftrace_ops@{ftrace_ops.vol.offset:#x} callback {callback:#x} module origin.", ) @@ -264,16 +217,15 @@ if the "hidden_modules" key is present in known_modules. kernel = self.context.modules[kernel_name] if not kernel.has_symbol("ftrace_ops_list"): - raise exceptions.SymbolError( - "ftrace_ops_list", - kernel.symbol_table_name, - 'The provided symbol table does not include the "ftrace_ops_list" symbol. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupted.', + vollog.error( + 'The provided symbol table does not include the "ftrace_ops_list" symbol. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupted.' ) + return - # Do not run hidden_modules by default, but only on failure to find a module - known_modules = modxview.Modxview.run_modules_scanners( - self.context, kernel_name, run_hidden_modules=False + known_modules = linux_utilities_modules.Modules.run_modules_scanners( + self.context, kernel_name, run_hidden_modules=True ) + for ftrace_ops in self.iterate_ftrace_ops_list(self.context, kernel_name): for ftrace_ops_parsed in self.parse_ftrace_ops( self.context, diff --git a/volatility3/framework/plugins/linux/tracing/tracepoints.py b/volatility3/framework/plugins/linux/tracing/tracepoints.py index f8edcc5b7..fe6d11af9 100644 --- a/volatility3/framework/plugins/linux/tracing/tracepoints.py +++ b/volatility3/framework/plugins/linux/tracing/tracepoints.py @@ -9,11 +9,9 @@ from typing import Dict, Iterable, List, Optional from dataclasses import dataclass import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules -from volatility3.plugins.linux import hidden_modules, modxview from volatility3.framework import constants, exceptions, interfaces from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints, NotAvailableValue, TreeGrid -from volatility3.framework.symbols.linux import extensions from volatility3.framework.objects import utility from volatility3.framework.constants import architectures @@ -54,15 +52,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(1, 1, 0), - ), - requirements.PluginRequirement( - name="modxview", plugin=modxview.Modxview, version=(1, 0, 0) - ), - requirements.PluginRequirement( - name="hidden_modules", - plugin=hidden_modules.Hidden_modules, - version=(1, 0, 0), + version=(2, 0, 0), ), ] @@ -105,15 +95,15 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): def parse_tracepoint( cls, context: interfaces.context.ContextInterface, - kernel_name: str, - known_modules: Dict[str, List[extensions.module]], + kernel_module_name: str, + known_modules: Dict[str, List[linux_utilities_modules.Modules.ModuleInfo]], tracepoint: interfaces.objects.ObjectInterface, run_hidden_modules: bool = True, ) -> Optional[Iterable[ParsedTracepointFunc]]: """Parse a tracepoint struct to highlight tracepoints kernel hooking. Args: - known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through modxview.run_modules_scanners(). + known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through run_modules_scanners(). tracepoint: The tracepoint 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. @@ -121,11 +111,10 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): Yields: An iterable of ParsedTracepointFunc dataclasses, containing a selection of useful fields related to a tracepoint struct """ - kernel = context.modules[kernel_name] - kernel_layer = context.layers[kernel.layer_name] + kernel = context.modules[kernel_module_name] for tracepoint_func in cls.iterate_tracepoint_funcs( - context, kernel_layer.name, tracepoint + context, kernel.layer_name, tracepoint ): try: tracepoint_name = utility.pointer_to_string(tracepoint.name, count=512) @@ -139,56 +128,19 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): probe_handler_symbol = module_address = module_name = None # Try to lookup within the known modules if the probe_handler address fits - module = linux_utilities_modules.Modules.module_lookup_by_address( - context, - kernel.layer_name, - modxview.Modxview.flatten_run_modules_results(known_modules), - probe_handler_address, - ) - # Run hidden_modules plugin if a probe handler origin couldn't be determined (only done once, results are re-used afterwards) - if ( - module is None - and run_hidden_modules - and "hidden_modules" not in known_modules - ): - vollog.info( - "A probe handler module origin could not be determined. hidden_modules plugin will be run to detect additional modules.", - ) - known_modules_addresses = set( - kernel_layer.canonicalize(module.vol.offset) - for module in modxview.Modxview.flatten_run_modules_results( - known_modules - ) - ) - modules_memory_boundaries = ( - hidden_modules.Hidden_modules.get_modules_memory_boundaries( - context, kernel_name - ) - ) - known_modules["hidden_modules"] = list( - hidden_modules.Hidden_modules.get_hidden_modules( - context, - kernel_name, - known_modules_addresses, - modules_memory_boundaries, - ) - ) - # Lookup the updated list to see if hidden_modules was able - # to find the missing module - module = linux_utilities_modules.Modules.module_lookup_by_address( + mod_info, probe_handler_symbol = ( + linux_utilities_modules.Modules.module_lookup_by_address( context, - kernel.layer_name, - modxview.Modxview.flatten_run_modules_results(known_modules), + kernel_module_name, + known_modules, probe_handler_address, ) + ) # Fetch more information about the module - if module is not None: - module_address = module.vol.offset - module_name = module.get_name() - probe_handler_symbol = module.get_symbol_by_address( - probe_handler_address - ) + if mod_info is not None: + module_address = mod_info.offset + module_name = mod_info.name else: vollog.debug( f"Could not determine tracepoint@{tracepoint.vol.offset:#x} probe handler {probe_handler_address:#x} module origin.", @@ -276,7 +228,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface): ) return - known_modules = modxview.Modxview.run_modules_scanners( + known_modules = linux_utilities_modules.Modules.run_modules_scanners( self.context, kernel_name, run_hidden_modules=False ) 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 742bacb0b..281d46eda 100644 --- a/volatility3/framework/plugins/linux/tty_check.py +++ b/volatility3/framework/plugins/linux/tty_check.py @@ -33,7 +33,7 @@ class tty_check(plugins.PluginInterface): requirements.VersionRequirement( name="linux_utilities_modules", component=linux_utilities_modules.Modules, - version=(1, 0, 0), + version=(2, 0, 0), ), requirements.PluginRequirement( name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0) diff --git a/volatility3/framework/symbols/linux/__init__.py b/volatility3/framework/symbols/linux/__init__.py index c443c37e1..758e142aa 100644 --- a/volatility3/framework/symbols/linux/__init__.py +++ b/volatility3/framework/symbols/linux/__init__.py @@ -1,6 +1,7 @@ # This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 # + import math import string import contextlib @@ -14,9 +15,9 @@ from volatility3 import framework from volatility3.framework import ( constants, exceptions, + deprecation, interfaces, objects, - Deprecation, ) from volatility3.framework.objects import utility from volatility3.framework.symbols import intermed @@ -367,9 +368,9 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface): yield fd_num, filp, full_path @classmethod - @Deprecation.deprecated_method( - replacement=linux_utilities_modules.Modules.mask_mods_list, - replacement_version=(1, 0, 0), + @deprecation.method_being_removed( + removal_date="2025-09-25", + message="Callers to this method should adapt `linux_utilities_modules.Modules.run_module_scanners`", ) def mask_mods_list( cls, @@ -385,6 +386,10 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface): return linux_utilities_modules.Modules.mask_mods_list(context, layer_name, mods) @classmethod + @deprecation.method_being_removed( + removal_date="2025-09-25", + message="Callers to this method should adapt `linux_utilities_modules.Modules.run_module_scanners`", + ) def generate_kernel_handler_info( cls, context: interfaces.context.ContextInterface, @@ -392,6 +397,8 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface): mods_list: Iterator[interfaces.objects.ObjectInterface], ) -> List[Tuple[str, int, int]]: """ + This method is being deprecated. Use `linux_utilities_modules.Modules.run_module_scanners` to map kernel pointers to modules") + A helper function that gets the beginning and end address of the kernel module """ @@ -412,9 +419,10 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface): ) @classmethod - @Deprecation.deprecated_method( + @deprecation.deprecated_method( replacement=linux_utilities_modules.Modules.lookup_module_address, - replacement_version=(1, 0, 0), + removal_date="2025-09-25", + replacement_version=(2, 0, 0), ) def lookup_module_address( cls, diff --git a/volatility3/framework/symbols/linux/utilities/modules.py b/volatility3/framework/symbols/linux/utilities/modules.py index d03e76c88..0eeb2b33c 100644 --- a/volatility3/framework/symbols/linux/utilities/modules.py +++ b/volatility3/framework/symbols/linux/utilities/modules.py @@ -1,32 +1,54 @@ +import logging import warnings -from typing import Iterable, Iterator, List, Optional, Tuple +from typing import Iterable, Iterator, List, Optional, Tuple, NamedTuple, Dict, Set from volatility3 import framework -from volatility3.framework import constants, interfaces +from volatility3.framework import ( + constants, + interfaces, + deprecation, + exceptions, + objects, +) 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.""" - _version = (1, 1, 0) + _version = (2, 0, 0) _required_framework_version = (2, 0, 0) framework.require_interface_version(*_required_framework_version) - @classmethod + class ModuleInfo(NamedTuple): + """ + Used to track the name and boundary of a kernel module + """ + + offset: int + name: str + start: int + end: int + + @staticmethod def module_lookup_by_address( - cls, context: interfaces.context.ContextInterface, - layer_name: str, - modules: Iterable[extensions.module], + kernel_module_name: str, + modules: Iterable[ModuleInfo], target_address: int, - ) -> Optional[extensions.module]: + run_hidden_modules: bool = True, + ) -> Optional[Tuple[ModuleInfo, Optional[str]]]: """ Determine if a target address lies in a module memory space. Returns the module where the provided address lies. + `modules` must be non-empty and contain masked addresses via `get_module_info_for_module` or + a ValueError will be thrown + Args: context: The context on which to operate layer_name: The name of the layer on which to operate @@ -34,44 +56,70 @@ class Modules(interfaces.configuration.VersionableInterface): target_address: The address to check for a match Returns: - The first memory module in which the address fits + The first memory module in which the address fits and the symbol name for `target_address` Kernel documentation: "within_module" and "within_module_mem_type" functions """ + kernel = context.modules[kernel_module_name] + + kernel_layer = context.layers[kernel.layer_name] + + if not modules: + raise ValueError("Empty list sent to `module_lookup_by_address`") + matches = [] - seen_addresses = set() for module in modules: - _, start, end = cls.mask_mods_list(context, layer_name, [module])[0] - if ( - start <= target_address < end - and module.vol.offset not in seen_addresses - ): + if module.start != module.start & kernel_layer.address_mask: + raise ValueError( + "Modules list must be gathered from `run_modules_scanners` to be used in this function" + ) + + if module.start <= target_address < module.end: matches.append(module) - seen_addresses.add(module.vol.offset) - if len(matches) > 1: - warnings.warn( - f"Address {hex(target_address)} fits in modules at {[hex(module.vol.offset) for module in matches]}, indicating potential modules memory space overlap.", - UserWarning, - ) - return matches[0] - elif len(matches) == 1: - return matches[0] + if len(matches) >= 1: + if len(matches) > 1: + warnings.warn( + f"Address {hex(target_address)} fits in modules at {[hex(module.start) for module in matches]}, indicating potential modules memory space overlap. The first matching entry {matches[0].name} will be used", + UserWarning, + ) - return None + symbol_name = None + + match = matches[0] + + if match.name == constants.linux.KERNEL_NAME: + symbols = list(kernel.get_symbols_by_absolute_location(target_address)) + + if len(symbols): + symbol_name = symbols[0] + else: + module = kernel.object("module", offset=module.offset, absolute=True) + symbol_name = module.get_symbol_by_address(target_address) + + if symbol_name: + symbol_name = symbol_name.split(constants.BANG)[1] + + return match, symbol_name + + return None, None @classmethod + @deprecation.method_being_removed( + removal_date="2025-09-25", + message="Code using this function should adapt `linux_utilities_modules.Modules.run_module_scanners`", + ) def mask_mods_list( cls, context: interfaces.context.ContextInterface, - layer_name: str, - mods: Iterator[interfaces.objects.ObjectInterface], + kernel_layer_name: str, + mods: Iterator[extensions.module], ) -> List[Tuple[str, int, int]]: """ A helper function to mask the starting and end address of kernel modules """ - mask = context.layers[layer_name].address_mask + mask = context.layers[kernel_layer_name].address_mask return [ ( @@ -83,6 +131,10 @@ class Modules(interfaces.configuration.VersionableInterface): ] @classmethod + @deprecation.method_being_removed( + removal_date="2025-09-25", + message="Use `module_lookup_by_address` to map address to their hosting kernel module and symbol.", + ) def lookup_module_address( cls, context: interfaces.context.ContextInterface, @@ -116,3 +168,369 @@ class Modules(interfaces.configuration.VersionableInterface): break return mod_name, symbol_name + + @classmethod + def get_module_info_for_module( + cls, address_mask: int, module: extensions.module + ) -> Optional[ModuleInfo]: + """ + Returns a ModuleInfo instance for `module` + + This performs address masking to avoid endless calls to `mask_mods_list` + + Returns None if the name is smeared + """ + try: + mod_name = utility.array_to_string(module.name) + except exceptions.InvalidAddressException: + return None + + start = module.get_module_base() & address_mask + + 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 + ) + + @classmethod + def run_modules_scanners( + cls, + context: interfaces.context.ContextInterface, + kernel_name: str, + run_hidden_modules: bool = True, + 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 + """ + + kernel = context.modules[kernel_name] + + address_mask = context.layers[kernel.layer_name].address_mask + + run_results = {} + + # the kernel module boundaries + run_results["kernel"] = [cls.get_kernel_module_info(context, kernel_name)] + + # lsmod + run_results["lsmod"] = [] + + 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) + + # 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) + + if flatten: + return cls.flatten_run_modules_results(run_results) + + return run_results + + @staticmethod + def get_modules_memory_boundaries( + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + ) -> Tuple[int, int]: + """Determine the boundaries of the module allocation area + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + vmlinux_module_name: The name of the kernel module on which to operate + + Returns: + A tuple containing the minimum and maximum addresses for the module allocation area. + """ + vmlinux = context.modules[vmlinux_module_name] + if vmlinux.has_symbol("mod_tree"): + # Kernel >= 5.19 58d208de3e8d87dbe196caf0b57cc58c7a3836ca + mod_tree = vmlinux.object_from_symbol("mod_tree") + modules_addr_min = mod_tree.addr_min + modules_addr_max = mod_tree.addr_max + elif vmlinux.has_symbol("module_addr_min"): + # 2.6.27 <= kernel < 5.19 3a642e99babe0617febb6f402e1e063479f489db + modules_addr_min = vmlinux.object_from_symbol("module_addr_min") + modules_addr_max = vmlinux.object_from_symbol("module_addr_max") + + if isinstance(modules_addr_min, objects.Void): + raise exceptions.VolatilityException( + "Your ISF symbols lack type information. You may need to update the" + "ISF using the latest version of dwarf2json" + ) + else: + raise exceptions.VolatilityException( + "Cannot find the module memory allocation area. Unsupported kernel" + ) + + return modules_addr_min, modules_addr_max + + @classmethod + def flatten_run_modules_results( + cls, run_results: Dict[str, List[ModuleInfo]], deduplicate: bool = True + ) -> List[ModuleInfo]: + """Flatten a dictionary mapping plugin names and modules list, to a single merged list. + This is useful to get a generic lookup list of all the detected modules. + + Args: + run_results: dictionary of plugin names mapping a list of detected modules + deduplicate: remove duplicate modules, based on their offsets + + Returns: + List of ModuleInfo objects + """ + uniq_modules: List[Modules.ModuleInfo] = [] + + seen_addresses: int = set() + + for modules in run_results.values(): + for module in modules: + if deduplicate and (module.start in seen_addresses): + continue + seen_addresses.add(module.start) + uniq_modules.append(module) + + return uniq_modules + + @classmethod + def get_hidden_modules( + cls, + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + known_module_addresses: Set[int], + modules_memory_boundaries: Tuple, + ) -> Iterable[interfaces.objects.ObjectInterface]: + """Enumerate hidden modules by taking advantage of memory address alignment patterns + + This technique is much faster and uses less memory than the traditional scan method + in Volatility2, but it doesn't work with older kernels. + + From kernels 4.2 struct module allocation are aligned to the L1 cache line size. + In i386/amd64/arm64 this is typically 64 bytes. However, this can be changed in + the Linux kernel configuration via CONFIG_X86_L1_CACHE_SHIFT. The alignment can + also be obtained from the DWARF info i.e. DW_AT_alignment<64>, but dwarf2json + doesn't support this feature yet. + In kernels < 4.2, alignment attributes are absent in the struct module, meaning + alignment cannot be guaranteed. Therefore, for older kernels, it's better to use + the traditional scan technique. + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + vmlinux_module_name: The name of the kernel module on which to operate + known_module_addresses: Set with known module addresses + modules_memory_boundaries: Minimum and maximum address boundaries for module allocation. + Yields: + module objects + """ + vmlinux = context.modules[vmlinux_module_name] + vmlinux_layer = context.layers[vmlinux.layer_name] + + module_addr_min, module_addr_max = modules_memory_boundaries + module_address_alignment = cls.get_module_address_alignment( + context, vmlinux_module_name + ) + if not cls.validate_alignment_patterns( + known_module_addresses, module_address_alignment + ): + vollog.warning( + f"Module addresses aren't aligned to {module_address_alignment} bytes. " + "Switching to 1 byte aligment scan method." + ) + module_address_alignment = 1 + + mkobj_offset = vmlinux.get_type("module").relative_child_offset("mkobj") + mod_offset = vmlinux.get_type("module_kobject").relative_child_offset("mod") + offset_to_mkobj_mod = mkobj_offset + mod_offset + mod_member_template = vmlinux.get_type("module_kobject").child_template("mod") + mod_size = mod_member_template.size + mod_member_data_format = mod_member_template.data_format + + for module_addr in range( + module_addr_min, module_addr_max, module_address_alignment + ): + if module_addr in known_module_addresses: + continue + + try: + # This is just a pre-filter. Module readability and consistency are verified in module.is_valid() + self_referential_bytes = vmlinux_layer.read( + module_addr + offset_to_mkobj_mod, mod_size + ) + self_referential = objects.convert_data_to_value( + self_referential_bytes, int, mod_member_data_format + ) + if self_referential != module_addr: + continue + except ( + exceptions.PagedInvalidAddressException, + exceptions.InvalidAddressException, + ): + continue + + module = vmlinux.object("module", offset=module_addr, absolute=True) + if module and module.is_valid(): + yield module + + @classmethod + def get_module_address_alignment( + cls, + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + ) -> int: + """Obtain the module memory address alignment. + + struct module is aligned to the L1 cache line, which is typically 64 bytes for most + common i386/AMD64/ARM64 configurations. In some cases, it can be 128 bytes, but this + will still work. + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + vmlinux_module_name: The name of the kernel module on which to operate + + Returns: + The struct module alignment + """ + # FIXME: When dwarf2json/ISF supports type alignments. Read it directly from the type metadata + # Additionally, while 'context' and 'vmlinux_module_name' are currently unused, they will be + # essential for retrieving type metadata in the future. + return 64 + + @classmethod + def list_modules( + cls, context: interfaces.context.ContextInterface, vmlinux_module_name: str + ) -> Iterable[interfaces.objects.ObjectInterface]: + """Lists all the modules in the primary layer. + + Args: + context: The context to retrieve required elements (layers, symbol tables) from + layer_name: The name of the layer on which to operate + vmlinux_symbols: The name of the table containing the kernel symbols + + Yields: + The modules present in the `layer_name` layer's modules list + + This function will throw a SymbolError exception if kernel module support is not enabled. + """ + vmlinux = context.modules[vmlinux_module_name] + + modules = vmlinux.object_from_symbol(symbol_name="modules").cast("list_head") + + table_name = vmlinux.symbol_table_name + + yield from modules.to_list(table_name + constants.BANG + "module", "list") + + @classmethod + def get_kset_modules( + cls, context: interfaces.context.ContextInterface, vmlinux_name: str + ) -> Dict[str, extensions.module]: + vmlinux = context.modules[vmlinux_name] + + try: + module_kset = vmlinux.object_from_symbol("module_kset") + except exceptions.SymbolError: + module_kset = None + + if not module_kset: + raise TypeError( + "This plugin requires the module_kset structure. This structure is not present in the supplied symbol table. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupt." + ) + + ret = {} + + kobj_off = vmlinux.get_type("module_kobject").relative_child_offset("kobj") + + for kobj in module_kset.list.to_list( + vmlinux.symbol_table_name + constants.BANG + "kobject", "entry" + ): + mod_kobj = vmlinux.object( + object_type="module_kobject", + offset=kobj.vol.offset - kobj_off, + absolute=True, + ) + + mod = mod_kobj.mod + + try: + name = utility.pointer_to_string(kobj.name, 32) + except exceptions.InvalidAddressException: + continue + + if kobj.name and kobj.reference_count() > 2: + ret[name] = mod + + return ret + + @staticmethod + def validate_alignment_patterns( + addresses: Iterable[int], + address_alignment: int, + ) -> bool: + """Check if the memory addresses meet our alignments patterns + + Args: + addresses: Iterable with the address values + address_alignment: Number of bytes for alignment validation + + Returns: + True if all the addresses meet the alignment + """ + return all(addr % address_alignment == 0 for addr in addresses)