Adjust run_modules_scanners to avoid special handling of hidden modules, adjust modxview to new API, add classes and version requirements on module gathering interface

This commit is contained in:
Andrew Case
2025-03-14 20:27:25 +00:00
parent 4606495ded
commit cf8bb3dfbc
8 changed files with 229 additions and 204 deletions
@@ -35,6 +35,11 @@ class Check_idt(interfaces.plugins.PluginInterface):
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)
),
@@ -81,7 +86,7 @@ class Check_idt(interfaces.plugins.PluginInterface):
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier,
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
idt_table_size = 256
@@ -31,6 +31,11 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
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)
),
@@ -58,7 +63,7 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier,
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
knl = vmlinux.object(
@@ -36,6 +36,11 @@ class Kthreads(plugins.PluginInterface):
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, 1, 0)
),
@@ -57,7 +62,7 @@ class Kthreads(plugins.PluginInterface):
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier,
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
for task in pslist.PsList.list_tasks(
+23 -20
View File
@@ -36,6 +36,11 @@ spot modules presence and taints."""
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="linux-tainting", component=tainting.Tainting, version=(1, 0, 0)
),
@@ -89,44 +94,42 @@ spot modules presence and taints."""
)
def _generator(self):
kernel_name = self.config["kernel"]
kernel = self.context.modules[self.config["kernel"]]
kernel = self.context.modules[kernel_name]
wanted_sources = [
linux_utilities_modules.Modules.source_lsmod_identifier,
linux_utilities_modules.Modules.source_sysfs_identifier,
linux_utilities_modules.Modules.source_hidden_identifier,
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_sources=wanted_sources,
caller_wanted_gatherers=wanted_gatherers,
flatten=False,
)
aggregated_modules = {}
# We want to be explicit on the plugins results we are interested in
for plugin_name in wanted_sources:
for gatherer in wanted_gatherers:
# Iterate over each recovered module
for mod_info in run_results[plugin_name]:
for mod_info in run_results[gatherer]:
# 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(plugin_name)
aggregated_modules[mod_info.offset].append(gatherer)
else:
aggregated_modules[mod_info.offset] = [plugin_name]
aggregated_modules[mod_info.offset] = [gatherer]
for module_offset, originating_plugins in aggregated_modules.items():
# Tainting parsing capabilities applied to the module
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,
kernel_name,
self.config["kernel"],
module.taints,
True,
)
@@ -134,7 +137,7 @@ spot modules presence and taints."""
taints = ",".join(
tainting.Tainting.get_taints_parsed(
self.context,
kernel_name,
self.config["kernel"],
module.taints,
True,
)
@@ -145,9 +148,9 @@ spot modules presence and taints."""
(
module.get_name() or NotAvailableValue(),
format_hints.Hex(module_offset),
"lsmod" in originating_plugins,
"check_modules" in originating_plugins,
"hidden_modules" in originating_plugins,
linux_utilities_modules.ModuleGathererLsmod in gatherers,
linux_utilities_modules.ModuleGathererSysFs in gatherers,
linux_utilities_modules.ModuleGathererScanner in gatherers,
taints or NotAvailableValue(),
),
)
@@ -158,7 +161,7 @@ spot modules presence and taints."""
("Address", format_hints.Hex),
("In procfs", bool),
("In sysfs", bool),
("Hidden", bool),
("In scan", bool),
("Taints", str),
]
@@ -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, Generator
from typing import List, Generator
from enum import Enum
from dataclasses import dataclass
@@ -65,7 +65,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 = (3, 0, 0)
_version = (4, 0, 0)
_required_framework_version = (2, 19, 0)
@classmethod
@@ -81,6 +81,11 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
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.BooleanRequirement(
name="show_ftrace_flags",
description="Show ftrace flags associated with an ftrace_ops struct",
@@ -127,9 +132,8 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
known_modules: Dict[str, List[linux_utilities_modules.Modules.ModuleInfo]],
known_modules: List[linux_utilities_modules.ModuleInfo],
ftrace_ops: interfaces.objects.ObjectInterface,
run_hidden_modules: bool = True,
) -> 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.
@@ -137,8 +141,6 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
Args:
known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through run_modules_scanners().
ftrace_ops: The ftrace_ops 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.
Yields:
An iterable of ParsedFtraceOps dataclasses, containing a selection of useful fields (callback, hook, module) related to an ftrace_ops struct
@@ -225,7 +227,7 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier,
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
for ftrace_ops in self.iterate_ftrace_ops_list(self.context, kernel_name):
@@ -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, Iterable, List, Optional
from typing import Iterable, List, Optional
from dataclasses import dataclass
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
@@ -38,7 +38,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
Investigate the tracepoints subsystem to uncover kernel attached probes, 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
@@ -54,6 +54,11 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
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),
),
]
@classmethod
@@ -96,7 +101,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
known_modules: Dict[str, List[linux_utilities_modules.Modules.ModuleInfo]],
known_modules: List[linux_utilities_modules.ModuleInfo],
tracepoint: interfaces.objects.ObjectInterface,
run_hidden_modules: bool = True,
) -> Optional[Iterable[ParsedTracepointFunc]]:
@@ -231,7 +236,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier,
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
tracepoints = self.iterate_tracepoints_array(self.context, kernel_name)
@@ -34,6 +34,11 @@ class tty_check(plugins.PluginInterface):
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)
),
@@ -57,7 +62,7 @@ class tty_check(plugins.PluginInterface):
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_sources=linux_utilities_modules.Modules.all_sources_identifier,
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
for tty in tty_drivers.to_list(
@@ -1,6 +1,18 @@
import logging
import warnings
from typing import Iterable, Iterator, List, Optional, Tuple, NamedTuple, Dict, Set
from typing import (
Iterable,
Iterator,
List,
Optional,
Tuple,
NamedTuple,
Dict,
Set,
Generator,
Union,
)
from abc import ABCMeta, abstractmethod
from volatility3 import framework
from volatility3.framework import (
@@ -10,12 +22,44 @@ from volatility3.framework import (
exceptions,
objects,
)
from volatility3.framework.objects import utility
from volatility3.framework.symbols.linux import extensions
vollog = logging.getLogger(__name__)
class ModuleInfo(NamedTuple):
"""
Used to track the name and boundary of a kernel module
"""
offset: int
name: str
start: int
end: int
class ModuleGathererInterface(
interfaces.configuration.VersionableInterface, metaclass=ABCMeta
):
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
gatherer_return_type = Generator[Union[ModuleInfo, "extensions.module"], None, None]
@classmethod
@abstractmethod
def gather_modules(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> gatherer_return_type:
"""
This method must return a generator (yield) of each `gatherer_return_type` found from its source
"""
class Modules(interfaces.configuration.VersionableInterface):
"""Kernel modules related utilities."""
@@ -24,31 +68,6 @@ class Modules(interfaces.configuration.VersionableInterface):
framework.require_interface_version(*_required_framework_version)
# Valid sources of kernel modules to send to `run_module_scanners`
source_kernel_identifier = "kernel"
source_lsmod_identifier = "lsmod"
source_sysfs_identifier = "check_modules"
source_hidden_identifier = "hidden_modules"
# With few exceptions, rootkit checking plugins want all sources
# This provides a stable identifier as new sources are added over time
all_sources_identifier = [
source_kernel_identifier,
source_lsmod_identifier,
source_sysfs_identifier,
source_hidden_identifier,
]
class ModuleInfo(NamedTuple):
"""
Used to track the name and boundary of a kernel module
"""
offset: int
name: str
start: int
end: int
@classmethod
def module_lookup_by_address(
cls,
@@ -204,138 +223,41 @@ class Modules(interfaces.configuration.VersionableInterface):
end = start + module.get_core_size()
return Modules.ModuleInfo(module.vol.offset, mod_name, start, end)
@staticmethod
def get_kernel_module_info(
context: interfaces.context.ContextInterface,
kernel_module_name: str,
) -> Iterator[ModuleInfo]:
"""
Returns a ModuleInfo instance that encodes the kernel
This is required to map function pointers to the kerenl executable
"""
kernel = context.modules[kernel_module_name]
address_mask = context.layers[kernel.layer_name].address_mask
start_addr = kernel.object_from_symbol("_text")
start_addr = start_addr.vol.offset & address_mask
end_addr = kernel.object_from_symbol("_etext")
end_addr = end_addr.vol.offset & address_mask
return [
Modules.ModuleInfo(
start_addr, constants.linux.KERNEL_NAME, start_addr, end_addr
)
]
return ModuleInfo(module.vol.offset, mod_name, start, end)
@classmethod
def _get_hidden_modules_results(
cls,
context: str,
kernel_module_name: str,
run_results: Dict[str, List[ModuleInfo]],
):
known_modules_addresses = set()
kernel = context.modules[kernel_module_name]
# Walk each sources' results
for results in run_results.values():
for modinfo in results:
address = context.layers[kernel.layer_name].canonicalize(modinfo.start)
known_modules_addresses.add(address)
modules_memory_boundaries = cls.get_modules_memory_boundaries(
context, kernel_module_name
)
hidden_results = []
address_mask = context.layers[kernel.layer_name].address_mask
for module in cls.get_hidden_modules(
context,
kernel_module_name,
known_modules_addresses,
modules_memory_boundaries,
):
modinfo = cls.get_module_info_for_module(address_mask, module)
if modinfo:
hidden_results.append(modinfo)
return hidden_results
@classmethod
def _get_list_modules(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> List[ModuleInfo]:
def _validate_gatherers(cls, caller_wanted_gatherers) -> List[str]:
"""
Gather `module` instances from lsmod
Called by `run_modules_scanners` to validate the caller supplied gatherers list
An exception is thrown if an empty gatherers list is given or a list containing an invalid source
"""
yield from cls.list_modules(context, kernel_module_name)
@classmethod
def _get_sysfs_modules(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> List[ModuleInfo]:
"""
Gather the `module` instances from sysfs
"""
kernel = context.modules[kernel_module_name]
sysfs_modules: dict = cls.get_kset_modules(context, kernel_module_name)
for m_offset in sysfs_modules.values():
yield kernel.object(object_type="module", offset=m_offset, absolute=True)
@classmethod
def _validated_sources(cls, caller_wanted_sources) -> List[str]:
"""
Called by `run_modules_scanners` to validate the caller supplied sources list
An exception is thrown if an empty source list is given or a list containing an invalid source
"""
if not caller_wanted_sources:
raise ValueError("`caller_wanted_sources` must have at least one source.")
if (
len(caller_wanted_sources) == 1
and caller_wanted_sources[0] == Modules.source_hidden_identifier
):
if not caller_wanted_gatherers:
raise ValueError(
f"{Modules.source_hidden_identifier} cannot be the only source or there is nothing to compare against."
"`caller_wanted_gatherers` must have at least one gatherer."
)
wanted_sources = []
for source in caller_wanted_sources:
if source not in Modules.all_sources_identifier:
for gatherer in caller_wanted_gatherers:
if gatherer not in ModuleGatherers.all_gatherers_identifier:
raise ValueError(
f"Invalid source sent through `caller_wanted_sources`: {source}"
f"Invalid gatherer sent through `caller_wanted_gatherers`: {gatherer}"
)
wanted_sources.append(source)
return wanted_sources
@classmethod
def run_modules_scanners(
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
caller_wanted_sources: List[str],
caller_wanted_gatherers: List[ModuleGathererInterface],
flatten: bool = True,
) -> Dict[str, List[ModuleInfo]]:
) -> Dict[ModuleGathererInterface, List[ModuleInfo]]:
"""Run module scanning plugins and aggregate the results. It is designed
to not operate any inter-plugin results triage.
Rules for `caller_wanted_sources`:
If `Modules.all_sources_identifier` is specified then every source will be populated
If `ModuleGathers.all_gathers_identifier` is specified then every source will be populated
If `Modules.source_hidden_identifier` is in the list, then at least one other sources must be
If `ModuleGathers.Scanner` is in the list, then at least one other sources must be
specified so a comparison will be populated
If empty or an invalid source is specified then a ValueError is thrown
@@ -346,52 +268,30 @@ class Modules(interfaces.configuration.VersionableInterface):
Returns:
Dictionary mapping each plugin to its corresponding result
"""
module_gatherers = {
Modules.source_kernel_identifier: cls.get_kernel_module_info,
Modules.source_lsmod_identifier: cls._get_list_modules,
Modules.source_sysfs_identifier: cls._get_sysfs_modules,
}
# Throws ValueError if invalid gatherers sent in
Modules._validate_gatherers(caller_wanted_gatherers)
kernel = context.modules[kernel_module_name]
address_mask = context.layers[kernel.layer_name].address_mask
wanted_sources = Modules._validated_sources(caller_wanted_sources)
run_results: Dict[ModuleGathererInterface, List[ModuleInfo]] = {}
run_results = {}
run_hidden_modules = False
# Special case hidden modules since it gathers modules on its own
if Modules.source_hidden_identifier in wanted_sources:
run_hidden_modules = True
wanted_sources.remove(Modules.source_hidden_identifier)
# Walk each source, gathering modules
for wanted_source in wanted_sources:
run_results[wanted_source] = []
gatherer = module_gatherers[wanted_source]
# Walk each source gathering modules
for gatherer in caller_wanted_gatherers:
run_results[gatherer] = []
# process each module coming from back the current source
for module in gatherer(context, kernel_module_name):
for module in gatherer.gather_modules(context, kernel_module_name):
# the kernel sends back a ModuleInfo directly
if wanted_source == Modules.source_kernel_identifier:
if gatherer == ModuleGathererKernel:
modinfo = module
else:
modinfo = cls.get_module_info_for_module(address_mask, module)
if modinfo:
run_results[wanted_source].append(modinfo)
# run hidden modules against the other sources
if run_hidden_modules:
run_results[Modules.source_hidden_identifier] = (
cls._get_hidden_modules_results(
context, kernel_module_name, run_results
)
)
run_results[gatherer].append(modinfo)
if flatten:
return cls.flatten_run_modules_results(run_results)
@@ -449,7 +349,7 @@ class Modules(interfaces.configuration.VersionableInterface):
Returns:
List of ModuleInfo objects
"""
uniq_modules: List[Modules.ModuleInfo] = []
uniq_modules: List[ModuleInfo] = []
seen_addresses: int = set()
@@ -645,3 +545,98 @@ class Modules(interfaces.configuration.VersionableInterface):
True if all the addresses meet the alignment
"""
return all(addr % address_alignment == 0 for addr in addresses)
class ModuleGathererLsmod(ModuleGathererInterface):
"""
Gathers modules from the main kernel list
"""
@classmethod
def gather_modules(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> ModuleGathererInterface.gatherer_return_type:
yield from Modules.list_modules(context, kernel_module_name)
class ModuleGathererSysFs(ModuleGathererInterface):
"""
Gathers modules from the sysfs /sys/modules objects
"""
@classmethod
def gather_modules(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> ModuleGathererInterface.gatherer_return_type:
kernel = context.modules[kernel_module_name]
sysfs_modules: dict = Modules.get_kset_modules(context, kernel_module_name)
for m_offset in sysfs_modules.values():
yield kernel.object(object_type="module", offset=m_offset, absolute=True)
class ModuleGathererScanner(ModuleGathererInterface):
"""
Gathers modules by scanning memory
"""
@classmethod
def gather_modules(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> ModuleGathererInterface.gatherer_return_type:
modules_memory_boundaries = Modules.get_modules_memory_boundaries(
context, kernel_module_name
)
# Send in an empty list to not filter on any modules
yield from Modules.get_hidden_modules(
context=context,
vmlinux_module_name=kernel_module_name,
known_module_addresses=[],
modules_memory_boundaries=modules_memory_boundaries,
)
class ModuleGathererKernel(ModuleGathererInterface):
"""
Creates a ModuleInfo instance for the kernel so that plugins
can determine when function pointers reference the kernel
"""
@classmethod
def gather_modules(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> ModuleGathererInterface.gatherer_return_type:
"""
Returns a ModuleInfo instance that encodes the kernel
This is required to map function pointers to the kerenl executable
"""
kernel = context.modules[kernel_module_name]
address_mask = context.layers[kernel.layer_name].address_mask
start_addr = kernel.object_from_symbol("_text")
start_addr = start_addr.vol.offset & address_mask
end_addr = kernel.object_from_symbol("_etext")
end_addr = end_addr.vol.offset & address_mask
yield ModuleInfo(start_addr, constants.linux.KERNEL_NAME, start_addr, end_addr)
class ModuleGatherers(interfaces.configuration.VersionableInterface):
_version = (1, 0, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
# Valid sources of cores kernel module gatherers to send to `run_module_scanners`
# With few exceptions, rootkit checking plugins want all sources
# This provides a stable identifier as new sources are added over time
all_gatherers_identifier = [
ModuleGathererLsmod,
ModuleGathererSysFs,
ModuleGathererScanner,
ModuleGathererKernel,
]