Merge pull request #1671 from volatilityfoundation/fix_tracing_plugins

Fix several bugs found in the tracing plugins during mass testing
This commit is contained in:
ikelos
2025-03-06 18:11:17 +00:00
committed by GitHub
2 changed files with 35 additions and 22 deletions
@@ -5,7 +5,7 @@
# 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 Dict, List, Iterable, Optional
from typing import Dict, List, Generator
from enum import Enum
from dataclasses import dataclass
@@ -67,7 +67,7 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
Investigate the ftrace infrastructure to uncover kernel attached callbacks, which can be leveraged
to hook kernel functions and modify their behaviour."""
_version = (1, 0, 0)
_version = (2, 0, 0)
_required_framework_version = (2, 19, 0)
@classmethod
@@ -103,32 +103,35 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
def extract_hash_table_filters(
cls,
ftrace_ops: interfaces.objects.ObjectInterface,
) -> Optional[Iterable[interfaces.objects.ObjectInterface]]:
) -> Generator[interfaces.objects.ObjectInterface, None, None]:
"""Wrap the process of walking to every ftrace_func_entry of an ftrace_ops.
Those are stored in a hash table of filters that indicates the addresses hooked.
Args:
ftrace_ops: The ftrace_ops struct to walk through
Returns:
Return, None, None:
An iterable of ftrace_func_entry structs
"""
if hasattr(ftrace_ops, "func_hash"):
ftrace_hash = ftrace_ops.func_hash.filter_hash
else:
ftrace_hash = ftrace_ops.filter_hash
try:
current_bucket_ptr = ftrace_ops.func_hash.filter_hash.buckets.first
current_bucket_ptr = ftrace_hash.buckets.first
except exceptions.InvalidAddressException:
vollog.log(
constants.LOGLEVEL_VV,
f"ftrace_func_entry list of ftrace_ops@{ftrace_ops.vol.offset:#x} is empty/invalid. Skipping it...",
)
return []
return
while current_bucket_ptr.is_readable():
yield current_bucket_ptr.dereference().cast("ftrace_func_entry")
current_bucket_ptr = current_bucket_ptr.next
return None
@classmethod
def parse_ftrace_ops(
cls,
@@ -137,7 +140,7 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
known_modules: Dict[str, List[extensions.module]],
ftrace_ops: interfaces.objects.ObjectInterface,
run_hidden_modules: bool = True,
) -> Optional[Iterable[ParsedFtraceOps]]:
) -> Generator[ParsedFtraceOps, None, None]:
"""Parse an ftrace_ops struct to highlight ftrace kernel hooking.
Iterates over embedded ftrace_func_entry entries, which point to hooked memory areas.
@@ -234,12 +237,10 @@ if the "hidden_modules" key is present in known_modules.
formatted_ftrace_flags,
)
return None
@classmethod
def iterate_ftrace_ops_list(
cls, context: interfaces.context.ContextInterface, kernel_name: str
) -> Optional[Iterable[interfaces.objects.ObjectInterface]]:
) -> Generator[interfaces.objects.ObjectInterface, None, None]:
"""Iterate over (ftrace_ops *)ftrace_ops_list.
Returns:
@@ -116,18 +116,25 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through modxview.run_modules_scanners().
tracepoint: The tracepoint struct to parse
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.
if the "hidden_modules" key is present in known_modules.
Yields:
An iterable of ParsedTracepointFunc dataclasses, containing a selection of useful fields related to a tracepoint struct
"""
kernel = context.modules[kernel_name]
kernel_layer = context.layers[kernel.layer_name]
for tracepoint_func in cls.iterate_tracepoint_funcs(
context, kernel_layer.name, tracepoint
):
try:
tracepoint_name = utility.pointer_to_string(tracepoint.name, count=512)
except exceptions.InvalidAddressException:
vollog.debug(
f"Tracepoint function at {tracepoint.vol.offset:#x} is smeared."
)
continue
probe_handler_address = tracepoint_func.func
probe_handler_symbol = module_address = module_name = None
@@ -183,16 +190,21 @@ if the "hidden_modules" key is present in known_modules.
probe_handler_address
)
else:
vollog.warning(
vollog.debug(
f"Could not determine tracepoint@{tracepoint.vol.offset:#x} probe handler {probe_handler_address:#x} module origin.",
)
if hasattr(tracepoint_func, "prio"):
prio = tracepoint_func.prio
else:
prio = None
yield ParsedTracepointFunc(
utility.pointer_to_string(tracepoint.name, count=512),
tracepoint_name,
tracepoint.vol.offset,
probe_handler_symbol,
probe_handler_address,
tracepoint_func.prio,
prio,
module_name,
module_address,
)
@@ -258,11 +270,11 @@ if the "hidden_modules" key is present in known_modules.
kernel_layer = self.context.layers[kernel.layer_name]
if not kernel.has_symbol("__start___tracepoints_ptrs"):
raise exceptions.SymbolError(
"__start___tracepoints_ptrs",
self.vmlinux.symbol_table_name,
'The provided symbol table does not include the "__start___tracepoints_ptrs" symbol. This means you are either analyzing an unsupported kernel version or that your symbol table is corrupted.',
vollog.error(
'The provided symbol table does not include the "__start___tracepoints_ptrs" symbol.'
"This means you are either analyzing an unsupported kernel version or that your symbol table is corrupted."
)
return
known_modules = modxview.Modxview.run_modules_scanners(
self.context, kernel_name, run_hidden_modules=False
@@ -281,7 +293,7 @@ if the "hidden_modules" key is present in known_modules.
format_hints.Hex(tracepoint_parsed.tracepoint_address),
tracepoint_parsed.probe_name or NotAvailableValue(),
format_hints.Hex(tracepoint_parsed.probe_address),
tracepoint_parsed.probe_priority,
tracepoint_parsed.probe_priority or NotAvailableValue(),
tracepoint_parsed.module_name or NotAvailableValue(),
(
format_hints.Hex(tracepoint_parsed.module_address)