diff --git a/volatility3/framework/plugins/linux/check_unlinked_modules.py b/volatility3/framework/plugins/linux/check_unlinked_modules.py new file mode 100644 index 000000000..197fd98c8 --- /dev/null +++ b/volatility3/framework/plugins/linux/check_unlinked_modules.py @@ -0,0 +1,403 @@ +# This file is Copyright 2023 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# +import logging +from typing import List, Dict +from volatility3.plugins.linux import check_modules +from volatility3.framework import interfaces, exceptions +from volatility3.framework.configuration import requirements +from volatility3.framework.renderers import format_hints, TreeGrid +from volatility3.framework.symbols import linux +from volatility3.framework.objects import ( + utility, + PrimitiveObject, + Boolean, + Enumeration, + templates, +) +from enum import Enum +from volatility3.framework.layers import scanners + +vollog = logging.getLogger(__name__) +MAX_KERNEL_MEMORY_SEARCH_LIMIT = 2**20 # Arbitrary constant + + +class Check_unlinked_modules(interfaces.plugins.PluginInterface): + """Scan memory for unlinked modules""" + + _version = (1, 0, 1) + _required_framework_version = (2, 5, 2) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Linux kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.PluginRequirement( + name="check_modules", + plugin=check_modules.Check_modules, + version=(0, 0, 0), + ), + requirements.IntRequirement( + name="leaked_address", + description="Optimized memory scan, around a leaked address from an hidden module (e.g. ftrace callback)", + optional=True, + default=0, + ), + ] + + @classmethod + def wrapper_get_sysfs_modules( + cls, + context: interfaces.context.ContextInterface, + config_path: str, + vmlinux_name: str, + ): + """Wrapper for check_modules plugin, return a list of modules (similarly to the lsmod plugin)""" + + vmlinux = context.modules[vmlinux_name] + sysfs_modules: dict = check_modules.Check_modules( + context, config_path + ).get_kset_modules(context, vmlinux_name) + # Convert get_kset_modules() offsets back to module objects + for m_offset in sysfs_modules.values(): + yield vmlinux.object(object_type="module", offset=m_offset, absolute=True) + + @classmethod + def lookup_sysfs_hidden_modules( + cls, + context: interfaces.context.ContextInterface, + vmlinux_name: str, + modules_handlers: list, + leaked_address: int = 0, + ): + """ + Some rootkits unlink themselves from list_head AND module_kobject. However, due to the use of a callback (e.g. via ftrace), they reveal an address inside their module memory. + Doing so, we can scan the close memory range of the callback for any "module" structs. + Most of the time, a hidden rootkit memory range will be located between two other modules, giving us some minimum and maximum range to search in. + Note : callbacks with values like 0xffffffffffffffff will fail with DEBUG : "Scan Failure: Sections have no size, nothing to scan". This ensure that pottential OOB reads won't crash the plugin. + + Current framework implementation does not take into account modules with non-printable characters in their name. + To avoid changing too much of the framework, we'll follow the same path. + However, a module might have a name like "\xde\xad\xbe\xef", which would need manual investigation with volshell to uncover. + In the case of multiple modules on a same memory dump nulling their name (e.g. rootkit "osom"), it could result in multiple modules with empty names, overlapping in the handers list. + """ + + vmlinux = context.modules[vmlinux_name] + flattened_handlers = flatten_modules_handlers(modules_handlers) + flattened_handlers.sort() + if leaked_address != 0: + ( + low_address, + high_address, + ) = calculate_closest_modules_from_address( + target_address=leaked_address, + flattened_handlers=flattened_handlers, + ) + if low_address == None: + low_address = leaked_address - MAX_KERNEL_MEMORY_SEARCH_LIMIT + if high_address == None: + high_address = leaked_address + MAX_KERNEL_MEMORY_SEARCH_LIMIT + + # Search : between previous module and leaked address ; between leaked address and next module + sections_to_scan = [ + (low_address, leaked_address - low_address), + (leaked_address, high_address - leaked_address), + ] + vollog.info( + f"Searching modules structs from {hex(low_address)} to {hex(high_address)}, based on provided address {hex(leaked_address)}..." + ) + else: + low_address = flattened_handlers[0] - MAX_KERNEL_MEMORY_SEARCH_LIMIT + high_address = flattened_handlers[-1] + MAX_KERNEL_MEMORY_SEARCH_LIMIT + # Search : between first and last modules ; before first module ; after last module + sections_to_scan = [ + (flattened_handlers[0], flattened_handlers[-1] - flattened_handlers[0]), + (low_address, flattened_handlers[0] - low_address), + (flattened_handlers[-1], high_address - flattened_handlers[-1]), + ] + vollog.info( + f"Searching modules structs, in (start, size) : {[(hex(x[0]), x[1]) for x in sections_to_scan]}..." + ) + scanned_modules = scan_memory_for_modules( + context=context, + kernel_module_name=vmlinux.name, + sections_to_scan=sections_to_scan, + ) + + return_modules = [] + for scanned_module in list(scanned_modules): + scanned_module_handler = linux.LinuxUtilities.generate_kernel_handler_info( + context, vmlinux.name, (scanned_module,) + )[ + 1 + ] # Skip __kernel__ + + # Check if scanned module already exists in our lists + if not any( + scanned_module_handler == existing_handler + for existing_handler in modules_handlers + ): + return_modules.append(scanned_module) + vollog.info( + f'Found sysfs non-listed module "{utility.array_to_string(scanned_module.name)}" at {hex(scanned_module.vol.offset)}' + ) + + return return_modules + + def _generator( + self, sysfs_handlers: list, sysfs_modules: list, leaked_address: int = 0 + ): + vmlinux = self.context.modules[self.config["kernel"]] + + # Detect unlinked modules + sysfs_unlinked_modules = self.lookup_sysfs_hidden_modules( + self.context, + vmlinux.name, + sysfs_handlers, + leaked_address, + ) + + dict_sysfs_modules = dict( + (str(utility.array_to_string(module.name)), module) + for module in sysfs_modules + ) + dict_sysfs_unlinked_modules = dict( + (str(utility.array_to_string(module.name)), module) + for module in sysfs_unlinked_modules + ) + for mod in set(dict_sysfs_unlinked_modules.items()).difference( + set(dict_sysfs_modules.items()) + ): + yield mod + + def run(self): + vmlinux = self.context.modules[self.config["kernel"]] + + # Get /sys/module/ listed modules + sysfs_modules = list( + self.wrapper_get_sysfs_modules(self.context, self.config_path, vmlinux.name) + ) + # Calculate boundaries for each module + sysfs_handlers = linux.LinuxUtilities.generate_kernel_handler_info( + self.context, vmlinux.name, sysfs_modules + )[ + 1: + ] # Skip __kernel__, else the range to search in will be huge (_stext to _etext) + + detected_modules = self._generator( + sysfs_handlers=sysfs_handlers, + sysfs_modules=sysfs_modules, + leaked_address=self.config.get("leaked_address"), + ) + return TreeGrid( + [("Module Address", format_hints.Hex), ("Module Name", str)], + [ + (0, (format_hints.Hex(mod[1].vol.offset), mod[0])) + for mod in detected_modules + ], + ) + + +## UTILITIES ## +def regex_from_struct_members( + context: interfaces.context.ContextInterface, + kernel_module_name: str, + vol_struct: templates.ObjectTemplate, + path: List = [], + overrides: Dict[str, bytes] = {}, + stop_key: str = "", +): + """Walk an ObjectTemplate members to create a matching RegEx. + This is useful to search for a struct artifacts in memory, while taking into account variants from one symbol table to another (one kernel to another). + """ + kernel = context.modules[kernel_module_name] + + # Sort struct members and flatten dict + struct_members = dict( + sorted(vol_struct.vol.members.items(), key=lambda item: item[1][0]) + ) + struct_members_list = [] + for name, (offset, obj) in struct_members.items(): + struct_members_list.append((name, offset, obj)) + if name == stop_key and path == []: + break + struct_size = vol_struct.size + struct_regex = [] + consumed = 0 + + # Iterate over every member of the struct + for i, member in enumerate(struct_members_list): + name, offset, obj = member + object_class: PrimitiveObject | None = obj.vol.get("object_class") + if i < len(struct_members_list) - 1: + member_len = struct_members_list[i + 1][1] - offset + else: + member_len = struct_size - offset + consumed += member_len + # Keep track of the depth and path of this element, starting from the root object + path.append(name) + # Determine what regex to use, depending on the element type + if ".".join(path) in overrides: + round_regex = overrides[".".join(path)] + elif object_class == Enumeration: + choices = [] + for choice in obj.vol.choices.values(): + choices.append(int.to_bytes(choice, member_len, "little", signed=True)) + round_regex = b"(?:" + b"|".join(choices) + b")" + elif object_class == Boolean: + possible_values = [] + for byte_order in ["little", "big"]: + for possible_value in [0, 1]: + possible_values.append( + int.to_bytes(possible_value, member_len, byte_order) + ) + # don't bother with endianness on 1 byte + if member_len == 1: + break + + round_regex = b"(?:" + b"|".join(possible_values) + b")" + + # # Possible alignement problem when not using lookbehind here + # if member_len > 1: + # round_regex = b"(?=" + round_regex + b")" + else: + type_name = obj.vol.type_name + # Recursive introspection, call this function again to analyze a sub-element + if kernel.has_type(type_name) and kernel.get_type(type_name).vol.get( + "members" + ): + type = kernel.get_type(type_name) + tmp_regex, ret_consumed = regex_from_struct_members( + context, kernel_module_name, type, path, overrides + ) + # Detect missing bytes (padding) + if member_len > ret_consumed: + padding = member_len - ret_consumed + tmp_regex += f".{{{padding}}}".encode() + round_regex = tmp_regex + # We can't make assumptions about this element + else: + round_regex = f".{{{member_len}}}".encode() + + # vollog.debug(f"path : {path} : {round_regex}") + struct_regex.append(round_regex) + # We are done with this element + path.remove(name) + + # Use a lookahead to allow overlapping matches (only around final struct_regex) + if len(path) == 0: + return ( + b"(?=" + b"".join(struct_regex) + b")", + consumed, + ) + else: + return b"".join(struct_regex), consumed + + +class RegexOverrides(Enum): + """Custom overrides""" + + # Match anything but \x00{ptr_size} + NON_NULL_POINTER = ( + lambda ptr_size: b"(?:(?!\x00" + f"{{{ptr_size}}}).{{{ptr_size}}})".encode() + ) + + +def scan_memory_for_modules( + context: interfaces.context.ContextInterface, + kernel_module_name: str, + sections_to_scan: list, +): + """Scan a memory region to uncover modules structs + + Args: + context: The current context + kernel_module_name: The name of the kernel module + sections_to_scan: A list of tuples including a start address and a size + Yields: + A module object + """ + kernel = context.modules[kernel_module_name] + m_struct = kernel.get_symbol("module").type.vol.subtype + ptr_size = kernel.get_type("pointer").size + + # We assume module.mkobj.mod is a non-null pointer to the module itself, allowing us to drastically reduce regex candidates + overrides = { + "mkobj.mod": RegexOverrides.NON_NULL_POINTER(ptr_size), + "init_layout.mnt.mod": RegexOverrides.NON_NULL_POINTER(ptr_size), + } + + # Using a regex too long will eventually result in overlapping and alignements problems, so stop at mkobj + stop_key = "mkobj" + module_regex, s = regex_from_struct_members( + context, kernel_module_name, m_struct, overrides=overrides, stop_key=stop_key + ) + scanner = scanners.RegExScanner(module_regex) + scanned = context.layers[kernel.layer_name].scan( + context=context, scanner=scanner, sections=sections_to_scan + ) + # Iterate over candidates structs + for module_candidate_offset in scanned: + # Use a try-except block to avoid crashing on OOB read + try: + m = kernel.object( + object_type="module", offset=module_candidate_offset, absolute=True + ) + # Check if module mkobj.mod points to the candidate offset + if m.mkobj.mod == module_candidate_offset: + vollog.info( + f'Found module "{utility.array_to_string(m.name)}" at {hex(m.vol.offset)}' + ) + yield m + except exceptions.InvalidAddressException: + continue + + +def calculate_closest_modules_from_address( + target_address: int, flattened_handlers: list +): + """Determine closest modules for a given address. + + Args: + target_address: The target address to search boundaries for + handlers_set: A list containing an unordered list with flattened (module_start, module_end) from handlers. See flatten_modules_handlers() for reference. + Returns: + Tuple containing previous and next boundary + """ + # Insert target_address and sort + flattened_handlers.append(target_address) + flattened_handlers.sort() + # Determine position of target_address in list + target_address_index = flattened_handlers.index(target_address) + # Check if target_address index in list is : top of the list > bottom of the list > any other case + if target_address_index + 1 >= len(flattened_handlers): + return ( + flattened_handlers[target_address_index - 1], + None, + ) # No next boundary + elif target_address_index - 1 < 0: + return ( + None, + flattened_handlers[target_address_index + 1], + ) # No previous boundary + else: + return ( + flattened_handlers[target_address_index - 1], + flattened_handlers[target_address_index + 1], + ) + + +def flatten_modules_handlers( + handlers: list, +): + """Flatten a list of previously calculated modules handlers boundaries (extract all "start" and "end")""" + return list( + sum( + [(h[1], h[2]) for h in set(handlers)], + (), + ) + ) diff --git a/volatility3/framework/plugins/linux/hidden_modules.py b/volatility3/framework/plugins/linux/hidden_modules.py new file mode 100644 index 000000000..679f2b00a --- /dev/null +++ b/volatility3/framework/plugins/linux/hidden_modules.py @@ -0,0 +1,248 @@ +# 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 logging +from typing import List, Set, Tuple, 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) + ), + ] + + @staticmethod + def get_modules_memory_boundaries( + 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"): + 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 + + @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 + + @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) + + @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").vol.members["mod"][1] + 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_lsmod_module_addresses( + cls, + context: interfaces.context.ContextInterface, + vmlinux_module_name: str, + ) -> Set[int]: + """Obtain a set the known module addresses from linux.lsmod plugin + + 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 set containing known kernel module addresses + """ + 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) + } + return known_module_addresses + + def _generator(self): + vmlinux_module_name = self.config["kernel"] + 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 + ) + for module in self.get_hidden_modules( + self.context, + vmlinux_module_name, + known_module_addresses, + modules_memory_boundaries, + ): + 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/plugins/linux/ftrace.py b/volatility3/plugins/linux/ftrace.py new file mode 100644 index 000000000..1ccbf56bb --- /dev/null +++ b/volatility3/plugins/linux/ftrace.py @@ -0,0 +1,317 @@ +# This file is Copyright 2023 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# +import logging +from typing import List, Optional +from enum import Enum +from volatility3.plugins.linux import lsmod, check_unlinked_modules +from volatility3.framework import constants, exceptions, interfaces +from volatility3.framework.configuration import requirements +from volatility3.framework.renderers import format_hints, TreeGrid +from volatility3.framework.symbols import linux +from volatility3.framework.objects import utility + +vollog = logging.getLogger(__name__) +UNKNOWN = "UNKNOWN" + + +class FTRACEFLAGS(Enum): + FTRACE_OPS_FL_ENABLED = 1 << 0 + FTRACE_OPS_FL_DYNAMIC = 1 << 1 + FTRACE_OPS_FL_SAVE_REGS = 1 << 2 + FTRACE_OPS_FL_SAVE_REGS_IF_SUPPORTED = 1 << 3 + FTRACE_OPS_FL_RECURSION = 1 << 4 + FTRACE_OPS_FL_STUB = 1 << 5 + FTRACE_OPS_FL_INITIALIZED = 1 << 6 + FTRACE_OPS_FL_DELETED = 1 << 7 + FTRACE_OPS_FL_ADDING = 1 << 8 + FTRACE_OPS_FL_REMOVING = 1 << 9 + FTRACE_OPS_FL_MODIFYING = 1 << 10 + FTRACE_OPS_FL_ALLOC_TRAMP = 1 << 11 + FTRACE_OPS_FL_IPMODIFY = 1 << 12 + FTRACE_OPS_FL_PID = 1 << 13 + FTRACE_OPS_FL_RCU = 1 << 14 + FTRACE_OPS_FL_TRACE_ARRAY = 1 << 15 + FTRACE_OPS_FL_PERMANENT = 1 << 16 + FTRACE_OPS_FL_DIRECT = 1 << 17 + + +class Check_ftrace(interfaces.plugins.PluginInterface): + """Detect ftrace hooking""" + + _version = (1, 0, 0) + _required_framework_version = (2, 5, 2) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Linux kernel", + architectures=["Intel32", "Intel64", "AArch64"], + ), + requirements.PluginRequirement( + name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0) + ), + requirements.PluginRequirement( + name="check_unlinked_modules", + plugin=check_unlinked_modules.Check_unlinked_modules, + version=(1, 0, 0), + ), + requirements.BooleanRequirement( + name="show_ftrace_flags", + description="Show ftrace flags associated with an ftrace_ops", + optional=True, + default=False, + ), + ] + + def run(self): + """Plugin output format : + + ftrace_ops : hex("ftrace_ops struct offset") + Callback : "callback offset" ["callback symbol" | "UNKNOWN"] + Hooked symbol : "hooked symbol" | "UNKNOWN" + Module : hex("module offset") ["associated module"] | "UNKNOWN" + Callback out of kernel .text : True | False + ftrace flags : list("ftrace_ops flags") + """ + + # Naturally ordered by "ftrace_ops" (the same order as we walked the "ftrace_ops_list") + columns = [ + ("ftrace_ops", format_hints.Hex), + ("Callback", str), + ("Hooked symbol", str), + ("Module", str), + ("Callback out of kernel .text", bool), + ] + + if self.config.get("show_ftrace_flags"): + columns.append(("ftrace_ops flags", str)) + + return TreeGrid( + columns, + self._generator(), + ) + + def _generator(self): + """Iterate over ftrace_ops_list struct""" + + self.vmlinux = self.context.modules[self.config["kernel"]] + if not self.vmlinux.has_symbol("ftrace_ops_list"): + raise exceptions.SymbolError( + "ftrace_ops_list", + self.vmlinux.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 corrupt.', + ) + + self.checked_callbacks = {} + self.set_compiled_kernel_space_boundaries() + self.setup_modules_and_handlers() + + # Access head of ftrace_ops_list + ftrace_ops_head = self.vmlinux.object_from_symbol( + "ftrace_ops_list" + ).dereference() + local_ftrace_ops_list = [] + results = [] + + while True: + local_ftrace_ops_list.append(ftrace_ops_head) + if ftrace_ops_head.next.is_readable(): + ftrace_ops_head = ftrace_ops_head.next.dereference() + else: + break + + for i, ftrace_ops in enumerate(local_ftrace_ops_list): + self._progress_callback( + (i / len(local_ftrace_ops_list)) * 100, f"Scanning ftrace_ops_list..." + ) + try: + parse_result = self.parse_ftrace_ops(ftrace_ops=ftrace_ops) + if parse_result: + results.append((0, parse_result)) + except Exception as e: + vollog.exception(f"Unhandled exception : {e}") + + # Preferred to "yield", otherwise progress_callback and results overlap... + return results + + def parse_ftrace_ops(self, ftrace_ops): + """Main parser for an ftrace_ops struct""" + ftrace_ops_addr = ftrace_ops.vol.offset + ftrace_func_entries = self.walk_to_ftrace_func_entry(ftrace_ops) + + for ftrace_func_entry in ftrace_func_entries: + callback = int(ftrace_ops.func) + hook_symbols = wrapper_get_symbols_by_absolute_location( + self.vmlinux, ftrace_func_entry.ip.cast("pointer") + ) + + # Avoid running the aggressive module finder twice for an address, if it wasn't found previously + if self.checked_callbacks.get(callback): + module_name = self.checked_callbacks[callback] + else: + module_name = self.wrapper_lookup_module_address(callback) + self.checked_callbacks[callback] = module_name + + # Useful information allowing to detect if a module was inserted dynamically or if it is part of the compiled kernel + callback_out_of_kernel_range = ( + callback < self.kernel_space_start or callback > self.kernel_space_end + ) + ftrace_flags = self.parse_ftrace_flags(ftrace_ops.flags) + + ### Format results ### + callback_symbol = UNKNOWN + f_module = UNKNOWN + # Fetch more informations about the module + if module_name != UNKNOWN: + module_obj = get_module_object_from_name(module_name, self.modules) + if module_obj is None: + continue + + module_address = module_obj.vol.offset + f_module = f"{hex(module_address)} [{module_name}]" + callback_symbol = module_obj.get_symbol_by_address(callback) or UNKNOWN + + result = ( + format_hints.Hex(ftrace_ops_addr), + f"{hex(callback)} [{callback_symbol}]", + str(hook_symbols), + f_module, + callback_out_of_kernel_range, + ) + + if self.config.get("show_ftrace_flags"): + result += (str(ftrace_flags),) + + return result + + def parse_ftrace_flags(self, ftrace_flags_value: int): + """Parse flags set on a hook structure""" + ret = [] + for couple in FTRACEFLAGS: + if ftrace_flags_value & couple.value: + ret.append(couple.name) + + return ret + + def walk_to_ftrace_func_entry(self, ftrace_ops): + """Function wrapping the process of walking to every ftrace_func_entry for an ftrace_ops""" + + # Decompose walk for better debugging + try: + func_hash = ftrace_ops.func_hash.dereference() + except: + vollog.debug( + f"No func_hash for ftrace_ops@{hex(ftrace_ops.vol.offset)}, skipping..." + ) + return None + + try: + filter_hash = func_hash.filter_hash.dereference() + except: + vollog.debug( + f"No func_hash.filter_hash for ftrace_ops@{hex(ftrace_ops.vol.offset)}, skipping..." + ) + return None + + try: + bucket_head = filter_hash.buckets.dereference().first.dereference() + except exceptions.InvalidAddressException: + vollog.debug( + f"No func_hash.filter_hash.buckets for ftrace_ops@{hex(ftrace_ops.vol.offset)}, skipping..." + ) + return None + + while True: + yield bucket_head.cast("ftrace_func_entry") + if bucket_head.next.is_readable(): + bucket_head = bucket_head.next.dereference() + else: + break + + def set_compiled_kernel_space_boundaries(self): + """Set compiler kernel address spaces. Preferred to linux.LinuxUtilities.generate_kernel_handler_info()[0] for convenience""" + self.kernel_space_start = self.vmlinux.get_absolute_symbol_address("_stext") + self.kernel_space_end = self.vmlinux.get_absolute_symbol_address("_etext") + + def get_all_handlers(self): + """Concatenate all handlers ("/proc/modules", "/sys/module/" and "unlinked kobject modules from sysfs hierarchy")""" + return self.proc_handlers + self.sysfs_handlers + self.sysfs_unlinked_handlers + + def wrapper_lookup_module_address(self, leaked_address: int): + # Detect module name based on leaked_address address + module_name, _ = linux.LinuxUtilities.lookup_module_address( + self.vmlinux, + self.get_all_handlers(), + leaked_address, + ) + # Aggressive module finder, for deeply hidden rootkits (try to detect usage of kobject_del) + if module_name == UNKNOWN: + sysfs_unlinked_modules = check_unlinked_modules.Check_unlinked_modules( + self.context, self.config_path + )._generator(self.sysfs_handlers, self.sysfs_modules, leaked_address) + if sysfs_unlinked_modules: + sysfs_unlinked_modules = [m[1] for m in sysfs_unlinked_modules] + self.sysfs_unlinked_handlers = ( + linux.LinuxUtilities.generate_kernel_handler_info( + self.context, self.vmlinux.name, sysfs_unlinked_modules + ) + ) + self.modules.extend(sysfs_unlinked_modules) + + # Search handlers again with the new informations, to see if leaked_address fits now + module_name, _ = linux.LinuxUtilities.lookup_module_address( + self.vmlinux, + self.get_all_handlers(), + leaked_address, + ) + + return module_name + + def setup_modules_and_handlers(self): + # Get /proc/modules and /sys/module/ listed modules + self.proc_modules = list( + lsmod.Lsmod.list_modules(self.context, self.vmlinux.name) + ) + self.sysfs_modules = list( + check_unlinked_modules.Check_unlinked_modules.wrapper_get_sysfs_modules( + self.context, self.config_path, self.vmlinux.name + ) + ) + # Calculate boundaries for each module + self.proc_handlers = linux.LinuxUtilities.generate_kernel_handler_info( + self.context, self.vmlinux.name, self.proc_modules + ) + self.sysfs_handlers = linux.LinuxUtilities.generate_kernel_handler_info( + self.context, self.vmlinux.name, self.sysfs_modules + ) + self.sysfs_unlinked_handlers = [] + self.modules = self.proc_modules + self.sysfs_modules + + +## UTILITIES ## +def wrapper_get_symbols_by_absolute_location( + vmlinux: interfaces.context.ModuleInterface, target_address: int +): + """List symbols related to a specified address""" + + symbols = list(vmlinux.get_symbols_by_absolute_location(target_address)) + + if len(symbols) == 0: + return "UNKNOWN" + else: + return [ + symbol if constants.BANG not in symbol else symbol.split(constants.BANG)[1] + for symbol in symbols + ] + + +def get_module_object_from_name( + wanted_module_name: str, modules: list +) -> Optional[linux.extensions.module]: + """Return a module object based on a module name""" + for m in modules: + if utility.array_to_string(m.name) == wanted_module_name: + return m