remove self-contained hidden_modules check, switch to dataclass

This commit is contained in:
Abyss Watcher
2025-01-22 00:52:41 +01:00
parent d297693876
commit 3ba60d55f7
@@ -5,8 +5,10 @@
# Public researches: https://i.blackhat.com/USA21/Wednesday-Handouts/us-21-Fixing-A-Memory-Forensics-Blind-Spot-Linux-Kernel-Tracing-wp.pdf
import logging
from typing import List, Iterable, Optional, Tuple, Set
from typing import Dict, List, Iterable, Optional
from enum import auto, IntFlag
from dataclasses import dataclass
from volatility3.plugins.linux import hidden_modules, modxview
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework.configuration import requirements
@@ -19,7 +21,7 @@ vollog = logging.getLogger(__name__)
# https://docs.python.org/3.13/library/enum.html#enum.IntFlag
class FTRACE_OPS_FLAGS(IntFlag):
class FtraceOpsFlags(IntFlag):
"""Denote the state of an ftrace_ops struct.
Based on https://elixir.bootlin.com/linux/v6.13-rc3/source/include/linux/ftrace.h#L255.
"""
@@ -45,16 +47,27 @@ class FTRACE_OPS_FLAGS(IntFlag):
FTRACE_OPS_FL_SUBOP = auto()
class Check_ftrace(interfaces.plugins.PluginInterface):
@dataclass
class ParsedFtraceOps:
"""Parsed ftrace_ops struct representation, containing a selection of forensics valuable
informations."""
ftrace_ops_offset: int
callback_symbol: str
callback_address: int
hooked_symbols: str
module_name: str
module_address: int
flags: str
class CheckFtrace(interfaces.plugins.PluginInterface):
"""Detect ftrace hooking"""
_version = (1, 0, 0)
_required_framework_version = (2, 19, 0)
additional_description = """Investigate the ftrace infrastructure to uncover kernel attached callbacks, which can be leveraged
to hook kernel functions and modify their behaviour."""
_hidden_modules_run = False
"""Flag to determine if the hidden_modules plugin was run,
in the context of this plugin."""
@staticmethod
def get_requirements() -> List[interfaces.configuration.RequirementInterface]:
@@ -85,11 +98,6 @@ class Check_ftrace(interfaces.plugins.PluginInterface):
),
]
@classmethod
def _set_hidden_modules_run(cls) -> None:
"""Use a self-contained setter, to prevent running hidden_modules multiple times."""
cls._hidden_modules_run = True
@staticmethod
def extract_hash_table_filters(
ftrace_ops: interfaces.objects.ObjectInterface,
@@ -124,43 +132,54 @@ class Check_ftrace(interfaces.plugins.PluginInterface):
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
known_modules: Set[extensions.module],
known_modules: Dict[str, List[extensions.module]],
ftrace_ops: interfaces.objects.ObjectInterface,
parse_flags: bool = False,
) -> Optional[Tuple]:
run_hidden_modules: bool = True,
) -> Optional[Iterable[ParsedFtraceOps]]:
"""Parse an ftrace_ops struct to highlight ftrace kernel hooking.
Iterates over embedded ftrace_func_entry entries, which point to hooked memory areas.
Args:
known_modules: A set of known modules to iterate over, used to locate callbacks origin
known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through modxview.run_modules_scanners().
ftrace_ops: The ftrace_ops struct to parse
parse_flags: Whether to parse ftrace_ops flags or not
run_hidden_modules: Whether to run the hidden_modules plugin or not. Note: it won't be run, even if specified, \
if the "hidden_modules" key is present in known_modules.
Yields:
A tuple containing a selection of useful fields (callback, hook, module) related to an ftrace_func_entry struct
An iterable of ParsedFtraceOps dataclasses, containing a selection of useful fields (callback, hook, module) related to an ftrace_ops struct
"""
kernel = context.modules[kernel_name]
callback = ftrace_ops.func
callback_symbol = module_address = module_name = None
# Try to lookup within the known modules if the callback address fits
module = modules_utilities.Modules.module_lookup_by_address(
context, kernel.layer_name, known_modules, callback
context,
kernel.layer_name,
modxview.Modxview.flatten_run_modules_results(known_modules),
callback,
)
# Run hidden_modules plugin if a callback origin couldn't be determined (only done once, results are re-used afterwards)
if module is None and not cls._hidden_modules_run:
if (
module is None
and run_hidden_modules
and "hidden_modules" not in known_modules
):
vollog.info(
"A callback module origin could not be determined. hidden_modules plugin will be run to detect additional modules.",
)
known_modules_addresses = set(
context.layers[kernel.layer_name].canonicalize(module.vol.offset)
for module in known_modules
for module in modxview.Modxview.flatten_run_modules_results(
known_modules
)
)
modules_memory_boundaries = (
hidden_modules.Hidden_modules.get_modules_memory_boundaries(
context, kernel_name
)
)
known_modules.update(
known_modules["hidden_modules"] = list(
hidden_modules.Hidden_modules.get_hidden_modules(
context,
kernel_name,
@@ -168,27 +187,24 @@ class Check_ftrace(interfaces.plugins.PluginInterface):
modules_memory_boundaries,
)
)
cls._set_hidden_modules_run()
# Lookup the updated list to see if hidden_modules was able
# to find the missing module
module = modules_utilities.Modules.module_lookup_by_address(
context, kernel.layer_name, known_modules, callback
context,
kernel.layer_name,
modxview.Modxview.flatten_run_modules_results(known_modules),
callback,
)
# Fetch more information about the module
if module:
module_address = format_hints.Hex(module.vol.offset)
module_name = module.get_name() or NotAvailableValue()
callback_symbol = (
module.get_symbol_by_address(callback) or NotAvailableValue()
)
if module is not None:
module_address = module.vol.offset
module_name = module.get_name()
callback_symbol = module.get_symbol_by_address(callback)
else:
vollog.warning(
f"Could not determine ftrace_ops@{ftrace_ops.vol.offset:#x} callback {callback:#x} module origin.",
)
module_address = NotAvailableValue()
module_name = NotAvailableValue()
callback_symbol = NotAvailableValue()
# Iterate over ftrace_func_entry list
for ftrace_func_entry in cls.extract_hash_table_filters(ftrace_ops):
@@ -199,23 +215,20 @@ class Check_ftrace(interfaces.plugins.PluginInterface):
hooked_symbols = ",".join(
[s.split(constants.BANG)[-1] for s in hooked_symbols]
)
parsed_entry = (
format_hints.Hex(ftrace_ops.vol.offset),
yield ParsedFtraceOps(
ftrace_ops.vol.offset,
callback_symbol,
format_hints.Hex(callback),
hooked_symbols or NotAvailableValue(),
callback,
hooked_symbols,
module_name,
module_address,
# FtraceOpsFlags(ftrace_ops.flags).name is valid in > Python3.10, but
# returns None <= Python 3.10. We need to manipulate it like so to ensure compatibility:
# FtraceOpsFlags.FTRACE_OPS_FL_IPMODIFY|FTRACE_OPS_FL_ALLOC_TRAMP
# -> FTRACE_OPS_FL_IPMODIFY,FTRACE_OPS_FL_ALLOC_TRAMP
str(FtraceOpsFlags(ftrace_ops.flags)).split(".")[-1].replace("|", ","),
)
if parse_flags:
# e.g. FTRACE_OPS_FL_ENABLED,FTRACE_OPS_FL_DYNAMIC
parsed_entry += (
FTRACE_OPS_FLAGS(ftrace_ops.flags).name.replace("|", ","),
)
return parsed_entry
return None
@staticmethod
@@ -252,23 +265,31 @@ class Check_ftrace(interfaces.plugins.PluginInterface):
)
# Do not run hidden_modules by default, but only on failure to find a module
known_modules = set(
modxview.Modxview.flatten_run_modules_results(
modxview.Modxview.run_modules_scanners(
self.context, kernel_name, run_hidden_modules=False
)
)
known_modules = modxview.Modxview.run_modules_scanners(
self.context, kernel_name, run_hidden_modules=False
)
for ftrace_ops in self.iterate_ftrace_ops_list(self.context, kernel_name):
ftrace_ops_parsed = self.parse_ftrace_ops(
for ftrace_ops_parsed in self.parse_ftrace_ops(
self.context,
kernel_name,
known_modules,
ftrace_ops,
self.config.get("show_ftrace_flags"),
)
if ftrace_ops_parsed is not None:
yield (0, (ftrace_ops_parsed))
):
formatted_results = (
format_hints.Hex(ftrace_ops_parsed.ftrace_ops_offset),
ftrace_ops_parsed.callback_symbol or NotAvailableValue(),
format_hints.Hex(ftrace_ops_parsed.callback_address),
ftrace_ops_parsed.hooked_symbols or NotAvailableValue(),
ftrace_ops_parsed.module_name or NotAvailableValue(),
(
format_hints.Hex(ftrace_ops_parsed.module_address)
if ftrace_ops_parsed.module_address is not None
else NotAvailableValue()
),
)
if self.config["show_ftrace_flags"]:
formatted_results += (ftrace_ops_parsed.flags,)
yield (0, formatted_results)
def run(self):
columns = [