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

Merge pull request #1599 from gcmoreira/linux_kallsyms

Browse Source 
Linux - Introduce the Linux kallsyms API and related plugins
This commit is contained in:
ikelos
2025-02-02 23:32:01 +00:00
committed by GitHub
parent 9a01f8307f cef87e014c
commit b3760ce7e5
8 changed files with 2379 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/test_volatility.py
+41
View File
@@ -842,6 +842,47 @@ def test_linux_hidden_modules(image, volatility, python):
assert out.count(b"\n") >= 4
def test_linux_kallsyms(image, volatility, python):
rc, out, _err = runvol_plugin(
"linux.kallsyms.Kallsyms",
image,
volatility,
python,
pluginargs=["--modules"],
)
# linux-sample-1.bin has no hidden modules.
# This validates that plugin requirements are met and exceptions are not raised.
assert rc == 0
assert out.count(b"\n") > 1000
# Addr Type Size Exported SubSystem ModuleName SymbolName Description
# 0xffffa009eba9 t 28 False module usbcore usb_mon_register Symbol is in the text (code) section
assert re.search(
rb"0xffffa009eba9\s+t\s+28\s+False\s+module\s+usbcore\s+usb_mon_register\s+Symbol is in the text \(code\) section",
out,
)
def test_linux_pscallstack(image, volatility, python):
rc, out, _err = runvol_plugin(
"linux.pscallstack.PsCallStack",
image,
volatility,
python,
pluginargs=["--pid", "1"],
)
assert rc == 0
assert out.count(b"\n") > 30
# TID Comm Position Address Value Name Type Module
# 1 init 39 0x88001f999a40 0xffff81109039 do_select T kernel
assert re.search(
rb"1\s+init\s+39\s+0x88001f999a40.*?0xffff81109039\s+do_select\s+T\s+kernel",
out,
)
# MAC
volatility3/framework/constants/_version.py
+1 -1
View File
@@ -1,6 +1,6 @@
# We use the SemVer 2.0.0 versioning scheme
VERSION_MAJOR = 2 # Number of releases of the library with a breaking change
VERSION_MINOR = 19 # Number of changes that only add to the interface
VERSION_MINOR = 20 # Number of changes that only add to the interface
VERSION_PATCH = 0 # Number of changes that do not change the interface
VERSION_SUFFIX = ""
volatility3/framework/constants/linux/__init__.py
+21
View File
@@ -356,6 +356,27 @@ MODULE_MINIMUM_SIZE = 4096
# Kallsyms
KSYM_NAME_LEN = 512
NM_TYPES_DESC = {
"a": "Symbol is absolute and doesn't change during linking",
"b": "Symbol in the BSS section, typically holding zero-initialized or uninitialized data",
"c": "Symbol is common, typically holding uninitialized data",
"d": "Symbol is in the initialized data section",
"g": "Symbol is in an initialized data section for small objects",
"i": "Symbol is an indirect reference to another symbol",
"N": "Symbol is a debugging symbol",
"n": "Symbol is in a non-data, non-code, non-debug read-only section",
"p": "Symbol is in a stack unwind section",
"r": "Symbol is in a read only data section",
"s": "Symbol is in an uninitialized or zero-initialized data section for small objects",
"t": "Symbol is in the text (code) section",
"U": "Symbol is undefined",
"u": "Symbol is a unique global symbol",
"V": "Symbol is a weak object, with a default value",
"v": "Symbol is a weak object",
"W": "Symbol is a weak symbol but not marked as a weak object symbol, with a default value",
"w": "Symbol is a weak symbol but not marked as a weak object symbol",
"?": "Symbol type is unknown",
}
# VMCOREINFO
VMCOREINFO_MAGIC = b"VMCOREINFO\x00"
volatility3/framework/plugins/linux/kallsyms.py
+138
View File
@@ -0,0 +1,138 @@
# 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, Union
from volatility3.framework import interfaces, renderers
from volatility3.framework.interfaces import plugins
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints
from volatility3.framework.constants import architectures
from volatility3.framework.symbols.linux import kallsyms
vollog = logging.getLogger(__name__)
class Kallsyms(plugins.PluginInterface):
"""Kallsyms symbols enumeration plugin.
If no arguments are provided, all symbols are included: core, modules, ftrace, and BPF.
Alternatively, you can use any combination of --core, --modules, --ftrace, and --bpf
to customize the output.
"""
_required_framework_version = (2, 19, 0)
_version = (1, 0, 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="Kallsyms", component=kallsyms.Kallsyms, version=(1, 0, 0)
),
requirements.BooleanRequirement(
name="core",
description="Include core symbols",
default=False,
optional=True,
),
requirements.BooleanRequirement(
name="modules",
description="Include module symbols",
default=False,
optional=True,
),
requirements.BooleanRequirement(
name="ftrace",
description="Include ftrace symbols",
default=False,
optional=True,
),
requirements.BooleanRequirement(
name="bpf",
description="Include BPF symbols",
default=False,
optional=True,
),
]
def _get_symbol_size(
self, kassymbol: kallsyms.KASSymbol
) -> Union[int, interfaces.renderers.BaseAbsentValue]:
# Symbol sizes are calculated using the address of the next non-aliased
# symbol or the end of the kernel text area _end/_etext. However, some kernel
# symbols live beyond that area. For these symbols, the size will be negative,
# resulting in incorrect values. Unfortunately, there isn't much that can be done
# in such cases.
# See comments on .init.scratch in arch/x86/kernel/vmlinux.lds.S for details
return kassymbol.size if kassymbol.size >= 0 else renderers.NotAvailableValue()
def _generator(self):
module_name = self.config["kernel"]
vmlinux = self.context.modules[module_name]
kas = kallsyms.Kallsyms(
context=self.context,
layer_name=vmlinux.layer_name,
module_name=module_name,
)
include_core = self.config.get("core", False)
include_modules = self.config.get("modules", False)
include_ftrace = self.config.get("ftrace", False)
include_bpf = self.config.get("bpf", False)
symbols_flags = (include_core, include_modules, include_ftrace, include_bpf)
if not any(symbols_flags):
include_core = include_modules = include_ftrace = include_bpf = True
symbol_generators = []
if include_core:
symbol_generators.append(kas.get_core_symbols())
if include_modules:
symbol_generators.append(kas.get_modules_symbols())
if include_ftrace:
symbol_generators.append(kas.get_ftrace_symbols())
if include_bpf:
symbol_generators.append(kas.get_bpf_symbols())
for symbols_generator in symbol_generators:
for kassymbol in symbols_generator:
# Symbol sizes are calculated using the address of the next non-aliased
# symbol or the end of the kernel text area _end/_etext. However, some kernel
# symbols are located beyond that area, which causes this method to fail for
# the last symbol, resulting in a negative size.
# See comments on .init.scratch in arch/x86/kernel/vmlinux.lds.S for details
symbol_size = self._get_symbol_size(kassymbol)
fields = (
format_hints.Hex(kassymbol.address),
kassymbol.type,
symbol_size,
kassymbol.exported,
kassymbol.subsystem,
kassymbol.module_name,
kassymbol.name,
kassymbol.type_description or renderers.NotAvailableValue(),
)
yield 0, fields
def run(self):
headers = [
("Addr", format_hints.Hex),
("Type", str),
("Size", int),
("Exported", bool),
("SubSystem", str),
("ModuleName", str),
("SymbolName", str),
("Description", str),
]
return renderers.TreeGrid(headers, self._generator())
volatility3/framework/plugins/linux/pscallstack.py
+198
View File
@@ -0,0 +1,198 @@
# 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
import dataclasses
from typing import List, Iterator
from volatility3.framework import interfaces, renderers, exceptions
from volatility3.framework.configuration import requirements
from volatility3.framework.interfaces import plugins
from volatility3.framework.renderers import format_hints
from volatility3.framework.constants import architectures
from volatility3.framework.objects import utility
from volatility3.framework.symbols.linux import kallsyms
from volatility3.plugins.linux import pslist
vollog = logging.getLogger(__name__)
@dataclasses.dataclass
class StackEntry:
position: int
address: int
value: int
name: str = renderers.NotAvailableValue()
type: str = renderers.NotAvailableValue()
module: str = renderers.NotAvailableValue()
class PsCallStack(plugins.PluginInterface):
"""Enumerates the call stack of each task"""
_required_framework_version = (2, 19, 0)
_version = (1, 0, 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="Kallsyms", component=kallsyms.Kallsyms, version=(1, 0, 0)
),
requirements.PluginRequirement(
name="pslist", plugin=pslist.PsList, version=(4, 0, 0)
),
requirements.ListRequirement(
name="pid",
description="Filter on specific process IDs",
element_type=int,
optional=True,
),
requirements.BooleanRequirement(
name="unresolved",
description="Include unresolved stack values",
default=False,
optional=True,
),
]
@classmethod
def get_task_callstack(
cls,
context: interfaces.context.ContextInterface,
module_name: str,
task: interfaces.objects.ObjectInterface,
kas: kallsyms.Kallsyms = None,
include_unresolved=False,
) -> Iterator[StackEntry]:
"""Retrieves the call stack for a given task
Args:
context: The context used to access memory layers and symbols
module_name: The name of the kernel module on which to operate
task: The task object whose stack is being retrieved
kas: Kallsyms instance for symbol resolution. If not provided or None, a new
instance will be created each time
include_unresolved: If True, includes stack values that could not be resolved
to known symbols. Defaults to False.
Yields:
StackEntry objects
"""
task_layer = task.get_address_space_layer()
if not task_layer:
return None
vmlinux = context.modules[module_name]
vmlinux_layer = context.layers[vmlinux.layer_name]
if not kas:
kas = kallsyms.Kallsyms(
context=context,
layer_name=vmlinux.layer_name,
module_name=module_name,
)
pointer_size = vmlinux.get_type("pointer").size
thread_size_order = 2 # Safe since kernel 3.15
# thread_size_order +=1 # If CONFIG_KASAN is enabled in kernels >= 4.0, default: DISABLED
# thread_size_order +=1 # If CONFIG_KASAN_EXTRA is enabled in kernels >= 4.19, default: DISABLED
thread_size = vmlinux_layer.page_size << thread_size_order
task_base_of_stack = vmlinux_layer.canonicalize(task.stack)
task_top_of_stack = task_base_of_stack + thread_size
byte_order = task.files.vol.data_format.byteorder
rsp_start = task.thread.sp
if not (task_base_of_stack <= rsp_start < task_top_of_stack):
raise exceptions.VolatilityException(
f"Invalid stack pointer {rsp_start:#x} for task {task.pid}"
)
current_sp = rsp_start
idx = 0
while current_sp < task_top_of_stack:
stack_value_bytes = task_layer.read(current_sp, pointer_size)
stack_value = int.from_bytes(stack_value_bytes, byteorder=byte_order)
kassymbol = kas.lookup_address(stack_value)
sp_address = current_sp & vmlinux_layer.address_mask
stack_value &= vmlinux_layer.address_mask
if kassymbol:
module_name = kassymbol.module_name or renderers.NotAvailableValue()
yield StackEntry(
position=idx,
address=sp_address,
value=stack_value,
name=kassymbol.name,
type=kassymbol.type,
module=module_name,
)
elif include_unresolved:
yield StackEntry(
position=idx,
address=sp_address,
value=stack_value,
)
idx += 1
current_sp += pointer_size
def _generator(self):
module_name = self.config["kernel"]
vmlinux = self.context.modules[module_name]
kas = kallsyms.Kallsyms(
context=self.context,
layer_name=vmlinux.layer_name,
module_name=self.config["kernel"],
)
include_unresolved = self.config.get("unresolved", False)
pids = self.config.get("pid", None)
filter_func = pslist.PsList.create_pid_filter(pids)
for task in pslist.PsList.list_tasks(
self.context, vmlinux.name, filter_func=filter_func, include_threads=True
):
task_name = utility.array_to_string(task.comm)
for stack_entry in self.get_task_callstack(
context=self.context,
module_name=vmlinux.name,
task=task,
kas=kas,
include_unresolved=include_unresolved,
):
fields = (
task.pid,
task_name,
stack_entry.position,
format_hints.Hex(stack_entry.address),
format_hints.Hex(stack_entry.value),
stack_entry.name,
stack_entry.type,
stack_entry.module,
)
yield 0, fields
def run(self):
return renderers.TreeGrid(
[
("TID", int),
("Comm", str),
("Position", int),
("Address", format_hints.Hex),
("Value", format_hints.Hex),
("Name", str),
("Type", str),
("Module", str),
],
self._generator(),
)
volatility3/framework/symbols/linux/__init__.py
+3
View File
@@ -87,6 +87,9 @@ class LinuxKernelIntermedSymbols(intermed.IntermediateSymbolTable):
# Only found in 6.1+ kernels
self.optional_set_type_class("maple_tree", extensions.maple_tree)
self.optional_set_type_class("latch_tree_root", extensions.latch_tree_root)
self.optional_set_type_class("kernel_symbol", extensions.kernel_symbol)
class LinuxUtilities(interfaces.configuration.VersionableInterface):
"""Class with multiple useful linux functions."""
volatility3/framework/symbols/linux/extensions/__init__.py
+309 -13
View File
@@ -10,7 +10,17 @@ import binascii
import stat
import datetime
import socket as socket_module
from typing import Generator, Iterable, Iterator, Optional, Tuple, List, Union, Dict
from typing import (
Generator,
Iterable,
Iterator,
Optional,
Tuple,
List,
Union,
Dict,
Callable,
)
from volatility3.framework import constants, exceptions, objects, interfaces, symbols
from volatility3.framework.renderers import conversion
@@ -166,37 +176,43 @@ class module(generic.GenericIntelProcess):
"""Get the name of the module as a string"""
return utility.array_to_string(self.name)
def _get_sect_count(self, grp):
def _get_sect_count(self, grp: interfaces.objects.ObjectInterface) -> int:
"""Try to determine the number of valid sections"""
symbol_table_name = self.get_symbol_table_name()
arr = self._context.object(
self.get_symbol_table_name() + constants.BANG + "array",
symbol_table_name + constants.BANG + "array",
layer_name=self.vol.layer_name,
offset=grp.attrs,
subtype=self._context.symbol_space.get_type(
self.get_symbol_table_name() + constants.BANG + "pointer"
symbol_table_name + constants.BANG + "pointer"
),
count=25,
)
idx = 0
while arr[idx]:
while arr[idx] and arr[idx].is_readable():
idx = idx + 1
return idx
def get_sections(self):
"""Get sections of the module"""
@functools.cached_property
def number_of_sections(self) -> int:
if self.sect_attrs.has_member("nsections"):
num_sects = self.sect_attrs.nsections
else:
num_sects = self._get_sect_count(self.sect_attrs.grp)
return self.sect_attrs.nsections
return self._get_sect_count(self.sect_attrs.grp)
def get_sections(self) -> Iterable[interfaces.objects.ObjectInterface]:
"""Get a list of section attributes for the given module."""
symbol_table_name = self.get_symbol_table_name()
arr = self._context.object(
self.get_symbol_table_name() + constants.BANG + "array",
symbol_table_name + constants.BANG + "array",
layer_name=self.vol.layer_name,
offset=self.sect_attrs.attrs.vol.offset,
subtype=self._context.symbol_space.get_type(
self.get_symbol_table_name() + constants.BANG + "module_sect_attr"
symbol_table_name + constants.BANG + "module_sect_attr"
),
count=num_sects,
count=self.number_of_sections,
)
yield from arr
@@ -254,6 +270,43 @@ class module(generic.GenericIntelProcess):
sym_address = elf_sym_obj.st_value & layer.address_mask
yield (sym_name, sym_address)
@functools.lru_cache
def get_module_address_boundaries(self) -> Tuple[int, int]:
"""Return the module address boundaries based on its symbol addresses"""
if not self.section_strtab or self.num_symtab < 1:
return None
elf_table_name = self.get_elf_table_name()
symbol_table_name = self.get_symbol_table_name()
is_64bit = symbols.symbol_table_is_64bit(self._context, symbol_table_name)
sym_name = "Elf64_Sym" if is_64bit else "Elf32_Sym"
sym_type = self._context.symbol_space.get_type(
elf_table_name + constants.BANG + sym_name
)
elf_syms = self._context.object(
symbol_table_name + constants.BANG + "array",
layer_name=self.vol.layer_name,
offset=self.section_symtab,
subtype=sym_type,
count=self.num_symtab,
)
# They should be sorted, but just in case
elf_syms_sorted = sorted(elf_syms, key=lambda x: x.st_value)
layer = self._context.layers[self.vol.layer_name]
# The first elf_sym is null
first_symbol = elf_syms_sorted[1]
last_symbol = elf_syms_sorted[-1]
minimum_address = first_symbol.st_value & layer.address_mask
maximum_address = (
last_symbol.st_value & layer.address_mask + last_symbol.st_size
)
return minimum_address, maximum_address
def get_symbol(self, wanted_sym_name) -> Optional[int]:
"""Get symbol address for a given symbol name"""
for sym_name, sym_address in self.get_symbols_names_and_addresses():
@@ -299,6 +352,38 @@ class module(generic.GenericIntelProcess):
raise AttributeError("Unable to get strtab")
@property
def section_typetab(self):
if self.has_member("kallsyms") and self.kallsyms.has_member("typetab"):
# kernels >= 4.5 8244062ef1e54502ef55f54cced659913f244c3e: kallsyms was added
# kernels >= 5.2 1c7651f43777cdd59c1aaa82c87324d3e7438c7b: types have its own array
return self.kallsyms.typetab
raise AttributeError("Unable to get typetab section, it needs a kernel >= 5.2")
def get_symbol_type(
self, symbol: interfaces.objects.ObjectInterface, symbol_index: int
) -> str:
"""Determines the type of a given ELF symbol.
Args:
symbol: The ELF symbol object (elf_sym)
symbol_index: The index of the symbol within the type table
Returns:
A single-character string representing the symbol type
"""
if self.has_member("kallsyms") and self.kallsyms.has_member("typetab"):
# kernels >= 5.2 1c7651f43777cdd59c1aaa82c87324d3e7438c7b types have its own array
layer = self._context.layers[self.vol.layer_name]
sym_type = layer.read(self.section_typetab + symbol_index, 1)
sym_type = sym_type.decode("utf-8", errors="ignore")
else:
# kernels < 5.2 the type was stored in the st_info
sym_type = chr(symbol.st_info)
return sym_type
class task_struct(generic.GenericIntelProcess):
def is_valid(self) -> bool:
@@ -370,6 +455,19 @@ class task_struct(generic.GenericIntelProcess):
self._context, dtb, config_prefix, preferred_name
)
def get_address_space_layer(
self,
) -> Optional[interfaces.layers.TranslationLayerInterface]:
"""Returns the task layer for this task's address space."""
task_layer_name = (
self.vol.layer_name if self.is_kernel_thread else self.add_process_layer()
)
if not task_layer_name:
return None
return self._context.layers[task_layer_name]
def get_process_memory_sections(
self, heap_only: bool = False
) -> Generator[Tuple[int, int], None, None]:
@@ -480,6 +578,15 @@ class task_struct(generic.GenericIntelProcess):
else None
)
@property
def state(self):
if self.has_member("__state"):
return self.member("__state")
elif self.has_member("state"):
return self.member("state")
else:
raise AttributeError("Unsupported task_struct: Cannot find state")
def _get_task_start_time(self) -> datetime.timedelta:
"""Returns the task's monotonic start_time as a timedelta.
@@ -2055,6 +2162,10 @@ class xdp_sock(objects.StructType):
class bpf_prog(objects.StructType):
_BPF_PROG_CHUNK_SHIFT = 6
_BPF_PROG_CHUNK_SIZE = 1 << _BPF_PROG_CHUNK_SHIFT
_BPF_PROG_CHUNK_MASK = ~(_BPF_PROG_CHUNK_SIZE - 1)
def get_type(self) -> Union[str, None]:
"""Returns a string with the eBPF program type"""
@@ -2099,6 +2210,58 @@ class bpf_prog(objects.StructType):
return self.aux.get_name()
def bpf_jit_binary_hdr_address(self) -> int:
"""Return the jitted BPF program start address
Based on bpf_jit_binary_hdr()
Returns:
The BPF program address
"""
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
# In 5.18 (33c9805860e584b194199cab1a1e81f4e6395408) <= kernels < 6.0 (1d5f82d9dd477d5c66e0214a68c3e4f308eadd6d)
# 'bpf_prog_aux' has a 'use_bpf_prog_pack' member
bpf_prog_aux_has_use_bpf_prog_pack = vmlinux.get_type(
"bpf_prog_aux"
).has_member("use_bpf_prog_pack")
if bpf_prog_aux_has_use_bpf_prog_pack and self.aux.use_bpf_prog_pack:
long_mask = (1 << vmlinux_layer.bits_per_register) - 1
addr_mask = self._BPF_PROG_CHUNK_MASK & long_mask
else:
addr_mask = vmlinux_layer.page_mask
real_start = self.bpf_func
return real_start & addr_mask
def get_address_region(self) -> Tuple[int, int]:
"""Returns the start and end memory addresses of the BPF program.
Based on bpf_get_prog_addr_region()
Returns:
A tuple with the addresses representing the memory range (start, end) of the BPF program.
"""
vmlinux = linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
vmlinux_layer = vmlinux.context.layers[vmlinux.layer_name]
# Based on bpf_get_prog_addr_region()
bpf_start_address = self.bpf_jit_binary_hdr_address()
if vmlinux.has_type("bpf_binary_header"):
# kernels >= 3.11 314beb9bcabfd6b4542ccbced2402af2c6f6142a
bpf_binary_header = vmlinux.object(
object_type="bpf_binary_header", offset=bpf_start_address, absolute=True
)
pages = bpf_binary_header.pages
else:
# kernels < 3.11 The first member is always the size
pages = vmlinux.object(
object_type="unsigned int", offset=bpf_start_address, absolute=True
)
bpf_end_address = bpf_start_address + pages * vmlinux_layer.page_size
return bpf_start_address, bpf_end_address
class bpf_prog_aux(objects.StructType):
def get_name(self) -> Union[str, None]:
@@ -2977,3 +3140,136 @@ class scatterlist(objects.StructType):
physical_layer = self._context.layers[physical_layer_name]
for sg in self.for_each_sg():
yield from physical_layer.read(sg.dma_address, sg._sg_dma_len())
class latch_tree_root(objects.StructType):
"""Latched RB-trees implementation"""
@functools.cached_property
def _vmlinux(self):
return linux.LinuxUtilities.get_module_from_volobj_type(self._context, self)
@functools.lru_cache
def _get_type_cached(self, name):
return self._vmlinux.get_type(name)
def _get_lt_node_from_rb_node(
self, rb_node, index
) -> Optional[interfaces.objects.ObjectInterface]:
"""Gets the latch tree node from the RBTree node.
Based on __lt_from_rb()
"""
# Unfortunately, we cannot use our LinuxUtilities.container_of() here, since the
# member is indexed by the 'index' variable:
# ltn = container_of(node, struct latch_tree_node, node[idx])
pointer_size = self._get_type_cached("pointer").size
type_dec = self._get_type_cached("latch_tree_node")
member_offset = type_dec.relative_child_offset("node") + index * pointer_size
container_addr = rb_node.vol.offset - member_offset
return self._vmlinux.object(
object_type="latch_tree_node", offset=container_addr, absolute=True
)
def find(
self, key: int, comp_function: Callable
) -> Optional[interfaces.objects.ObjectInterface]:
"""Returns a pointer to the node matching key or None.
Based on latch_tree_find() and __lt_find()
Args:
key (int): Typically an address
comp_function: Callback comparison function to provide the order between the
search key and an element. It's works like the kernel's latch_tree_ops::comp
i.e.: comp_function(key, latch_tree_node)
Returns:
latch_tree_node: A pointer to the node matching key or None.
"""
# latch_tree_root >= 4.2 ade3f510f93a5613b672febe88eff8ea7f1c63b7
# Use the lowest sequence bit as an index for picking which data copy to read
if self.seq.has_member("seqcount"):
# kernels >= 5.10 0c9794c8b6781eb7dad8e19b78c5d4557790597a
sequence = self.seq.seqcount.sequence
elif self.seq.has_member("sequence"):
# 4.2 <= kernel < 5.10
sequence = self.seq.sequence
else:
raise AttributeError("Unsupported sequence type implementation")
idx = sequence & 1
rb_node_ptr = self.tree[idx].rb_node
while rb_node_ptr and rb_node_ptr.is_readable():
rb_node = rb_node_ptr.dereference()
lt_node = self._get_lt_node_from_rb_node(rb_node, idx)
c = comp_function(key, lt_node)
if c < 0:
rb_node_ptr = rb_node.rb_left
elif c > 0:
rb_node_ptr = rb_node.rb_right
else:
return lt_node
return None
class kernel_symbol(objects.StructType):
def _offset_to_ptr(self, off) -> int:
layer = self._context.layers[self.vol.layer_name]
long_mask = (1 << layer.bits_per_register) - 1
return (self.vol.offset + off) & long_mask
def get_name(self) -> str:
if self.has_member("name_offset"):
# kernel >= 4.19 and CONFIG_HAVE_ARCH_PREL32_RELOCATIONS=y
# See 7290d58095712a89f845e1bca05334796dd49ed2
name_offset = self._offset_to_ptr(self.name_offset)
elif self.has_member("name"):
# kernel < 4.19 or CONFIG_HAVE_ARCH_PREL32_RELOCATIONS=n
name_offset = self.member("name")
else:
raise AttributeError("Unsupported kernel_symbol type implementation")
layer = self._context.layers[self.vol.layer_name]
name_bytes = layer.read(name_offset, linux_constants.KSYM_NAME_LEN)
idx = name_bytes.find(b"\x00")
if idx != -1:
name_bytes = name_bytes[:idx]
return name_bytes.decode("utf-8", errors="ignore")
def get_value(self) -> int:
if self.has_member("value_offset"):
# kernel >= 4.19 and CONFIG_HAVE_ARCH_PREL32_RELOCATIONS=y
# See 7290d58095712a89f845e1bca05334796dd49ed2
return self._offset_to_ptr(self.value_offset)
elif self.has_member("value"):
# kernel < 4.19 or CONFIG_HAVE_ARCH_PREL32_RELOCATIONS=n
return self.member("value")
raise AttributeError("Unsupported kernel_symbol type implementation")
def get_namespace(self) -> str:
if self.has_member("namespace_offset"):
# kernel >= 4.19 and CONFIG_HAVE_ARCH_PREL32_RELOCATIONS=y
# See 7290d58095712a89f845e1bca05334796dd49ed2
namespace_offset = self._offset_to_ptr(self.namespace_offset)
elif self.has_member("namespace"):
# kernel < 4.19 or CONFIG_HAVE_ARCH_PREL32_RELOCATIONS=n
namespace_offset = self.member("namespace")
else:
raise AttributeError("Unsupported kernel_symbol type implementation")
layer = self._context.layers[self.vol.layer_name]
namespace_bytes = layer.read(namespace_offset, linux_constants.KSYM_NAME_LEN)
idx = namespace_bytes.find(b"\x00")
if idx != -1:
namespace_bytes = namespace_bytes[:idx]
return namespace_bytes.decode("utf-8", errors="ignore")
volatility3/framework/symbols/linux/kallsyms.py
+1668
View File
File diff suppressed because it is too large Load Diff