diff --git a/test/plugins/linux/linux.py b/test/plugins/linux/linux.py index 745c432f7..cbda81c48 100644 --- a/test/plugins/linux/linux.py +++ b/test/plugins/linux/linux.py @@ -89,7 +89,7 @@ class TestLinuxProcMaps: class TestLinuxTtyCheck: def test_linux_generic_tty_check(self, image, volatility, python): rc, out, _err = test_volatility.runvol_plugin( - "linux.tty_check.tty_check", image, volatility, python + "linux.malware.tty_check.Tty_Check", image, volatility, python ) assert rc == 0 @@ -504,7 +504,7 @@ class TestLinuxKmesg: class TestLinuxNetfilter: def test_linux_generic_netfilter(self, image, volatility, python): rc, out, _err = test_volatility.runvol_plugin( - "linux.netfilter.Netfilter", image, volatility, python + "linux.malware.netfilter.Netfilter", image, volatility, python ) # linux-sample-1.bin has no suspicious results for this plugin. diff --git a/volatility3/framework/plugins/linux/malware/modxview.py b/volatility3/framework/plugins/linux/malware/modxview.py new file mode 100644 index 000000000..c1707d26f --- /dev/null +++ b/volatility3/framework/plugins/linux/malware/modxview.py @@ -0,0 +1,181 @@ +# 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, Dict, Iterator + +import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules + +from volatility3.framework import interfaces, deprecation, renderers +from volatility3.framework.configuration import requirements +from volatility3.framework.renderers import format_hints +from volatility3.framework.symbols.linux import extensions +from volatility3.framework.constants import architectures +from volatility3.framework.symbols.linux.utilities import tainting + +vollog = logging.getLogger(__name__) + + +class Modxview(interfaces.plugins.PluginInterface): + """Centralize lsmod, check_modules and hidden_modules results to efficiently \ +spot modules presence and taints.""" + + _version = (1, 0, 0) + _required_framework_version = (2, 17, 0) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Linux kernel", + architectures=architectures.LINUX_ARCHS, + ), + requirements.VersionRequirement( + name="linux_utilities_modules", + component=linux_utilities_modules.Modules, + version=(3, 0, 0), + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherer_lsmod", + component=linux_utilities_modules.ModuleGathererLsmod, + version=(1, 0, 0), + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherer_sysfs", + component=linux_utilities_modules.ModuleGathererSysFs, + version=(1, 0, 0), + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherer_scanner", + component=linux_utilities_modules.ModuleGathererScanner, + version=(1, 0, 0), + ), + requirements.VersionRequirement( + name="linux-tainting", component=tainting.Tainting, version=(1, 0, 0) + ), + requirements.BooleanRequirement( + name="plain_taints", + description="Display the plain taints string for each module.", + optional=True, + default=False, + ), + ] + + @classmethod + @deprecation.deprecated_method( + replacement=linux_utilities_modules.Modules.flatten_run_modules_results, + replacement_version=(3, 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]: + """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: + Iterator of modules objects + """ + 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=(3, 0, 0), + removal_date="2025-09-25", + ) + def run_modules_scanners( + cls, + context: interfaces.context.ContextInterface, + kernel_name: str, + 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.""" + return linux_utilities_modules.Modules.run_modules_scanners( + context, kernel_name, run_hidden_modules + ) + + def _generator(self): + kernel = self.context.modules[self.config["kernel"]] + + wanted_gatherers = [ + linux_utilities_modules.ModuleGathererLsmod, + linux_utilities_modules.ModuleGathererSysFs, + linux_utilities_modules.ModuleGathererScanner, + ] + + run_results = linux_utilities_modules.Modules.run_modules_scanners( + context=self.context, + kernel_module_name=self.config["kernel"], + caller_wanted_gatherers=wanted_gatherers, + flatten=False, + ) + + aggregated_modules = {} + # We want to be explicit on the plugins results we are interested in + for gatherer in wanted_gatherers: + # Iterate over each recovered module + for mod_info in run_results[gatherer.name]: + # Use offsets as unique keys, whether a module + # appears in many plugin runs or not + if aggregated_modules.get(mod_info.offset, None) is not None: + # Append the plugin to the list of originating plugins + aggregated_modules[mod_info.offset].append(gatherer.name) + else: + aggregated_modules[mod_info.offset] = [gatherer.name] + + for module_offset, gatherers in aggregated_modules.items(): + module = kernel.object("module", offset=module_offset, absolute=True) + + # Tainting parsing capabilities applied to the module + if self.config.get("plain_taints"): + taints = tainting.Tainting.get_taints_as_plain_string( + self.context, + self.config["kernel"], + module.taints, + True, + ) + else: + taints = ",".join( + tainting.Tainting.get_taints_parsed( + self.context, + self.config["kernel"], + module.taints, + True, + ) + ) + + yield ( + 0, + ( + module.get_name() or renderers.NotAvailableValue(), + format_hints.Hex(module_offset), + linux_utilities_modules.ModuleGathererLsmod.name in gatherers, + linux_utilities_modules.ModuleGathererSysFs.name in gatherers, + linux_utilities_modules.ModuleGathererScanner.name in gatherers, + taints or renderers.NotAvailableValue(), + ), + ) + + def run(self): + columns = [ + ("Name", str), + ("Address", format_hints.Hex), + ("In procfs", bool), + ("In sysfs", bool), + ("In scan", bool), + ("Taints", str), + ] + + return renderers.TreeGrid( + columns, + self._generator(), + ) diff --git a/volatility3/framework/plugins/linux/malware/netfilter.py b/volatility3/framework/plugins/linux/malware/netfilter.py new file mode 100644 index 000000000..d724d4296 --- /dev/null +++ b/volatility3/framework/plugins/linux/malware/netfilter.py @@ -0,0 +1,806 @@ +# 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 +# +from dataclasses import dataclass, field +from abc import ABC, abstractmethod +import logging + +import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules +from typing import Iterator, List, Tuple, Optional +from volatility3 import framework +from volatility3.framework import ( + constants, + interfaces, + renderers, + exceptions, + deprecation, +) +from volatility3.framework.renderers import format_hints +from volatility3.framework.configuration import requirements +from volatility3.framework.symbols.linux import network + +vollog = logging.getLogger(__name__) + + +@dataclass +class Proto: + name: str + hooks: Tuple[str] = field(default_factory=tuple) + + +PROTO_NOT_IMPLEMENTED = Proto(name="UNSPEC") + +NF_INET_HOOKS = ("PRE_ROUTING", "LOCAL_IN", "FORWARD", "LOCAL_OUT", "POST_ROUTING") +NF_DEC_HOOKS = ( + "PRE_ROUTING", + "LOCAL_IN", + "FORWARD", + "LOCAL_OUT", + "POST_ROUTING", + "HELLO", + "ROUTE", +) +NF_ARP_HOOKS = ("IN", "OUT", "FORWARD") +NF_NETDEV_HOOKS = ("INGRESS", "EGRESS") +LARGEST_HOOK_NUMBER = max( + len(NF_INET_HOOKS), len(NF_DEC_HOOKS), len(NF_ARP_HOOKS), len(NF_NETDEV_HOOKS) +) + + +class AbstractNetfilter(ABC): + """Netfilter Abstract Base Classes handling details across various + Netfilter implementations, including constants, helpers, and common + routines. + """ + + PROTO_HOOKS = ( + PROTO_NOT_IMPLEMENTED, # NFPROTO_UNSPEC + Proto(name="INET", hooks=NF_INET_HOOKS), # From kernels 3.14 + Proto(name="IPV4", hooks=NF_INET_HOOKS), + Proto(name="ARP", hooks=NF_ARP_HOOKS), + PROTO_NOT_IMPLEMENTED, + Proto(name="NETDEV", hooks=NF_NETDEV_HOOKS), + PROTO_NOT_IMPLEMENTED, + Proto(name="BRIDGE", hooks=NF_INET_HOOKS), + PROTO_NOT_IMPLEMENTED, + PROTO_NOT_IMPLEMENTED, + Proto(name="IPV6", hooks=NF_INET_HOOKS), + PROTO_NOT_IMPLEMENTED, + Proto(name="DECNET", hooks=NF_DEC_HOOKS), # Removed in kernel 6.1 + ) + NF_MAX_HOOKS = LARGEST_HOOK_NUMBER + 1 + + def __init__( + self, context: interfaces.context.ContextInterface, kernel_module_name: str + ): + self._context = context + self.vmlinux = context.modules[kernel_module_name] + self.layer_name = self.vmlinux.layer_name + + # Set data sizes + self.ptr_size = self.vmlinux.get_type("pointer").size + self.list_head_size = self.vmlinux.get_type("list_head").size + + linuxutils_modulegatherers_required_version = ( + Netfilter._required_linuxutils_gatherers_version + ) + linuxutils_modulegatherers_current_version = ( + linux_utilities_modules.ModuleGatherers.version + ) + if not requirements.VersionRequirement.matches_required( + linuxutils_modulegatherers_required_version, + linuxutils_modulegatherers_current_version, + ): + raise exceptions.PluginRequirementException( + f"linux_utilities_modules.ModuleGatherer version not suitable: required {linuxutils_modulegatherers_required_version} found {linuxutils_modulegatherers_current_version}" + ) + + linux_net_required_version = Netfilter._required_linuxnet_version + linux_net_current_version = network.NetSymbols.version + if not requirements.VersionRequirement.matches_required( + linux_net_required_version, linux_net_current_version + ): + raise exceptions.PluginRequirementException( + f"symbols.linux.net.NetSymbols version not suitable: required {linux_net_required_version} found {linux_net_current_version}" + ) + + linux_utilities_modules_required_version = ( + Netfilter._required_linux_utilities_modules_version + ) + linux_utilities_modules_current_version = ( + linux_utilities_modules.Modules.version + ) + if not requirements.VersionRequirement.matches_required( + linux_utilities_modules_required_version, + linux_utilities_modules_current_version, + ): + raise exceptions.PluginRequirementException( + f"linux_utilities_modules.Modules version not suitable: required {linux_utilities_modules_required_version} found {linux_utilities_modules_current_version}" + ) + + symbol_table = context.symbol_space[self.vmlinux.symbol_table_name] + network.NetSymbols.apply(symbol_table) + + self.handlers = linux_utilities_modules.Modules.run_modules_scanners( + context=context, + kernel_module_name=kernel_module_name, + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, + ) + + @classmethod + def run_all( + cls, context: interfaces.context.ContextInterface, kernel_module_name: str + ) -> Iterator[Tuple[int, str, str, int, int, str, bool]]: + """It calls each subclass symtab_checks() to test the required + conditions to that specific kernel implementation. + + Args: + context: The volatility3 context on which to operate + kernel_module_name: The name of the table containing the kernel symbols + + Yields: + The kmsg records. Same as _run() + """ + vmlinux = context.modules[kernel_module_name] + + implementation_inst = None # type: ignore + for subclass in framework.class_subclasses(cls): + if not subclass.symtab_checks(vmlinux=vmlinux): + vollog.log( + constants.LOGLEVEL_VVVV, + "Netfilter implementation '%s' doesn't match this memory dump", + subclass.__name__, + ) + continue + + vollog.log( + constants.LOGLEVEL_VVVV, + "Netfilter implementation '%s' matches!", + subclass.__name__, + ) + implementation_inst = subclass( + context=context, kernel_module_name=kernel_module_name + ) + # More than one class could be executed for an specific kernel version + # For instance: Netfilter Ingress hooks + yield from implementation_inst._run() + + if implementation_inst is None: + vollog.error("Unsupported Netfilter kernel implementation") + + def _run(self) -> Iterator[Tuple[int, str, str, int, int, str, bool]]: + """Iterates over namespaces and protocols, executing various callbacks that + allow customization of the code to the specific data structure used in a + particular kernel implementation + + get_hooks_container(net, proto_name, hook_name) + It returns the data structure used in a specific kernel implementation + to store the hooks for a respective namespace and protocol, basically: + For Ingress hooks: + network_namespace[] -> net_device[] -> nf_hooks_ingress[] + For egress hooks: + network_namespace[] -> net_device[] -> nf_hooks_egress[] + For all the other Netfilter hooks: + <= 4.2.8 + nf_hooks[] + >= 4.3 + network_namespace[] -> nf.hooks[] + + get_hook_ops(hook_container, proto_idx, hook_idx) + Give the 'hook_container' got in get_hooks_container(), it + returns an iterable of 'nf_hook_ops' elements for a respective protocol + and hook type. + + Returns: + netns [int]: Network namespace id + proto_name [str]: Protocol name + hook_name [str]: Hook name + priority [int]: Priority + hook_ops_hook [int]: Hook address + module_name [str]: Linux kernel module name + hooked [bool]: "True" if the network stack has been hijacked + """ + for netns, net in self.get_net_namespaces(): + for proto_idx, proto_name, hook_idx, hook_name in self._proto_hook_loop(): + hooks_container = self.get_hooks_container(net, proto_name, hook_name) + + for hook_container in hooks_container: + for hook_ops in self.get_hook_ops( + hook_container, proto_idx, hook_idx + ): + if not hook_ops: + continue + + priority = int(hook_ops.priority) + hook_ops_hook = hook_ops.hook + module_info, symbol_name = ( + linux_utilities_modules.Modules.module_lookup_by_address( + self._context, + self.vmlinux.name, + self.handlers, + hook_ops_hook, + ) + ) + hooked = module_info is None + + yield netns, proto_name, hook_name, priority, hook_ops_hook, module_info, symbol_name, hooked + + @classmethod + @abstractmethod + def symtab_checks(cls, vmlinux: interfaces.context.ModuleInterface) -> bool: + """This method on each sublasss will be called to evaluate if the kernel + being analyzed fulfill the type & symbols requirements for the implementation. + The first class returning True will be instantiated and called via the + run() method. + + Returns: + bool: True if the kernel being analyzed fulfill the class requirements. + """ + + def _proto_hook_loop(self) -> Iterator[Tuple[int, str, int, str]]: + """Flattens the protocol families and hooks""" + for proto_idx, proto in enumerate(AbstractNetfilter.PROTO_HOOKS): + if proto == PROTO_NOT_IMPLEMENTED: + continue + if proto.name not in self.subscribed_protocols(): + # This protocol is not managed in this object + continue + for hook_idx, hook_name in enumerate(proto.hooks): + yield proto_idx, proto.name, hook_idx, hook_name + + def build_nf_hook_ops_array( + self, nf_hook_entries + ) -> Optional[interfaces.objects.ObjectInterface]: + """Function helper to build the nf_hook_ops array when it is not part of the + struct 'nf_hook_entries' definition. + + nf_hook_ops was stored adjacent in memory to the nf_hook_entry array, in the + new struct 'nf_hook_entries'. However, this 'nf_hooks_ops' array 'orig_ops' is + not part of the 'nf_hook_entries' struct. So, we need to calculate the offset. + + struct nf_hook_entries { + u16 num_hook_entries; /* plus padding */ + struct nf_hook_entry hooks[]; + //const struct nf_hook_ops *orig_ops[]; + } + """ + nf_hook_entry_size = self.vmlinux.get_type("nf_hook_entry").size + + try: + num_hook_entries = nf_hook_entries.num_hook_entries + except exceptions.InvalidAddressException: + return None + + orig_ops_addr = ( + nf_hook_entries.hooks.vol.offset + nf_hook_entry_size * num_hook_entries + ) + + if not self.vmlinux._context.layers[self.vmlinux.layer_name].is_valid( + orig_ops_addr + ): + return None + + orig_ops = self._context.object( + object_type=self.get_symbol_fullname("array"), + offset=orig_ops_addr, + subtype=self.vmlinux.get_type("pointer"), + layer_name=self.layer_name, + count=num_hook_entries, + ) + + return orig_ops + + def subscribed_protocols(self) -> Tuple[str]: + """Allows to select which PROTO_HOOKS protocols will be processed by the + Netfiler subclass. + """ + + # Most implementation handlers respond to these protocols, except for + # the ingress hook, which specifically handles the 'NETDEV' protocol. + # However, there is no corresponding Netfilter hook implementation for + # the INET protocol in the kernel. AFAIU, this is used as + # 'NFPROTO_INET = NFPROTO_IPV4 || NFPROTO_IPV6' + # in other parts of the kernel source code. + return ("IPV4", "ARP", "BRIDGE", "IPV6", "DECNET") + + @deprecation.method_being_removed( + removal_date="2025-09-25", + message="Callers to this method should adapt `linux_utilities_modules.Modules.run_module_scanners`", + ) + def get_module_name_for_address(self, addr) -> str: + """Helper to obtain the module and symbol name in the format needed for the + output of this plugin. + """ + module_name, symbol_name = ( + linux_utilities_modules.Modules.lookup_module_address( + self._context, self.vmlinux.name, self.handlers, addr + ) + ) + + if module_name == "UNKNOWN": + module_name = None + + if symbol_name != "N/A": + module_name = f"[{symbol_name}]" + + return module_name + + def get_net_namespaces(self): + """Common function to retrieve the different namespaces. + From 4.3 on, all the implementations use network namespaces. + """ + nethead = self.vmlinux.object_from_symbol("net_namespace_list") + symbol_net_name = self.get_symbol_fullname("net") + for net in nethead.to_list(symbol_net_name, "list"): + net_ns_id = net.ns.inum + yield net_ns_id, net + + def get_hooks_container(self, net, proto_name, hook_name): + """Returns the data structure used in a specific kernel implementation to store + the hooks for a respective namespace and protocol. + + Except for kernels < 4.3, all the implementations use network namespaces. + Also the data structure which contains the hooks, even though it changes its + implementation and/or data type, it is always in this location. + """ + yield net.nf.hooks + + def get_hook_ops(self, hook_container, proto_idx, hook_idx): + """Given the hook_container obtained from get_hooks_container(), it + returns an iterable of 'nf_hook_ops' elements for a corresponding protocol + and hook type. + + This is the most variable/unstable part of all Netfilter hook designs, it + changes almost in every single implementation. + """ + raise NotImplementedError("You must implement this method") + + def get_symbol_fullname(self, symbol_basename: str) -> str: + """Given a short symbol or type name, it returns its full name""" + return self.vmlinux.symbol_table_name + constants.BANG + symbol_basename + + @staticmethod + def get_member_type( + vol_type: interfaces.objects.Template, member_name: str + ) -> List[str]: + """Returns a list of types/subtypes belonging to the given type member. + + Args: + vol_type (interfaces.objects.Template): A vol3 type object + member_name (str): The member name + + Returns: + list: A list of types/subtypes + """ + _size, vol_obj = vol_type.vol.members[member_name] + type_name = vol_obj.type_name + type_basename = type_name.split(constants.BANG)[1] + member_type = [type_basename] + cur_type = vol_obj + while hasattr(cur_type, "subtype"): + subtype_name = cur_type.subtype.type_name + subtype_basename = subtype_name.split(constants.BANG)[1] + member_type.append(subtype_basename) + cur_type = cur_type.subtype + + return member_type + + +class NetfilterImp_to_4_3(AbstractNetfilter): + """At this point, Netfilter hooks were implemented as a linked list of struct + 'nf_hook_ops' type. One linked list per protocol per hook type. + It was like that until 4.2.8. + + struct list_head nf_hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + return vmlinux.has_symbol("nf_hooks") + + def get_net_namespaces(self): + # In kernels <= 4.2.8 netfilter hooks are not implemented per namespaces + netns, net = renderers.NotAvailableValue(), renderers.NotAvailableValue() + yield netns, net + + def get_hooks_container(self, net, proto_name, hook_name): + nf_hooks = self.vmlinux.object_from_symbol("nf_hooks") + if not nf_hooks: + return + + yield nf_hooks + + def get_hook_ops(self, hook_container, proto_idx, hook_idx): + list_head = hook_container[proto_idx][hook_idx] + nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops") + return list_head.to_list(nf_hooks_ops_name, "list") + + +class NetfilterImp_4_3_to_4_9(AbstractNetfilter): + """Netfilter hooks were added to network namespaces in 4.3. + It is still implemented as a linked list of 'struct nf_hook_ops' type but inside a + network namespace. One linked list per protocol per hook type. + + struct net { ... struct netns_nf nf; ... } + struct netns_nf { ... + struct list_head hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... } + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + return ( + vmlinux.has_symbol("net_namespace_list") + and vmlinux.has_type("netns_nf") + and vmlinux.get_type("netns_nf").has_member("hooks") + and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") + == ["array", "array", "list_head"] + ) + + def get_hook_ops(self, hook_container, proto_idx, hook_idx): + list_head = hook_container[proto_idx][hook_idx] + nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops") + return list_head.to_list(nf_hooks_ops_name, "list") + + +class NetfilterImp_4_9_to_4_14(AbstractNetfilter): + """In this range of kernel versions, the doubly-linked lists of netfilter hooks were + replaced by an array of arrays of 'nf_hook_entry' pointers in a singly-linked lists. + struct net { ... struct netns_nf nf; ... } + struct netns_nf { .. + struct nf_hook_entry __rcu *hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... } + + Also in v4.10 the struct nf_hook_entry changed, a hook function pointer was added to + it. However, for simplicity of this design, we will still take the hook address from + the 'nf_hook_ops'. As per v5.0-rc2, the hook address is duplicated in both sides. + - v4.9: + struct nf_hook_entry { + struct nf_hook_entry *next; + struct nf_hook_ops ops; + const struct nf_hook_ops *orig_ops; }; + - v4.10: + struct nf_hook_entry { + struct nf_hook_entry *next; + nf_hookfn *hook; + void *priv; + const struct nf_hook_ops *orig_ops; }; + (*) Even though the hook address is in the struct 'nf_hook_entry', we use the + original 'nf_hook_ops' hook address value, the one which was filled by the user, to + make it uniform to all the implementations. + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + hooks_type = ["array", "array", "pointer", "nf_hook_entry"] + return ( + vmlinux.has_symbol("net_namespace_list") + and vmlinux.has_type("netns_nf") + and vmlinux.get_type("netns_nf").has_member("hooks") + and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") == hooks_type + ) + + def _get_hook_ops(self, hook_container, proto_idx, hook_idx): + list_head = hook_container[proto_idx][hook_idx] + nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops") + return list_head.to_list(nf_hooks_ops_name, "list") + + def get_hook_ops(self, hook_container, proto_idx, hook_idx): + nf_hook_entry_list = hook_container[proto_idx][hook_idx] + while nf_hook_entry_list: + yield nf_hook_entry_list.orig_ops + nf_hook_entry_list = nf_hook_entry_list.next + + +class NetfilterImp_4_14_to_4_16(AbstractNetfilter): + """'nf_hook_ops' was removed from struct 'nf_hook_entry'. Instead, it was stored + adjacent in memory to the 'nf_hook_entry' array, in the new struct 'nf_hook_entries' + However, 'orig_ops' is not part of the 'nf_hook_entries' struct definition. So, we + have to craft it by hand. + + struct net { ... struct netns_nf nf; ... } + struct netns_nf { + struct nf_hook_entries *hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... } + struct nf_hook_entries { + u16 num_hook_entries; /* plus padding */ + struct nf_hook_entry hooks[]; + //const struct nf_hook_ops *orig_ops[]; } + struct nf_hook_entry { + nf_hookfn *hook; + void *priv; } + + (*) Even though the hook address is in the struct 'nf_hook_entry', we use the + original 'nf_hook_ops' hook address value, the one which was filled by the user, to + make it uniform to all the implementations. + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + hooks_type = ["array", "array", "pointer", "nf_hook_entries"] + return ( + vmlinux.has_symbol("net_namespace_list") + and vmlinux.has_type("netns_nf") + and vmlinux.get_type("netns_nf").has_member("hooks") + and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") == hooks_type + ) + + def get_nf_hook_entries(self, nf_hooks_addr, proto_idx, hook_idx): + """This allows to support different hook array implementations from this version + on. For instance, in kernels >= 4.16 this multi-dimensional array is split in + one-dimensional array of pointers to 'nf_hooks_entries' per each protocol.""" + return nf_hooks_addr[proto_idx][hook_idx] + + def get_hook_ops(self, hook_container, proto_idx, hook_idx): + nf_hook_entries = self.get_nf_hook_entries(hook_container, proto_idx, hook_idx) + if not nf_hook_entries: + return + + nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops") + nf_hook_ops_ptr_arr = self.build_nf_hook_ops_array(nf_hook_entries) + if not nf_hook_ops_ptr_arr: + return + + for nf_hook_ops_ptr in nf_hook_ops_ptr_arr: + nf_hook_ops = nf_hook_ops_ptr.dereference().cast(nf_hook_ops_name) + yield nf_hook_ops + + +class NetfilterImp_4_16_to_latest(NetfilterImp_4_14_to_4_16): + """The multidimensional array of nf_hook_entries was split in a one-dimensional + array per each protocol. + + struct net { + struct netns_nf nf; ... } + struct netns_nf { + struct nf_hook_entries * hooks_ipv4[NF_INET_NUMHOOKS]; + struct nf_hook_entries * hooks_ipv6[NF_INET_NUMHOOKS]; + struct nf_hook_entries * hooks_arp[NF_ARP_NUMHOOKS]; + struct nf_hook_entries * hooks_bridge[NF_INET_NUMHOOKS]; + struct nf_hook_entries * hooks_decnet[NF_DN_NUMHOOKS]; ... } + struct nf_hook_entries { + u16 num_hook_entries; /* plus padding */ + struct nf_hook_entry hooks[]; + //const struct nf_hook_ops *orig_ops[]; } + struct nf_hook_entry { + nf_hookfn *hook; + void *priv; } + + (*) Even though the hook address is in the struct nf_hook_entry, we use the original + nf_hook_ops hook address value, the one which was filled by the user, to make it + uniform to all the implementations. + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + return ( + vmlinux.has_symbol("net_namespace_list") + and vmlinux.has_type("netns_nf") + and vmlinux.get_type("netns_nf").has_member("hooks_ipv4") + ) + + def get_hooks_container(self, net, proto_name, hook_name): + try: + if proto_name == "IPV4": + net_nf_hooks = net.nf.hooks_ipv4 + elif proto_name == "ARP": + net_nf_hooks = net.nf.hooks_arp + elif proto_name == "BRIDGE": + net_nf_hooks = net.nf.hooks_bridge + elif proto_name == "IPV6": + net_nf_hooks = net.nf.hooks_ipv6 + elif proto_name == "DECNET": + net_nf_hooks = net.nf.hooks_decnet + else: + return + + yield net_nf_hooks + + except AttributeError: + # Protocol family disabled at kernel compilation + # CONFIG_NETFILTER_FAMILY_ARP=n || + # CONFIG_NETFILTER_FAMILY_BRIDGE=n || + # CONFIG_DECNET=n + pass + + def _get_nf_hook_entries_ptr(self, nf_hooks_addr, proto_idx, hook_idx): + nf_hook_entries_ptr = nf_hooks_addr[hook_idx] + return nf_hook_entries_ptr + + def get_nf_hook_entries(self, nf_hooks_addr, proto_idx, hook_idx): + return nf_hooks_addr[hook_idx] + + +class AbstractNetfilterNetDev(AbstractNetfilter): + """Base class to handle the Netfilter NetDev hooks. + It won't be executed. It has some common functions to all Netfilter NetDev hook + implementations. + + Netfilter NetDev hooks are set per network device which belongs to a network + namespace. + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + return False + + def subscribed_protocols(self): + return ("NETDEV",) + + def get_hooks_container(self, net, proto_name, hook_name): + net_device_type = self.vmlinux.get_type("net_device") + net_device_name = self.get_symbol_fullname("net_device") + for net_device in net.dev_base_head.to_list(net_device_name, "dev_list"): + if hook_name == "INGRESS": + if net_device_type.has_member("nf_hooks_ingress"): + # CONFIG_NETFILTER_INGRESS=y + yield net_device.nf_hooks_ingress + + elif hook_name == "EGRESS": + if net_device_type.has_member("nf_hooks_egress"): + # CONFIG_NETFILTER_EGRESS=y + yield net_device.nf_hooks_egress + + +class NetfilterNetDevImp_4_2_to_4_9(AbstractNetfilterNetDev): + """This is the first version of Netfilter Ingress hooks which was implemented using + a doubly-linked list of 'nf_hook_ops'. + struct list_head nf_hooks_ingress; + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + hooks_type = ["list_head"] + return ( + vmlinux.has_symbol("net_namespace_list") + and vmlinux.has_type("net_device") + and vmlinux.get_type("net_device").has_member("nf_hooks_ingress") + and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress") + == hooks_type + ) + + def get_hook_ops(self, hook_container, proto_idx, hook_idx): + nf_hooks_ingress = hook_container + nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops") + return nf_hooks_ingress.to_list(nf_hook_ops_name, "list") + + +class NetfilterNetDevImp_4_9_to_4_14(AbstractNetfilterNetDev): + """In 4.9 it was changed to a simple singly-linked list. + struct nf_hook_entry * nf_hooks_ingress; + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + hooks_type = ["pointer", "nf_hook_entry"] + return ( + vmlinux.has_symbol("net_namespace_list") + and vmlinux.has_type("net_device") + and vmlinux.get_type("net_device").has_member("nf_hooks_ingress") + and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress") + == hooks_type + ) + + def get_hook_ops(self, hook_container, proto_idx, hook_idx): + nf_hooks_ingress_ptr = hook_container + if not nf_hooks_ingress_ptr: + return + + while nf_hooks_ingress_ptr: + nf_hook_entry = nf_hooks_ingress_ptr.dereference() + orig_ops = nf_hook_entry.orig_ops.dereference() + yield orig_ops + nf_hooks_ingress_ptr = nf_hooks_ingress_ptr.next + + +class NetfilterNetDevImp_4_14_to_latest(AbstractNetfilterNetDev): + """In 4.14 the hook list was converted to an array of pointers inside the struct + 'nf_hook_entries': + struct nf_hook_entries * nf_hooks_ingress; + struct nf_hook_entries { + u16 num_hook_entries; + struct nf_hook_entry hooks[]; + //const struct nf_hook_ops *orig_ops[]; } + """ + + @classmethod + def symtab_checks(cls, vmlinux) -> bool: + hooks_type = ["pointer", "nf_hook_entries"] + return ( + vmlinux.has_symbol("net_namespace_list") + and vmlinux.has_type("net_device") + and vmlinux.get_type("net_device").has_member("nf_hooks_ingress") + and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress") + == hooks_type + ) + + def get_hook_ops(self, hook_container, proto_idx, hook_idx): + nf_hook_entries = hook_container + if not nf_hook_entries: + return + + nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops") + nf_hook_ops_ptr_arr = self.build_nf_hook_ops_array(nf_hook_entries) + if not nf_hook_ops_ptr_arr: + return + + for nf_hook_ops_ptr in nf_hook_ops_ptr_arr: + nf_hook_ops = nf_hook_ops_ptr.dereference().cast(nf_hook_ops_name) + yield nf_hook_ops + + +class Netfilter(interfaces.plugins.PluginInterface): + """Lists Netfilter hooks.""" + + _required_framework_version = (2, 22, 0) + + _version = (2, 0, 0) + + _required_linux_utilities_modules_version = (3, 0, 0) + _required_linuxutils_gatherers_version = (1, 0, 0) + _required_linuxnet_version = (1, 0, 0) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Linux kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=cls._required_linuxutils_gatherers_version, + ), + requirements.VersionRequirement( + name="linuxnet", + component=network.NetSymbols, + version=cls._required_linuxnet_version, + ), + ] + + def _format_fields(self, fields): + ( + netns, + proto_name, + hook_name, + priority, + hook_func, + module_info, + symbol_name, + hooked, + ) = fields + + if module_info: + module_name = module_info.name + else: + module_name = renderers.NotAvailableValue() + + return ( + netns, + proto_name, + hook_name, + priority, + format_hints.Hex(hook_func), + module_name, + symbol_name or renderers.NotAvailableValue(), + str(hooked), + ) + + def _generator(self): + kernel_module_name = self.config["kernel"] + for fields in AbstractNetfilter.run_all( + context=self.context, kernel_module_name=kernel_module_name + ): + yield (0, self._format_fields(fields)) + + def run(self): + headers = [ + ("Net NS", int), + ("Proto", str), + ("Hook", str), + ("Priority", int), + ("Handler", format_hints.Hex), + ("Module", str), + ("Symbol", str), + ("Is Hooked", str), + ] + return renderers.TreeGrid(headers, self._generator()) diff --git a/volatility3/framework/plugins/linux/malware/tty_check.py b/volatility3/framework/plugins/linux/malware/tty_check.py new file mode 100644 index 000000000..1547c5cc6 --- /dev/null +++ b/volatility3/framework/plugins/linux/malware/tty_check.py @@ -0,0 +1,119 @@ +# This file is Copyright 2020 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 + +import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules +from volatility3.framework import interfaces, renderers, exceptions, constants +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.framework.symbols import linux + +vollog = logging.getLogger(__name__) + + +class Tty_Check(plugins.PluginInterface): + """Checks tty devices for hooks""" + + _required_framework_version = (2, 0, 0) + _version = (1, 0, 0) + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Linux kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="linux_utilities_modules", + component=linux_utilities_modules.Modules, + version=(3, 0, 0), + ), + requirements.VersionRequirement( + name="linux_utilities_module_gatherers", + component=linux_utilities_modules.ModuleGatherers, + version=(1, 0, 0), + ), + requirements.VersionRequirement( + name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0) + ), + ] + + def _generator(self): + vmlinux = self.context.modules[self.config["kernel"]] + + try: + tty_drivers = vmlinux.object_from_symbol("tty_drivers").cast("list_head") + except exceptions.SymbolError: + tty_drivers = None + + if not tty_drivers: + raise TypeError( + "This plugin requires the tty_drivers 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." + ) + + known_modules = linux_utilities_modules.Modules.run_modules_scanners( + context=self.context, + kernel_module_name=self.config["kernel"], + caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, + ) + + for tty in tty_drivers.to_list( + vmlinux.symbol_table_name + constants.BANG + "tty_driver", "tty_drivers" + ): + try: + ttys = utility.array_of_pointers( + tty.ttys.dereference(), + count=tty.num, + subtype=vmlinux.symbol_table_name + constants.BANG + "tty_struct", + context=self.context, + ) + except exceptions.PagedInvalidAddressException: + continue + + for tty_dev in ttys: + if tty_dev == 0: + continue + + try: + name = utility.array_to_string(tty_dev.name) + recv_buf = tty_dev.ldisc.ops.receive_buf + except exceptions.InvalidAddressException: + continue + + module_info, symbol_name = ( + linux_utilities_modules.Modules.module_lookup_by_address( + self.context, vmlinux.name, known_modules, recv_buf + ) + ) + + if module_info: + module_name = module_info.name + else: + module_name = renderers.NotAvailableValue() + + yield 0, ( + name, + format_hints.Hex(recv_buf), + module_name, + symbol_name or renderers.NotAvailableValue(), + ) + + def run(self): + return renderers.TreeGrid( + [ + ("Name", str), + ("Address", format_hints.Hex), + ("Module", str), + ("Symbol", str), + ], + self._generator(), + ) diff --git a/volatility3/framework/plugins/linux/modxview.py b/volatility3/framework/plugins/linux/modxview.py index c1707d26f..d91f37587 100644 --- a/volatility3/framework/plugins/linux/modxview.py +++ b/volatility3/framework/plugins/linux/modxview.py @@ -1,181 +1,21 @@ -# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# 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 # import logging -from typing import List, Dict, Iterator - -import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules - -from volatility3.framework import interfaces, deprecation, renderers -from volatility3.framework.configuration import requirements -from volatility3.framework.renderers import format_hints -from volatility3.framework.symbols.linux import extensions -from volatility3.framework.constants import architectures -from volatility3.framework.symbols.linux.utilities import tainting +from volatility3.framework import interfaces, deprecation +from volatility3.plugins.linux.malware import modxview vollog = logging.getLogger(__name__) -class Modxview(interfaces.plugins.PluginInterface): +class Modxview( + interfaces.plugins.PluginInterface, + deprecation.PluginRenameClass, + replacement_class=modxview.Modxview, + removal_date="2026-06-07", +): """Centralize lsmod, check_modules and hidden_modules results to efficiently \ -spot modules presence and taints.""" +spot modules presence and taints (deprecated).""" _version = (1, 0, 0) _required_framework_version = (2, 17, 0) - - @classmethod - def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: - return [ - requirements.ModuleRequirement( - name="kernel", - description="Linux kernel", - architectures=architectures.LINUX_ARCHS, - ), - requirements.VersionRequirement( - name="linux_utilities_modules", - component=linux_utilities_modules.Modules, - version=(3, 0, 0), - ), - requirements.VersionRequirement( - name="linux_utilities_module_gatherer_lsmod", - component=linux_utilities_modules.ModuleGathererLsmod, - version=(1, 0, 0), - ), - requirements.VersionRequirement( - name="linux_utilities_module_gatherer_sysfs", - component=linux_utilities_modules.ModuleGathererSysFs, - version=(1, 0, 0), - ), - requirements.VersionRequirement( - name="linux_utilities_module_gatherer_scanner", - component=linux_utilities_modules.ModuleGathererScanner, - version=(1, 0, 0), - ), - requirements.VersionRequirement( - name="linux-tainting", component=tainting.Tainting, version=(1, 0, 0) - ), - requirements.BooleanRequirement( - name="plain_taints", - description="Display the plain taints string for each module.", - optional=True, - default=False, - ), - ] - - @classmethod - @deprecation.deprecated_method( - replacement=linux_utilities_modules.Modules.flatten_run_modules_results, - replacement_version=(3, 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]: - """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: - Iterator of modules objects - """ - 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=(3, 0, 0), - removal_date="2025-09-25", - ) - def run_modules_scanners( - cls, - context: interfaces.context.ContextInterface, - kernel_name: str, - 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.""" - return linux_utilities_modules.Modules.run_modules_scanners( - context, kernel_name, run_hidden_modules - ) - - def _generator(self): - kernel = self.context.modules[self.config["kernel"]] - - wanted_gatherers = [ - linux_utilities_modules.ModuleGathererLsmod, - linux_utilities_modules.ModuleGathererSysFs, - linux_utilities_modules.ModuleGathererScanner, - ] - - run_results = linux_utilities_modules.Modules.run_modules_scanners( - context=self.context, - kernel_module_name=self.config["kernel"], - caller_wanted_gatherers=wanted_gatherers, - flatten=False, - ) - - aggregated_modules = {} - # We want to be explicit on the plugins results we are interested in - for gatherer in wanted_gatherers: - # Iterate over each recovered module - for mod_info in run_results[gatherer.name]: - # Use offsets as unique keys, whether a module - # appears in many plugin runs or not - if aggregated_modules.get(mod_info.offset, None) is not None: - # Append the plugin to the list of originating plugins - aggregated_modules[mod_info.offset].append(gatherer.name) - else: - aggregated_modules[mod_info.offset] = [gatherer.name] - - for module_offset, gatherers in aggregated_modules.items(): - module = kernel.object("module", offset=module_offset, absolute=True) - - # Tainting parsing capabilities applied to the module - if self.config.get("plain_taints"): - taints = tainting.Tainting.get_taints_as_plain_string( - self.context, - self.config["kernel"], - module.taints, - True, - ) - else: - taints = ",".join( - tainting.Tainting.get_taints_parsed( - self.context, - self.config["kernel"], - module.taints, - True, - ) - ) - - yield ( - 0, - ( - module.get_name() or renderers.NotAvailableValue(), - format_hints.Hex(module_offset), - linux_utilities_modules.ModuleGathererLsmod.name in gatherers, - linux_utilities_modules.ModuleGathererSysFs.name in gatherers, - linux_utilities_modules.ModuleGathererScanner.name in gatherers, - taints or renderers.NotAvailableValue(), - ), - ) - - def run(self): - columns = [ - ("Name", str), - ("Address", format_hints.Hex), - ("In procfs", bool), - ("In sysfs", bool), - ("In scan", bool), - ("Taints", str), - ] - - return renderers.TreeGrid( - columns, - self._generator(), - ) diff --git a/volatility3/framework/plugins/linux/netfilter.py b/volatility3/framework/plugins/linux/netfilter.py index d724d4296..741241039 100644 --- a/volatility3/framework/plugins/linux/netfilter.py +++ b/volatility3/framework/plugins/linux/netfilter.py @@ -1,806 +1,20 @@ -# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# 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 # -from dataclasses import dataclass, field -from abc import ABC, abstractmethod import logging - -import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules -from typing import Iterator, List, Tuple, Optional -from volatility3 import framework -from volatility3.framework import ( - constants, - interfaces, - renderers, - exceptions, - deprecation, -) -from volatility3.framework.renderers import format_hints -from volatility3.framework.configuration import requirements -from volatility3.framework.symbols.linux import network +from volatility3.framework import interfaces, deprecation +from volatility3.plugins.linux.malware import netfilter vollog = logging.getLogger(__name__) -@dataclass -class Proto: - name: str - hooks: Tuple[str] = field(default_factory=tuple) - - -PROTO_NOT_IMPLEMENTED = Proto(name="UNSPEC") - -NF_INET_HOOKS = ("PRE_ROUTING", "LOCAL_IN", "FORWARD", "LOCAL_OUT", "POST_ROUTING") -NF_DEC_HOOKS = ( - "PRE_ROUTING", - "LOCAL_IN", - "FORWARD", - "LOCAL_OUT", - "POST_ROUTING", - "HELLO", - "ROUTE", -) -NF_ARP_HOOKS = ("IN", "OUT", "FORWARD") -NF_NETDEV_HOOKS = ("INGRESS", "EGRESS") -LARGEST_HOOK_NUMBER = max( - len(NF_INET_HOOKS), len(NF_DEC_HOOKS), len(NF_ARP_HOOKS), len(NF_NETDEV_HOOKS) -) - - -class AbstractNetfilter(ABC): - """Netfilter Abstract Base Classes handling details across various - Netfilter implementations, including constants, helpers, and common - routines. - """ - - PROTO_HOOKS = ( - PROTO_NOT_IMPLEMENTED, # NFPROTO_UNSPEC - Proto(name="INET", hooks=NF_INET_HOOKS), # From kernels 3.14 - Proto(name="IPV4", hooks=NF_INET_HOOKS), - Proto(name="ARP", hooks=NF_ARP_HOOKS), - PROTO_NOT_IMPLEMENTED, - Proto(name="NETDEV", hooks=NF_NETDEV_HOOKS), - PROTO_NOT_IMPLEMENTED, - Proto(name="BRIDGE", hooks=NF_INET_HOOKS), - PROTO_NOT_IMPLEMENTED, - PROTO_NOT_IMPLEMENTED, - Proto(name="IPV6", hooks=NF_INET_HOOKS), - PROTO_NOT_IMPLEMENTED, - Proto(name="DECNET", hooks=NF_DEC_HOOKS), # Removed in kernel 6.1 - ) - NF_MAX_HOOKS = LARGEST_HOOK_NUMBER + 1 - - def __init__( - self, context: interfaces.context.ContextInterface, kernel_module_name: str - ): - self._context = context - self.vmlinux = context.modules[kernel_module_name] - self.layer_name = self.vmlinux.layer_name - - # Set data sizes - self.ptr_size = self.vmlinux.get_type("pointer").size - self.list_head_size = self.vmlinux.get_type("list_head").size - - linuxutils_modulegatherers_required_version = ( - Netfilter._required_linuxutils_gatherers_version - ) - linuxutils_modulegatherers_current_version = ( - linux_utilities_modules.ModuleGatherers.version - ) - if not requirements.VersionRequirement.matches_required( - linuxutils_modulegatherers_required_version, - linuxutils_modulegatherers_current_version, - ): - raise exceptions.PluginRequirementException( - f"linux_utilities_modules.ModuleGatherer version not suitable: required {linuxutils_modulegatherers_required_version} found {linuxutils_modulegatherers_current_version}" - ) - - linux_net_required_version = Netfilter._required_linuxnet_version - linux_net_current_version = network.NetSymbols.version - if not requirements.VersionRequirement.matches_required( - linux_net_required_version, linux_net_current_version - ): - raise exceptions.PluginRequirementException( - f"symbols.linux.net.NetSymbols version not suitable: required {linux_net_required_version} found {linux_net_current_version}" - ) - - linux_utilities_modules_required_version = ( - Netfilter._required_linux_utilities_modules_version - ) - linux_utilities_modules_current_version = ( - linux_utilities_modules.Modules.version - ) - if not requirements.VersionRequirement.matches_required( - linux_utilities_modules_required_version, - linux_utilities_modules_current_version, - ): - raise exceptions.PluginRequirementException( - f"linux_utilities_modules.Modules version not suitable: required {linux_utilities_modules_required_version} found {linux_utilities_modules_current_version}" - ) - - symbol_table = context.symbol_space[self.vmlinux.symbol_table_name] - network.NetSymbols.apply(symbol_table) - - self.handlers = linux_utilities_modules.Modules.run_modules_scanners( - context=context, - kernel_module_name=kernel_module_name, - caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, - ) - - @classmethod - def run_all( - cls, context: interfaces.context.ContextInterface, kernel_module_name: str - ) -> Iterator[Tuple[int, str, str, int, int, str, bool]]: - """It calls each subclass symtab_checks() to test the required - conditions to that specific kernel implementation. - - Args: - context: The volatility3 context on which to operate - kernel_module_name: The name of the table containing the kernel symbols - - Yields: - The kmsg records. Same as _run() - """ - vmlinux = context.modules[kernel_module_name] - - implementation_inst = None # type: ignore - for subclass in framework.class_subclasses(cls): - if not subclass.symtab_checks(vmlinux=vmlinux): - vollog.log( - constants.LOGLEVEL_VVVV, - "Netfilter implementation '%s' doesn't match this memory dump", - subclass.__name__, - ) - continue - - vollog.log( - constants.LOGLEVEL_VVVV, - "Netfilter implementation '%s' matches!", - subclass.__name__, - ) - implementation_inst = subclass( - context=context, kernel_module_name=kernel_module_name - ) - # More than one class could be executed for an specific kernel version - # For instance: Netfilter Ingress hooks - yield from implementation_inst._run() - - if implementation_inst is None: - vollog.error("Unsupported Netfilter kernel implementation") - - def _run(self) -> Iterator[Tuple[int, str, str, int, int, str, bool]]: - """Iterates over namespaces and protocols, executing various callbacks that - allow customization of the code to the specific data structure used in a - particular kernel implementation - - get_hooks_container(net, proto_name, hook_name) - It returns the data structure used in a specific kernel implementation - to store the hooks for a respective namespace and protocol, basically: - For Ingress hooks: - network_namespace[] -> net_device[] -> nf_hooks_ingress[] - For egress hooks: - network_namespace[] -> net_device[] -> nf_hooks_egress[] - For all the other Netfilter hooks: - <= 4.2.8 - nf_hooks[] - >= 4.3 - network_namespace[] -> nf.hooks[] - - get_hook_ops(hook_container, proto_idx, hook_idx) - Give the 'hook_container' got in get_hooks_container(), it - returns an iterable of 'nf_hook_ops' elements for a respective protocol - and hook type. - - Returns: - netns [int]: Network namespace id - proto_name [str]: Protocol name - hook_name [str]: Hook name - priority [int]: Priority - hook_ops_hook [int]: Hook address - module_name [str]: Linux kernel module name - hooked [bool]: "True" if the network stack has been hijacked - """ - for netns, net in self.get_net_namespaces(): - for proto_idx, proto_name, hook_idx, hook_name in self._proto_hook_loop(): - hooks_container = self.get_hooks_container(net, proto_name, hook_name) - - for hook_container in hooks_container: - for hook_ops in self.get_hook_ops( - hook_container, proto_idx, hook_idx - ): - if not hook_ops: - continue - - priority = int(hook_ops.priority) - hook_ops_hook = hook_ops.hook - module_info, symbol_name = ( - linux_utilities_modules.Modules.module_lookup_by_address( - self._context, - self.vmlinux.name, - self.handlers, - hook_ops_hook, - ) - ) - hooked = module_info is None - - yield netns, proto_name, hook_name, priority, hook_ops_hook, module_info, symbol_name, hooked - - @classmethod - @abstractmethod - def symtab_checks(cls, vmlinux: interfaces.context.ModuleInterface) -> bool: - """This method on each sublasss will be called to evaluate if the kernel - being analyzed fulfill the type & symbols requirements for the implementation. - The first class returning True will be instantiated and called via the - run() method. - - Returns: - bool: True if the kernel being analyzed fulfill the class requirements. - """ - - def _proto_hook_loop(self) -> Iterator[Tuple[int, str, int, str]]: - """Flattens the protocol families and hooks""" - for proto_idx, proto in enumerate(AbstractNetfilter.PROTO_HOOKS): - if proto == PROTO_NOT_IMPLEMENTED: - continue - if proto.name not in self.subscribed_protocols(): - # This protocol is not managed in this object - continue - for hook_idx, hook_name in enumerate(proto.hooks): - yield proto_idx, proto.name, hook_idx, hook_name - - def build_nf_hook_ops_array( - self, nf_hook_entries - ) -> Optional[interfaces.objects.ObjectInterface]: - """Function helper to build the nf_hook_ops array when it is not part of the - struct 'nf_hook_entries' definition. - - nf_hook_ops was stored adjacent in memory to the nf_hook_entry array, in the - new struct 'nf_hook_entries'. However, this 'nf_hooks_ops' array 'orig_ops' is - not part of the 'nf_hook_entries' struct. So, we need to calculate the offset. - - struct nf_hook_entries { - u16 num_hook_entries; /* plus padding */ - struct nf_hook_entry hooks[]; - //const struct nf_hook_ops *orig_ops[]; - } - """ - nf_hook_entry_size = self.vmlinux.get_type("nf_hook_entry").size - - try: - num_hook_entries = nf_hook_entries.num_hook_entries - except exceptions.InvalidAddressException: - return None - - orig_ops_addr = ( - nf_hook_entries.hooks.vol.offset + nf_hook_entry_size * num_hook_entries - ) - - if not self.vmlinux._context.layers[self.vmlinux.layer_name].is_valid( - orig_ops_addr - ): - return None - - orig_ops = self._context.object( - object_type=self.get_symbol_fullname("array"), - offset=orig_ops_addr, - subtype=self.vmlinux.get_type("pointer"), - layer_name=self.layer_name, - count=num_hook_entries, - ) - - return orig_ops - - def subscribed_protocols(self) -> Tuple[str]: - """Allows to select which PROTO_HOOKS protocols will be processed by the - Netfiler subclass. - """ - - # Most implementation handlers respond to these protocols, except for - # the ingress hook, which specifically handles the 'NETDEV' protocol. - # However, there is no corresponding Netfilter hook implementation for - # the INET protocol in the kernel. AFAIU, this is used as - # 'NFPROTO_INET = NFPROTO_IPV4 || NFPROTO_IPV6' - # in other parts of the kernel source code. - return ("IPV4", "ARP", "BRIDGE", "IPV6", "DECNET") - - @deprecation.method_being_removed( - removal_date="2025-09-25", - message="Callers to this method should adapt `linux_utilities_modules.Modules.run_module_scanners`", - ) - def get_module_name_for_address(self, addr) -> str: - """Helper to obtain the module and symbol name in the format needed for the - output of this plugin. - """ - module_name, symbol_name = ( - linux_utilities_modules.Modules.lookup_module_address( - self._context, self.vmlinux.name, self.handlers, addr - ) - ) - - if module_name == "UNKNOWN": - module_name = None - - if symbol_name != "N/A": - module_name = f"[{symbol_name}]" - - return module_name - - def get_net_namespaces(self): - """Common function to retrieve the different namespaces. - From 4.3 on, all the implementations use network namespaces. - """ - nethead = self.vmlinux.object_from_symbol("net_namespace_list") - symbol_net_name = self.get_symbol_fullname("net") - for net in nethead.to_list(symbol_net_name, "list"): - net_ns_id = net.ns.inum - yield net_ns_id, net - - def get_hooks_container(self, net, proto_name, hook_name): - """Returns the data structure used in a specific kernel implementation to store - the hooks for a respective namespace and protocol. - - Except for kernels < 4.3, all the implementations use network namespaces. - Also the data structure which contains the hooks, even though it changes its - implementation and/or data type, it is always in this location. - """ - yield net.nf.hooks - - def get_hook_ops(self, hook_container, proto_idx, hook_idx): - """Given the hook_container obtained from get_hooks_container(), it - returns an iterable of 'nf_hook_ops' elements for a corresponding protocol - and hook type. - - This is the most variable/unstable part of all Netfilter hook designs, it - changes almost in every single implementation. - """ - raise NotImplementedError("You must implement this method") - - def get_symbol_fullname(self, symbol_basename: str) -> str: - """Given a short symbol or type name, it returns its full name""" - return self.vmlinux.symbol_table_name + constants.BANG + symbol_basename - - @staticmethod - def get_member_type( - vol_type: interfaces.objects.Template, member_name: str - ) -> List[str]: - """Returns a list of types/subtypes belonging to the given type member. - - Args: - vol_type (interfaces.objects.Template): A vol3 type object - member_name (str): The member name - - Returns: - list: A list of types/subtypes - """ - _size, vol_obj = vol_type.vol.members[member_name] - type_name = vol_obj.type_name - type_basename = type_name.split(constants.BANG)[1] - member_type = [type_basename] - cur_type = vol_obj - while hasattr(cur_type, "subtype"): - subtype_name = cur_type.subtype.type_name - subtype_basename = subtype_name.split(constants.BANG)[1] - member_type.append(subtype_basename) - cur_type = cur_type.subtype - - return member_type - - -class NetfilterImp_to_4_3(AbstractNetfilter): - """At this point, Netfilter hooks were implemented as a linked list of struct - 'nf_hook_ops' type. One linked list per protocol per hook type. - It was like that until 4.2.8. - - struct list_head nf_hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - return vmlinux.has_symbol("nf_hooks") - - def get_net_namespaces(self): - # In kernels <= 4.2.8 netfilter hooks are not implemented per namespaces - netns, net = renderers.NotAvailableValue(), renderers.NotAvailableValue() - yield netns, net - - def get_hooks_container(self, net, proto_name, hook_name): - nf_hooks = self.vmlinux.object_from_symbol("nf_hooks") - if not nf_hooks: - return - - yield nf_hooks - - def get_hook_ops(self, hook_container, proto_idx, hook_idx): - list_head = hook_container[proto_idx][hook_idx] - nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops") - return list_head.to_list(nf_hooks_ops_name, "list") - - -class NetfilterImp_4_3_to_4_9(AbstractNetfilter): - """Netfilter hooks were added to network namespaces in 4.3. - It is still implemented as a linked list of 'struct nf_hook_ops' type but inside a - network namespace. One linked list per protocol per hook type. - - struct net { ... struct netns_nf nf; ... } - struct netns_nf { ... - struct list_head hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... } - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - return ( - vmlinux.has_symbol("net_namespace_list") - and vmlinux.has_type("netns_nf") - and vmlinux.get_type("netns_nf").has_member("hooks") - and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") - == ["array", "array", "list_head"] - ) - - def get_hook_ops(self, hook_container, proto_idx, hook_idx): - list_head = hook_container[proto_idx][hook_idx] - nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops") - return list_head.to_list(nf_hooks_ops_name, "list") - - -class NetfilterImp_4_9_to_4_14(AbstractNetfilter): - """In this range of kernel versions, the doubly-linked lists of netfilter hooks were - replaced by an array of arrays of 'nf_hook_entry' pointers in a singly-linked lists. - struct net { ... struct netns_nf nf; ... } - struct netns_nf { .. - struct nf_hook_entry __rcu *hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... } - - Also in v4.10 the struct nf_hook_entry changed, a hook function pointer was added to - it. However, for simplicity of this design, we will still take the hook address from - the 'nf_hook_ops'. As per v5.0-rc2, the hook address is duplicated in both sides. - - v4.9: - struct nf_hook_entry { - struct nf_hook_entry *next; - struct nf_hook_ops ops; - const struct nf_hook_ops *orig_ops; }; - - v4.10: - struct nf_hook_entry { - struct nf_hook_entry *next; - nf_hookfn *hook; - void *priv; - const struct nf_hook_ops *orig_ops; }; - (*) Even though the hook address is in the struct 'nf_hook_entry', we use the - original 'nf_hook_ops' hook address value, the one which was filled by the user, to - make it uniform to all the implementations. - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - hooks_type = ["array", "array", "pointer", "nf_hook_entry"] - return ( - vmlinux.has_symbol("net_namespace_list") - and vmlinux.has_type("netns_nf") - and vmlinux.get_type("netns_nf").has_member("hooks") - and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") == hooks_type - ) - - def _get_hook_ops(self, hook_container, proto_idx, hook_idx): - list_head = hook_container[proto_idx][hook_idx] - nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops") - return list_head.to_list(nf_hooks_ops_name, "list") - - def get_hook_ops(self, hook_container, proto_idx, hook_idx): - nf_hook_entry_list = hook_container[proto_idx][hook_idx] - while nf_hook_entry_list: - yield nf_hook_entry_list.orig_ops - nf_hook_entry_list = nf_hook_entry_list.next - - -class NetfilterImp_4_14_to_4_16(AbstractNetfilter): - """'nf_hook_ops' was removed from struct 'nf_hook_entry'. Instead, it was stored - adjacent in memory to the 'nf_hook_entry' array, in the new struct 'nf_hook_entries' - However, 'orig_ops' is not part of the 'nf_hook_entries' struct definition. So, we - have to craft it by hand. - - struct net { ... struct netns_nf nf; ... } - struct netns_nf { - struct nf_hook_entries *hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... } - struct nf_hook_entries { - u16 num_hook_entries; /* plus padding */ - struct nf_hook_entry hooks[]; - //const struct nf_hook_ops *orig_ops[]; } - struct nf_hook_entry { - nf_hookfn *hook; - void *priv; } - - (*) Even though the hook address is in the struct 'nf_hook_entry', we use the - original 'nf_hook_ops' hook address value, the one which was filled by the user, to - make it uniform to all the implementations. - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - hooks_type = ["array", "array", "pointer", "nf_hook_entries"] - return ( - vmlinux.has_symbol("net_namespace_list") - and vmlinux.has_type("netns_nf") - and vmlinux.get_type("netns_nf").has_member("hooks") - and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") == hooks_type - ) - - def get_nf_hook_entries(self, nf_hooks_addr, proto_idx, hook_idx): - """This allows to support different hook array implementations from this version - on. For instance, in kernels >= 4.16 this multi-dimensional array is split in - one-dimensional array of pointers to 'nf_hooks_entries' per each protocol.""" - return nf_hooks_addr[proto_idx][hook_idx] - - def get_hook_ops(self, hook_container, proto_idx, hook_idx): - nf_hook_entries = self.get_nf_hook_entries(hook_container, proto_idx, hook_idx) - if not nf_hook_entries: - return - - nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops") - nf_hook_ops_ptr_arr = self.build_nf_hook_ops_array(nf_hook_entries) - if not nf_hook_ops_ptr_arr: - return - - for nf_hook_ops_ptr in nf_hook_ops_ptr_arr: - nf_hook_ops = nf_hook_ops_ptr.dereference().cast(nf_hook_ops_name) - yield nf_hook_ops - - -class NetfilterImp_4_16_to_latest(NetfilterImp_4_14_to_4_16): - """The multidimensional array of nf_hook_entries was split in a one-dimensional - array per each protocol. - - struct net { - struct netns_nf nf; ... } - struct netns_nf { - struct nf_hook_entries * hooks_ipv4[NF_INET_NUMHOOKS]; - struct nf_hook_entries * hooks_ipv6[NF_INET_NUMHOOKS]; - struct nf_hook_entries * hooks_arp[NF_ARP_NUMHOOKS]; - struct nf_hook_entries * hooks_bridge[NF_INET_NUMHOOKS]; - struct nf_hook_entries * hooks_decnet[NF_DN_NUMHOOKS]; ... } - struct nf_hook_entries { - u16 num_hook_entries; /* plus padding */ - struct nf_hook_entry hooks[]; - //const struct nf_hook_ops *orig_ops[]; } - struct nf_hook_entry { - nf_hookfn *hook; - void *priv; } - - (*) Even though the hook address is in the struct nf_hook_entry, we use the original - nf_hook_ops hook address value, the one which was filled by the user, to make it - uniform to all the implementations. - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - return ( - vmlinux.has_symbol("net_namespace_list") - and vmlinux.has_type("netns_nf") - and vmlinux.get_type("netns_nf").has_member("hooks_ipv4") - ) - - def get_hooks_container(self, net, proto_name, hook_name): - try: - if proto_name == "IPV4": - net_nf_hooks = net.nf.hooks_ipv4 - elif proto_name == "ARP": - net_nf_hooks = net.nf.hooks_arp - elif proto_name == "BRIDGE": - net_nf_hooks = net.nf.hooks_bridge - elif proto_name == "IPV6": - net_nf_hooks = net.nf.hooks_ipv6 - elif proto_name == "DECNET": - net_nf_hooks = net.nf.hooks_decnet - else: - return - - yield net_nf_hooks - - except AttributeError: - # Protocol family disabled at kernel compilation - # CONFIG_NETFILTER_FAMILY_ARP=n || - # CONFIG_NETFILTER_FAMILY_BRIDGE=n || - # CONFIG_DECNET=n - pass - - def _get_nf_hook_entries_ptr(self, nf_hooks_addr, proto_idx, hook_idx): - nf_hook_entries_ptr = nf_hooks_addr[hook_idx] - return nf_hook_entries_ptr - - def get_nf_hook_entries(self, nf_hooks_addr, proto_idx, hook_idx): - return nf_hooks_addr[hook_idx] - - -class AbstractNetfilterNetDev(AbstractNetfilter): - """Base class to handle the Netfilter NetDev hooks. - It won't be executed. It has some common functions to all Netfilter NetDev hook - implementations. - - Netfilter NetDev hooks are set per network device which belongs to a network - namespace. - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - return False - - def subscribed_protocols(self): - return ("NETDEV",) - - def get_hooks_container(self, net, proto_name, hook_name): - net_device_type = self.vmlinux.get_type("net_device") - net_device_name = self.get_symbol_fullname("net_device") - for net_device in net.dev_base_head.to_list(net_device_name, "dev_list"): - if hook_name == "INGRESS": - if net_device_type.has_member("nf_hooks_ingress"): - # CONFIG_NETFILTER_INGRESS=y - yield net_device.nf_hooks_ingress - - elif hook_name == "EGRESS": - if net_device_type.has_member("nf_hooks_egress"): - # CONFIG_NETFILTER_EGRESS=y - yield net_device.nf_hooks_egress - - -class NetfilterNetDevImp_4_2_to_4_9(AbstractNetfilterNetDev): - """This is the first version of Netfilter Ingress hooks which was implemented using - a doubly-linked list of 'nf_hook_ops'. - struct list_head nf_hooks_ingress; - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - hooks_type = ["list_head"] - return ( - vmlinux.has_symbol("net_namespace_list") - and vmlinux.has_type("net_device") - and vmlinux.get_type("net_device").has_member("nf_hooks_ingress") - and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress") - == hooks_type - ) - - def get_hook_ops(self, hook_container, proto_idx, hook_idx): - nf_hooks_ingress = hook_container - nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops") - return nf_hooks_ingress.to_list(nf_hook_ops_name, "list") - - -class NetfilterNetDevImp_4_9_to_4_14(AbstractNetfilterNetDev): - """In 4.9 it was changed to a simple singly-linked list. - struct nf_hook_entry * nf_hooks_ingress; - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - hooks_type = ["pointer", "nf_hook_entry"] - return ( - vmlinux.has_symbol("net_namespace_list") - and vmlinux.has_type("net_device") - and vmlinux.get_type("net_device").has_member("nf_hooks_ingress") - and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress") - == hooks_type - ) - - def get_hook_ops(self, hook_container, proto_idx, hook_idx): - nf_hooks_ingress_ptr = hook_container - if not nf_hooks_ingress_ptr: - return - - while nf_hooks_ingress_ptr: - nf_hook_entry = nf_hooks_ingress_ptr.dereference() - orig_ops = nf_hook_entry.orig_ops.dereference() - yield orig_ops - nf_hooks_ingress_ptr = nf_hooks_ingress_ptr.next - - -class NetfilterNetDevImp_4_14_to_latest(AbstractNetfilterNetDev): - """In 4.14 the hook list was converted to an array of pointers inside the struct - 'nf_hook_entries': - struct nf_hook_entries * nf_hooks_ingress; - struct nf_hook_entries { - u16 num_hook_entries; - struct nf_hook_entry hooks[]; - //const struct nf_hook_ops *orig_ops[]; } - """ - - @classmethod - def symtab_checks(cls, vmlinux) -> bool: - hooks_type = ["pointer", "nf_hook_entries"] - return ( - vmlinux.has_symbol("net_namespace_list") - and vmlinux.has_type("net_device") - and vmlinux.get_type("net_device").has_member("nf_hooks_ingress") - and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress") - == hooks_type - ) - - def get_hook_ops(self, hook_container, proto_idx, hook_idx): - nf_hook_entries = hook_container - if not nf_hook_entries: - return - - nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops") - nf_hook_ops_ptr_arr = self.build_nf_hook_ops_array(nf_hook_entries) - if not nf_hook_ops_ptr_arr: - return - - for nf_hook_ops_ptr in nf_hook_ops_ptr_arr: - nf_hook_ops = nf_hook_ops_ptr.dereference().cast(nf_hook_ops_name) - yield nf_hook_ops - - -class Netfilter(interfaces.plugins.PluginInterface): - """Lists Netfilter hooks.""" - - _required_framework_version = (2, 22, 0) +class Netfilter( + interfaces.plugins.PluginInterface, + deprecation.PluginRenameClass, + replacement_class=netfilter.Netfilter, + removal_date="2026-06-07", +): + """Lists Netfilter hooks (deprecated).""" _version = (2, 0, 0) - - _required_linux_utilities_modules_version = (3, 0, 0) - _required_linuxutils_gatherers_version = (1, 0, 0) - _required_linuxnet_version = (1, 0, 0) - - @classmethod - def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: - return [ - requirements.ModuleRequirement( - name="kernel", - description="Linux kernel", - architectures=["Intel32", "Intel64"], - ), - requirements.VersionRequirement( - name="linux_utilities_module_gatherers", - component=linux_utilities_modules.ModuleGatherers, - version=cls._required_linuxutils_gatherers_version, - ), - requirements.VersionRequirement( - name="linuxnet", - component=network.NetSymbols, - version=cls._required_linuxnet_version, - ), - ] - - def _format_fields(self, fields): - ( - netns, - proto_name, - hook_name, - priority, - hook_func, - module_info, - symbol_name, - hooked, - ) = fields - - if module_info: - module_name = module_info.name - else: - module_name = renderers.NotAvailableValue() - - return ( - netns, - proto_name, - hook_name, - priority, - format_hints.Hex(hook_func), - module_name, - symbol_name or renderers.NotAvailableValue(), - str(hooked), - ) - - def _generator(self): - kernel_module_name = self.config["kernel"] - for fields in AbstractNetfilter.run_all( - context=self.context, kernel_module_name=kernel_module_name - ): - yield (0, self._format_fields(fields)) - - def run(self): - headers = [ - ("Net NS", int), - ("Proto", str), - ("Hook", str), - ("Priority", int), - ("Handler", format_hints.Hex), - ("Module", str), - ("Symbol", str), - ("Is Hooked", str), - ] - return renderers.TreeGrid(headers, self._generator()) + _required_framework_version = (2, 22, 0) diff --git a/volatility3/framework/plugins/linux/tty_check.py b/volatility3/framework/plugins/linux/tty_check.py index 7d30b84ee..36bfb1b5a 100644 --- a/volatility3/framework/plugins/linux/tty_check.py +++ b/volatility3/framework/plugins/linux/tty_check.py @@ -1,118 +1,20 @@ -# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0 +# 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 # - import logging -from typing import List - -import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules -from volatility3.framework import interfaces, renderers, exceptions, constants -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.framework.symbols import linux +from volatility3.framework import interfaces, deprecation +from volatility3.plugins.linux.malware import tty_check as ttycheck vollog = logging.getLogger(__name__) -class tty_check(plugins.PluginInterface): - """Checks tty devices for hooks""" +class tty_check( + interfaces.plugins.PluginInterface, + deprecation.PluginRenameClass, + replacement_class=ttycheck.Tty_Check, + removal_date="2026-06-07", +): + """Checks tty devices for hooks (deprecated).""" _required_framework_version = (2, 0, 0) - - @classmethod - def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: - return [ - requirements.ModuleRequirement( - name="kernel", - description="Linux kernel", - architectures=["Intel32", "Intel64"], - ), - requirements.VersionRequirement( - name="linux_utilities_modules", - component=linux_utilities_modules.Modules, - version=(3, 0, 0), - ), - requirements.VersionRequirement( - name="linux_utilities_module_gatherers", - component=linux_utilities_modules.ModuleGatherers, - version=(1, 0, 0), - ), - requirements.VersionRequirement( - name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0) - ), - ] - - def _generator(self): - vmlinux = self.context.modules[self.config["kernel"]] - - try: - tty_drivers = vmlinux.object_from_symbol("tty_drivers").cast("list_head") - except exceptions.SymbolError: - tty_drivers = None - - if not tty_drivers: - raise TypeError( - "This plugin requires the tty_drivers 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." - ) - - known_modules = linux_utilities_modules.Modules.run_modules_scanners( - context=self.context, - kernel_module_name=self.config["kernel"], - caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier, - ) - - for tty in tty_drivers.to_list( - vmlinux.symbol_table_name + constants.BANG + "tty_driver", "tty_drivers" - ): - try: - ttys = utility.array_of_pointers( - tty.ttys.dereference(), - count=tty.num, - subtype=vmlinux.symbol_table_name + constants.BANG + "tty_struct", - context=self.context, - ) - except exceptions.PagedInvalidAddressException: - continue - - for tty_dev in ttys: - if tty_dev == 0: - continue - - try: - name = utility.array_to_string(tty_dev.name) - recv_buf = tty_dev.ldisc.ops.receive_buf - except exceptions.InvalidAddressException: - continue - - module_info, symbol_name = ( - linux_utilities_modules.Modules.module_lookup_by_address( - self.context, vmlinux.name, known_modules, recv_buf - ) - ) - - if module_info: - module_name = module_info.name - else: - module_name = renderers.NotAvailableValue() - - yield 0, ( - name, - format_hints.Hex(recv_buf), - module_name, - symbol_name or renderers.NotAvailableValue(), - ) - - def run(self): - return renderers.TreeGrid( - [ - ("Name", str), - ("Address", format_hints.Hex), - ("Module", str), - ("Symbol", str), - ], - self._generator(), - ) + _version = (1, 0, 0)