WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-09-12 12:47:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1673 from volatilityfoundation/linux_kernel_api_overhaul

Browse Source 
Merge two Linux APIs for mapping kernel modules to pointers. Fix bugs…
This commit is contained in:
ikelos
2025-03-10 19:36:56 +00:00
committed by GitHub
parent d3799022ef 9e4fdbf8ef
commit 1c5fdccf1e
15 changed files with 737 additions and 473 deletions
Download Patch File Download Diff File Expand all files Collapse all files
volatility3/framework/__init__.py
+2 -60
View File
@@ -11,12 +11,9 @@ import inspect
import logging
import os
import traceback
import functools
import warnings
from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, Type, TypeVar
from typing import Any, Dict, Generator, List, Optional, Tuple, Type, TypeVar
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework import constants, interfaces
if (
sys.version_info.major != constants.REQUIRED_PYTHON_VERSION[0]
@@ -66,61 +63,6 @@ def require_interface_version(*args) -> None:
)
class Deprecation:
"""Deprecation related methods."""
@staticmethod
def deprecated_method(
replacement: Callable,
replacement_version: Tuple[int, int, int] = None,
additional_information: str = "",
):
"""A decorator for marking functions as deprecated.
Args:
replacement: The replacement function overriding the deprecated API, in the form of a Callable (typically a method)
replacement_version: The "replacement" base class version that the deprecated method expects before proxying to it. This implies that "replacement" is a method from a class that inherits from VersionableInterface.
additional_information: Information appended at the end of the deprecation message
"""
def decorator(deprecated_func):
@functools.wraps(deprecated_func)
def wrapper(*args, **kwargs):
nonlocal replacement, replacement_version, additional_information
# Prevent version mismatches between deprecated (proxy) methods and the ones they proxy
if (
replacement_version is not None
and callable(replacement)
and hasattr(replacement, "__self__")
):
replacement_base_class = replacement.__self__
# Verify that the base class inherits from VersionableInterface
if inspect.isclass(replacement_base_class) and issubclass(
replacement_base_class,
interfaces.configuration.VersionableInterface,
):
# SemVer check
if not requirements.VersionRequirement.matches_required(
replacement_version, replacement_base_class.version
):
raise exceptions.VersionMismatchException(
deprecated_func,
replacement_base_class,
replacement_version,
"This is a bug, the deprecated call needs to be removed and the caller needs to update their code to use the new method.",
)
deprecation_msg = f"Method \"{deprecated_func.__module__ + '.' + deprecated_func.__qualname__}\" is deprecated, use \"{replacement.__module__ + '.' + replacement.__qualname__}\" instead. {additional_information}"
warnings.warn(deprecation_msg, FutureWarning)
# Return the wrapped function with its original arguments
return deprecated_func(*args, **kwargs)
return wrapper
return decorator
class NonInheritable:
def __init__(self, value: Any, cls: Type) -> None:
self.default_value = value
volatility3/framework/deprecation.py
+90
View File
@@ -0,0 +1,90 @@
# 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
#
# This file contains the Deprecation class used to deprecate methods in an orderly manner
import warnings
import functools
import inspect
from typing import Callable, Tuple
from volatility3.framework import interfaces, exceptions
from volatility3.framework.configuration import requirements
def method_being_removed(message: str, removal_date: str):
"""A decorator for marking functions as being removed in the future and without a replacement.
Callers to this function should explicitly list the API paths that should be used instead.
Args:
message: A message added to the standard deprecation warning. Should include the replacement API paths
removal_date: A YYYY-MM-DD formatted date of when the function will be removed from the framework
"""
def decorator(deprecated_func):
@functools.wraps(deprecated_func)
def wrapper(*args, **kwargs):
warnings.warn(
f"This API ({deprecated_func.__module__}.{deprecated_func.__qualname__}) will be removed in the first release after {removal_date}. {message}",
FutureWarning,
)
return deprecated_func(*args, **kwargs)
return wrapper
return decorator
def deprecated_method(
replacement: Callable,
removal_date: str,
replacement_version: Tuple[int, int, int] = None,
additional_information: str = "",
):
"""A decorator for marking functions as deprecated.
Args:
replacement: The replacement function overriding the deprecated API, in the form of a Callable (typically a method)
removal_date: A YYYY-MM-DD formatted date of when the function will be removed from the framework
replacement_version: The "replacement" base class version that the deprecated method expects before proxying to it. This implies that "replacement" is a method from a class that inherits from VersionableInterface.
additional_information: Information appended at the end of the deprecation message
"""
def decorator(deprecated_func):
@functools.wraps(deprecated_func)
def wrapper(*args, **kwargs):
nonlocal replacement, replacement_version, additional_information
# Prevent version mismatches between deprecated (proxy) methods and the ones they proxy
if (
replacement_version is not None
and callable(replacement)
and hasattr(replacement, "__self__")
):
replacement_base_class = replacement.__self__
# Verify that the base class inherits from VersionableInterface
if inspect.isclass(replacement_base_class) and issubclass(
replacement_base_class,
interfaces.configuration.VersionableInterface,
):
# SemVer check
if not requirements.VersionRequirement.matches_required(
replacement_version, replacement_base_class.version
):
raise exceptions.VersionMismatchException(
deprecated_func,
replacement_base_class,
replacement_version,
"This is a bug, the deprecated call needs to be removed and the caller needs to update their code to use the new method.",
)
deprecation_msg = f"Method \"{deprecated_func.__module__ + '.' + deprecated_func.__qualname__}\" is deprecated and will be removed in the first release after {removal_date}, use \"{replacement.__module__ + '.' + replacement.__qualname__}\" instead. {additional_information}"
warnings.warn(deprecation_msg, FutureWarning)
# Return the wrapped function with its original arguments
return deprecated_func(*args, **kwargs)
return wrapper
return decorator
volatility3/framework/plugins/linux/check_idt.py
+1 -1
View File
@@ -34,7 +34,7 @@ class Check_idt(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 0, 0),
version=(2, 0, 0),
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 0, 0)
volatility3/framework/plugins/linux/check_modules.py
+22 -45
View File
@@ -3,14 +3,15 @@
#
import logging
from typing import List
from typing import List, Dict
from volatility3.framework import interfaces, renderers, exceptions, constants
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, renderers, deprecation
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.plugins.linux import lsmod
from volatility3.framework.symbols.linux import extensions
vollog = logging.getLogger(__name__)
@@ -18,7 +19,7 @@ vollog = logging.getLogger(__name__)
class Check_modules(plugins.PluginInterface):
"""Compares module list to sysfs info, if available"""
_version = (1, 0, 0)
_version = (2, 0, 0)
_required_framework_version = (2, 0, 0)
@classmethod
@@ -29,58 +30,34 @@ class Check_modules(plugins.PluginInterface):
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
),
]
@classmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.get_kset_modules,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
)
def get_kset_modules(
cls, context: interfaces.context.ContextInterface, vmlinux_name: str
):
vmlinux = context.modules[vmlinux_name]
try:
module_kset = vmlinux.object_from_symbol("module_kset")
except exceptions.SymbolError:
module_kset = None
if not module_kset:
raise TypeError(
"This plugin requires the module_kset 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."
)
ret = {}
kobj_off = vmlinux.get_type("module_kobject").relative_child_offset("kobj")
for kobj in module_kset.list.to_list(
vmlinux.symbol_table_name + constants.BANG + "kobject", "entry"
):
mod_kobj = vmlinux.object(
object_type="module_kobject",
offset=kobj.vol.offset - kobj_off,
absolute=True,
)
mod = mod_kobj.mod
try:
name = utility.pointer_to_string(kobj.name, 32)
except exceptions.InvalidAddressException:
continue
if kobj.name and kobj.reference_count() > 2:
ret[name] = mod
return ret
) -> Dict[str, extensions.module]:
return linux_utilities_modules.Modules.get_kset_modules(context, vmlinux_name)
def _generator(self):
kset_modules = self.get_kset_modules(self.context, self.config["kernel"])
kset_modules = linux_utilities_modules.Modules.get_kset_modules(
self.context, self.config["kernel"]
)
lsmod_modules = set(
str(utility.array_to_string(modules.name))
for modules in lsmod.Lsmod.list_modules(self.context, self.config["kernel"])
for modules in linux_utilities_modules.Modules.list_modules(
self.context, self.config["kernel"]
)
)
for mod_name in set(kset_modules.keys()).difference(lsmod_modules):
volatility3/framework/plugins/linux/hidden_modules.py
+61 -103
View File
@@ -3,7 +3,10 @@
#
import logging
from typing import List, Set, Tuple, Iterable
from volatility3.framework import renderers, interfaces, exceptions, objects
from volatility3.framework.symbols.linux.utilities import (
modules as linux_utilities_modules,
)
from volatility3.framework import renderers, interfaces, exceptions, deprecation
from volatility3.framework.constants import architectures
from volatility3.framework.renderers import format_hints
from volatility3.framework.configuration import requirements
@@ -16,7 +19,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
"""Carves memory to find hidden kernel modules"""
_required_framework_version = (2, 10, 0)
_version = (1, 0, 1)
_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
@@ -29,46 +32,32 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
),
]
@classmethod
@staticmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.get_modules_memory_boundaries,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
)
def get_modules_memory_boundaries(
cls,
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"):
# Kernel >= 5.19 58d208de3e8d87dbe196caf0b57cc58c7a3836ca
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"):
# 2.6.27 <= kernel < 5.19 3a642e99babe0617febb6f402e1e063479f489db
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):
raise exceptions.VolatilityException(
"Your ISF symbols lack type information. You may need to 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
) -> Tuple[int, int]:
return linux_utilities_modules.Modules.get_modules_memory_boundaries(
context, vmlinux_module_name
)
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.get_module_address_alignment,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
)
@classmethod
def _get_module_address_alignment(
cls,
@@ -88,27 +77,15 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
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)
return linux_utilities_modules.get_module_address_alignment(
context, vmlinux_module_name
)
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.get_hidden_modules,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
)
@classmethod
def get_hidden_modules(
cls,
@@ -139,54 +116,32 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
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
return linux_utilities_modules.get_hidden_modules(
vmlinux_module_name, known_module_addresses, modules_memory_boundaries
)
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").child_template("mod")
mod_size = mod_member_template.size
mod_member_data_format = mod_member_template.data_format
@staticmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.validate_alignment_patterns,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
)
def _validate_alignment_patterns(
addresses: Iterable[int],
address_alignment: int,
) -> bool:
"""Check if the memory addresses meet our alignments patterns
for module_addr in range(
module_addr_min, module_addr_max, module_address_alignment
):
if module_addr in known_module_addresses:
continue
Args:
addresses: Iterable with the address values
address_alignment: Number of bytes for alignment validation
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
Returns:
True if all the addresses meet the alignment
"""
return linux_utilities_modules.validate_alignment_patterns(
addresses, address_alignment
)
@classmethod
def get_lsmod_module_addresses(
@@ -217,10 +172,13 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
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
modules_memory_boundaries = (
linux_utilities_modules.Modules.get_modules_memory_boundaries(
self.context, vmlinux_module_name
)
)
for module in self.get_hidden_modules(
for module in linux_utilities_modules.Modules.get_hidden_modules(
self.context,
vmlinux_module_name,
known_module_addresses,
volatility3/framework/plugins/linux/keyboard_notifiers.py
+1 -1
View File
@@ -30,7 +30,7 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 0, 0),
version=(2, 0, 0),
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
volatility3/framework/plugins/linux/kthreads.py
+1 -1
View File
@@ -34,7 +34,7 @@ class Kthreads(plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 0, 0),
version=(2, 0, 0),
),
requirements.VersionRequirement(
name="linuxutils", component=linux.LinuxUtilities, version=(2, 1, 0)
volatility3/framework/plugins/linux/lsmod.py
+19 -22
View File
@@ -6,7 +6,8 @@
import logging
from typing import List, Iterable
from volatility3.framework import exceptions, renderers, constants, interfaces
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import exceptions, renderers, interfaces, deprecation
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
@@ -29,35 +30,31 @@ class Lsmod(plugins.PluginInterface):
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
),
]
@classmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.list_modules,
replacement_version=(2, 0, 0),
removal_date="2025-09-25",
)
def list_modules(
cls, context: interfaces.context.ContextInterface, vmlinux_module_name: str
) -> Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the modules in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
vmlinux_symbols: The name of the table containing the kernel symbols
Yields:
The modules present in the `layer_name` layer's modules list
This function will throw a SymbolError exception if kernel module support is not enabled.
"""
vmlinux = context.modules[vmlinux_module_name]
modules = vmlinux.object_from_symbol(symbol_name="modules").cast("list_head")
table_name = modules.vol.type_name.split(constants.BANG)[0]
yield from modules.to_list(table_name + constants.BANG + "module", "list")
return linux_utilities_modules.Modules.list_modules(
context, vmlinux_module_name
)
def _generator(self):
try:
for module in self.list_modules(self.context, self.config["kernel"]):
for module in linux_utilities_modules.Modules.list_modules(
self.context, self.config["kernel"]
):
mod_size = module.get_init_size() + module.get_core_size()
mod_name = utility.array_to_string(module.name)
@@ -65,7 +62,7 @@ class Lsmod(plugins.PluginInterface):
yield 0, (format_hints.Hex(module.vol.offset), mod_name, mod_size)
except exceptions.SymbolError:
vollog.debug(
vollog.warning(
"The required symbol 'module' is not present in symbol table. Please check that kernel modules are enabled for the system under analysis."
)
volatility3/framework/plugins/linux/modxview.py
+39 -69
View File
@@ -3,8 +3,10 @@
#
import logging
from typing import List, Dict, Iterator
from volatility3.plugins.linux import lsmod, check_modules, hidden_modules
from volatility3.framework import interfaces
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, deprecation
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue
from volatility3.framework.symbols.linux import extensions
@@ -29,22 +31,14 @@ spot modules presence and taints."""
description="Linux kernel",
architectures=architectures.LINUX_ARCHS,
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
),
requirements.VersionRequirement(
name="linux-tainting", component=tainting.Tainting, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
),
requirements.PluginRequirement(
name="check_modules",
plugin=check_modules.Check_modules,
version=(1, 0, 0),
),
requirements.PluginRequirement(
name="hidden_modules",
plugin=hidden_modules.Hidden_modules,
version=(1, 0, 0),
),
requirements.BooleanRequirement(
name="plain_taints",
description="Display the plain taints string for each module.",
@@ -54,6 +48,11 @@ spot modules presence and taints."""
]
@classmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.flatten_run_modules_results,
replacement_version=(2, 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]:
@@ -67,15 +66,16 @@ spot modules presence and taints."""
Returns:
Iterator of modules objects
"""
seen_addresses = set()
for modules in run_results.values():
for module in modules:
if deduplicate and module.vol.offset in seen_addresses:
continue
seen_addresses.add(module.vol.offset)
yield module
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=(2, 0, 0),
removal_date="2025-09-25",
)
def run_modules_scanners(
cls,
context: interfaces.context.ContextInterface,
@@ -83,67 +83,37 @@ spot modules presence and taints."""
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.
Args:
run_hidden_modules: specify if the hidden_modules plugin should be run
Returns:
Dictionary mapping each plugin to its corresponding result
"""
kernel = context.modules[kernel_name]
run_results = {}
# lsmod
run_results["lsmod"] = list(lsmod.Lsmod.list_modules(context, kernel_name))
# check_modules
sysfs_modules: dict = check_modules.Check_modules.get_kset_modules(
context, kernel_name
to not operate any inter-plugin results triage."""
return linux_utilities_modules.Modules.run_modules_scanners(
context, kernel_name, run_hidden_modules
)
## Convert get_kset_modules() offsets back to module objects
run_results["check_modules"] = [
kernel.object(object_type="module", offset=m_offset, absolute=True)
for m_offset in sysfs_modules.values()
]
# hidden_modules
if run_hidden_modules:
known_modules_addresses = set(
context.layers[kernel.layer_name].canonicalize(module.vol.offset)
for module in run_results["lsmod"] + run_results["check_modules"]
)
modules_memory_boundaries = (
hidden_modules.Hidden_modules.get_modules_memory_boundaries(
context, kernel_name
)
)
run_results["hidden_modules"] = list(
hidden_modules.Hidden_modules.get_hidden_modules(
context,
kernel_name,
known_modules_addresses,
modules_memory_boundaries,
)
)
return run_results
def _generator(self):
kernel_name = self.config["kernel"]
run_results = self.run_modules_scanners(self.context, kernel_name)
kernel = self.context.modules[kernel_name]
run_results = linux_utilities_modules.Modules.run_modules_scanners(
self.context, kernel_name, flatten=False
)
aggregated_modules = {}
# We want to be explicit on the plugins results we are interested in
for plugin_name in ["lsmod", "check_modules", "hidden_modules"]:
# Iterate over each recovered module
for module in run_results[plugin_name]:
for mod_info in run_results[plugin_name]:
# Use offsets as unique keys, whether a module
# appears in many plugin runs or not
if aggregated_modules.get(module.vol.offset, None) is not None:
if aggregated_modules.get(mod_info.offset, None) is not None:
# Append the plugin to the list of originating plugins
aggregated_modules[module.vol.offset][1].append(plugin_name)
aggregated_modules[mod_info.offset].append(plugin_name)
else:
aggregated_modules[module.vol.offset] = (module, [plugin_name])
aggregated_modules[mod_info.offset] = [plugin_name]
for module_offset, (module, originating_plugins) in aggregated_modules.items():
for module_offset, originating_plugins in aggregated_modules.items():
# Tainting parsing capabilities applied to the module
module = kernel.object("module", offset=module_offset, absolute=True)
if self.config.get("plain_taints"):
taints = tainting.Tainting.get_taints_as_plain_string(
self.context,
volatility3/framework/plugins/linux/netfilter.py
+1 -1
View File
@@ -726,7 +726,7 @@ class Netfilter(interfaces.plugins.PluginInterface):
_version = (1, 1, 1)
_required_linux_utilities_modules_version = (1, 0, 0)
_required_linux_utilities_modules_version = (2, 0, 0)
_required_linuxutils_version = (2, 1, 0)
_required_lsmod_version = (2, 0, 0)
_required_linuxnet_version = (1, 0, 0)
volatility3/framework/plugins/linux/tracing/ftrace.py
+24 -72
View File
@@ -10,11 +10,9 @@ from enum import Enum
from dataclasses import dataclass
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.plugins.linux import hidden_modules, modxview
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints, TreeGrid, NotAvailableValue
from volatility3.framework.symbols.linux import extensions
from volatility3.framework.constants import architectures
vollog = logging.getLogger(__name__)
@@ -67,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 = (2, 0, 0)
_version = (3, 0, 0)
_required_framework_version = (2, 19, 0)
@classmethod
@@ -81,15 +79,7 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 1, 0),
),
requirements.PluginRequirement(
name="modxview", plugin=modxview.Modxview, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="hidden_modules",
plugin=hidden_modules.Hidden_modules,
version=(1, 0, 0),
version=(2, 0, 0),
),
requirements.BooleanRequirement(
name="show_ftrace_flags",
@@ -136,8 +126,8 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
def parse_ftrace_ops(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
known_modules: Dict[str, List[extensions.module]],
kernel_module_name: str,
known_modules: Dict[str, List[linux_utilities_modules.Modules.ModuleInfo]],
ftrace_ops: interfaces.objects.ObjectInterface,
run_hidden_modules: bool = True,
) -> Generator[ParsedFtraceOps, None, None]:
@@ -145,70 +135,33 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
Iterates over embedded ftrace_func_entry entries, which point to hooked memory areas.
Args:
known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through modxview.run_modules_scanners().
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.
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
"""
kernel = context.modules[kernel_name]
kernel_layer = context.layers[kernel.layer_name]
kernel = context.modules[kernel_module_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 = linux_utilities_modules.Modules.module_lookup_by_address(
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 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(
kernel_layer.canonicalize(module.vol.offset)
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["hidden_modules"] = list(
hidden_modules.Hidden_modules.get_hidden_modules(
context,
kernel_name,
known_modules_addresses,
modules_memory_boundaries,
)
)
# Lookup the updated list to see if hidden_modules was able
# to find the missing module
module = linux_utilities_modules.Modules.module_lookup_by_address(
mod_info, callback_symbol = (
linux_utilities_modules.Modules.module_lookup_by_address(
context,
kernel.layer_name,
modxview.Modxview.flatten_run_modules_results(known_modules),
kernel_module_name,
known_modules,
callback,
)
)
# Fetch more information about the module
if module is not None:
module_address = module.vol.offset
module_name = module.get_name()
callback_symbol = module.get_symbol_by_address(callback)
if mod_info:
module_address = mod_info.start
module_name = mod_info.name
else:
vollog.warning(
callback_symbol = module_address = module_name = None
vollog.debug(
f"Could not determine ftrace_ops@{ftrace_ops.vol.offset:#x} callback {callback:#x} module origin.",
)
@@ -264,16 +217,15 @@ if the "hidden_modules" key is present in known_modules.
kernel = self.context.modules[kernel_name]
if not kernel.has_symbol("ftrace_ops_list"):
raise exceptions.SymbolError(
"ftrace_ops_list",
kernel.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 corrupted.',
vollog.error(
'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 corrupted.'
)
return
# Do not run hidden_modules by default, but only on failure to find a module
known_modules = modxview.Modxview.run_modules_scanners(
self.context, kernel_name, run_hidden_modules=False
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
self.context, kernel_name, run_hidden_modules=True
)
for ftrace_ops in self.iterate_ftrace_ops_list(self.context, kernel_name):
for ftrace_ops_parsed in self.parse_ftrace_ops(
self.context,
volatility3/framework/plugins/linux/tracing/tracepoints.py
+15 -63
View File
@@ -9,11 +9,9 @@ from typing import Dict, Iterable, List, Optional
from dataclasses import dataclass
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.plugins.linux import hidden_modules, modxview
from volatility3.framework import constants, exceptions, interfaces
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints, NotAvailableValue, TreeGrid
from volatility3.framework.symbols.linux import extensions
from volatility3.framework.objects import utility
from volatility3.framework.constants import architectures
@@ -54,15 +52,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 1, 0),
),
requirements.PluginRequirement(
name="modxview", plugin=modxview.Modxview, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="hidden_modules",
plugin=hidden_modules.Hidden_modules,
version=(1, 0, 0),
version=(2, 0, 0),
),
]
@@ -105,15 +95,15 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
def parse_tracepoint(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
known_modules: Dict[str, List[extensions.module]],
kernel_module_name: str,
known_modules: Dict[str, List[linux_utilities_modules.Modules.ModuleInfo]],
tracepoint: interfaces.objects.ObjectInterface,
run_hidden_modules: bool = True,
) -> Optional[Iterable[ParsedTracepointFunc]]:
"""Parse a tracepoint struct to highlight tracepoints kernel hooking.
Args:
known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through modxview.run_modules_scanners().
known_modules: A dict of known modules, used to locate callbacks origin. Typically obtained through 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.
@@ -121,11 +111,10 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
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]
kernel = context.modules[kernel_module_name]
for tracepoint_func in cls.iterate_tracepoint_funcs(
context, kernel_layer.name, tracepoint
context, kernel.layer_name, tracepoint
):
try:
tracepoint_name = utility.pointer_to_string(tracepoint.name, count=512)
@@ -139,56 +128,19 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
probe_handler_symbol = module_address = module_name = None
# Try to lookup within the known modules if the probe_handler address fits
module = linux_utilities_modules.Modules.module_lookup_by_address(
context,
kernel.layer_name,
modxview.Modxview.flatten_run_modules_results(known_modules),
probe_handler_address,
)
# Run hidden_modules plugin if a probe handler origin couldn't be determined (only done once, results are re-used afterwards)
if (
module is None
and run_hidden_modules
and "hidden_modules" not in known_modules
):
vollog.info(
"A probe handler module origin could not be determined. hidden_modules plugin will be run to detect additional modules.",
)
known_modules_addresses = set(
kernel_layer.canonicalize(module.vol.offset)
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["hidden_modules"] = list(
hidden_modules.Hidden_modules.get_hidden_modules(
context,
kernel_name,
known_modules_addresses,
modules_memory_boundaries,
)
)
# Lookup the updated list to see if hidden_modules was able
# to find the missing module
module = linux_utilities_modules.Modules.module_lookup_by_address(
mod_info, probe_handler_symbol = (
linux_utilities_modules.Modules.module_lookup_by_address(
context,
kernel.layer_name,
modxview.Modxview.flatten_run_modules_results(known_modules),
kernel_module_name,
known_modules,
probe_handler_address,
)
)
# Fetch more information about the module
if module is not None:
module_address = module.vol.offset
module_name = module.get_name()
probe_handler_symbol = module.get_symbol_by_address(
probe_handler_address
)
if mod_info is not None:
module_address = mod_info.offset
module_name = mod_info.name
else:
vollog.debug(
f"Could not determine tracepoint@{tracepoint.vol.offset:#x} probe handler {probe_handler_address:#x} module origin.",
@@ -276,7 +228,7 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
)
return
known_modules = modxview.Modxview.run_modules_scanners(
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
self.context, kernel_name, run_hidden_modules=False
)
tracepoints = self.iterate_tracepoints_array(self.context, kernel_name)
volatility3/framework/plugins/linux/tty_check.py
+1 -1
View File
@@ -33,7 +33,7 @@ class tty_check(plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(1, 0, 0),
version=(2, 0, 0),
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
volatility3/framework/symbols/linux/__init__.py
+14 -6
View File
@@ -1,6 +1,7 @@
# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import math
import string
import contextlib
@@ -14,9 +15,9 @@ from volatility3 import framework
from volatility3.framework import (
constants,
exceptions,
deprecation,
interfaces,
objects,
Deprecation,
)
from volatility3.framework.objects import utility
from volatility3.framework.symbols import intermed
@@ -367,9 +368,9 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
yield fd_num, filp, full_path
@classmethod
@Deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.mask_mods_list,
replacement_version=(1, 0, 0),
@deprecation.method_being_removed(
removal_date="2025-09-25",
message="Callers to this method should adapt `linux_utilities_modules.Modules.run_module_scanners`",
)
def mask_mods_list(
cls,
@@ -385,6 +386,10 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
return linux_utilities_modules.Modules.mask_mods_list(context, layer_name, mods)
@classmethod
@deprecation.method_being_removed(
removal_date="2025-09-25",
message="Callers to this method should adapt `linux_utilities_modules.Modules.run_module_scanners`",
)
def generate_kernel_handler_info(
cls,
context: interfaces.context.ContextInterface,
@@ -392,6 +397,8 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
mods_list: Iterator[interfaces.objects.ObjectInterface],
) -> List[Tuple[str, int, int]]:
"""
This method is being deprecated. Use `linux_utilities_modules.Modules.run_module_scanners` to map kernel pointers to modules")
A helper function that gets the beginning and end address of the kernel module
"""
@@ -412,9 +419,10 @@ class LinuxUtilities(interfaces.configuration.VersionableInterface):
)
@classmethod
@Deprecation.deprecated_method(
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.lookup_module_address,
replacement_version=(1, 0, 0),
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
)
def lookup_module_address(
cls,
volatility3/framework/symbols/linux/utilities/modules.py
+446 -28
View File
@@ -1,32 +1,54 @@
import logging
import warnings
from typing import Iterable, Iterator, List, Optional, Tuple
from typing import Iterable, Iterator, List, Optional, Tuple, NamedTuple, Dict, Set
from volatility3 import framework
from volatility3.framework import constants, interfaces
from volatility3.framework import (
constants,
interfaces,
deprecation,
exceptions,
objects,
)
from volatility3.framework.objects import utility
from volatility3.framework.symbols.linux import extensions
vollog = logging.getLogger(__name__)
class Modules(interfaces.configuration.VersionableInterface):
"""Kernel modules related utilities."""
_version = (1, 1, 0)
_version = (2, 0, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
@classmethod
class ModuleInfo(NamedTuple):
"""
Used to track the name and boundary of a kernel module
"""
offset: int
name: str
start: int
end: int
@staticmethod
def module_lookup_by_address(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
modules: Iterable[extensions.module],
kernel_module_name: str,
modules: Iterable[ModuleInfo],
target_address: int,
) -> Optional[extensions.module]:
run_hidden_modules: bool = True,
) -> Optional[Tuple[ModuleInfo, Optional[str]]]:
"""
Determine if a target address lies in a module memory space.
Returns the module where the provided address lies.
`modules` must be non-empty and contain masked addresses via `get_module_info_for_module` or
a ValueError will be thrown
Args:
context: The context on which to operate
layer_name: The name of the layer on which to operate
@@ -34,44 +56,70 @@ class Modules(interfaces.configuration.VersionableInterface):
target_address: The address to check for a match
Returns:
The first memory module in which the address fits
The first memory module in which the address fits and the symbol name for `target_address`
Kernel documentation:
"within_module" and "within_module_mem_type" functions
"""
kernel = context.modules[kernel_module_name]
kernel_layer = context.layers[kernel.layer_name]
if not modules:
raise ValueError("Empty list sent to `module_lookup_by_address`")
matches = []
seen_addresses = set()
for module in modules:
_, start, end = cls.mask_mods_list(context, layer_name, [module])[0]
if (
start <= target_address < end
and module.vol.offset not in seen_addresses
):
if module.start != module.start & kernel_layer.address_mask:
raise ValueError(
"Modules list must be gathered from `run_modules_scanners` to be used in this function"
)
if module.start <= target_address < module.end:
matches.append(module)
seen_addresses.add(module.vol.offset)
if len(matches) > 1:
warnings.warn(
f"Address {hex(target_address)} fits in modules at {[hex(module.vol.offset) for module in matches]}, indicating potential modules memory space overlap.",
UserWarning,
)
return matches[0]
elif len(matches) == 1:
return matches[0]
if len(matches) >= 1:
if len(matches) > 1:
warnings.warn(
f"Address {hex(target_address)} fits in modules at {[hex(module.start) for module in matches]}, indicating potential modules memory space overlap. The first matching entry {matches[0].name} will be used",
UserWarning,
)
return None
symbol_name = None
match = matches[0]
if match.name == constants.linux.KERNEL_NAME:
symbols = list(kernel.get_symbols_by_absolute_location(target_address))
if len(symbols):
symbol_name = symbols[0]
else:
module = kernel.object("module", offset=module.offset, absolute=True)
symbol_name = module.get_symbol_by_address(target_address)
if symbol_name:
symbol_name = symbol_name.split(constants.BANG)[1]
return match, symbol_name
return None, None
@classmethod
@deprecation.method_being_removed(
removal_date="2025-09-25",
message="Code using this function should adapt `linux_utilities_modules.Modules.run_module_scanners`",
)
def mask_mods_list(
cls,
context: interfaces.context.ContextInterface,
layer_name: str,
mods: Iterator[interfaces.objects.ObjectInterface],
kernel_layer_name: str,
mods: Iterator[extensions.module],
) -> List[Tuple[str, int, int]]:
"""
A helper function to mask the starting and end address of kernel modules
"""
mask = context.layers[layer_name].address_mask
mask = context.layers[kernel_layer_name].address_mask
return [
(
@@ -83,6 +131,10 @@ class Modules(interfaces.configuration.VersionableInterface):
]
@classmethod
@deprecation.method_being_removed(
removal_date="2025-09-25",
message="Use `module_lookup_by_address` to map address to their hosting kernel module and symbol.",
)
def lookup_module_address(
cls,
context: interfaces.context.ContextInterface,
@@ -116,3 +168,369 @@ class Modules(interfaces.configuration.VersionableInterface):
break
return mod_name, symbol_name
@classmethod
def get_module_info_for_module(
cls, address_mask: int, module: extensions.module
) -> Optional[ModuleInfo]:
"""
Returns a ModuleInfo instance for `module`
This performs address masking to avoid endless calls to `mask_mods_list`
Returns None if the name is smeared
"""
try:
mod_name = utility.array_to_string(module.name)
except exceptions.InvalidAddressException:
return None
start = module.get_module_base() & address_mask
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,
) -> 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
)
@classmethod
def run_modules_scanners(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
run_hidden_modules: bool = True,
flatten: bool = True,
) -> Dict[str, List[ModuleInfo]]:
"""Run module scanning plugins and aggregate the results. It is designed
to not operate any inter-plugin results triage.
Args:
run_hidden_modules: specify if the hidden_modules plugin should be run
Returns:
Dictionary mapping each plugin to its corresponding result
"""
kernel = context.modules[kernel_name]
address_mask = context.layers[kernel.layer_name].address_mask
run_results = {}
# the kernel module boundaries
run_results["kernel"] = [cls.get_kernel_module_info(context, kernel_name)]
# lsmod
run_results["lsmod"] = []
for module in cls.list_modules(context, kernel_name):
modinfo = cls.get_module_info_for_module(address_mask, module)
if modinfo:
run_results["lsmod"].append(modinfo)
# check_modules
run_results["check_modules"] = []
sysfs_modules: dict = cls.get_kset_modules(context, kernel_name)
for m_offset in sysfs_modules.values():
module = kernel.object(object_type="module", offset=m_offset, absolute=True)
modinfo = cls.get_module_info_for_module(address_mask, module)
if modinfo:
run_results["check_modules"].append(modinfo)
# hidden_modules
if run_hidden_modules:
known_modules_addresses = set(
context.layers[kernel.layer_name].canonicalize(modinfo.start)
for modinfo in run_results["kernel"]
+ run_results["lsmod"]
+ run_results["check_modules"]
)
modules_memory_boundaries = cls.get_modules_memory_boundaries(
context, kernel_name
)
run_results["hidden_modules"] = []
for module in cls.get_hidden_modules(
context,
kernel_name,
known_modules_addresses,
modules_memory_boundaries,
):
modinfo = cls.get_module_info_for_module(address_mask, module)
if modinfo:
run_results["hidden_modules"].append(modinfo)
if flatten:
return cls.flatten_run_modules_results(run_results)
return run_results
@staticmethod
def get_modules_memory_boundaries(
context: interfaces.context.ContextInterface,
vmlinux_module_name: str,
) -> Tuple[int, 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"):
# Kernel >= 5.19 58d208de3e8d87dbe196caf0b57cc58c7a3836ca
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"):
# 2.6.27 <= kernel < 5.19 3a642e99babe0617febb6f402e1e063479f489db
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):
raise exceptions.VolatilityException(
"Your ISF symbols lack type information. You may need to 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 flatten_run_modules_results(
cls, run_results: Dict[str, List[ModuleInfo]], deduplicate: bool = True
) -> List[ModuleInfo]:
"""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:
List of ModuleInfo objects
"""
uniq_modules: List[Modules.ModuleInfo] = []
seen_addresses: int = set()
for modules in run_results.values():
for module in modules:
if deduplicate and (module.start in seen_addresses):
continue
seen_addresses.add(module.start)
uniq_modules.append(module)
return uniq_modules
@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").child_template("mod")
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_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
@classmethod
def list_modules(
cls, context: interfaces.context.ContextInterface, vmlinux_module_name: str
) -> Iterable[interfaces.objects.ObjectInterface]:
"""Lists all the modules in the primary layer.
Args:
context: The context to retrieve required elements (layers, symbol tables) from
layer_name: The name of the layer on which to operate
vmlinux_symbols: The name of the table containing the kernel symbols
Yields:
The modules present in the `layer_name` layer's modules list
This function will throw a SymbolError exception if kernel module support is not enabled.
"""
vmlinux = context.modules[vmlinux_module_name]
modules = vmlinux.object_from_symbol(symbol_name="modules").cast("list_head")
table_name = vmlinux.symbol_table_name
yield from modules.to_list(table_name + constants.BANG + "module", "list")
@classmethod
def get_kset_modules(
cls, context: interfaces.context.ContextInterface, vmlinux_name: str
) -> Dict[str, extensions.module]:
vmlinux = context.modules[vmlinux_name]
try:
module_kset = vmlinux.object_from_symbol("module_kset")
except exceptions.SymbolError:
module_kset = None
if not module_kset:
raise TypeError(
"This plugin requires the module_kset 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."
)
ret = {}
kobj_off = vmlinux.get_type("module_kobject").relative_child_offset("kobj")
for kobj in module_kset.list.to_list(
vmlinux.symbol_table_name + constants.BANG + "kobject", "entry"
):
mod_kobj = vmlinux.object(
object_type="module_kobject",
offset=kobj.vol.offset - kobj_off,
absolute=True,
)
mod = mod_kobj.mod
try:
name = utility.pointer_to_string(kobj.name, 32)
except exceptions.InvalidAddressException:
continue
if kobj.name and kobj.reference_count() > 2:
ret[name] = mod
return ret
@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)