WEBDIENSTE/volatility3
1
0
Fork 0
mirror of https://github.com/volatilityfoundation/volatility3.git synced 2026-08-17 20:35:40 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1850 from SolitudePy/malware_plugins_category

Browse Source 
Malware categorization: linux.tty_check, linux.netfilter, linux.modxview
This commit is contained in:
ikelos
2025-06-10 21:20:12 +01:00
committed by GitHub
parent 43ed10a4f8 8040c049e0
commit 58f441186c
7 changed files with 1140 additions and 1078 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/plugins/linux/linux.py
+2 -2
View File
@@ -89,7 +89,7 @@ class TestLinuxProcMaps:
class TestLinuxTtyCheck:
def test_linux_generic_tty_check(self, image, volatility, python):
rc, out, _err = test_volatility.runvol_plugin(
"linux.tty_check.tty_check", image, volatility, python
"linux.malware.tty_check.Tty_Check", image, volatility, python
)
assert rc == 0
@@ -504,7 +504,7 @@ class TestLinuxKmesg:
class TestLinuxNetfilter:
def test_linux_generic_netfilter(self, image, volatility, python):
rc, out, _err = test_volatility.runvol_plugin(
"linux.netfilter.Netfilter", image, volatility, python
"linux.malware.netfilter.Netfilter", image, volatility, python
)
# linux-sample-1.bin has no suspicious results for this plugin.
volatility3/framework/plugins/linux/malware/modxview.py
+181
View File
@@ -0,0 +1,181 @@
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from typing import List, Dict, Iterator
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, deprecation, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols.linux import extensions
from volatility3.framework.constants import architectures
from volatility3.framework.symbols.linux.utilities import tainting
vollog = logging.getLogger(__name__)
class Modxview(interfaces.plugins.PluginInterface):
"""Centralize lsmod, check_modules and hidden_modules results to efficiently \
spot modules presence and taints."""
_version = (1, 0, 0)
_required_framework_version = (2, 17, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=architectures.LINUX_ARCHS,
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
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)
),
requirements.BooleanRequirement(
name="plain_taints",
description="Display the plain taints string for each module.",
optional=True,
default=False,
),
]
@classmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.flatten_run_modules_results,
replacement_version=(3, 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]:
"""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:
Iterator of modules objects
"""
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=(3, 0, 0),
removal_date="2025-09-25",
)
def run_modules_scanners(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
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."""
return linux_utilities_modules.Modules.run_modules_scanners(
context, kernel_name, run_hidden_modules
)
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
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_gatherers=wanted_gatherers,
flatten=False,
)
aggregated_modules = {}
# We want to be explicit on the plugins results we are interested in
for gatherer in wanted_gatherers:
# Iterate over each recovered module
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(gatherer.name)
else:
aggregated_modules[mod_info.offset] = [gatherer.name]
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,
self.config["kernel"],
module.taints,
True,
)
else:
taints = ",".join(
tainting.Tainting.get_taints_parsed(
self.context,
self.config["kernel"],
module.taints,
True,
)
)
yield (
0,
(
module.get_name() or renderers.NotAvailableValue(),
format_hints.Hex(module_offset),
linux_utilities_modules.ModuleGathererLsmod.name in gatherers,
linux_utilities_modules.ModuleGathererSysFs.name in gatherers,
linux_utilities_modules.ModuleGathererScanner.name in gatherers,
taints or renderers.NotAvailableValue(),
),
)
def run(self):
columns = [
("Name", str),
("Address", format_hints.Hex),
("In procfs", bool),
("In sysfs", bool),
("In scan", bool),
("Taints", str),
]
return renderers.TreeGrid(
columns,
self._generator(),
)
volatility3/framework/plugins/linux/malware/netfilter.py
+806
View File
@@ -0,0 +1,806 @@
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
import logging
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from typing import Iterator, List, Tuple, Optional
from volatility3 import framework
from volatility3.framework import (
constants,
interfaces,
renderers,
exceptions,
deprecation,
)
from volatility3.framework.renderers import format_hints
from volatility3.framework.configuration import requirements
from volatility3.framework.symbols.linux import network
vollog = logging.getLogger(__name__)
@dataclass
class Proto:
name: str
hooks: Tuple[str] = field(default_factory=tuple)
PROTO_NOT_IMPLEMENTED = Proto(name="UNSPEC")
NF_INET_HOOKS = ("PRE_ROUTING", "LOCAL_IN", "FORWARD", "LOCAL_OUT", "POST_ROUTING")
NF_DEC_HOOKS = (
"PRE_ROUTING",
"LOCAL_IN",
"FORWARD",
"LOCAL_OUT",
"POST_ROUTING",
"HELLO",
"ROUTE",
)
NF_ARP_HOOKS = ("IN", "OUT", "FORWARD")
NF_NETDEV_HOOKS = ("INGRESS", "EGRESS")
LARGEST_HOOK_NUMBER = max(
len(NF_INET_HOOKS), len(NF_DEC_HOOKS), len(NF_ARP_HOOKS), len(NF_NETDEV_HOOKS)
)
class AbstractNetfilter(ABC):
"""Netfilter Abstract Base Classes handling details across various
Netfilter implementations, including constants, helpers, and common
routines.
"""
PROTO_HOOKS = (
PROTO_NOT_IMPLEMENTED, # NFPROTO_UNSPEC
Proto(name="INET", hooks=NF_INET_HOOKS), # From kernels 3.14
Proto(name="IPV4", hooks=NF_INET_HOOKS),
Proto(name="ARP", hooks=NF_ARP_HOOKS),
PROTO_NOT_IMPLEMENTED,
Proto(name="NETDEV", hooks=NF_NETDEV_HOOKS),
PROTO_NOT_IMPLEMENTED,
Proto(name="BRIDGE", hooks=NF_INET_HOOKS),
PROTO_NOT_IMPLEMENTED,
PROTO_NOT_IMPLEMENTED,
Proto(name="IPV6", hooks=NF_INET_HOOKS),
PROTO_NOT_IMPLEMENTED,
Proto(name="DECNET", hooks=NF_DEC_HOOKS), # Removed in kernel 6.1
)
NF_MAX_HOOKS = LARGEST_HOOK_NUMBER + 1
def __init__(
self, context: interfaces.context.ContextInterface, kernel_module_name: str
):
self._context = context
self.vmlinux = context.modules[kernel_module_name]
self.layer_name = self.vmlinux.layer_name
# Set data sizes
self.ptr_size = self.vmlinux.get_type("pointer").size
self.list_head_size = self.vmlinux.get_type("list_head").size
linuxutils_modulegatherers_required_version = (
Netfilter._required_linuxutils_gatherers_version
)
linuxutils_modulegatherers_current_version = (
linux_utilities_modules.ModuleGatherers.version
)
if not requirements.VersionRequirement.matches_required(
linuxutils_modulegatherers_required_version,
linuxutils_modulegatherers_current_version,
):
raise exceptions.PluginRequirementException(
f"linux_utilities_modules.ModuleGatherer version not suitable: required {linuxutils_modulegatherers_required_version} found {linuxutils_modulegatherers_current_version}"
)
linux_net_required_version = Netfilter._required_linuxnet_version
linux_net_current_version = network.NetSymbols.version
if not requirements.VersionRequirement.matches_required(
linux_net_required_version, linux_net_current_version
):
raise exceptions.PluginRequirementException(
f"symbols.linux.net.NetSymbols version not suitable: required {linux_net_required_version} found {linux_net_current_version}"
)
linux_utilities_modules_required_version = (
Netfilter._required_linux_utilities_modules_version
)
linux_utilities_modules_current_version = (
linux_utilities_modules.Modules.version
)
if not requirements.VersionRequirement.matches_required(
linux_utilities_modules_required_version,
linux_utilities_modules_current_version,
):
raise exceptions.PluginRequirementException(
f"linux_utilities_modules.Modules version not suitable: required {linux_utilities_modules_required_version} found {linux_utilities_modules_current_version}"
)
symbol_table = context.symbol_space[self.vmlinux.symbol_table_name]
network.NetSymbols.apply(symbol_table)
self.handlers = linux_utilities_modules.Modules.run_modules_scanners(
context=context,
kernel_module_name=kernel_module_name,
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
@classmethod
def run_all(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> Iterator[Tuple[int, str, str, int, int, str, bool]]:
"""It calls each subclass symtab_checks() to test the required
conditions to that specific kernel implementation.
Args:
context: The volatility3 context on which to operate
kernel_module_name: The name of the table containing the kernel symbols
Yields:
The kmsg records. Same as _run()
"""
vmlinux = context.modules[kernel_module_name]
implementation_inst = None # type: ignore
for subclass in framework.class_subclasses(cls):
if not subclass.symtab_checks(vmlinux=vmlinux):
vollog.log(
constants.LOGLEVEL_VVVV,
"Netfilter implementation '%s' doesn't match this memory dump",
subclass.__name__,
)
continue
vollog.log(
constants.LOGLEVEL_VVVV,
"Netfilter implementation '%s' matches!",
subclass.__name__,
)
implementation_inst = subclass(
context=context, kernel_module_name=kernel_module_name
)
# More than one class could be executed for an specific kernel version
# For instance: Netfilter Ingress hooks
yield from implementation_inst._run()
if implementation_inst is None:
vollog.error("Unsupported Netfilter kernel implementation")
def _run(self) -> Iterator[Tuple[int, str, str, int, int, str, bool]]:
"""Iterates over namespaces and protocols, executing various callbacks that
allow customization of the code to the specific data structure used in a
particular kernel implementation
get_hooks_container(net, proto_name, hook_name)
It returns the data structure used in a specific kernel implementation
to store the hooks for a respective namespace and protocol, basically:
For Ingress hooks:
network_namespace[] -> net_device[] -> nf_hooks_ingress[]
For egress hooks:
network_namespace[] -> net_device[] -> nf_hooks_egress[]
For all the other Netfilter hooks:
<= 4.2.8
nf_hooks[]
>= 4.3
network_namespace[] -> nf.hooks[]
get_hook_ops(hook_container, proto_idx, hook_idx)
Give the 'hook_container' got in get_hooks_container(), it
returns an iterable of 'nf_hook_ops' elements for a respective protocol
and hook type.
Returns:
netns [int]: Network namespace id
proto_name [str]: Protocol name
hook_name [str]: Hook name
priority [int]: Priority
hook_ops_hook [int]: Hook address
module_name [str]: Linux kernel module name
hooked [bool]: "True" if the network stack has been hijacked
"""
for netns, net in self.get_net_namespaces():
for proto_idx, proto_name, hook_idx, hook_name in self._proto_hook_loop():
hooks_container = self.get_hooks_container(net, proto_name, hook_name)
for hook_container in hooks_container:
for hook_ops in self.get_hook_ops(
hook_container, proto_idx, hook_idx
):
if not hook_ops:
continue
priority = int(hook_ops.priority)
hook_ops_hook = hook_ops.hook
module_info, symbol_name = (
linux_utilities_modules.Modules.module_lookup_by_address(
self._context,
self.vmlinux.name,
self.handlers,
hook_ops_hook,
)
)
hooked = module_info is None
yield netns, proto_name, hook_name, priority, hook_ops_hook, module_info, symbol_name, hooked
@classmethod
@abstractmethod
def symtab_checks(cls, vmlinux: interfaces.context.ModuleInterface) -> bool:
"""This method on each sublasss will be called to evaluate if the kernel
being analyzed fulfill the type & symbols requirements for the implementation.
The first class returning True will be instantiated and called via the
run() method.
Returns:
bool: True if the kernel being analyzed fulfill the class requirements.
"""
def _proto_hook_loop(self) -> Iterator[Tuple[int, str, int, str]]:
"""Flattens the protocol families and hooks"""
for proto_idx, proto in enumerate(AbstractNetfilter.PROTO_HOOKS):
if proto == PROTO_NOT_IMPLEMENTED:
continue
if proto.name not in self.subscribed_protocols():
# This protocol is not managed in this object
continue
for hook_idx, hook_name in enumerate(proto.hooks):
yield proto_idx, proto.name, hook_idx, hook_name
def build_nf_hook_ops_array(
self, nf_hook_entries
) -> Optional[interfaces.objects.ObjectInterface]:
"""Function helper to build the nf_hook_ops array when it is not part of the
struct 'nf_hook_entries' definition.
nf_hook_ops was stored adjacent in memory to the nf_hook_entry array, in the
new struct 'nf_hook_entries'. However, this 'nf_hooks_ops' array 'orig_ops' is
not part of the 'nf_hook_entries' struct. So, we need to calculate the offset.
struct nf_hook_entries {
u16 num_hook_entries; /* plus padding */
struct nf_hook_entry hooks[];
//const struct nf_hook_ops *orig_ops[];
}
"""
nf_hook_entry_size = self.vmlinux.get_type("nf_hook_entry").size
try:
num_hook_entries = nf_hook_entries.num_hook_entries
except exceptions.InvalidAddressException:
return None
orig_ops_addr = (
nf_hook_entries.hooks.vol.offset + nf_hook_entry_size * num_hook_entries
)
if not self.vmlinux._context.layers[self.vmlinux.layer_name].is_valid(
orig_ops_addr
):
return None
orig_ops = self._context.object(
object_type=self.get_symbol_fullname("array"),
offset=orig_ops_addr,
subtype=self.vmlinux.get_type("pointer"),
layer_name=self.layer_name,
count=num_hook_entries,
)
return orig_ops
def subscribed_protocols(self) -> Tuple[str]:
"""Allows to select which PROTO_HOOKS protocols will be processed by the
Netfiler subclass.
"""
# Most implementation handlers respond to these protocols, except for
# the ingress hook, which specifically handles the 'NETDEV' protocol.
# However, there is no corresponding Netfilter hook implementation for
# the INET protocol in the kernel. AFAIU, this is used as
# 'NFPROTO_INET = NFPROTO_IPV4 || NFPROTO_IPV6'
# in other parts of the kernel source code.
return ("IPV4", "ARP", "BRIDGE", "IPV6", "DECNET")
@deprecation.method_being_removed(
removal_date="2025-09-25",
message="Callers to this method should adapt `linux_utilities_modules.Modules.run_module_scanners`",
)
def get_module_name_for_address(self, addr) -> str:
"""Helper to obtain the module and symbol name in the format needed for the
output of this plugin.
"""
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self._context, self.vmlinux.name, self.handlers, addr
)
)
if module_name == "UNKNOWN":
module_name = None
if symbol_name != "N/A":
module_name = f"[{symbol_name}]"
return module_name
def get_net_namespaces(self):
"""Common function to retrieve the different namespaces.
From 4.3 on, all the implementations use network namespaces.
"""
nethead = self.vmlinux.object_from_symbol("net_namespace_list")
symbol_net_name = self.get_symbol_fullname("net")
for net in nethead.to_list(symbol_net_name, "list"):
net_ns_id = net.ns.inum
yield net_ns_id, net
def get_hooks_container(self, net, proto_name, hook_name):
"""Returns the data structure used in a specific kernel implementation to store
the hooks for a respective namespace and protocol.
Except for kernels < 4.3, all the implementations use network namespaces.
Also the data structure which contains the hooks, even though it changes its
implementation and/or data type, it is always in this location.
"""
yield net.nf.hooks
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
"""Given the hook_container obtained from get_hooks_container(), it
returns an iterable of 'nf_hook_ops' elements for a corresponding protocol
and hook type.
This is the most variable/unstable part of all Netfilter hook designs, it
changes almost in every single implementation.
"""
raise NotImplementedError("You must implement this method")
def get_symbol_fullname(self, symbol_basename: str) -> str:
"""Given a short symbol or type name, it returns its full name"""
return self.vmlinux.symbol_table_name + constants.BANG + symbol_basename
@staticmethod
def get_member_type(
vol_type: interfaces.objects.Template, member_name: str
) -> List[str]:
"""Returns a list of types/subtypes belonging to the given type member.
Args:
vol_type (interfaces.objects.Template): A vol3 type object
member_name (str): The member name
Returns:
list: A list of types/subtypes
"""
_size, vol_obj = vol_type.vol.members[member_name]
type_name = vol_obj.type_name
type_basename = type_name.split(constants.BANG)[1]
member_type = [type_basename]
cur_type = vol_obj
while hasattr(cur_type, "subtype"):
subtype_name = cur_type.subtype.type_name
subtype_basename = subtype_name.split(constants.BANG)[1]
member_type.append(subtype_basename)
cur_type = cur_type.subtype
return member_type
class NetfilterImp_to_4_3(AbstractNetfilter):
"""At this point, Netfilter hooks were implemented as a linked list of struct
'nf_hook_ops' type. One linked list per protocol per hook type.
It was like that until 4.2.8.
struct list_head nf_hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS];
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
return vmlinux.has_symbol("nf_hooks")
def get_net_namespaces(self):
# In kernels <= 4.2.8 netfilter hooks are not implemented per namespaces
netns, net = renderers.NotAvailableValue(), renderers.NotAvailableValue()
yield netns, net
def get_hooks_container(self, net, proto_name, hook_name):
nf_hooks = self.vmlinux.object_from_symbol("nf_hooks")
if not nf_hooks:
return
yield nf_hooks
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
list_head = hook_container[proto_idx][hook_idx]
nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops")
return list_head.to_list(nf_hooks_ops_name, "list")
class NetfilterImp_4_3_to_4_9(AbstractNetfilter):
"""Netfilter hooks were added to network namespaces in 4.3.
It is still implemented as a linked list of 'struct nf_hook_ops' type but inside a
network namespace. One linked list per protocol per hook type.
struct net { ... struct netns_nf nf; ... }
struct netns_nf { ...
struct list_head hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... }
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("netns_nf")
and vmlinux.get_type("netns_nf").has_member("hooks")
and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks")
== ["array", "array", "list_head"]
)
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
list_head = hook_container[proto_idx][hook_idx]
nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops")
return list_head.to_list(nf_hooks_ops_name, "list")
class NetfilterImp_4_9_to_4_14(AbstractNetfilter):
"""In this range of kernel versions, the doubly-linked lists of netfilter hooks were
replaced by an array of arrays of 'nf_hook_entry' pointers in a singly-linked lists.
struct net { ... struct netns_nf nf; ... }
struct netns_nf { ..
struct nf_hook_entry __rcu *hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... }
Also in v4.10 the struct nf_hook_entry changed, a hook function pointer was added to
it. However, for simplicity of this design, we will still take the hook address from
the 'nf_hook_ops'. As per v5.0-rc2, the hook address is duplicated in both sides.
- v4.9:
struct nf_hook_entry {
struct nf_hook_entry *next;
struct nf_hook_ops ops;
const struct nf_hook_ops *orig_ops; };
- v4.10:
struct nf_hook_entry {
struct nf_hook_entry *next;
nf_hookfn *hook;
void *priv;
const struct nf_hook_ops *orig_ops; };
(*) Even though the hook address is in the struct 'nf_hook_entry', we use the
original 'nf_hook_ops' hook address value, the one which was filled by the user, to
make it uniform to all the implementations.
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["array", "array", "pointer", "nf_hook_entry"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("netns_nf")
and vmlinux.get_type("netns_nf").has_member("hooks")
and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") == hooks_type
)
def _get_hook_ops(self, hook_container, proto_idx, hook_idx):
list_head = hook_container[proto_idx][hook_idx]
nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops")
return list_head.to_list(nf_hooks_ops_name, "list")
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hook_entry_list = hook_container[proto_idx][hook_idx]
while nf_hook_entry_list:
yield nf_hook_entry_list.orig_ops
nf_hook_entry_list = nf_hook_entry_list.next
class NetfilterImp_4_14_to_4_16(AbstractNetfilter):
"""'nf_hook_ops' was removed from struct 'nf_hook_entry'. Instead, it was stored
adjacent in memory to the 'nf_hook_entry' array, in the new struct 'nf_hook_entries'
However, 'orig_ops' is not part of the 'nf_hook_entries' struct definition. So, we
have to craft it by hand.
struct net { ... struct netns_nf nf; ... }
struct netns_nf {
struct nf_hook_entries *hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... }
struct nf_hook_entries {
u16 num_hook_entries; /* plus padding */
struct nf_hook_entry hooks[];
//const struct nf_hook_ops *orig_ops[]; }
struct nf_hook_entry {
nf_hookfn *hook;
void *priv; }
(*) Even though the hook address is in the struct 'nf_hook_entry', we use the
original 'nf_hook_ops' hook address value, the one which was filled by the user, to
make it uniform to all the implementations.
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["array", "array", "pointer", "nf_hook_entries"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("netns_nf")
and vmlinux.get_type("netns_nf").has_member("hooks")
and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") == hooks_type
)
def get_nf_hook_entries(self, nf_hooks_addr, proto_idx, hook_idx):
"""This allows to support different hook array implementations from this version
on. For instance, in kernels >= 4.16 this multi-dimensional array is split in
one-dimensional array of pointers to 'nf_hooks_entries' per each protocol."""
return nf_hooks_addr[proto_idx][hook_idx]
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hook_entries = self.get_nf_hook_entries(hook_container, proto_idx, hook_idx)
if not nf_hook_entries:
return
nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops")
nf_hook_ops_ptr_arr = self.build_nf_hook_ops_array(nf_hook_entries)
if not nf_hook_ops_ptr_arr:
return
for nf_hook_ops_ptr in nf_hook_ops_ptr_arr:
nf_hook_ops = nf_hook_ops_ptr.dereference().cast(nf_hook_ops_name)
yield nf_hook_ops
class NetfilterImp_4_16_to_latest(NetfilterImp_4_14_to_4_16):
"""The multidimensional array of nf_hook_entries was split in a one-dimensional
array per each protocol.
struct net {
struct netns_nf nf; ... }
struct netns_nf {
struct nf_hook_entries * hooks_ipv4[NF_INET_NUMHOOKS];
struct nf_hook_entries * hooks_ipv6[NF_INET_NUMHOOKS];
struct nf_hook_entries * hooks_arp[NF_ARP_NUMHOOKS];
struct nf_hook_entries * hooks_bridge[NF_INET_NUMHOOKS];
struct nf_hook_entries * hooks_decnet[NF_DN_NUMHOOKS]; ... }
struct nf_hook_entries {
u16 num_hook_entries; /* plus padding */
struct nf_hook_entry hooks[];
//const struct nf_hook_ops *orig_ops[]; }
struct nf_hook_entry {
nf_hookfn *hook;
void *priv; }
(*) Even though the hook address is in the struct nf_hook_entry, we use the original
nf_hook_ops hook address value, the one which was filled by the user, to make it
uniform to all the implementations.
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("netns_nf")
and vmlinux.get_type("netns_nf").has_member("hooks_ipv4")
)
def get_hooks_container(self, net, proto_name, hook_name):
try:
if proto_name == "IPV4":
net_nf_hooks = net.nf.hooks_ipv4
elif proto_name == "ARP":
net_nf_hooks = net.nf.hooks_arp
elif proto_name == "BRIDGE":
net_nf_hooks = net.nf.hooks_bridge
elif proto_name == "IPV6":
net_nf_hooks = net.nf.hooks_ipv6
elif proto_name == "DECNET":
net_nf_hooks = net.nf.hooks_decnet
else:
return
yield net_nf_hooks
except AttributeError:
# Protocol family disabled at kernel compilation
# CONFIG_NETFILTER_FAMILY_ARP=n ||
# CONFIG_NETFILTER_FAMILY_BRIDGE=n ||
# CONFIG_DECNET=n
pass
def _get_nf_hook_entries_ptr(self, nf_hooks_addr, proto_idx, hook_idx):
nf_hook_entries_ptr = nf_hooks_addr[hook_idx]
return nf_hook_entries_ptr
def get_nf_hook_entries(self, nf_hooks_addr, proto_idx, hook_idx):
return nf_hooks_addr[hook_idx]
class AbstractNetfilterNetDev(AbstractNetfilter):
"""Base class to handle the Netfilter NetDev hooks.
It won't be executed. It has some common functions to all Netfilter NetDev hook
implementations.
Netfilter NetDev hooks are set per network device which belongs to a network
namespace.
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
return False
def subscribed_protocols(self):
return ("NETDEV",)
def get_hooks_container(self, net, proto_name, hook_name):
net_device_type = self.vmlinux.get_type("net_device")
net_device_name = self.get_symbol_fullname("net_device")
for net_device in net.dev_base_head.to_list(net_device_name, "dev_list"):
if hook_name == "INGRESS":
if net_device_type.has_member("nf_hooks_ingress"):
# CONFIG_NETFILTER_INGRESS=y
yield net_device.nf_hooks_ingress
elif hook_name == "EGRESS":
if net_device_type.has_member("nf_hooks_egress"):
# CONFIG_NETFILTER_EGRESS=y
yield net_device.nf_hooks_egress
class NetfilterNetDevImp_4_2_to_4_9(AbstractNetfilterNetDev):
"""This is the first version of Netfilter Ingress hooks which was implemented using
a doubly-linked list of 'nf_hook_ops'.
struct list_head nf_hooks_ingress;
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["list_head"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("net_device")
and vmlinux.get_type("net_device").has_member("nf_hooks_ingress")
and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress")
== hooks_type
)
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hooks_ingress = hook_container
nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops")
return nf_hooks_ingress.to_list(nf_hook_ops_name, "list")
class NetfilterNetDevImp_4_9_to_4_14(AbstractNetfilterNetDev):
"""In 4.9 it was changed to a simple singly-linked list.
struct nf_hook_entry * nf_hooks_ingress;
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["pointer", "nf_hook_entry"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("net_device")
and vmlinux.get_type("net_device").has_member("nf_hooks_ingress")
and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress")
== hooks_type
)
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hooks_ingress_ptr = hook_container
if not nf_hooks_ingress_ptr:
return
while nf_hooks_ingress_ptr:
nf_hook_entry = nf_hooks_ingress_ptr.dereference()
orig_ops = nf_hook_entry.orig_ops.dereference()
yield orig_ops
nf_hooks_ingress_ptr = nf_hooks_ingress_ptr.next
class NetfilterNetDevImp_4_14_to_latest(AbstractNetfilterNetDev):
"""In 4.14 the hook list was converted to an array of pointers inside the struct
'nf_hook_entries':
struct nf_hook_entries * nf_hooks_ingress;
struct nf_hook_entries {
u16 num_hook_entries;
struct nf_hook_entry hooks[];
//const struct nf_hook_ops *orig_ops[]; }
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["pointer", "nf_hook_entries"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("net_device")
and vmlinux.get_type("net_device").has_member("nf_hooks_ingress")
and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress")
== hooks_type
)
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hook_entries = hook_container
if not nf_hook_entries:
return
nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops")
nf_hook_ops_ptr_arr = self.build_nf_hook_ops_array(nf_hook_entries)
if not nf_hook_ops_ptr_arr:
return
for nf_hook_ops_ptr in nf_hook_ops_ptr_arr:
nf_hook_ops = nf_hook_ops_ptr.dereference().cast(nf_hook_ops_name)
yield nf_hook_ops
class Netfilter(interfaces.plugins.PluginInterface):
"""Lists Netfilter hooks."""
_required_framework_version = (2, 22, 0)
_version = (2, 0, 0)
_required_linux_utilities_modules_version = (3, 0, 0)
_required_linuxutils_gatherers_version = (1, 0, 0)
_required_linuxnet_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherers",
component=linux_utilities_modules.ModuleGatherers,
version=cls._required_linuxutils_gatherers_version,
),
requirements.VersionRequirement(
name="linuxnet",
component=network.NetSymbols,
version=cls._required_linuxnet_version,
),
]
def _format_fields(self, fields):
(
netns,
proto_name,
hook_name,
priority,
hook_func,
module_info,
symbol_name,
hooked,
) = fields
if module_info:
module_name = module_info.name
else:
module_name = renderers.NotAvailableValue()
return (
netns,
proto_name,
hook_name,
priority,
format_hints.Hex(hook_func),
module_name,
symbol_name or renderers.NotAvailableValue(),
str(hooked),
)
def _generator(self):
kernel_module_name = self.config["kernel"]
for fields in AbstractNetfilter.run_all(
context=self.context, kernel_module_name=kernel_module_name
):
yield (0, self._format_fields(fields))
def run(self):
headers = [
("Net NS", int),
("Proto", str),
("Hook", str),
("Priority", int),
("Handler", format_hints.Hex),
("Module", str),
("Symbol", str),
("Is Hooked", str),
]
return renderers.TreeGrid(headers, self._generator())
volatility3/framework/plugins/linux/malware/tty_check.py
+119
View File
@@ -0,0 +1,119 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0
#
import logging
from typing import List
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, renderers, exceptions, constants
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.framework.symbols import linux
vollog = logging.getLogger(__name__)
class Tty_Check(plugins.PluginInterface):
"""Checks tty devices for hooks"""
_required_framework_version = (2, 0, 0)
_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_modules",
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)
),
]
def _generator(self):
vmlinux = self.context.modules[self.config["kernel"]]
try:
tty_drivers = vmlinux.object_from_symbol("tty_drivers").cast("list_head")
except exceptions.SymbolError:
tty_drivers = None
if not tty_drivers:
raise TypeError(
"This plugin requires the tty_drivers 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."
)
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"
):
try:
ttys = utility.array_of_pointers(
tty.ttys.dereference(),
count=tty.num,
subtype=vmlinux.symbol_table_name + constants.BANG + "tty_struct",
context=self.context,
)
except exceptions.PagedInvalidAddressException:
continue
for tty_dev in ttys:
if tty_dev == 0:
continue
try:
name = utility.array_to_string(tty_dev.name)
recv_buf = tty_dev.ldisc.ops.receive_buf
except exceptions.InvalidAddressException:
continue
module_info, symbol_name = (
linux_utilities_modules.Modules.module_lookup_by_address(
self.context, vmlinux.name, known_modules, recv_buf
)
)
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(
[
("Name", str),
("Address", format_hints.Hex),
("Module", str),
("Symbol", str),
],
self._generator(),
)
volatility3/framework/plugins/linux/modxview.py
+10 -170
View File
@@ -1,181 +1,21 @@
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
# 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
#
import logging
from typing import List, Dict, Iterator
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, deprecation, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.framework.symbols.linux import extensions
from volatility3.framework.constants import architectures
from volatility3.framework.symbols.linux.utilities import tainting
from volatility3.framework import interfaces, deprecation
from volatility3.plugins.linux.malware import modxview
vollog = logging.getLogger(__name__)
class Modxview(interfaces.plugins.PluginInterface):
class Modxview(
interfaces.plugins.PluginInterface,
deprecation.PluginRenameClass,
replacement_class=modxview.Modxview,
removal_date="2026-06-07",
):
"""Centralize lsmod, check_modules and hidden_modules results to efficiently \
spot modules presence and taints."""
spot modules presence and taints (deprecated)."""
_version = (1, 0, 0)
_required_framework_version = (2, 17, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=architectures.LINUX_ARCHS,
),
requirements.VersionRequirement(
name="linux_utilities_modules",
component=linux_utilities_modules.Modules,
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)
),
requirements.BooleanRequirement(
name="plain_taints",
description="Display the plain taints string for each module.",
optional=True,
default=False,
),
]
@classmethod
@deprecation.deprecated_method(
replacement=linux_utilities_modules.Modules.flatten_run_modules_results,
replacement_version=(3, 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]:
"""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:
Iterator of modules objects
"""
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=(3, 0, 0),
removal_date="2025-09-25",
)
def run_modules_scanners(
cls,
context: interfaces.context.ContextInterface,
kernel_name: str,
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."""
return linux_utilities_modules.Modules.run_modules_scanners(
context, kernel_name, run_hidden_modules
)
def _generator(self):
kernel = self.context.modules[self.config["kernel"]]
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_gatherers=wanted_gatherers,
flatten=False,
)
aggregated_modules = {}
# We want to be explicit on the plugins results we are interested in
for gatherer in wanted_gatherers:
# Iterate over each recovered module
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(gatherer.name)
else:
aggregated_modules[mod_info.offset] = [gatherer.name]
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,
self.config["kernel"],
module.taints,
True,
)
else:
taints = ",".join(
tainting.Tainting.get_taints_parsed(
self.context,
self.config["kernel"],
module.taints,
True,
)
)
yield (
0,
(
module.get_name() or renderers.NotAvailableValue(),
format_hints.Hex(module_offset),
linux_utilities_modules.ModuleGathererLsmod.name in gatherers,
linux_utilities_modules.ModuleGathererSysFs.name in gatherers,
linux_utilities_modules.ModuleGathererScanner.name in gatherers,
taints or renderers.NotAvailableValue(),
),
)
def run(self):
columns = [
("Name", str),
("Address", format_hints.Hex),
("In procfs", bool),
("In sysfs", bool),
("In scan", bool),
("Taints", str),
]
return renderers.TreeGrid(
columns,
self._generator(),
)
volatility3/framework/plugins/linux/netfilter.py
+11 -797
View File
@@ -1,806 +1,20 @@
# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0
# 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
#
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
import logging
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from typing import Iterator, List, Tuple, Optional
from volatility3 import framework
from volatility3.framework import (
constants,
interfaces,
renderers,
exceptions,
deprecation,
)
from volatility3.framework.renderers import format_hints
from volatility3.framework.configuration import requirements
from volatility3.framework.symbols.linux import network
from volatility3.framework import interfaces, deprecation
from volatility3.plugins.linux.malware import netfilter
vollog = logging.getLogger(__name__)
@dataclass
class Proto:
name: str
hooks: Tuple[str] = field(default_factory=tuple)
PROTO_NOT_IMPLEMENTED = Proto(name="UNSPEC")
NF_INET_HOOKS = ("PRE_ROUTING", "LOCAL_IN", "FORWARD", "LOCAL_OUT", "POST_ROUTING")
NF_DEC_HOOKS = (
"PRE_ROUTING",
"LOCAL_IN",
"FORWARD",
"LOCAL_OUT",
"POST_ROUTING",
"HELLO",
"ROUTE",
)
NF_ARP_HOOKS = ("IN", "OUT", "FORWARD")
NF_NETDEV_HOOKS = ("INGRESS", "EGRESS")
LARGEST_HOOK_NUMBER = max(
len(NF_INET_HOOKS), len(NF_DEC_HOOKS), len(NF_ARP_HOOKS), len(NF_NETDEV_HOOKS)
)
class AbstractNetfilter(ABC):
"""Netfilter Abstract Base Classes handling details across various
Netfilter implementations, including constants, helpers, and common
routines.
"""
PROTO_HOOKS = (
PROTO_NOT_IMPLEMENTED, # NFPROTO_UNSPEC
Proto(name="INET", hooks=NF_INET_HOOKS), # From kernels 3.14
Proto(name="IPV4", hooks=NF_INET_HOOKS),
Proto(name="ARP", hooks=NF_ARP_HOOKS),
PROTO_NOT_IMPLEMENTED,
Proto(name="NETDEV", hooks=NF_NETDEV_HOOKS),
PROTO_NOT_IMPLEMENTED,
Proto(name="BRIDGE", hooks=NF_INET_HOOKS),
PROTO_NOT_IMPLEMENTED,
PROTO_NOT_IMPLEMENTED,
Proto(name="IPV6", hooks=NF_INET_HOOKS),
PROTO_NOT_IMPLEMENTED,
Proto(name="DECNET", hooks=NF_DEC_HOOKS), # Removed in kernel 6.1
)
NF_MAX_HOOKS = LARGEST_HOOK_NUMBER + 1
def __init__(
self, context: interfaces.context.ContextInterface, kernel_module_name: str
):
self._context = context
self.vmlinux = context.modules[kernel_module_name]
self.layer_name = self.vmlinux.layer_name
# Set data sizes
self.ptr_size = self.vmlinux.get_type("pointer").size
self.list_head_size = self.vmlinux.get_type("list_head").size
linuxutils_modulegatherers_required_version = (
Netfilter._required_linuxutils_gatherers_version
)
linuxutils_modulegatherers_current_version = (
linux_utilities_modules.ModuleGatherers.version
)
if not requirements.VersionRequirement.matches_required(
linuxutils_modulegatherers_required_version,
linuxutils_modulegatherers_current_version,
):
raise exceptions.PluginRequirementException(
f"linux_utilities_modules.ModuleGatherer version not suitable: required {linuxutils_modulegatherers_required_version} found {linuxutils_modulegatherers_current_version}"
)
linux_net_required_version = Netfilter._required_linuxnet_version
linux_net_current_version = network.NetSymbols.version
if not requirements.VersionRequirement.matches_required(
linux_net_required_version, linux_net_current_version
):
raise exceptions.PluginRequirementException(
f"symbols.linux.net.NetSymbols version not suitable: required {linux_net_required_version} found {linux_net_current_version}"
)
linux_utilities_modules_required_version = (
Netfilter._required_linux_utilities_modules_version
)
linux_utilities_modules_current_version = (
linux_utilities_modules.Modules.version
)
if not requirements.VersionRequirement.matches_required(
linux_utilities_modules_required_version,
linux_utilities_modules_current_version,
):
raise exceptions.PluginRequirementException(
f"linux_utilities_modules.Modules version not suitable: required {linux_utilities_modules_required_version} found {linux_utilities_modules_current_version}"
)
symbol_table = context.symbol_space[self.vmlinux.symbol_table_name]
network.NetSymbols.apply(symbol_table)
self.handlers = linux_utilities_modules.Modules.run_modules_scanners(
context=context,
kernel_module_name=kernel_module_name,
caller_wanted_gatherers=linux_utilities_modules.ModuleGatherers.all_gatherers_identifier,
)
@classmethod
def run_all(
cls, context: interfaces.context.ContextInterface, kernel_module_name: str
) -> Iterator[Tuple[int, str, str, int, int, str, bool]]:
"""It calls each subclass symtab_checks() to test the required
conditions to that specific kernel implementation.
Args:
context: The volatility3 context on which to operate
kernel_module_name: The name of the table containing the kernel symbols
Yields:
The kmsg records. Same as _run()
"""
vmlinux = context.modules[kernel_module_name]
implementation_inst = None # type: ignore
for subclass in framework.class_subclasses(cls):
if not subclass.symtab_checks(vmlinux=vmlinux):
vollog.log(
constants.LOGLEVEL_VVVV,
"Netfilter implementation '%s' doesn't match this memory dump",
subclass.__name__,
)
continue
vollog.log(
constants.LOGLEVEL_VVVV,
"Netfilter implementation '%s' matches!",
subclass.__name__,
)
implementation_inst = subclass(
context=context, kernel_module_name=kernel_module_name
)
# More than one class could be executed for an specific kernel version
# For instance: Netfilter Ingress hooks
yield from implementation_inst._run()
if implementation_inst is None:
vollog.error("Unsupported Netfilter kernel implementation")
def _run(self) -> Iterator[Tuple[int, str, str, int, int, str, bool]]:
"""Iterates over namespaces and protocols, executing various callbacks that
allow customization of the code to the specific data structure used in a
particular kernel implementation
get_hooks_container(net, proto_name, hook_name)
It returns the data structure used in a specific kernel implementation
to store the hooks for a respective namespace and protocol, basically:
For Ingress hooks:
network_namespace[] -> net_device[] -> nf_hooks_ingress[]
For egress hooks:
network_namespace[] -> net_device[] -> nf_hooks_egress[]
For all the other Netfilter hooks:
<= 4.2.8
nf_hooks[]
>= 4.3
network_namespace[] -> nf.hooks[]
get_hook_ops(hook_container, proto_idx, hook_idx)
Give the 'hook_container' got in get_hooks_container(), it
returns an iterable of 'nf_hook_ops' elements for a respective protocol
and hook type.
Returns:
netns [int]: Network namespace id
proto_name [str]: Protocol name
hook_name [str]: Hook name
priority [int]: Priority
hook_ops_hook [int]: Hook address
module_name [str]: Linux kernel module name
hooked [bool]: "True" if the network stack has been hijacked
"""
for netns, net in self.get_net_namespaces():
for proto_idx, proto_name, hook_idx, hook_name in self._proto_hook_loop():
hooks_container = self.get_hooks_container(net, proto_name, hook_name)
for hook_container in hooks_container:
for hook_ops in self.get_hook_ops(
hook_container, proto_idx, hook_idx
):
if not hook_ops:
continue
priority = int(hook_ops.priority)
hook_ops_hook = hook_ops.hook
module_info, symbol_name = (
linux_utilities_modules.Modules.module_lookup_by_address(
self._context,
self.vmlinux.name,
self.handlers,
hook_ops_hook,
)
)
hooked = module_info is None
yield netns, proto_name, hook_name, priority, hook_ops_hook, module_info, symbol_name, hooked
@classmethod
@abstractmethod
def symtab_checks(cls, vmlinux: interfaces.context.ModuleInterface) -> bool:
"""This method on each sublasss will be called to evaluate if the kernel
being analyzed fulfill the type & symbols requirements for the implementation.
The first class returning True will be instantiated and called via the
run() method.
Returns:
bool: True if the kernel being analyzed fulfill the class requirements.
"""
def _proto_hook_loop(self) -> Iterator[Tuple[int, str, int, str]]:
"""Flattens the protocol families and hooks"""
for proto_idx, proto in enumerate(AbstractNetfilter.PROTO_HOOKS):
if proto == PROTO_NOT_IMPLEMENTED:
continue
if proto.name not in self.subscribed_protocols():
# This protocol is not managed in this object
continue
for hook_idx, hook_name in enumerate(proto.hooks):
yield proto_idx, proto.name, hook_idx, hook_name
def build_nf_hook_ops_array(
self, nf_hook_entries
) -> Optional[interfaces.objects.ObjectInterface]:
"""Function helper to build the nf_hook_ops array when it is not part of the
struct 'nf_hook_entries' definition.
nf_hook_ops was stored adjacent in memory to the nf_hook_entry array, in the
new struct 'nf_hook_entries'. However, this 'nf_hooks_ops' array 'orig_ops' is
not part of the 'nf_hook_entries' struct. So, we need to calculate the offset.
struct nf_hook_entries {
u16 num_hook_entries; /* plus padding */
struct nf_hook_entry hooks[];
//const struct nf_hook_ops *orig_ops[];
}
"""
nf_hook_entry_size = self.vmlinux.get_type("nf_hook_entry").size
try:
num_hook_entries = nf_hook_entries.num_hook_entries
except exceptions.InvalidAddressException:
return None
orig_ops_addr = (
nf_hook_entries.hooks.vol.offset + nf_hook_entry_size * num_hook_entries
)
if not self.vmlinux._context.layers[self.vmlinux.layer_name].is_valid(
orig_ops_addr
):
return None
orig_ops = self._context.object(
object_type=self.get_symbol_fullname("array"),
offset=orig_ops_addr,
subtype=self.vmlinux.get_type("pointer"),
layer_name=self.layer_name,
count=num_hook_entries,
)
return orig_ops
def subscribed_protocols(self) -> Tuple[str]:
"""Allows to select which PROTO_HOOKS protocols will be processed by the
Netfiler subclass.
"""
# Most implementation handlers respond to these protocols, except for
# the ingress hook, which specifically handles the 'NETDEV' protocol.
# However, there is no corresponding Netfilter hook implementation for
# the INET protocol in the kernel. AFAIU, this is used as
# 'NFPROTO_INET = NFPROTO_IPV4 || NFPROTO_IPV6'
# in other parts of the kernel source code.
return ("IPV4", "ARP", "BRIDGE", "IPV6", "DECNET")
@deprecation.method_being_removed(
removal_date="2025-09-25",
message="Callers to this method should adapt `linux_utilities_modules.Modules.run_module_scanners`",
)
def get_module_name_for_address(self, addr) -> str:
"""Helper to obtain the module and symbol name in the format needed for the
output of this plugin.
"""
module_name, symbol_name = (
linux_utilities_modules.Modules.lookup_module_address(
self._context, self.vmlinux.name, self.handlers, addr
)
)
if module_name == "UNKNOWN":
module_name = None
if symbol_name != "N/A":
module_name = f"[{symbol_name}]"
return module_name
def get_net_namespaces(self):
"""Common function to retrieve the different namespaces.
From 4.3 on, all the implementations use network namespaces.
"""
nethead = self.vmlinux.object_from_symbol("net_namespace_list")
symbol_net_name = self.get_symbol_fullname("net")
for net in nethead.to_list(symbol_net_name, "list"):
net_ns_id = net.ns.inum
yield net_ns_id, net
def get_hooks_container(self, net, proto_name, hook_name):
"""Returns the data structure used in a specific kernel implementation to store
the hooks for a respective namespace and protocol.
Except for kernels < 4.3, all the implementations use network namespaces.
Also the data structure which contains the hooks, even though it changes its
implementation and/or data type, it is always in this location.
"""
yield net.nf.hooks
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
"""Given the hook_container obtained from get_hooks_container(), it
returns an iterable of 'nf_hook_ops' elements for a corresponding protocol
and hook type.
This is the most variable/unstable part of all Netfilter hook designs, it
changes almost in every single implementation.
"""
raise NotImplementedError("You must implement this method")
def get_symbol_fullname(self, symbol_basename: str) -> str:
"""Given a short symbol or type name, it returns its full name"""
return self.vmlinux.symbol_table_name + constants.BANG + symbol_basename
@staticmethod
def get_member_type(
vol_type: interfaces.objects.Template, member_name: str
) -> List[str]:
"""Returns a list of types/subtypes belonging to the given type member.
Args:
vol_type (interfaces.objects.Template): A vol3 type object
member_name (str): The member name
Returns:
list: A list of types/subtypes
"""
_size, vol_obj = vol_type.vol.members[member_name]
type_name = vol_obj.type_name
type_basename = type_name.split(constants.BANG)[1]
member_type = [type_basename]
cur_type = vol_obj
while hasattr(cur_type, "subtype"):
subtype_name = cur_type.subtype.type_name
subtype_basename = subtype_name.split(constants.BANG)[1]
member_type.append(subtype_basename)
cur_type = cur_type.subtype
return member_type
class NetfilterImp_to_4_3(AbstractNetfilter):
"""At this point, Netfilter hooks were implemented as a linked list of struct
'nf_hook_ops' type. One linked list per protocol per hook type.
It was like that until 4.2.8.
struct list_head nf_hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS];
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
return vmlinux.has_symbol("nf_hooks")
def get_net_namespaces(self):
# In kernels <= 4.2.8 netfilter hooks are not implemented per namespaces
netns, net = renderers.NotAvailableValue(), renderers.NotAvailableValue()
yield netns, net
def get_hooks_container(self, net, proto_name, hook_name):
nf_hooks = self.vmlinux.object_from_symbol("nf_hooks")
if not nf_hooks:
return
yield nf_hooks
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
list_head = hook_container[proto_idx][hook_idx]
nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops")
return list_head.to_list(nf_hooks_ops_name, "list")
class NetfilterImp_4_3_to_4_9(AbstractNetfilter):
"""Netfilter hooks were added to network namespaces in 4.3.
It is still implemented as a linked list of 'struct nf_hook_ops' type but inside a
network namespace. One linked list per protocol per hook type.
struct net { ... struct netns_nf nf; ... }
struct netns_nf { ...
struct list_head hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... }
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("netns_nf")
and vmlinux.get_type("netns_nf").has_member("hooks")
and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks")
== ["array", "array", "list_head"]
)
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
list_head = hook_container[proto_idx][hook_idx]
nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops")
return list_head.to_list(nf_hooks_ops_name, "list")
class NetfilterImp_4_9_to_4_14(AbstractNetfilter):
"""In this range of kernel versions, the doubly-linked lists of netfilter hooks were
replaced by an array of arrays of 'nf_hook_entry' pointers in a singly-linked lists.
struct net { ... struct netns_nf nf; ... }
struct netns_nf { ..
struct nf_hook_entry __rcu *hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... }
Also in v4.10 the struct nf_hook_entry changed, a hook function pointer was added to
it. However, for simplicity of this design, we will still take the hook address from
the 'nf_hook_ops'. As per v5.0-rc2, the hook address is duplicated in both sides.
- v4.9:
struct nf_hook_entry {
struct nf_hook_entry *next;
struct nf_hook_ops ops;
const struct nf_hook_ops *orig_ops; };
- v4.10:
struct nf_hook_entry {
struct nf_hook_entry *next;
nf_hookfn *hook;
void *priv;
const struct nf_hook_ops *orig_ops; };
(*) Even though the hook address is in the struct 'nf_hook_entry', we use the
original 'nf_hook_ops' hook address value, the one which was filled by the user, to
make it uniform to all the implementations.
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["array", "array", "pointer", "nf_hook_entry"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("netns_nf")
and vmlinux.get_type("netns_nf").has_member("hooks")
and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") == hooks_type
)
def _get_hook_ops(self, hook_container, proto_idx, hook_idx):
list_head = hook_container[proto_idx][hook_idx]
nf_hooks_ops_name = self.get_symbol_fullname("nf_hook_ops")
return list_head.to_list(nf_hooks_ops_name, "list")
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hook_entry_list = hook_container[proto_idx][hook_idx]
while nf_hook_entry_list:
yield nf_hook_entry_list.orig_ops
nf_hook_entry_list = nf_hook_entry_list.next
class NetfilterImp_4_14_to_4_16(AbstractNetfilter):
"""'nf_hook_ops' was removed from struct 'nf_hook_entry'. Instead, it was stored
adjacent in memory to the 'nf_hook_entry' array, in the new struct 'nf_hook_entries'
However, 'orig_ops' is not part of the 'nf_hook_entries' struct definition. So, we
have to craft it by hand.
struct net { ... struct netns_nf nf; ... }
struct netns_nf {
struct nf_hook_entries *hooks[NFPROTO_NUMPROTO][NF_MAX_HOOKS]; ... }
struct nf_hook_entries {
u16 num_hook_entries; /* plus padding */
struct nf_hook_entry hooks[];
//const struct nf_hook_ops *orig_ops[]; }
struct nf_hook_entry {
nf_hookfn *hook;
void *priv; }
(*) Even though the hook address is in the struct 'nf_hook_entry', we use the
original 'nf_hook_ops' hook address value, the one which was filled by the user, to
make it uniform to all the implementations.
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["array", "array", "pointer", "nf_hook_entries"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("netns_nf")
and vmlinux.get_type("netns_nf").has_member("hooks")
and cls.get_member_type(vmlinux.get_type("netns_nf"), "hooks") == hooks_type
)
def get_nf_hook_entries(self, nf_hooks_addr, proto_idx, hook_idx):
"""This allows to support different hook array implementations from this version
on. For instance, in kernels >= 4.16 this multi-dimensional array is split in
one-dimensional array of pointers to 'nf_hooks_entries' per each protocol."""
return nf_hooks_addr[proto_idx][hook_idx]
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hook_entries = self.get_nf_hook_entries(hook_container, proto_idx, hook_idx)
if not nf_hook_entries:
return
nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops")
nf_hook_ops_ptr_arr = self.build_nf_hook_ops_array(nf_hook_entries)
if not nf_hook_ops_ptr_arr:
return
for nf_hook_ops_ptr in nf_hook_ops_ptr_arr:
nf_hook_ops = nf_hook_ops_ptr.dereference().cast(nf_hook_ops_name)
yield nf_hook_ops
class NetfilterImp_4_16_to_latest(NetfilterImp_4_14_to_4_16):
"""The multidimensional array of nf_hook_entries was split in a one-dimensional
array per each protocol.
struct net {
struct netns_nf nf; ... }
struct netns_nf {
struct nf_hook_entries * hooks_ipv4[NF_INET_NUMHOOKS];
struct nf_hook_entries * hooks_ipv6[NF_INET_NUMHOOKS];
struct nf_hook_entries * hooks_arp[NF_ARP_NUMHOOKS];
struct nf_hook_entries * hooks_bridge[NF_INET_NUMHOOKS];
struct nf_hook_entries * hooks_decnet[NF_DN_NUMHOOKS]; ... }
struct nf_hook_entries {
u16 num_hook_entries; /* plus padding */
struct nf_hook_entry hooks[];
//const struct nf_hook_ops *orig_ops[]; }
struct nf_hook_entry {
nf_hookfn *hook;
void *priv; }
(*) Even though the hook address is in the struct nf_hook_entry, we use the original
nf_hook_ops hook address value, the one which was filled by the user, to make it
uniform to all the implementations.
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("netns_nf")
and vmlinux.get_type("netns_nf").has_member("hooks_ipv4")
)
def get_hooks_container(self, net, proto_name, hook_name):
try:
if proto_name == "IPV4":
net_nf_hooks = net.nf.hooks_ipv4
elif proto_name == "ARP":
net_nf_hooks = net.nf.hooks_arp
elif proto_name == "BRIDGE":
net_nf_hooks = net.nf.hooks_bridge
elif proto_name == "IPV6":
net_nf_hooks = net.nf.hooks_ipv6
elif proto_name == "DECNET":
net_nf_hooks = net.nf.hooks_decnet
else:
return
yield net_nf_hooks
except AttributeError:
# Protocol family disabled at kernel compilation
# CONFIG_NETFILTER_FAMILY_ARP=n ||
# CONFIG_NETFILTER_FAMILY_BRIDGE=n ||
# CONFIG_DECNET=n
pass
def _get_nf_hook_entries_ptr(self, nf_hooks_addr, proto_idx, hook_idx):
nf_hook_entries_ptr = nf_hooks_addr[hook_idx]
return nf_hook_entries_ptr
def get_nf_hook_entries(self, nf_hooks_addr, proto_idx, hook_idx):
return nf_hooks_addr[hook_idx]
class AbstractNetfilterNetDev(AbstractNetfilter):
"""Base class to handle the Netfilter NetDev hooks.
It won't be executed. It has some common functions to all Netfilter NetDev hook
implementations.
Netfilter NetDev hooks are set per network device which belongs to a network
namespace.
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
return False
def subscribed_protocols(self):
return ("NETDEV",)
def get_hooks_container(self, net, proto_name, hook_name):
net_device_type = self.vmlinux.get_type("net_device")
net_device_name = self.get_symbol_fullname("net_device")
for net_device in net.dev_base_head.to_list(net_device_name, "dev_list"):
if hook_name == "INGRESS":
if net_device_type.has_member("nf_hooks_ingress"):
# CONFIG_NETFILTER_INGRESS=y
yield net_device.nf_hooks_ingress
elif hook_name == "EGRESS":
if net_device_type.has_member("nf_hooks_egress"):
# CONFIG_NETFILTER_EGRESS=y
yield net_device.nf_hooks_egress
class NetfilterNetDevImp_4_2_to_4_9(AbstractNetfilterNetDev):
"""This is the first version of Netfilter Ingress hooks which was implemented using
a doubly-linked list of 'nf_hook_ops'.
struct list_head nf_hooks_ingress;
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["list_head"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("net_device")
and vmlinux.get_type("net_device").has_member("nf_hooks_ingress")
and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress")
== hooks_type
)
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hooks_ingress = hook_container
nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops")
return nf_hooks_ingress.to_list(nf_hook_ops_name, "list")
class NetfilterNetDevImp_4_9_to_4_14(AbstractNetfilterNetDev):
"""In 4.9 it was changed to a simple singly-linked list.
struct nf_hook_entry * nf_hooks_ingress;
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["pointer", "nf_hook_entry"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("net_device")
and vmlinux.get_type("net_device").has_member("nf_hooks_ingress")
and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress")
== hooks_type
)
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hooks_ingress_ptr = hook_container
if not nf_hooks_ingress_ptr:
return
while nf_hooks_ingress_ptr:
nf_hook_entry = nf_hooks_ingress_ptr.dereference()
orig_ops = nf_hook_entry.orig_ops.dereference()
yield orig_ops
nf_hooks_ingress_ptr = nf_hooks_ingress_ptr.next
class NetfilterNetDevImp_4_14_to_latest(AbstractNetfilterNetDev):
"""In 4.14 the hook list was converted to an array of pointers inside the struct
'nf_hook_entries':
struct nf_hook_entries * nf_hooks_ingress;
struct nf_hook_entries {
u16 num_hook_entries;
struct nf_hook_entry hooks[];
//const struct nf_hook_ops *orig_ops[]; }
"""
@classmethod
def symtab_checks(cls, vmlinux) -> bool:
hooks_type = ["pointer", "nf_hook_entries"]
return (
vmlinux.has_symbol("net_namespace_list")
and vmlinux.has_type("net_device")
and vmlinux.get_type("net_device").has_member("nf_hooks_ingress")
and cls.get_member_type(vmlinux.get_type("net_device"), "nf_hooks_ingress")
== hooks_type
)
def get_hook_ops(self, hook_container, proto_idx, hook_idx):
nf_hook_entries = hook_container
if not nf_hook_entries:
return
nf_hook_ops_name = self.get_symbol_fullname("nf_hook_ops")
nf_hook_ops_ptr_arr = self.build_nf_hook_ops_array(nf_hook_entries)
if not nf_hook_ops_ptr_arr:
return
for nf_hook_ops_ptr in nf_hook_ops_ptr_arr:
nf_hook_ops = nf_hook_ops_ptr.dereference().cast(nf_hook_ops_name)
yield nf_hook_ops
class Netfilter(interfaces.plugins.PluginInterface):
"""Lists Netfilter hooks."""
_required_framework_version = (2, 22, 0)
class Netfilter(
interfaces.plugins.PluginInterface,
deprecation.PluginRenameClass,
replacement_class=netfilter.Netfilter,
removal_date="2026-06-07",
):
"""Lists Netfilter hooks (deprecated)."""
_version = (2, 0, 0)
_required_linux_utilities_modules_version = (3, 0, 0)
_required_linuxutils_gatherers_version = (1, 0, 0)
_required_linuxnet_version = (1, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_module_gatherers",
component=linux_utilities_modules.ModuleGatherers,
version=cls._required_linuxutils_gatherers_version,
),
requirements.VersionRequirement(
name="linuxnet",
component=network.NetSymbols,
version=cls._required_linuxnet_version,
),
]
def _format_fields(self, fields):
(
netns,
proto_name,
hook_name,
priority,
hook_func,
module_info,
symbol_name,
hooked,
) = fields
if module_info:
module_name = module_info.name
else:
module_name = renderers.NotAvailableValue()
return (
netns,
proto_name,
hook_name,
priority,
format_hints.Hex(hook_func),
module_name,
symbol_name or renderers.NotAvailableValue(),
str(hooked),
)
def _generator(self):
kernel_module_name = self.config["kernel"]
for fields in AbstractNetfilter.run_all(
context=self.context, kernel_module_name=kernel_module_name
):
yield (0, self._format_fields(fields))
def run(self):
headers = [
("Net NS", int),
("Proto", str),
("Hook", str),
("Priority", int),
("Handler", format_hints.Hex),
("Module", str),
("Symbol", str),
("Is Hooked", str),
]
return renderers.TreeGrid(headers, self._generator())
_required_framework_version = (2, 22, 0)
volatility3/framework/plugins/linux/tty_check.py
+11 -109
View File
@@ -1,118 +1,20 @@
# This file is Copyright 2020 Volatility Foundation and licensed under the Volatility Software License 1.0
# 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
#
import logging
from typing import List
import volatility3.framework.symbols.linux.utilities.modules as linux_utilities_modules
from volatility3.framework import interfaces, renderers, exceptions, constants
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.framework.symbols import linux
from volatility3.framework import interfaces, deprecation
from volatility3.plugins.linux.malware import tty_check as ttycheck
vollog = logging.getLogger(__name__)
class tty_check(plugins.PluginInterface):
"""Checks tty devices for hooks"""
class tty_check(
interfaces.plugins.PluginInterface,
deprecation.PluginRenameClass,
replacement_class=ttycheck.Tty_Check,
removal_date="2026-06-07",
):
"""Checks tty devices for hooks (deprecated)."""
_required_framework_version = (2, 0, 0)
@classmethod
def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]:
return [
requirements.ModuleRequirement(
name="kernel",
description="Linux kernel",
architectures=["Intel32", "Intel64"],
),
requirements.VersionRequirement(
name="linux_utilities_modules",
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)
),
]
def _generator(self):
vmlinux = self.context.modules[self.config["kernel"]]
try:
tty_drivers = vmlinux.object_from_symbol("tty_drivers").cast("list_head")
except exceptions.SymbolError:
tty_drivers = None
if not tty_drivers:
raise TypeError(
"This plugin requires the tty_drivers 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."
)
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"
):
try:
ttys = utility.array_of_pointers(
tty.ttys.dereference(),
count=tty.num,
subtype=vmlinux.symbol_table_name + constants.BANG + "tty_struct",
context=self.context,
)
except exceptions.PagedInvalidAddressException:
continue
for tty_dev in ttys:
if tty_dev == 0:
continue
try:
name = utility.array_to_string(tty_dev.name)
recv_buf = tty_dev.ldisc.ops.receive_buf
except exceptions.InvalidAddressException:
continue
module_info, symbol_name = (
linux_utilities_modules.Modules.module_lookup_by_address(
self.context, vmlinux.name, known_modules, recv_buf
)
)
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(
[
("Name", str),
("Address", format_hints.Hex),
("Module", str),
("Symbol", str),
],
self._generator(),
)
_version = (1, 0, 0)