From 5dee3aeeee5e6b08481c1e39d2f7cb37a17c086c Mon Sep 17 00:00:00 2001 From: Gustavo Moreira Date: Tue, 1 Oct 2024 19:30:50 +1000 Subject: [PATCH] Add linux.hidden_modules plugin --- .../framework/plugins/linux/hidden_modules.py | 331 ++++++++++++++++++ .../symbols/linux/extensions/__init__.py | 38 ++ 2 files changed, 369 insertions(+) create mode 100644 volatility3/framework/plugins/linux/hidden_modules.py diff --git a/volatility3/framework/plugins/linux/hidden_modules.py b/volatility3/framework/plugins/linux/hidden_modules.py new file mode 100644 index 000000000..1c2429449 --- /dev/null +++ b/volatility3/framework/plugins/linux/hidden_modules.py @@ -0,0 +1,331 @@ +# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# +import re +import functools +import logging +import contextlib +from typing import List, Iterable +from volatility3.framework import renderers, interfaces, exceptions, objects +from volatility3.framework.constants.architectures import LINUX_ARCHS +from volatility3.framework.renderers import format_hints +from volatility3.framework.configuration import requirements +from volatility3.plugins.linux import lsmod + +vollog = logging.getLogger(__name__) + + +class Hidden_modules(interfaces.plugins.PluginInterface): + """Carves memory to find hidden kernel modules""" + + _required_framework_version = (2, 10, 0) + + _version = (1, 0, 0) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Linux kernel", + architectures=LINUX_ARCHS, + ), + requirements.PluginRequirement( + name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0) + ), + requirements.BooleanRequirement( + name="fast", + description="Fast scan method. Recommended only for kernels 4.2 and above", + optional=True, + default=False, + ), + ] + + def _get_modules_memory_boundaries(self, vmlinux): + if vmlinux.has_symbol("mod_tree"): + 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"): + 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): + # Crap ISF! Here's my best-effort workaround + vollog.warning( + "Your ISF symbols are missing type information. You may need to update " + "the ISF using the latest version of dwarf2json" + ) + # See issue #1041. In the Linux kernel these are "unsigned long" + for type_name in ("long unsigned int", "unsigned long"): + if vmlinux.has_type(type_name): + modules_addr_min = modules_addr_min.cast(type_name) + modules_addr_max = modules_addr_max.cast(type_name) + break + else: + raise exceptions.VolatilityException( + "Bad ISF! Please 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 + + def _get_module_state_live_bytes( + self, + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + ) -> bytes: + """Retrieve the MODULE_STATE_LIVE value bytes by introspecting its enum type + + 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 MODULE_STATE_LIVE value bytes + """ + vmlinux = context.modules[vmlinux_module_name] + module_state_type_template = vmlinux.get_type("module").vol.members["state"][1] + module_state_live_val = module_state_type_template.choices["MODULE_STATE_LIVE"] + data_format = module_state_type_template.base_type.vol.data_format + module_state_live_bytes = objects.convert_value_to_data( + module_state_live_val, int, data_format + ) + return module_state_live_bytes + + def get_hidden_modules_vol2( + self, + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + known_module_addresses, + modules_memory_boundaries: tuple, + ) -> Iterable[interfaces.objects.ObjectInterface]: + """Enumerate hidden modules using the traditional implementation. + + This is a port of the Volatility2 plugin, with minor code improvements. + + 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 + Yields: + module objects + """ + vmlinux = context.modules[vmlinux_module_name] + vmlinux_layer = context.layers[vmlinux.layer_name] + + check_nums = ( + 3000, + 2800, + 2700, + 2500, + 2300, + 2100, + 2000, + 1500, + 1300, + 1200, + 1024, + 512, + 256, + 128, + 96, + 64, + 48, + 32, + 24, + ) + modules_addr_min, modules_addr_max = modules_memory_boundaries + modules_addr_min = modules_addr_min & ~0xFFF + modules_addr_max = (modules_addr_max & ~0xFFF) + vmlinux_layer.page_size + + check_bufs = [] + replace_bufs = [] + minus_size = vmlinux.get_type("pointer").size + null_pointer_bytes = b"\x00" * minus_size + for num in check_nums: + check_bufs.append(b"\x00" * num) + replace_bufs.append((b"\xff" * (num - minus_size)) + null_pointer_bytes) + + all_ffs = b"\xff" * 4096 + scan_list = [] + for page_addr in range( + modules_addr_min, modules_addr_max, vmlinux_layer.page_size + ): + content_fixed = all_ffs + with contextlib.suppress( + exceptions.InvalidAddressException, + exceptions.PagedInvalidAddressException, + ): + content = vmlinux_layer.read(page_addr, vmlinux_layer.page_size) + + all_nulls = all(x == 0 for x in content) + if content and not all_nulls: + content_fixed = content + for check_bytes, replace_bytes in zip(check_bufs, replace_bufs): + content_fixed = content_fixed.replace( + check_bytes, replace_bytes + ) + + scan_list.append(content_fixed) + + scan_buf = b"".join(scan_list) + del scan_list + + module_state_live_bytes = self._get_module_state_live_bytes( + context, vmlinux_module_name + ) + # f'strings cannot be combined with bytes literals + for cur_addr in re.finditer(b"(?=(%s))" % (module_state_live_bytes), scan_buf): + module_addr = modules_addr_min + cur_addr.start() + + if module_addr in known_module_addresses: + continue + + module = vmlinux.object("module", offset=module_addr, absolute=True) + if module and module.is_valid(): + yield module + + @functools.cached_property + def module_address_alignment(self) -> int: + """Obtain the module memory address alignment. This is only used with the fast scan method. + + 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. + + Returns: + The struct module alignment + """ + # FIXME: When dwarf2json/ISF supports type alignments. Read it directly from the type metadata + # The cached_property won't provide any benefits until then + return 64 + + def get_hidden_modules_fast( + self, + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + known_module_addresses, + 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 + 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_state_live_bytes = self._get_module_state_live_bytes( + context, vmlinux_module_name + ) + + for module_addr in range( + module_addr_min, module_addr_max, self.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() + module_state_bytes = vmlinux_layer.read( + module_addr, len(module_state_live_bytes) + ) + if module_state_bytes != module_state_live_bytes: + continue + except ( + exceptions.PagedInvalidAddressException, + exceptions.InvalidAddressException, + ): + continue + + module = vmlinux.object("module", offset=module_addr, absolute=True) + if module and module.is_valid(): + yield module + + def _validate_alignment_patterns(self, addresses: Iterable[int]) -> bool: + """Check if the memory addresses meet our alignments patterns + + Args: + addresses: Iterable with the address values + + Returns: + True if all the addresses meet the alignment + """ + return all(addr % self.module_address_alignment == 0 for addr in addresses) + + def get_hidden_modules( + self, + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + ) -> Iterable[interfaces.objects.ObjectInterface]: + """Enumerate hidden modules + + 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 + Yields: + module objects + """ + vmlinux = context.modules[vmlinux_module_name] + vmlinux_layer = context.layers[vmlinux.layer_name] + + known_module_addresses = { + vmlinux_layer.canonicalize(module.vol.offset) + for module in lsmod.Lsmod.list_modules(context, vmlinux_module_name) + } + + modules_memory_boundaries = self._get_modules_memory_boundaries(vmlinux) + + if self.config.get("fast"): + if self._validate_alignment_patterns(known_module_addresses): + scan_method = self.get_hidden_modules_fast + else: + vollog.warning( + f"Module addresses aren't aligned to {self.module_address_alignment} bytes. " + "Switching to the traditional scan method." + ) + scan_method = self.get_hidden_modules_vol2 + else: + scan_method = self.get_hidden_modules_vol2 + + yield from scan_method( + context, + vmlinux_module_name, + known_module_addresses, + modules_memory_boundaries, + ) + + def _generator(self): + vmlinux_module_name = self.config["kernel"] + for module in self.get_hidden_modules(self.context, vmlinux_module_name): + module_addr = module.vol.offset + module_name = module.get_name() or renderers.NotAvailableValue() + fields = (format_hints.Hex(module_addr), module_name) + yield (0, fields) + + def run(self): + headers = [ + ("Address", format_hints.Hex), + ("Name", str), + ] + return renderers.TreeGrid(headers, self._generator()) diff --git a/volatility3/framework/symbols/linux/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index 0a879249b..b30634ca3 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -35,6 +35,34 @@ class module(generic.GenericIntelProcess): super().__init__(*args, **kwargs) self._mod_mem_type = None # Initialize _mod_mem_type to None for memoization + def is_valid(self): + layer = self._context.layers[self.vol.layer_name] + # Make sure the entire module content is readable + if not layer.is_valid(self.vol.offset, self.vol.size): + return False + + if not self.state.is_valid_choice: + return False + + core_size = self.get_core_size() + if not ( + 1 <= core_size <= 20000000 + and core_size + self.get_init_size() >= 4096 + and 1 <= self.get_core_text_size() <= 20000000 + ): + return False + + if self.has_member("mkobj") and self.mkobj.has_member("mod"): + if not ( + self.mkobj + and self.mkobj.mod + and self.mkobj.mod.is_readable() + and self.mkobj.mod == self.vol.offset + ): + return False + + return True + @property def mod_mem_type(self): """Return the mod_mem_type enum choices if available or an empty dict if not""" @@ -112,6 +140,16 @@ class module(generic.GenericIntelProcess): raise AttributeError("Unable to determine core size of module") + def get_core_text_size(self): + if self.has_member("mem"): # kernels 6.4+ + return self._get_mem_size("MOD_TEXT") + elif self.has_member("core_layout"): + return self.core_layout.text_size + elif self.has_member("core_text_size"): + return self.core_text_size + + raise AttributeError("Unable to determine core text size of module") + def get_module_core(self): if self.has_member("mem"): # kernels 6.4+ return self._get_mem_base("MOD_TEXT")