Merge pull request #1710 from volatilityfoundation/update_deprecated_api_usage

Update nearly all callers of now deprecated Linux kernel APIs
This commit is contained in:
ikelos
2025-03-15 20:44:08 +00:00
committed by GitHub
12 changed files with 409 additions and 194 deletions
@@ -10,7 +10,6 @@ from volatility3.framework import interfaces, renderers, symbols
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import linux
from volatility3.plugins.linux import lsmod
vollog = logging.getLogger(__name__)
@@ -34,14 +33,16 @@ class Check_idt(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
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)
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
),
]
@staticmethod
@@ -82,10 +83,10 @@ class Check_idt(interfaces.plugins.PluginInterface):
vmlinux = self.context.modules[self.config["kernel"]]
modules = lsmod.Lsmod.list_modules(self.context, vmlinux.name)
handlers = linux.LinuxUtilities.generate_kernel_handler_info(
self.context, vmlinux.name, modules
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
idt_table_size = 256
@@ -134,19 +135,24 @@ class Check_idt(interfaces.plugins.PluginInterface):
module_name = renderers.NotAvailableValue()
symbol_name = renderers.NotAvailableValue()
else:
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self.context, vmlinux.name, handlers, idt_addr
module_info, symbol_name = (
linux_utilities_modules.Modules.module_lookup_by_address(
self.context, vmlinux.name, known_modules, idt_addr
)
)
if module_info:
module_name = module_info.name
else:
module_name = renderers.NotAvailableValue()
yield (
0,
[
format_hints.Hex(i),
format_hints.Hex(idt_addr),
module_name,
symbol_name,
symbol_name or renderers.NotAvailableValue(),
],
)
@@ -33,7 +33,7 @@ class Check_modules(plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
version=(3, 0, 0),
),
]
@@ -41,7 +41,7 @@ class Check_modules(plugins.PluginInterface):
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.get_kset_modules,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
replacement_version=(3, 0, 0),
)
def get_kset_modules(
cls, context: interfaces.context.ContextInterface, vmlinux_name: str
@@ -10,7 +10,6 @@ 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
from volatility3.plugins.linux import lsmod
vollog = logging.getLogger(__name__)
@@ -29,13 +28,10 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
description="Linux kernel",
architectures=architectures.LINUX_ARCHS,
),
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),
version=(3, 0, 0),
),
]
@@ -43,7 +39,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.get_modules_memory_boundaries,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
replacement_version=(3, 0, 0),
)
def get_modules_memory_boundaries(
context: interfaces.context.ContextInterface,
@@ -56,7 +52,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.get_module_address_alignment,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
replacement_version=(3, 0, 0),
)
@classmethod
def _get_module_address_alignment(
@@ -84,7 +80,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.get_hidden_modules,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
replacement_version=(3, 0, 0),
)
@classmethod
def get_hidden_modules(
@@ -124,7 +120,7 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.validate_alignment_patterns,
removal_date="2025-09-25",
replacement_version=(2, 0, 0),
replacement_version=(3, 0, 0),
)
def _validate_alignment_patterns(
addresses: Iterable[int],
@@ -163,7 +159,9 @@ class Hidden_modules(interfaces.plugins.PluginInterface):
known_module_addresses = {
vmlinux_layer.canonicalize(module.vol.offset)
for module in lsmod.Lsmod.list_modules(context, vmlinux_module_name)
for module in linux_utilities_modules.Modules.list_modules(
context, vmlinux_module_name
)
}
return known_module_addresses
@@ -9,7 +9,6 @@ from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import linux
from volatility3.plugins.linux import lsmod
vollog = logging.getLogger(__name__)
@@ -30,10 +29,12 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
version=(3, 0, 0),
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 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)
@@ -43,12 +44,6 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
def _generator(self):
vmlinux = self.context.modules[self.config["kernel"]]
modules = lsmod.Lsmod.list_modules(self.context, vmlinux.name)
handlers = linux.LinuxUtilities.generate_kernel_handler_info(
self.context, vmlinux.name, modules
)
try:
knl_addr = vmlinux.object_from_symbol("keyboard_notifier_list")
except exceptions.SymbolError:
@@ -65,6 +60,12 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
vollog.error("The head of the keyboard notifier list is paged out.")
return
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
knl = vmlinux.object(
object_type="atomic_notifier_head",
offset=knl_addr.vol.offset,
@@ -76,13 +77,25 @@ class Keyboard_notifiers(interfaces.plugins.PluginInterface):
):
call_addr = call_back.notifier_call
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self.context, vmlinux.name, handlers, call_addr
module_info, symbol_name = (
linux_utilities_modules.Modules.module_lookup_by_address(
self.context, vmlinux.name, known_modules, call_addr
)
)
yield (0, [format_hints.Hex(call_addr), module_name, symbol_name])
if module_info:
module_name = module_info.name
else:
module_name = renderers.NotAvailableValue()
yield (
0,
[
format_hints.Hex(call_addr),
module_name,
symbol_name or renderers.NotAvailableValue(),
],
)
def run(self):
return renderers.TreeGrid(
+25 -21
View File
@@ -5,14 +5,14 @@ import logging
from typing import List
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import constants, exceptions, interfaces, renderers
from volatility3.framework import exceptions, interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import linux
from volatility3.framework.constants import architectures
from volatility3.framework.objects import utility
from volatility3.plugins.linux import pslist, lsmod
from volatility3.plugins.linux import pslist
vollog = logging.getLogger(__name__)
@@ -34,7 +34,12 @@ class Kthreads(plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
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)
@@ -42,28 +47,24 @@ class Kthreads(plugins.PluginInterface):
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 0, 0)
),
]
def _generator(self):
vmlinux = self.context.modules[self.config["kernel"]]
modules = lsmod.Lsmod.list_modules(self.context, vmlinux.name)
handlers = linux.LinuxUtilities.generate_kernel_handler_info(
self.context, vmlinux.name, modules
)
kthread_type = vmlinux.get_type(
vmlinux.symbol_table_name + constants.BANG + "kthread"
)
kthread_type = vmlinux.get_type("kthread")
if not kthread_type.has_member("threadfn"):
raise exceptions.VolatilityException(
"Unsupported kthread implementation. This plugin only works with kernels >= 5.8"
)
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
for task in pslist.PsList.list_tasks(
self.context, vmlinux.name, include_threads=True
):
@@ -86,9 +87,7 @@ class Kthreads(plugins.PluginInterface):
if not (threadfn and threadfn.is_readable()):
continue
task_name = utility.array_to_string(task.comm)
thread_name = task_name
thread_name = utility.array_to_string(task.comm)
# kernels >= 5.17 in d6986ce24fc00b0638bd29efe8fb7ba7619ed2aa full_name was added to kthread
if kthread.has_member("full_name"):
@@ -101,18 +100,23 @@ class Kthreads(plugins.PluginInterface):
f"full_name pointer for thread at {kthread.vol.offset:#x} is paged out."
)
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self.context, vmlinux.name, handlers, threadfn
module_info, symbol_name = (
linux_utilities_modules.Modules.module_lookup_by_address(
self.context, vmlinux.name, known_modules, threadfn
)
)
if module_info:
module_name = module_info.name
else:
module_name = renderers.NotAvailableValue()
fields = [
task.pid,
thread_name,
format_hints.Hex(threadfn),
module_name,
symbol_name,
symbol_name or renderers.NotAvailableValue(),
]
yield 0, fields
+2 -2
View File
@@ -33,14 +33,14 @@ class Lsmod(plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
version=(3, 0, 0),
),
]
@classmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.list_modules,
replacement_version=(2, 0, 0),
replacement_version=(3, 0, 0),
removal_date="2025-09-25",
)
def list_modules(
+40 -18
View File
@@ -34,7 +34,22 @@ spot modules presence and taints."""
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
version=(3, 0, 0),
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherer_lsmod",
component=linux_utilities_modules.ModuleGathererLsmod,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherer_sysfs",
component=linux_utilities_modules.ModuleGathererSysFs,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherer_scanner",
component=linux_utilities_modules.ModuleGathererScanner,
version=(1, 0, 0),
),
requirements.VersionRequirement(
name="linux-tainting", component=tainting.Tainting, version=(1, 0, 0)
@@ -50,7 +65,7 @@ spot modules presence and taints."""
@classmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.flatten_run_modules_results,
replacement_version=(2, 0, 0),
replacement_version=(3, 0, 0),
removal_date="2025-09-25",
)
def flatten_run_modules_results(
@@ -73,7 +88,7 @@ spot modules presence and taints."""
@classmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.run_modules_scanners,
replacement_version=(2, 0, 0),
replacement_version=(3, 0, 0),
removal_date="2025-09-25",
)
def run_modules_scanners(
@@ -89,35 +104,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_gatherers = [
linux_utilities_modules.ModuleGathererLsmod,
linux_utilities_modules.ModuleGathererSysFs,
linux_utilities_modules.ModuleGathererScanner,
]
run_results = linux_utilities_modules.Modules.run_modules_scanners(
self.context, kernel_name, flatten=False
context=self.context,
kernel_module_name=self.config["kernel"],
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 ["lsmod", "check_modules", "hidden_modules"]:
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.name]:
# 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.name)
else:
aggregated_modules[mod_info.offset] = [plugin_name]
aggregated_modules[mod_info.offset] = [gatherer.name]
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,
)
@@ -125,7 +147,7 @@ spot modules presence and taints."""
taints = ",".join(
tainting.Tainting.get_taints_parsed(
self.context,
kernel_name,
self.config["kernel"],
module.taints,
True,
)
@@ -136,9 +158,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.name in gatherers,
linux_utilities_modules.ModuleGathererSysFs.name in gatherers,
linux_utilities_modules.ModuleGathererScanner.name in gatherers,
taints or NotAvailableValue(),
),
)
@@ -149,7 +171,7 @@ spot modules presence and taints."""
("Address", format_hints.Hex),
("In procfs", bool),
("In sysfs", bool),
("Hidden", bool),
("In scan", bool),
("Taints", str),
]
@@ -726,7 +726,7 @@ class Netfilter(interfaces.plugins.PluginInterface):
_version = (1, 1, 1)
_required_linux_utilities_modules_version = (2, 0, 0)
_required_linux_utilities_modules_version = (3, 0, 0)
_required_linuxutils_version = (2, 1, 0)
_required_lsmod_version = (2, 0, 0)
_required_linuxnet_version = (1, 0, 0)
@@ -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
@@ -79,7 +79,12 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
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",
@@ -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
@@ -223,7 +225,9 @@ class CheckFtrace(interfaces.plugins.PluginInterface):
return
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
self.context, kernel_name, run_hidden_modules=True
context=self.context,
kernel_module_name=self.config["kernel"],
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
@@ -52,7 +52,12 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
version=(3, 0, 0),
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherers",
component=linux_utilities_modules.ModuleGatherers,
version=(1, 0, 0),
),
]
@@ -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]]:
@@ -229,7 +234,9 @@ class CheckTracepoints(interfaces.plugins.PluginInterface):
return
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
self.context, kernel_name, run_hidden_modules=False
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
tracepoints = self.iterate_tracepoints_array(self.context, kernel_name)
@@ -12,7 +12,6 @@ from volatility3.framework.interfaces import plugins
from volatility3.framework.objects import utility
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols import linux
from volatility3.plugins.linux import lsmod
vollog = logging.getLogger(__name__)
@@ -33,10 +32,12 @@ class tty_check(plugins.PluginInterface):
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
version=(2, 0, 0),
version=(3, 0, 0),
),
requirements.PluginRequirement(
name="lsmod", plugin=lsmod.Lsmod, version=(2, 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)
@@ -46,12 +47,6 @@ class tty_check(plugins.PluginInterface):
def _generator(self):
vmlinux = self.context.modules[self.config["kernel"]]
modules = lsmod.Lsmod.list_modules(self.context, vmlinux.name)
handlers = linux.LinuxUtilities.generate_kernel_handler_info(
self.context, vmlinux.name, modules
)
try:
tty_drivers = vmlinux.object_from_symbol("tty_drivers").cast("list_head")
except exceptions.SymbolError:
@@ -64,6 +59,12 @@ class tty_check(plugins.PluginInterface):
"This means you are either analyzing an unsupported kernel version or that your symbol table is corrupt."
)
known_modules = linux_utilities_modules.Modules.run_modules_scanners(
context=self.context,
kernel_module_name=self.config["kernel"],
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
for tty in tty_drivers.to_list(
vmlinux.symbol_table_name + constants.BANG + "tty_driver", "tty_drivers"
):
@@ -87,13 +88,23 @@ class tty_check(plugins.PluginInterface):
except exceptions.InvalidAddressException:
continue
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self.context, vmlinux.name, handlers, recv_buf
module_info, symbol_name = (
linux_utilities_modules.Modules.module_lookup_by_address(
self.context, vmlinux.name, known_modules, recv_buf
)
)
yield (0, (name, format_hints.Hex(recv_buf), module_name, symbol_name))
if module_info:
module_name = module_info.name
else:
module_name = renderers.NotAvailableValue()
yield 0, (
name,
format_hints.Hex(recv_buf),
module_name,
symbol_name or renderers.NotAvailableValue(),
)
def run(self):
return renderers.TreeGrid(
@@ -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,37 +22,63 @@ from volatility3.framework import (
exceptions,
objects,
)
from volatility3.framework.configuration import requirements
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."""
class ModuleInfo(NamedTuple):
"""
Used to track the name and boundary of a kernel module
"""
_version = (2, 0, 0)
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)
class ModuleInfo(NamedTuple):
gatherer_return_type = Generator[Union[ModuleInfo, "extensions.module"], None, None]
# Must be set to a unique, descriptive name of the gathering technique or data structure source
name = None
@classmethod
@abstractmethod
def gather_modules(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> gatherer_return_type:
"""
Used to track the name and boundary of a kernel module
This method must return a generator (yield) of each `gatherer_return_type` found from its source
"""
offset: int
name: str
start: int
end: int
@staticmethod
class Modules(interfaces.configuration.VersionableInterface):
"""Kernel modules related utilities."""
_version = (3, 0, 0)
_required_framework_version = (2, 0, 0)
framework.require_interface_version(*_required_framework_version)
@classmethod
def module_lookup_by_address(
cls,
context: interfaces.context.ContextInterface,
kernel_module_name: str,
modules: Iterable[ModuleInfo],
target_address: int,
run_hidden_modules: bool = True,
) -> Optional[Tuple[ModuleInfo, Optional[str]]]:
"""
Determine if a target address lies in a module memory space.
@@ -189,98 +227,80 @@ 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,
) -> 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 run_modules_scanners(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
run_hidden_modules: bool = True,
kernel_module_name: str,
caller_wanted_gatherers: List[ModuleGathererInterface],
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
"""
Rules for `caller_wanted_gatherers`:
If `ModuleGatherers.all_gathers_identifier` is specified then every source will be populated
kernel = context.modules[kernel_name]
If empty or an invalid gatherer is specified then a ValueError is thrown
All gatherer names must be unique
Args:
called_wanted_sources: The list of sources to gather modules.
flatten: Whether to de-duplicate modules across gatherers
Returns:
Dictionary mapping each gatherer to its corresponding result
"""
if not caller_wanted_gatherers:
raise ValueError(
"`caller_wanted_gatherers` must have at least one gatherer."
)
if not isinstance(caller_wanted_gatherers, Iterable):
raise ValueError("`caller_wanted_gatherers` must be iterable")
seen_names = set()
for gatherer in caller_wanted_gatherers:
if not issubclass(gatherer, ModuleGathererInterface):
raise ValueError(
f"Invalid gatherer sent through `caller_wanted_gatherers`: {gatherer}"
)
if not gatherer.name:
raise ValueError(
f"{gatherer} does not have a valid name attribute, which is required. It must be a non-zero length string."
)
if gatherer.name in seen_names:
raise ValueError(
f"{gatherer} has a name {gatherer.name} which has already been processed. Names must be unique."
)
seen_names.add(gatherer.name)
kernel = context.modules[kernel_module_name]
address_mask = context.layers[kernel.layer_name].address_mask
run_results = {}
run_results: Dict[ModuleGathererInterface, List[ModuleInfo]] = {}
# the kernel module boundaries
run_results["kernel"] = [cls.get_kernel_module_info(context, kernel_name)]
# Walk each source gathering modules
for gatherer in caller_wanted_gatherers:
run_results[gatherer.name] = []
# lsmod
run_results["lsmod"] = []
# process each module coming from back the current source
for module in gatherer.gather_modules(context, kernel_module_name):
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)
# the kernel sends back a ModuleInfo directly
if isinstance(module, ModuleInfo):
modinfo = module
else:
modinfo = cls.get_module_info_for_module(address_mask, module)
# 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)
run_results[gatherer.name].append(modinfo)
if flatten:
return cls.flatten_run_modules_results(run_results)
@@ -338,7 +358,7 @@ class Modules(interfaces.configuration.VersionableInterface):
Returns:
List of ModuleInfo objects
"""
uniq_modules: List[Modules.ModuleInfo] = []
uniq_modules: List[ModuleInfo] = []
seen_addresses: int = set()
@@ -534,3 +554,133 @@ 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
"""
_version = (1, 0, 0)
name = "Lsmod"
@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
"""
_version = (1, 0, 0)
name = "SysFs"
@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
"""
_version = (1, 0, 0)
name = "Scanner"
@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
"""
_version = (1, 0, 0)
name = "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,
interfaces.configuration.ConfigurableInterface,
):
_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,
]
@classmethod
def get_requirements(cls):
reqs = []
# for now, all versions are 1, this will be broken out if/when that changes
for gatherer in ModuleGatherers.all_gatherers_identifier:
reqs.append(
requirements.VersionRequirement(
name=gatherer.name.replace(" ", ""),
component=gatherer,
version=(1, 0, 0),
)
)
return reqs