Add plugins missing from original commit

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)],
+            (),
+        )
+    )

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())

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