diff --git a/test/test_volatility.py b/test/test_volatility.py index 8676d1f3e..d5b59b15c 100644 --- a/test/test_volatility.py +++ b/test/test_volatility.py @@ -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 diff --git a/volatility3/framework/constants/_version.py b/volatility3/framework/constants/_version.py index f2403cf4a..cf7c7b51a 100644 --- a/volatility3/framework/constants/_version.py +++ b/volatility3/framework/constants/_version.py @@ -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 = "" diff --git a/volatility3/framework/constants/linux/__init__.py b/volatility3/framework/constants/linux/__init__.py index f7f3faf87..99867b1fd 100644 --- a/volatility3/framework/constants/linux/__init__.py +++ b/volatility3/framework/constants/linux/__init__.py @@ -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" diff --git a/volatility3/framework/plugins/linux/kallsyms.py b/volatility3/framework/plugins/linux/kallsyms.py new file mode 100644 index 000000000..7dd4f06e6 --- /dev/null +++ b/volatility3/framework/plugins/linux/kallsyms.py @@ -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()) diff --git a/volatility3/framework/plugins/linux/pscallstack.py b/volatility3/framework/plugins/linux/pscallstack.py new file mode 100644 index 000000000..8931ca581 --- /dev/null +++ b/volatility3/framework/plugins/linux/pscallstack.py @@ -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(), + ) diff --git a/volatility3/framework/symbols/linux/__init__.py b/volatility3/framework/symbols/linux/__init__.py index aa89e7a16..c8a22c7f5 100644 --- a/volatility3/framework/symbols/linux/__init__.py +++ b/volatility3/framework/symbols/linux/__init__.py @@ -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.""" diff --git a/volatility3/framework/symbols/linux/extensions/__init__.py b/volatility3/framework/symbols/linux/extensions/__init__.py index c24b495d8..d02794b1e 100644 --- a/volatility3/framework/symbols/linux/extensions/__init__.py +++ b/volatility3/framework/symbols/linux/extensions/__init__.py @@ -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") diff --git a/volatility3/framework/symbols/linux/kallsyms.py b/volatility3/framework/symbols/linux/kallsyms.py new file mode 100644 index 000000000..298725a7a --- /dev/null +++ b/volatility3/framework/symbols/linux/kallsyms.py @@ -0,0 +1,1668 @@ +# 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 dataclasses +import functools +import logging +from typing import Iterator, List, Optional, Tuple + +from volatility3.framework import constants, exceptions, interfaces +from volatility3.framework.configuration import requirements +from volatility3.framework.constants import linux as linux_constants +from volatility3.framework.objects import utility +from volatility3.framework.symbols import linux +from volatility3.plugins.linux import lsmod + +vollog = logging.getLogger(__name__) + + +@dataclasses.dataclass +class KASConfig: + """Kallsyms configuration class""" + + num_syms_address: int + names_address: int + token_table_address: int + token_index_address: int + offsets_address: int + relative_base_address: int + _stext: int + + # Usually not in VMCOREINFO, these are found during the bootstrap stage. + # If an ISF is available, they are fetched from there instead. + markers_address: int = None + addresses_address: int = None + _sinittext: int = None + _einittext: int = None + _etext: int = None + _end: int = None + mod_tree: int = None + module_addr_min: int = None + module_addr_max: int = None + start_ksymtab: int = None + stop_ksymtab: int = None + bpf_tree_address: int = None + seqs_of_names_address: int = None + + num_syms_type_size: int = None + markers_type_size: int = None + kernel_symbol_size: int = None + + @classmethod + def _get_symbol_address(cls, context, layer_name, module_name, symbol_name): + vmlinux = context.modules[module_name] + if not vmlinux.has_symbol(symbol_name): + return None + + layer = context.layers[layer_name] + address = vmlinux.get_symbol(symbol_name).address + address += layer.config["kernel_virtual_offset"] + return address + + @classmethod + def new_from_isf(cls, context, layer_name, module_name): + vmlinux = context.modules[module_name] + + # kallsyms_num_syms and kallsyms_markers types were updated from a unsigned long + # to unsigned int in 4.20 80ffbaa5b1bd98e80e3239a3b8cfda2da433009a + num_syms_type_size = vmlinux.get_symbol("kallsyms_num_syms").type.size + kernel_symbol_size = vmlinux.get_type("kernel_symbol").size + + def get_symbol_address(symbol_name): + return cls._get_symbol_address( + context, layer_name, module_name, symbol_name + ) + + kas_config = KASConfig( + num_syms_address=get_symbol_address("kallsyms_num_syms"), + names_address=get_symbol_address("kallsyms_names"), + token_table_address=get_symbol_address("kallsyms_token_table"), + token_index_address=get_symbol_address("kallsyms_token_index"), + offsets_address=get_symbol_address("kallsyms_offsets"), + relative_base_address=get_symbol_address("kallsyms_relative_base"), + markers_address=get_symbol_address("kallsyms_markers"), + addresses_address=get_symbol_address("kallsyms_addresses"), + _sinittext=get_symbol_address("_sinittext"), + _einittext=get_symbol_address("_einittext"), + _stext=get_symbol_address("_stext"), + _etext=get_symbol_address("_etext"), + _end=get_symbol_address("_end"), + mod_tree=get_symbol_address("mod_tree"), + module_addr_min=get_symbol_address("module_addr_min"), + module_addr_max=get_symbol_address("module_addr_max"), + start_ksymtab=get_symbol_address("__start___ksymtab"), + stop_ksymtab=get_symbol_address("__stop___ksymtab"), + bpf_tree_address=get_symbol_address("bpf_tree"), + seqs_of_names_address=get_symbol_address("kallsyms_seqs_of_names"), + num_syms_type_size=num_syms_type_size, + markers_type_size=num_syms_type_size, + kernel_symbol_size=kernel_symbol_size, + ) + return kas_config + + +class _KallsymsIO: + """Helper to interpret a memory address as a file pointer. + + For internal use within the Kallsyms API; external use is discouraged. + """ + + def __init__( + self, + context: interfaces.context.ContextInterface, + layer_name: str, + base=0, + endian="little", + ): + self._context = context + self._layer_name = layer_name + self._base = base + self._position = base + self._endian = endian + + def read(self, size: int) -> bytes: + """Return 'size' bytes from the current postion""" + layer = self._context.layers[self._layer_name] + buf = layer.read(offset=self._position, length=size) + self._position += size + return buf + + def read_str(self, size: int) -> str: + """Returns 'size' bytes as a string from the current position.""" + return self.read(size).decode() + + def read_int(self, size: int, signed: bool = False) -> int: + """Returns the integer stored in the current position using 'size' bytes. + Args: + size: Number of bytes to use for the int. + signed: Integer sign. + + Returns: + The integer stored in the current position. + """ + return int.from_bytes( + self.read(size), + byteorder=self._endian, + signed=signed, + ) + + def seek(self, offset: int) -> None: + """Seek the pointer to the given offset, based on the base address. + + Args: + offset: offset from the base address + """ + self._position = self._base + offset + + +@dataclasses.dataclass +class KASSymbolBasic: + name: str + type: str + + +@dataclasses.dataclass +class KASSymbol(KASSymbolBasic): + address: int + size: int + module_name: str + exported: bool = False + subsystem: str = None + + def __str__(self): + return ( + f"name:{self.name}, type:{self.type}, address:{self.address:#x}, " + f"size:{self.size}, exported:{self.exported}, subsystem:{self.subsystem}" + ) + + def set_exported_from_type(self) -> None: + """Updates the 'export' member based on the symbol's type. + + This method evaluates the symbol's type and sets the 'export' member + to indicate whether the object is exported. This code and Linux kernel follows + the nm symbol type logic. + """ + # As per the "nm" man page: + # If lowercase, the symbol is usually local; if uppercase, the symbol is + # global (external). There are however a few lowercase symbols that are shown + # for special global symbols ("u", "v" and "w"). + self.exported = bool(self.type.isupper() or self.type in ("u", "v", "w")) + + @functools.cached_property + def type_description(self) -> Optional[str]: + """Returns the interpreted meaning of the symbol type based on the nm tool. + + Returns: + A string with the type description. + """ + # If a symbol type exists with the original case, get it + symbol_type_description = linux_constants.NM_TYPES_DESC.get(self.type, None) + if symbol_type_description: + return symbol_type_description + + # Otherwise, use the lowercase version + symbol_type_description = linux_constants.NM_TYPES_DESC.get( + self.type.lower(), None + ) + return symbol_type_description + + +@dataclasses.dataclass +class KASFilter: + name: str + type: str + + +class Kallsyms(interfaces.configuration.VersionableInterface): + """Kallsyms API class""" + + _required_framework_version = (2, 19, 0) + _version = (1, 0, 0) + + # Internal kernel core constants + _CORE_SUBSYSTEM_NAME = "core" + _CORE_MODULE_NAME = "kernel" + + # Internal module constants + _MODULE_SUBSYSTEM_NAME = "module" + + # Internal FTrace constants + _FTRACE_SUBSYSTEM_NAME = "ftrace" + _FTRACE_MODULE_SYM_TYPE = "T" + _FTRACE_TRAMPOLINE_MODULE_NAME = "__builtin__ftrace" + _FTRACE_TRAMPOLINE_SYM = "ftrace_trampoline" + _FTRACE_TRAMPOLINE_SYM_TYPE = "t" + + # Internal BPF constants + _BPF_SUBSYSTEM_NAME = "bpf" + _BPF_MODULE_NAME = "bpf" + _BPF_SYM_TYPE = "t" + + def __init__( + self, + context: interfaces.context.ContextInterface, + layer_name: str, + module_name: str, + kas_config: KASConfig = None, + progress_callback: constants.ProgressCallback = None, + ) -> None: + """Initialize the Kallsyms API + + Args: + context: The context used to access memory layers and symbols + layer_name: The name of layer within the context in which the module exists + module_name: The name of the kernel module on which to operate + kas_config: The KAllSyms configuration + progress_callback: Method that is called periodically during scanning to + update progress + """ + super().__init__() + + self._assert_versions() + + self._context = context + self._layer_name = layer_name + self._module_name = module_name + self._kas_config = kas_config + self._progress_callback = progress_callback + if progress_callback and not callable(progress_callback): + raise TypeError("Progress_callback is not callable") + + if not kas_config: + self._kas_config = KASConfig.new_from_isf( + context=context, + layer_name=layer_name, + module_name=module_name, + ) + + layer = self._context.layers[self._layer_name] + # FIXME: The layer lacks this information. Could there be a better alternative? + self._endian = "little" if layer._entry_format[0] == "<" else "big" + self._long_size = layer.bits_per_register // 8 + + self._kallsyms_num_syms = None + self._kallsyms_relative_base = None + + self._kallsyms_token_index_address = None + self._kallsyms_offsets_address = None + self._kallsyms_names_io = _KallsymsIO( + context=self._context, + layer_name=self._layer_name, + base=self._kas_config.names_address, + endian=self._endian, + ) + + self._kallsyms_token_table_io = _KallsymsIO( + context=self._context, + layer_name=self._layer_name, + base=self._kas_config.token_table_address, + endian=self._endian, + ) + + self._bootstrap() + + @classmethod + def _assert_versions(cls) -> None: + """Verify versions of shared dependencies""" + lsmod_version_required = (2, 0, 0) + if not requirements.VersionRequirement.matches_required( + lsmod_version_required, lsmod.Lsmod.version + ): + raise exceptions.VolatilityException( + "Lsmod version not suitable: " + f"required {lsmod_version_required} found {lsmod.Lsmod.version}", + ) + + return None + + def _read_bytes(self, address: int, size: int) -> bytes: + layer = self._context.layers[self._layer_name] + return layer.read(address, size).decode() + + def _read_int(self, address: int, size: int, signed: bool = False) -> int: + layer = self._context.layers[self._layer_name] + return int.from_bytes( + layer.read(address, size), + byteorder=self._endian, + signed=signed, + ) + + def _bootstrap(self) -> None: + layer = self._context.layers[self._layer_name] + # kallsyms_num_syms and kallsyms_markers[] types were updated from a unsigned long + # to unsigned int in 4.20 80ffbaa5b1bd98e80e3239a3b8cfda2da433009a + self._kallsyms_num_syms = self._read_int( + self._kas_config.num_syms_address, + self._kas_config.num_syms_type_size, + signed=False, + ) + + if self._kas_config.relative_base_address: + # kernels >= 4.6 + self._kallsyms_relative_base = ( + self._read_int( + self._kas_config.relative_base_address, + self._long_size, + signed=False, + ) + & layer.address_mask + ) + + self._kallsyms_offsets_address = self._kas_config.offsets_address + self._kallsyms_token_index_address = self._kas_config.token_index_address + + # Preload the kallsyms_token_index array + short_size = 2 + self._kallsyms_token_index = [ + self._read_int( + self._kallsyms_token_index_address + index * short_size, + short_size, + signed=False, + ) + for index in range(256) + ] + + def _get_symbol( + self, + offset, + index, + filters: List[KASFilter] = None, + ) -> Optional[Tuple[KASSymbol, int]]: + kassymbolbasic, compressed_length = self._expand_symbol(offset, filters) + kassymbol = None + if kassymbolbasic: + sym_addr = self._get_symbol_address_by_index(index=index) + _, sym_size = self._get_symbol_pos(sym_addr) + + kassymbol = KASSymbol( + name=kassymbolbasic.name, + type=kassymbolbasic.type, + address=sym_addr, + size=sym_size, + module_name=self._CORE_MODULE_NAME, + subsystem=self._CORE_SUBSYSTEM_NAME, + ) + kassymbol.set_exported_from_type() + return kassymbol, compressed_length + + def get_core_symbols( + self, + progress_callback: constants.ProgressCallback = None, + ) -> Iterator[KASSymbol]: + """Yield each kernel core symbol + + Args: + progress_callback: Method that is called periodically during scanning to + update progress + + Based on kallsyms_on_each_symbol() + + Yields: + KASSymbol objects + """ + current_offset = 0 + for sym_idx in range(self._kallsyms_num_syms): + kassymbol, compressed_length = self._get_symbol(current_offset, sym_idx) + if kassymbol: + yield kassymbol + + if progress_callback: + progress_callback( + (sym_idx / self._kallsyms_num_syms) * 100, + "Populating Kallsyms core symbols", + ) + + current_offset += compressed_length + 1 + + def _expand_symbol( + self, + offset: int, + filters: List[KASFilter] = None, + ) -> Tuple[KASSymbolBasic, int]: + """Expand a compressed symbol using its offset in the stream + Based on kallsyms_expand_symbol() + + Args: + offset: Symbol offset in the kallsyms arrays. + filters: List of KASFilter filters + + Returns: + A tuple with a KASSymbolBasic object and the symbol name's compressed length. + """ + filters = filters if filters is not None else [] + type_filters = tuple(kassymbolfilter.type for kassymbolfilter in filters) + + self._kallsyms_names_io.seek(offset) + # The compressed symbol length is in the first byte + compressed_length = self._kallsyms_names_io.read_int(size=1) + if compressed_length & 0x80 != 0: + # kernels >= 6.1 73bbb94466fd3f8b313eeb0b0467314a262dddb3 + # MSB 1 means a 'big' symbol, we need an extra byte + lower_byte = compressed_length + upper_byte = self._kallsyms_names_io.read_int(size=1) + compressed_length = (upper_byte << 7) | (lower_byte & 0x7F) + + abort_decompression = False + sym_type = None + sym_name = "" + for _ in range(compressed_length): + token_index_index = self._kallsyms_names_io.read_int(size=1) + token_index = self._kallsyms_token_index[token_index_index] + self._kallsyms_token_table_io.seek(token_index) + token = self._kallsyms_token_table_io.read_str(1) + while token != "\x00": + if not sym_type: + sym_type = token + # We got the symbol type, we can abort this immediatelly + if type_filters and sym_type not in type_filters: + abort_decompression = True + break + else: + sym_name += token + for kassymbolfilter in filters: + if kassymbolfilter.type is not None: + if ( + sym_type == kassymbolfilter.type + and kassymbolfilter.name.startswith(sym_name) + ): + break + elif kassymbolfilter.name.startswith(sym_name): + break + + else: + if filters: + abort_decompression = True + + token = self._kallsyms_token_table_io.read_str(1) + + if abort_decompression: + break + + kassymbolbasic = ( + KASSymbolBasic(name=sym_name, type=sym_type) + if not abort_decompression + else None + ) + return kassymbolbasic, compressed_length + + def _get_symbol_address_by_index(self, index: int) -> int: + """Return symbol address based on the symbol index in the kallsyms arrays. + Based on kallsyms_sym_address() + + Args: + index: Symbol index + + Returns: + Symbol address + """ + layer = self._context.layers[self._layer_name] + if self._kallsyms_offsets_address: + # kernels >= 4.6 - Addresses are relative to kallsyms_relative_base + # It assumes: CONFIG_KALLSYMS_BASE_RELATIVE=y and CONFIG_KALLSYMS_ABSOLUTE_PERCPU=y + signed_int_size = 4 + sym_offset_ptr = self._kallsyms_offsets_address + (index * signed_int_size) + sym_addr = self._read_int(sym_offset_ptr, signed_int_size, signed=True) + + if sym_addr < 0: + # Negative offsets are relative to kallsyms_relative_base - 1 + return self._kallsyms_relative_base - 1 - sym_addr + + # Positive offsets are absolute values + return sym_addr & layer.address_mask + elif self._kas_config.addresses_address: + # kernels < 4.6 - Addresses are absolute + # unsigned long kallsyms_addresses[] + kallsyms_address = self._read_int( + self._kas_config.addresses_address + (index * self._long_size), + self._long_size, + signed=False, + ) + return kallsyms_address & layer.address_mask + else: + raise exceptions.VolatilityException("Unsupported kernel") + + @functools.lru_cache + def _get_symbol_pos(self, address: int) -> Tuple[int, int]: + """Returns the symbol position in the kallsyms arrays and its size.""" + low = 0 + high = self._kallsyms_num_syms + + while high - low > 1: + mid = low + (high - low) // 2 + if self._get_symbol_address_by_index(mid) <= address: + low = mid + else: + high = mid + + # Search for the first aliased symbol. *Aliased symbols* are symbols with the same address. + while low and self._get_symbol_address_by_index( + low - 1 + ) == self._get_symbol_address_by_index(low): + low -= 1 + + symbol_start = self._get_symbol_address_by_index(low) + symbol_end = 0 + + # Search for next non-aliased symbol. + for idx in range(low + 1, self._kallsyms_num_syms): + if self._get_symbol_address_by_index(idx) > symbol_start: + symbol_end = self._get_symbol_address_by_index(idx) + break + + # pylint: disable=protected-access + # If no next symbol is found, we default to using the end of the section + if not symbol_end: + if self._is_kernel_inittext(address): + symbol_end = self._kas_config._einittext + elif self._kas_config._end is not None: + # Assume CONFIG_KALLSYMS_ALL=y. Otherwise, symbol_end will be _etext + symbol_end = self._kas_config._end + else: + symbol_end = self._kas_config._etext + + symbol_size = symbol_end - symbol_start + + return low, symbol_size + + @functools.lru_cache + def _get_symbol_offset(self, index: int) -> int: + """Find the offset on the compressed stream given the index in the kallsyms array. + + Based on get_symbol_offset + + Returns: + Offset on the compressed stream + """ + + # Use the nearest marker, placed every 256 positions + kallsyms_markers_pos_ptr = ( + self._kas_config.markers_address + + (index >> 8) * self._kas_config.markers_type_size + ) + kallsyms_markers_pos = self._read_int( + kallsyms_markers_pos_ptr, self._kas_config.markers_type_size, signed=False + ) + name_addr = self._kas_config.names_address + kallsyms_markers_pos + + # Scan symbols sequentially until the target. Each symbol uses a + # [][ bytes of data] format, so we skip symbols by adding their length + # to the pointer value. + for _ in range(index & 0xFF): + compressed_length = self._read_int(name_addr, 1) + if compressed_length & 0x80 != 0: + # kernels >= 6.1 73bbb94466fd3f8b313eeb0b0467314a262dddb3 + # MSB 1 means a 'big' symbol, we need an extra byte + lower_byte = compressed_length + upper_byte = self._kallsyms_names_io.read_int(size=1) + compressed_length = (upper_byte << 7) | (lower_byte & 0x7F) + + name_addr += compressed_length + 1 + + return name_addr - self._kas_config.names_address + + def _is_kernel_inittext(self, addr: int) -> bool: + # pylint: disable=protected-access + if not (self._kas_config._sinittext and self._kas_config._einittext): + # We don't know + return False + + return self._kas_config._sinittext <= addr < self._kas_config._einittext + + def _is_kernel_text(self, addr: int) -> bool: + # pylint: disable=protected-access + return self._kas_config._stext <= addr < self._kas_config._etext + + def _is_core_ksym_addr(self, addr: int) -> bool: + return self._is_kernel_text(addr) or self._is_kernel_inittext(addr) + + def lookup_address(self, address: int) -> Optional[KASSymbol]: + """Search for a symbol by its memory address. + + This function scans kernel core, module symbols, BPF symbols, and Ftrace symbols + to locate the first symbol matching the specified address. Note that multiple + symbols (aliased symbols) can share the same memory address, so this method + returns the first match found. + + Based on kallsyms_lookup. + + Args: + address: The memory address to search for. + + Returns: + The matching symbol if found, or None if no match is found. + """ + layer = self._context.layers[self._layer_name] + address &= layer.address_mask + + kassymbol = self.core_lookup_address(address) + if not kassymbol: + kassymbol = self.module_lookup_address(address) + + if not kassymbol: + kassymbol = self.bpf_lookup_address(address) + + if not kassymbol: + kassymbol = self.ftrace_lookup_address(address) + + return kassymbol + + def core_lookup_address(self, address: int) -> Optional[KASSymbol]: + """Search for a symbol by its memory address within the kernel core. + + Based on kallsyms_lookup_buildid. + + Args: + address: The memory address to search for. + + Returns: + The matching symbol if found, or None if no match is found. + """ + layer = self._context.layers[self._layer_name] + address &= layer.address_mask + + if not self._is_core_ksym_addr(address): + return None + + pos, sym_size = self._get_symbol_pos(address) + offset = self._get_symbol_offset(pos) + sym_address = self._get_symbol_address_by_index(pos) + kassymbolbasic, _compressed_length = self._expand_symbol(offset) + + if not kassymbolbasic: + return None + + kas_symbol = KASSymbol( + name=kassymbolbasic.name, + type=kassymbolbasic.type, + address=sym_address, + size=sym_size, + module_name=self._CORE_MODULE_NAME, + subsystem=self._CORE_SUBSYSTEM_NAME, + ) + kas_symbol.set_exported_from_type() + return kas_symbol + + def _is_symbol_exported( + self, + name: int, + address: int, + module: Optional[interfaces.objects.ObjectInterface] = None, + ) -> bool: + """Check if the address belongs to an exported symbol. + If a module object is provided, it searches in that module symbols. + Otherwise, it searches in the global symbols. + + Bases on is_exported + + Args: + name: Symbol name + address: Symbol address + module: Module object. Defaults to None. + + Returns: + True if the symbol is exported; otherwise, returns False + """ + if module: + if module.num_syms <= 0: + return False + + start_mod_ksymtab = module.syms + stop_mod_ksymtab = ( + start_mod_ksymtab + + module.num_syms * self._kas_config.kernel_symbol_size + ) + kernel_symbol = self._find_exported_symbol_in_range( + name, start_mod_ksymtab, stop_mod_ksymtab + ) + else: + # Search the not GPL modules + kernel_symbol = self._find_exported_symbol_in_range( + name, + self._kas_config.start_ksymtab, + self._kas_config.stop_ksymtab, + ) + + return kernel_symbol is not None and kernel_symbol.get_value() == address + + def _elfsym_to_kassymbol( + self, + module: interfaces.objects.ObjectInterface, + elf_sym_obj: interfaces.objects.ObjectInterface, + elf_sym_index: int, + subsystem: str = None, + ) -> Optional[KASSymbol]: + """Returns a KASSymbol from a ElfSym + + Args: + module: Module object + elf_sym_obj: ElfSym object + elf_sym_index: ElfSym index + subsystem: Name of the sub-subtem: core, module, bpf, ftrace, etc + + Returns: + A KASSymbol object + """ + layer = self._context.layers[self._layer_name] + sym_name = elf_sym_obj.get_name() + if not sym_name: + return None + + # Normalize sym.st_value offset, which is an address pointing to the symbol value + sym_address = elf_sym_obj.st_value & layer.address_mask + sym_type = module.get_symbol_type(elf_sym_obj, elf_sym_index) + + kas_symbol = KASSymbol( + name=sym_name, + type=sym_type, + address=sym_address, + size=elf_sym_obj.st_size, + module_name=module.get_name(), + exported=False, + subsystem=subsystem, + ) + kas_symbol.set_exported_from_type() + return kas_symbol + + def _is_module_ksym_address(self, address: int) -> bool: + return self._modules_address_min <= address <= self._modules_address_max + + def module_lookup_address( + self, + address: int, + module: Optional[interfaces.objects.ObjectInterface] = None, + ) -> Optional[KASSymbol]: + """Search for a symbol within kernel modules based on its memory address. + If a module object is provided, it will only search in that module. Otherwise, + it will try to first find the module to where the provided address belong to. + + Based on module_address_lookup. + + Args: + address: The memory address of the symbol to search for + module [optional]: The module to search within. If not provided, the module + containing the address will be automatically determined + + Returns: + The matching KASSymbol if found; otherwise, returns None + """ + if not self._is_module_ksym_address(address): + return None + + module = module or self._get_module_by_address(address) + if not module: + # This may occur if the kernel lacks the mod_tree implementation. + for ( + cur_module, + minimum_address, + maximum_address, + ) in self._module_memory_region: + if minimum_address <= address < maximum_address: + module = cur_module + break + + if not module: + # We couldn't find the module + return None + + kassymbol = self._find_address_in_module_symbols(module, address) + if kassymbol: + return kassymbol + + return None + + def _find_address_in_module_symbols( + self, + module: interfaces.objects.ObjectInterface, + address: int, + ) -> Optional[KASSymbol]: + """Find the symbol corresponding to a given address within a module. + + Based on find_kallsyms_symbol + + Args: + module: The module where the address belongs to + address: The memory address to search for + + Returns: + The matching KASSymbol if found; otherwise, returns None + """ + # Before walking all the symbols, ensure the address belongs to this module + module_boundaries = module.get_module_address_boundaries() + if not module_boundaries: + return None + + minimum_address, maximum_address = module_boundaries + if not (minimum_address <= address < maximum_address): + return None + + layer = self._context.layers[self._layer_name] + for elf_sym_idx, elf_sym in enumerate(module.get_symbols()): + if not elf_sym.get_name(): + continue + + sym_address_start = elf_sym.st_value & layer.address_mask + sym_address_end = sym_address_start + elf_sym.st_size + + if sym_address_start <= address < sym_address_end: + return self._elfsym_to_kassymbol( + module, elf_sym, elf_sym_idx, subsystem=self._MODULE_SUBSYSTEM_NAME + ) + + return None + + @functools.cached_property + def _module_memory_region( + self, + ) -> List[Tuple[interfaces.objects.ObjectInterface, int, int]]: + modules_region = [] + for module in lsmod.Lsmod.list_modules(self._context, self._module_name): + minimum_address, maximum_address = module.get_module_address_boundaries() + module_region = module, minimum_address, maximum_address + modules_region.append(module_region) + + return modules_region + + @functools.lru_cache + def _get_modules_memory_boundaries(self) -> Tuple[int, int]: + """Determine the boundaries of the module allocation area + + Returns: + A tuple containing the minimum and maximum addresses for the kernel module + allocation area. + """ + + if self._kas_config.mod_tree: + # Kernel >= 5.19 58d208de3e8d87dbe196caf0b57cc58c7a3836ca + mod_tree_address = self._kas_config.mod_tree + vmlinux = self._context.modules[self._module_name] + mod_tree = vmlinux.object( + object_type="mod_tree_root", + offset=mod_tree_address, + absolute=True, + ) + addr_min, addr_max = mod_tree.addr_min, mod_tree.addr_max + elif self._kas_config.module_addr_min and self._kas_config.module_addr_max: + # 2.6.27 <= kernel < 5.19 3a642e99babe0617febb6f402e1e063479f489db + kas_config = self._kas_config + addr_min, addr_max = kas_config.module_addr_min, kas_config.module_addr_max + else: + raise exceptions.VolatilityException( + "Cannot find the module memory allocation area. Unsupported kernel" + ) + + layer = self._context.layers[self._layer_name] + return addr_min & layer.address_mask, addr_max & layer.address_mask + + @functools.cached_property + def _modules_address_min(self): + address_min, _address_max = self._get_modules_memory_boundaries() + return address_min + + @functools.cached_property + def _modules_address_max(self): + _address_min, address_max = self._get_modules_memory_boundaries() + return address_max + + def _get_module_by_address( + self, address: int + ) -> Optional[interfaces.objects.ObjectInterface]: + """Searches for the module that contains the given memory address within its range. + It uses a latch tree for optimized address range searching. + + Based on __module_address() + + Args: + address: The module memory address to search for. + + Returns: + The matching module if found; otherwise, returns None + """ + if not self._is_module_ksym_address(address): + return None + + return self._search_module_by_address(address) + + @functools.lru_cache + def _get_type_cache(self, name: str): + vmlinux = self._context.modules[self._module_name] + return vmlinux.get_type(name) + + def _mod_tree_comp( + self, address: int, latch_tree_node: interfaces.objects.ObjectInterface + ) -> int: + vmlinux = self._context.modules[self._module_name] + + module_memory_mtn_offset = self._get_type_cache( + "module_memory" + ).relative_child_offset("mtn") + mod_tree_node_mod_offset = self._get_type_cache( + "mod_tree_node" + ).relative_child_offset("mod") + + module_memory_offset = ( + latch_tree_node.vol.offset + + module_memory_mtn_offset + + mod_tree_node_mod_offset + ) + + module_memory = vmlinux.object( + object_type="module_memory", + offset=module_memory_offset, + absolute=True, + ) + start = module_memory.base + end = start + module_memory.size + + if address < start: + return -1 + elif address >= end: + return 1 + else: + return 0 + + def _search_module_by_address( + self, address: int + ) -> Optional[interfaces.objects.ObjectInterface]: + """Searches for the module that contains the given memory address within its range. + It uses a latch tree for optimized address range searching. + + Based on mod_find + + Args: + address: The module memory address to search for + + Returns: + The matching module if found; otherwise, returns None + """ + vmlinux = self._context.modules[self._module_name] + if self._kas_config.mod_tree: + mod_tree_address = self._kas_config.mod_tree + mod_tree = vmlinux.object( + object_type="mod_tree_root", + offset=mod_tree_address, + absolute=True, + ) + latch_tree_root = mod_tree.root + latch_tree_node = latch_tree_root.find(address, self._mod_tree_comp) + if latch_tree_node: + mod_tree_node = linux.LinuxUtilities.container_of( + latch_tree_node.vol.offset, "mod_tree_node", "node", vmlinux + ) + module_ptr = mod_tree_node.mod + if not module_ptr.is_readable(): + vollog.warning("Modules latch tree seems corrupt") + return None + + return module_ptr.dereference() + + return None + + def _find_exported_symbol_in_range( + self, name: str, start: int, stop: int + ) -> Optional[interfaces.objects.ObjectInterface]: + """Find an exported symbol within a specified range of kernel symbols. + + Based on lookup_exported_symbol + + Args: + name: Symbol name + start: Start address + stop: Stop address + + Returns: + The matching kernel_symbol object if found, or None if no match is found. + """ + + num_elems = (stop - start) // self._kas_config.kernel_symbol_size + + return self._search_kernel_symbol_object_by_name( + name, + base=start, + num_elems=num_elems, + ) + + def _cmp_kernel_symbol_name( + self, + name: str, + kernel_symbol: interfaces.objects.ObjectInterface, + ) -> int: + return self._cmp_symbol_name(name, kernel_symbol.get_name()) + + def _cmp_symbol_name( + self, + name: str, + other: str, + ) -> int: + if name == other: + return 0 + elif name < other: + return -1 + else: + return 1 + + def _search_kernel_symbol_object_by_name( + self, name: str, base: int, num_elems: int + ) -> Optional[interfaces.objects.ObjectInterface]: + """Search a kernel_symbol by name using binary search. + + Based on bsearch / __inline_bsearch() + + Args: + name: Symbol name + base: Base address + num_elems: Number of elements + + Returns: + A kernel_symbol object + """ + vmlinux = self._context.modules[self._module_name] + while num_elems > 0: + pivot = base + (num_elems // 2) * self._kas_config.kernel_symbol_size + + kernel_symbol_pivot = vmlinux.object( + object_type="kernel_symbol", + offset=pivot, + absolute=True, + ) + + result = self._cmp_kernel_symbol_name(name, kernel_symbol_pivot) + if result == 0: + return kernel_symbol_pivot + elif result > 0: + base = pivot + self._kas_config.kernel_symbol_size + num_elems -= 1 + + num_elems = num_elems // 2 + + return None + + def get_modules_symbols(self, name: str = None) -> Iterator[KASSymbol]: + """Yield each symbol from the kernel modules. + This function iterates over the symbols of the kernel modules and yields them as + KASSymbol objects. + + name (optional): If specified, the symbol name used to filter the symbols. + + Yields: + KASSymbol objects + """ + layer = self._context.layers[self._layer_name] + for module in lsmod.Lsmod.list_modules(self._context, self._module_name): + module_name = utility.array_to_string(module.name) + for elf_sym_idx, elf_sym_obj in enumerate(module.get_symbols()): + sym_name = elf_sym_obj.get_name() + if not sym_name: + continue + + if name and name != sym_name: + continue + + # Normalize sym.st_value offset, which is an address pointing to the symbol value + sym_address = elf_sym_obj.st_value & layer.address_mask + sym_size = elf_sym_obj.st_size + sym_type = module.get_symbol_type(elf_sym_obj, elf_sym_idx) + is_exported = self._is_symbol_exported(sym_name, sym_address, module) + sym_type = sym_type.upper() if is_exported else sym_type.lower() + + yield KASSymbol( + name=sym_name, + type=sym_type, + address=sym_address, + size=sym_size, + exported=is_exported, + module_name=module_name, + subsystem=self._MODULE_SUBSYSTEM_NAME, + ) + + def _ftrace_mod_get_symbols(self, address: int = None) -> Iterator[KASSymbol]: + """Yield each symbol from the ftrace modules. + This function iterates over the symbols of the ftrace modules and yields them as + KASSymbol objects. + + Based on ftrace_mod_get_kallsym + + Args: + address (optional): Address to filter symbols by + + Yields: + KASSymbol objects + """ + vmlinux = self._context.modules[self._module_name] + layer = self._context.layers[self._layer_name] + if not ( + vmlinux.has_type("ftrace_mod_map") and vmlinux.has_type("ftrace_mod_func") + ): + # kernel < 4.15 aba4b5c22cbac296f4081a0476d0c55828f135b4 + vollog.info( + "Unsupported Ftrace kallsyms implementation. Ignore this if it's a kernel < 4.15" + ) + return None + + symbol_table_name = vmlinux.symbol_table_name + ftrace_mod_map_symname = f"{symbol_table_name}{constants.BANG}ftrace_mod_map" + ftrace_mod_func_symname = f"{symbol_table_name}{constants.BANG}ftrace_mod_func" + ftrace_mod_maps = vmlinux.object_from_symbol("ftrace_mod_maps") + for mod_map in ftrace_mod_maps.to_list(ftrace_mod_map_symname, "list"): + for mod_func in mod_map.funcs.to_list(ftrace_mod_func_symname, "list"): + sym_name = utility.pointer_to_string( + mod_func.name, count=linux_constants.KSYM_NAME_LEN + ) + sym_addr = mod_func.ip & layer.address_mask + sym_size = mod_func.size + if address is not None and not ( + sym_addr <= address < sym_addr + sym_size + ): + continue + + module_name = utility.array_to_string(mod_map.mod.name) + kas_symbol = KASSymbol( + name=sym_name, + type=self._FTRACE_MODULE_SYM_TYPE, + address=sym_addr, + size=sym_size, + module_name=module_name, + subsystem=self._FTRACE_SUBSYSTEM_NAME, + ) + kas_symbol.set_exported_from_type() + yield kas_symbol + + def _ftrace_get_trampoline_symbols( + self, address: int = None + ) -> Iterator[KASSymbol]: + """Yield each symbol from the ftrace trampoline. + + Based on ftrace_get_trampoline_kallsym + + Args: + address (optional): Address to filter symbols by + + Yields: + KASSymbol objects + """ + # See kernel's ftrace_get_trampoline_kallsym() + vmlinux = self._context.modules[self._module_name] + if not vmlinux.has_type("ftrace_ops"): + # kernels < 2.6.27 16444a8a40d4c7b4f6de34af0cae1f76a4f6c901 + return None + + if not vmlinux.has_symbol("ftrace_ops_trampoline_list"): + # kernels < 5.9 fc0ea795f53c8d7040fa42471f74fe51d78d0834 + return None + + symbol_table_name = vmlinux.symbol_table_name + ftrace_ops_symname = f"{symbol_table_name}{constants.BANG}ftrace_ops" + ftrace_ops_trampoline_list = vmlinux.object_from_symbol( + "ftrace_ops_trampoline_list" + ) + + for ftrace_op in ftrace_ops_trampoline_list.to_list(ftrace_ops_symname, "list"): + sym_name = self._FTRACE_TRAMPOLINE_SYM + sym_addr = ftrace_op.trampoline + sym_size = ftrace_op.trampoline_size + + if address is not None and not (sym_addr <= address < sym_addr + sym_size): + continue + + kas_symbol = KASSymbol( + name=sym_name, + type=self._FTRACE_TRAMPOLINE_SYM_TYPE, + address=sym_addr, + size=sym_size, + module_name=self._FTRACE_TRAMPOLINE_MODULE_NAME, + subsystem=self._FTRACE_SUBSYSTEM_NAME, + ) + kas_symbol.set_exported_from_type() + yield kas_symbol + + def get_ftrace_symbols(self) -> Iterator[KASSymbol]: + """Yield each kernel ftrace symbol + + Yields: + KASSymbol objects + """ + yield from self._ftrace_mod_get_symbols() + yield from self._ftrace_get_trampoline_symbols() + + def get_bpf_symbols(self) -> Iterator[KASSymbol]: + """Yield each kernel BPF symbol + + Based on bpf_get_kallsym() + + Yields: + KASSymbol objects + """ + vmlinux = self._context.modules[self._module_name] + if vmlinux.has_type("bpf_ksym"): + # kernels >= 5.8 + list_type, list_head_member = "bpf_ksym", "lnode" + elif vmlinux.has_type("bpf_prog_aux"): + # 3.18 <= kernels < 5.8 + list_type, list_head_member = "bpf_prog_aux", "ksym_lnode" + else: + # kernels < 3.18 + vollog.info( + "Unsupported BPF kallsysms implementation. Don't worry if kernel < 3.18" + ) + return None + + symbol_table_name = vmlinux.symbol_table_name + list_type_symname = f"{symbol_table_name}{constants.BANG}{list_type}" + + layer = self._context.layers[self._layer_name] + + # Even when bpf_jit_kallsyms is disabled (/proc/sys/net/core/bpf_jit_kallsyms = 0), + # this function will still be able to gather the symbols. + bpf_kallsyms_list = vmlinux.object_from_symbol("bpf_kallsyms") + for elem in bpf_kallsyms_list.to_list(list_type_symname, list_head_member): + # See kernel's bpf_get_kallsym() + if list_type == "bpf_ksym": + # kernels >= 5.8 + bpf_ksym = elem + sym_name = utility.array_to_string(bpf_ksym.name) + sym_addr = bpf_ksym.start + sym_size = bpf_ksym.end - bpf_ksym.start + else: + # list_type == "bpf_prog_aux" 3.18 <= kernels < 5.8 + bpf_prog_aux = elem + bpf_prog = bpf_prog_aux.prog + sym_name = bpf_prog.get_name() + sym_addr = bpf_prog.bpf_func + sym_start, sym_end = bpf_prog.get_address_region() + sym_size = sym_end - sym_start + + # The following are also hardcoded in the Linux kernel + # see kernel's get_ksymbol_bpf(), bpf_get_kallsym() and BPF_SYM_ELF_TYPE + module_name = self._BPF_MODULE_NAME + sym_type = self._BPF_SYM_TYPE + sym_addr &= layer.address_mask + + kas_symbol = KASSymbol( + name=sym_name, + type=sym_type, + address=sym_addr, + size=sym_size, + module_name=module_name, + subsystem=self._BPF_SUBSYSTEM_NAME, + ) + kas_symbol.set_exported_from_type() + yield kas_symbol + + def get_all_symbols(self) -> Iterator[KASSymbol]: + """Enumerates each kallsym symbol + + Yields: + KASSymbol objects + """ + yield from self.get_core_symbols() + yield from self.get_modules_symbols() + yield from self.get_ftrace_symbols() + yield from self.get_bpf_symbols() + + def bpf_lookup_address(self, address: int) -> Optional[KASSymbol]: + """Search for a BPF symbol based on its memory address. + + Based on bpf_address_lookup() and __bpf_address_lookup() + + Args: + address: The memory address to search for + + Returns: + The matching KASSymbol if found; otherwise, returns None + """ + vmlinux = self._context.modules[self._module_name] + + if vmlinux.has_type("bpf_ksym"): + # kernels >= 5.7 535911c80ad4f5801700e9d827a1985bbff41519 + bpf_ksym = self._find_bpf_ksym(address) + if not bpf_ksym: + return None + symbol_start = bpf_ksym.start + symbol_end = bpf_ksym.end + sym_name = utility.array_to_string(bpf_ksym.name) + sym_size = symbol_end - symbol_start + elif vmlinux.has_type("latch_tree_root") and vmlinux.get_type( + "bpf_prog_aux" + ).child_template("ksym_tnode"): + # For 4.11 <= kernels < 5.7 + # latch_tree_root was added in kernels 4.2 ade3f510f93a5613b672febe88eff8ea7f1c63b7 + # BPF kallsyms support was added in kernels 4.11 74451e66d516c55e309e8d89a4a1e7596e46aacd + bpf_prog = self._find_bpf_prog(address) + if not bpf_prog: + return None + + symbol_start, symbol_end = bpf_prog.get_addr_region() + sym_name = bpf_prog.get_name() + sym_size = symbol_end - symbol_start + else: + # kernel < 4.11 + vollog.info( + "Unsupported BPF kallsyms implementation. Ignore this if it's a kernel < 4.11" + ) + return None + + layer = self._context.layers[self._layer_name] + symbol_start &= layer.address_mask + + kas_symbol = KASSymbol( + name=sym_name, + type=self._BPF_SYM_TYPE, + address=symbol_start, + size=sym_size, + module_name=self._BPF_MODULE_NAME, + subsystem=self._BPF_SUBSYSTEM_NAME, + ) + kas_symbol.set_exported_from_type() + return kas_symbol + + def _find_bpf_prog( + self, address: int + ) -> Optional[interfaces.objects.ObjectInterface]: + """Search for a BPF program based on its address. + Based on __bpf_address_lookup & bpf_prog_kallsyms_find() for kernels < 5.7 + + Args: + address: The BPF symbol address to search for + + Returns: + A bpf_prog object if found; otherwise, returns None. + """ + vmlinux = self._context.modules[self._module_name] + if not self._kas_config.bpf_tree_address: + return None + + bpf_latch_tree_root = vmlinux.object( + object_type="latch_tree_root", + offset=self._kas_config.bpf_tree_address, + absolute=True, + ) + latch_tree_node = bpf_latch_tree_root.find( + address, self._bpf_tree_comp_bpf_prog_aux + ) + + if not latch_tree_node: + return None + + bpf_prog_aux = linux.LinuxUtilities.container_of( + latch_tree_node.vol.offset, "bpf_prog_aux", "ksym_tnode", vmlinux + ) + bpf_prog = bpf_prog_aux.prog + return bpf_prog + + def _bpf_tree_comp_bpf_prog_aux( + self, address: int, latch_tree_node: interfaces.objects.ObjectInterface + ) -> int: + """Comparison function used by _find_bpf_prog() + Based on bpf_tree_comp for kernels < 5.7 + + Args: + address: The memory address to search for + latch_tree_node: A latch tree node + + Returns: + 0: equal, >0: key is greater, <0: key is less than this bpf_prog + """ + vmlinux = self._context.modules[self._module_name] + layer = self._context.layers[self._layer_name] + bpf_prog_aux = linux.LinuxUtilities.container_of( + latch_tree_node.vol.offset, "bpf_prog_aux", "ksym_tnode", vmlinux + ) + bpf_prog = bpf_prog_aux.prog + bpf_start, bpf_end = bpf_prog.get_address_region() + bpf_start &= layer.address_mask + bpf_end &= layer.address_mask + + if address < bpf_start: + return -1 + elif address > bpf_end: + # Keep 'key > end' instead of 'key >= end'. This detects return addresses + # within the program when the final instruction in a stack trace is a call. + return 1 + else: + return 0 + + def _find_bpf_ksym( + self, address: int + ) -> Optional[interfaces.objects.ObjectInterface]: + """Search for the respective bpf_ksym based on a symbol address. + Based on __bpf_address_lookup & bpf_ksym_find() for kernels >= 5.7 + + Args: + address: The memory address to search for + + Returns: + A bpf_ksym object if found; otherwise, returns None. + """ + vmlinux = self._context.modules[self._module_name] + if not self._kas_config.bpf_tree_address: + return None + + bpf_latch_tree_root = vmlinux.object( + object_type="latch_tree_root", + offset=self._kas_config.bpf_tree_address, + absolute=True, + ) + latch_tree_node = bpf_latch_tree_root.find( + address, self._bpf_tree_comp_bpf_ksym + ) + if not latch_tree_node: + return None + + bpf_ksym = linux.LinuxUtilities.container_of( + latch_tree_node.vol.offset, "bpf_ksym", "tnode", vmlinux + ) + return bpf_ksym + + def _bpf_tree_comp_bpf_ksym( + self, address: int, latch_tree_node: interfaces.objects.ObjectInterface + ) -> int: + """Comparison function used by _find_bpf_ksym. + + Based on bpf_tree_comp in kernels >= 5.7 + + Args: + address: The memory address to search for + latch_tree_node: A latch tree node + + Returns: + 0: equal, >0: key is greater, <0: key is less than this bpf_prog + """ + # + vmlinux = self._context.modules[self._module_name] + layer = self._context.layers[self._layer_name] + bpf_ksym = linux.LinuxUtilities.container_of( + latch_tree_node.vol.offset, "bpf_ksym", "tnode", vmlinux + ) + bpf_start = bpf_ksym.start & layer.address_mask + bpf_end = bpf_ksym.end & layer.address_mask + + if address < bpf_start: + return -1 + elif address > bpf_end: + # Keep 'address > bpf_end' instead of 'address >= bpf_end'. This detects return + # addresses within the program when the final instruction in a stack trace is a call. + return 1 + else: + return 0 + + def ftrace_lookup_address(self, address: int) -> Optional[KASSymbol]: + """Search for a ftrace symbol based on its address. + + Based on ftrace_mod_address_lookup() + + Args: + address: The memory address to search for + + Returns: + The matching KASSymbol if found, or None if no match is found. + """ + + # Filter by address and return only the first matching result. + for kassymbol in self._ftrace_mod_get_symbols(address): + return kassymbol + + for kassymbol in self._ftrace_get_trampoline_symbols(address): + return kassymbol + + return None + + def _core_lookup_name_slow(self, name) -> Optional[KASSymbol]: + """Search a core symbol by name + + Based on kallsyms_lookup_name in kernels < 6.2 + + Args: + name: The symbol name to search for. + + Returns: + A KASSymbol object + """ + # kernels < 6.2 60443c88f3a89fd303a9e8c0e84895910675c316 + current_offset = 0 + for sym_idx in range(self._kallsyms_num_syms): + kassymbol, compressed_length = self._get_symbol(current_offset, sym_idx) + if kassymbol and name == kassymbol.name: + return kassymbol + + current_offset += compressed_length + 1 + + return None + + @functools.cached_property + def _kallsyms_seqs_of_names(self): + vmlinux = self._context.modules[self._module_name] + symbol_table_name = vmlinux.symbol_table_name + unsigned_char_symname = symbol_table_name + constants.BANG + "unsigned char" + # See 19bd8981dc2ee35fdc81ab1b0104b607c917d470: 3 bytes per index + array_size = 3 * self._kallsyms_num_syms + kallsyms_seqs_of_names = vmlinux.object( + object_type="array", + offset=self._kas_config.seqs_of_names_address, + subtype=vmlinux.get_type(unsigned_char_symname), + count=array_size, + absolute=True, + ) + return kallsyms_seqs_of_names + + def _get_symbol_seq(self, index: int) -> int: + # See 19bd8981dc2ee35fdc81ab1b0104b607c917d470 + bits = 3 + seq = 0 + for i in range(bits): + seq = (seq << 8) | self._kallsyms_seqs_of_names[bits * index + i] + return seq + + def _get_symbol_by_index(self, index) -> Tuple[KASSymbolBasic, int]: + seq = self._get_symbol_seq(index) + offset = self._get_symbol_offset(seq) + kassymbolbasic, _compressed_length = self._expand_symbol(offset) + return kassymbolbasic + + def _lookup_name_index(self, name: str) -> Optional[int]: + # based on kallsyms_lookup_names + high = self._kallsyms_num_syms - 1 + low = 0 + + while low <= high: + mid = (low + high) // 2 + kassymbolbasic = self._get_symbol_by_index(mid) + if not kassymbolbasic: + return None + + ret = self._cmp_symbol_name(name, kassymbolbasic.name) + if ret > 0: + low = mid + 1 + elif ret < 0: + high = mid - 1 + else: + break + + if low > high: + # Not found + return None + + low = mid + while low: + kassymbolbasic = self._get_symbol_by_index(low - 1) + if not kassymbolbasic: + return None + if self._cmp_symbol_name(name, kassymbolbasic.name) != 0: + return low + low -= 1 + + return None + + def _core_lookup_name_fast(self, name: str) -> Optional[KASSymbol]: + """Search a core symbol by name + + Based on kallsyms_lookup_name in kernels >= 6.2 + + Args: + name: The symbol name to search for + + Returns: + A KASSymbol object + """ + # kernels >= 6.2 60443c88f3a89fd303a9e8c0e84895910675c316 + index = self._lookup_name_index(name) + if not index: + return None + + seq = self._get_symbol_seq(index) + offset = self._get_symbol_offset(seq) + kassymbolbasic, _compressed_length = self._expand_symbol(offset) + sym_address = self._get_symbol_address_by_index(seq) + _seq, sym_size = self._get_symbol_pos(sym_address) + + kas_symbol = KASSymbol( + name=kassymbolbasic.name, + type=kassymbolbasic.type, + address=sym_address, + size=sym_size, + module_name=self._CORE_MODULE_NAME, + subsystem=self._CORE_SUBSYSTEM_NAME, + ) + kas_symbol.set_exported_from_type() + return kas_symbol + + def _kallsyms_lookup_name_modules(self, name: str) -> Optional[KASSymbol]: + """_summary_ + + Based on module_kallsyms_lookup_name + + Args: + name: The symbol name to search for. + + Returns: + A KASSymbol object + """ + for kassymbol in self.get_modules_symbols(name): + if name == kassymbol.name: + # First match only + return kassymbol + return None + + def lookup_name(self, name: str) -> Optional[KASSymbol]: + """Search symbols by name. + WARNING: This function is super slow. The kernel does not index the symbols by + name, so the it is a linear search. + + Based on kallsyms_lookup_name + + Args: + name: The symbol name to search for. + + Returns: + A KASSymbol object + """ + if self._kas_config.seqs_of_names_address: + # kernels >= 6.2: + # 60443c88f3a89fd303a9e8c0e84895910675c316 and 19bd8981dc2ee35fdc81ab1b0104b607c917d470 + kassymbol = self._core_lookup_name_fast(name) + else: + # kernels < 6.2 + kassymbol = self._core_lookup_name_slow(name) + + if kassymbol: + return kassymbol + + return self._kallsyms_lookup_name_modules(name)