From 69e6e59daf39ec7a0cf9a82fb13762d4389c95a4 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 12:01:21 -0500 Subject: [PATCH 01/17] Add new pe_symbols API, debug registers plugin, unhooked system calls plugin --- .../plugins/windows/debugregisters.py | 223 ++++++ .../framework/plugins/windows/pe_symbols.py | 732 ++++++++++++++++++ .../plugins/windows/unhooked_system_calls.py | 183 +++++ .../framework/plugins/windows/vadinfo.py | 88 ++- 4 files changed, 1219 insertions(+), 7 deletions(-) create mode 100644 volatility3/framework/plugins/windows/debugregisters.py create mode 100644 volatility3/framework/plugins/windows/pe_symbols.py create mode 100644 volatility3/framework/plugins/windows/unhooked_system_calls.py diff --git a/volatility3/framework/plugins/windows/debugregisters.py b/volatility3/framework/plugins/windows/debugregisters.py new file mode 100644 index 000000000..5f3c2e519 --- /dev/null +++ b/volatility3/framework/plugins/windows/debugregisters.py @@ -0,0 +1,223 @@ +import logging + +from typing import Tuple, Optional, Generator, List, Dict + +from functools import partial + +from volatility3.framework import renderers, interfaces, exceptions +from volatility3.framework.configuration import requirements +from volatility3.framework.renderers import format_hints +import volatility3.plugins.windows.pslist as pslist +import volatility3.plugins.windows.threads as threads +import volatility3.plugins.windows.vadinfo as vadinfo +import volatility3.plugins.windows.pe_symbols as pe_symbols + +vollog = logging.getLogger(__name__) + + +class DebugRegisters(interfaces.plugins.PluginInterface): + # version 2.6.0 adds support for scanning for 'Ethread' structures by pool tags + _required_framework_version = (2, 6, 0) + _version = (1, 0, 0) + + @classmethod + def get_requirements(cls) -> List: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(2, 0, 0) + ), + requirements.VersionRequirement( + name="vadinfo", component=vadinfo.VadInfo, version=(2, 0, 0) + ), + requirements.VersionRequirement( + name="pe_symbols", component=pe_symbols.PESymbols, version=(1, 0, 0) + ), + ] + + def _get_debug_info( + self, ethread: interfaces.objects.ObjectInterface + ) -> Optional[Tuple[interfaces.objects.ObjectInterface, int, int, int, int, int]]: + """ + Gathers information related to the debug registers for the given thread + """ + try: + dr7 = ethread.Tcb.TrapFrame.Dr7 + state = ethread.Tcb.State + except exceptions.InvalidAddressException: + return None + + # 0 = debug registers not active + # 4 = terminated + if dr7 == 0 or state == 4: + return None + + try: + owner_proc = ethread.owning_process() + except (AttributeError, exceptions.InvalidAddressException): + return None + + dr0 = ethread.Tcb.TrapFrame.Dr0 + dr1 = ethread.Tcb.TrapFrame.Dr1 + dr2 = ethread.Tcb.TrapFrame.Dr2 + dr3 = ethread.Tcb.TrapFrame.Dr3 + + # bail if all are 0 + if not (dr0 or dr1 or dr2 or dr3): + return None + + return owner_proc, dr7, dr0, dr1, dr2, dr3 + + def _get_vads( + self, + vads_cache: Dict[int, List[Tuple[int, int, str]]], + owner_proc: interfaces.objects.ObjectInterface, + ) -> Optional[List[Tuple[int, int, str]]]: + if owner_proc.vol.offset in vads_cache: + vads = vads_cache[owner_proc.vol.offset] + else: + vads = vadinfo.VadInfo.get_proc_vads_with_file_paths(owner_proc) + vads_cache[owner_proc.vol.offset] = vads + + # smear or terminated process + if len(vads) == 0: + return None + + return vads + + def _generator( + self, + ) -> Generator[ + Tuple[ + int, + Tuple[ + str, + int, + int, + int, + int, + format_hints.Hex, + str, + str, + format_hints.Hex, + str, + str, + format_hints.Hex, + str, + str, + format_hints.Hex, + str, + str, + ], + ], + None, + None, + ]: + kernel = self.context.modules[self.config["kernel"]] + + vads_cache: Dict[int, List[Tuple[int, int, str]]] = {} + + proc_modules = None + + procs = pslist.PsList.list_processes( + context=self.context, + layer_name=kernel.layer_name, + symbol_table=kernel.symbol_table_name, + ) + + for proc in procs: + for thread in threads.Threads.list_threads(kernel, proc): + debug_info = self._get_debug_info(thread) + if not debug_info: + continue + + owner_proc, dr7, dr0, dr1, dr2, dr3 = debug_info + + vads = self._get_vads(vads_cache, owner_proc) + if not vads: + continue + + # this lookup takes a while, so only perform if we need to + if not proc_modules: + proc_modules = pe_symbols.PESymbols.get_process_modules( + self.context, kernel.layer_name, kernel.symbol_table_name, None + ) + path_and_symbol = partial( + pe_symbols.PESymbols.path_and_symbol_for_address, + self.context, + self.config_path, + proc_modules, + ) + + file0, sym0 = path_and_symbol(vads, dr0) + file1, sym1 = path_and_symbol(vads, dr1) + file2, sym2 = path_and_symbol(vads, dr2) + file3, sym3 = path_and_symbol(vads, dr3) + + # if none map to an actual file VAD then bail + if not ( + isinstance(file0, str) + or isinstance(file1, str) + or isinstance(file2, str) + or isinstance(file3, str) + ): + continue + + process_name = owner_proc.ImageFileName.cast( + "string", + max_length=owner_proc.ImageFileName.vol.count, + errors="replace", + ) + + thread_tid = thread.Cid.UniqueThread + + yield ( + 0, + ( + process_name, + owner_proc.UniqueProcessId, + thread_tid, + thread.Tcb.State, + dr7, + format_hints.Hex(dr0), + file0, + sym0, + format_hints.Hex(dr1), + file1, + sym1, + format_hints.Hex(dr2), + file2, + sym2, + format_hints.Hex(dr3), + file3, + sym3, + ), + ) + + def run(self) -> renderers.TreeGrid: + return renderers.TreeGrid( + [ + ("Process", str), + ("PID", int), + ("TID", int), + ("State", int), + ("Dr7", int), + ("Dr0", format_hints.Hex), + ("Range0", str), + ("Symbol0", str), + ("Dr1", format_hints.Hex), + ("Range1", str), + ("Symbol1", str), + ("Dr2", format_hints.Hex), + ("Range2", str), + ("Symbol2", str), + ("Dr3", format_hints.Hex), + ("Range3", str), + ("Symbol3", str), + ], + self._generator(), + ) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py new file mode 100644 index 000000000..b7faeaf55 --- /dev/null +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -0,0 +1,732 @@ +# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 + + +import io +import logging + +from typing import Dict, Tuple, Optional, List, Generator, Union + +import pefile + +from volatility3.framework import interfaces, exceptions +from volatility3.framework import renderers, constants +from volatility3.framework.configuration import requirements +from volatility3.framework.renderers import format_hints +from volatility3.framework.symbols import intermed +from volatility3.framework.symbols.windows import pdbutil +from volatility3.framework.symbols.windows.extensions import pe +from volatility3.plugins.windows import pslist, vadinfo, modules + +vollog = logging.getLogger(__name__) + + +class PESymbolFinder: + """ + Interface for PE symbol finding classes + This interface provides a standard way for the calling code to + lookup symbols by name or address + """ + + cached_str = Union[str, None] + cached_str_dict = Dict[str, cached_str] + + cached_int = Union[int, None] + cached_int_dict = Dict[str, cached_int] + + cached_value = Union[int, str, None] + cached_value_dict = Dict[str, Union[Dict[str, List[str]], Dict[str, List[int]]]] + + def __init__( + self, + layer_name: str, + mod_name: str, + module_start: int, + symbol_module: Union[interfaces.context.ModuleInterface, pefile.ExportDirData], + ): + self._layer_name = layer_name + self._mod_name = mod_name + self._module_start = module_start + self._symbol_module = symbol_module + + self._address_cache: PESymbolFinder.cached_int_dict = {} + self._name_cache: PESymbolFinder.cached_str_dict = {} + + def _get_cache_key(self, value: cached_value) -> str: + """ + Maintain a cache for symbol lookups to avoid re-walking of PDB symbols or export tables + within the same module for the same address in the same layer + """ + return f"{self._layer_name}|{self._mod_name}|{value}" + + def get_name_for_address(self, address: int) -> cached_str: + cached_key = self._get_cache_key(address) + if cached_key not in self._name_cache: + name = self._do_get_name(address) + self._name_cache[cached_key] = name + + return self._name_cache[cached_key] + + def get_address_for_name(self, name: str) -> cached_int: + cached_key = self._get_cache_key(name) + if cached_key not in self._address_cache: + address = self._do_get_address(name) + self._address_cache[cached_key] = address + + return self._address_cache[cached_key] + + def _do_get_name(self, address: int) -> cached_str: + raise NotImplementedError("_do_get_name must be overwritten") + + def _do_get_address(self, name: str) -> cached_int: + raise NotImplementedError("_do_get_address must be overwritten") + + +class PDBSymbolFinder(PESymbolFinder): + """ + PESymbolFinder implementation for PDB modules + """ + + def _do_get_address(self, name: str) -> PESymbolFinder.cached_int: + try: + return self._symbol_module.get_absolute_symbol_address(name) + except exceptions.SymbolError: + return None + + def _do_get_name(self, address: int) -> PESymbolFinder.cached_str: + try: + name = self._symbol_module.get_symbols_by_absolute_location(address)[0] + return name.split(constants.BANG)[1] + except (exceptions.SymbolError, IndexError): + return None + + +class ExportSymbolFinder(PESymbolFinder): + """ + PESymbolFinder implementation for PDB modules + """ + + def _get_name(self, export: pefile.ExportData) -> Optional[str]: + # AttributeError throws on empty or ordinal-only exports + try: + return export.name.decode("ascii") + except AttributeError: + return None + + def _do_get_name(self, address: int) -> PESymbolFinder.cached_str: + for export in self._symbol_module: + if export.address + self._module_start == address: + return self._get_name(export) + + return None + + def _do_get_address(self, name: str) -> PESymbolFinder.cached_int: + for export in self._symbol_module: + sym_name = self._get_name(export) + if sym_name and sym_name == name: + return self._module_start + export.address + + return None + + +class PESymbols(interfaces.plugins.PluginInterface): + """Prints symbols in PE files in process and kernel memory""" + + _required_framework_version = (2, 7, 0) + + _version = (1, 0, 0) + + # used for special handling of the kernel PDB file. See later notes + os_module_name = "ntoskrnl.exe" + + # keys for specifying wanted names and/or addresses + # used for consistent access between the API and plugins + wanted_names = "names" + wanted_addresses = "addresses" + + # how wanted modules/symbols are specified, such as: + # {"ntdll.dll" : {wanted_addresses : [42, 43, 43]}} + # {"ntdll.dll" : {wanted_names : ["NtCreateThread"]}} + filter_modules_type = Dict[str, Union[Dict[str, List[str]], Dict[str, List[int]]]] + + # holds resolved symbols + # {"ntdll.dll": [("Bob", 123), ("Alice", 456)]} + found_symbols_type = Dict[str, List[Tuple[str, int]]] + + # used to hold informatin about a range (VAD or kernel module) + # (start address, size, file path) + range_type = Tuple[int, int, str] + ranges_type = List[range_type] + + @classmethod + def get_requirements(cls) -> List: + # Since we're calling the plugin, make sure we have the plugin's requirements + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(2, 0, 0) + ), + requirements.VersionRequirement( + name="vadinfo", component=vadinfo.VadInfo, version=(2, 0, 0) + ), + requirements.VersionRequirement( + name="modules", component=modules.Modules, version=(2, 0, 0) + ), + requirements.VersionRequirement( + name="pdbutil", component=pdbutil.PDBUtility, version=(1, 0, 0) + ), + requirements.ChoiceRequirement( + name="source", + description="Where to resolve symbols.", + choices=["kernel", "processes"], + optional=False, + ), + requirements.StringRequirement( + name="module", + description='Module in which to resolve symbols. Use "ntoskrnl.exe" to resolve in the base kernel executable.', + optional=False, + ), + requirements.StringRequirement( + name="symbol", + description="Symbol name to resolve", + optional=True, + ), + requirements.IntRequirement( + name="address", + description="Address of symbol to resolve", + optional=True, + ), + ] + + @staticmethod + def _get_pefile_obj( + context: interfaces.context.ContextInterface, + pe_table_name: str, + layer_name: str, + base_address: int, + ) -> Optional[pefile.PE]: + """ + Attempts to pefile object from the bytes of the PE file + + Args: + pe_table_name: name of the pe types table + layer_name: name of the process layer + base_address: base address of the module + + Returns: + the constructed pefile object + """ + pe_data = io.BytesIO() + + try: + dos_header = context.object( + pe_table_name + constants.BANG + "_IMAGE_DOS_HEADER", + offset=base_address, + layer_name=layer_name, + ) + + for offset, data in dos_header.reconstruct(): + pe_data.seek(offset) + pe_data.write(data) + + pe_ret = pefile.PE(data=pe_data.getvalue(), fast_load=True) + + except exceptions.InvalidAddressException: + pe_ret = None + + return pe_ret + + @staticmethod + def range_info_for_address( + ranges: ranges_type, address: int + ) -> Optional[range_type]: + """ + Helper for getting the range information for an address + """ + for start, size, filepath in ranges: + if start <= address < start + size: + return start, size, filepath + + return None + + @staticmethod + def filepath_for_address(ranges: ranges_type, address: int) -> Optional[str]: + """ + Helper to get the file path for an address + """ + info = PESymbols.range_info_for_address(ranges, address) + if info: + return info[2] + + return None + + @staticmethod + def filename_for_path(filepath: str) -> str: + """ + Consistent way to get the filename + """ + return filepath.split("\\")[-1] + + @staticmethod + def addresses_for_process_symbols( + context: interfaces.context.ContextInterface, + config_path: str, + layer_name: str, + symbol_table_name: str, + symbols: filter_modules_type, + ) -> found_symbols_type: + collected_modules = PESymbols.get_process_modules( + context, layer_name, symbol_table_name, symbols + ) + + found_symbols = PESymbols.find_symbols( + context, config_path, symbols, collected_modules + ) + + for mod_name, unresolved_symbols in symbols.items(): + for symbol in unresolved_symbols: + vollog.debug(f"Unable to resolve symbol {symbol} in module {mod_name}") + + return found_symbols + + @staticmethod + def path_and_symbol_for_address( + context: interfaces.context.ContextInterface, + config_path: str, + collected_modules: Dict[str, List[Tuple[str, int, int]]], + ranges: ranges_type, + address: int, + ) -> Tuple[str, str]: + """ + Method for plugins to determine the file path and symbol name for a given address + + collected_modules: return value from `get_kernel_modules` or `get_process_modules` + ranges: the memory ranges to examine in this layer. + address: address to resolve to its symbol name + """ + + if not address: + return renderers.NotApplicableValue(), renderers.NotApplicableValue() + + filepath = PESymbols.filepath_for_address(ranges, address) + + if not filepath: + return renderers.NotAvailableValue(), renderers.NotAvailableValue() + + filename = PESymbols.filename_for_path(filepath).lower() + + # setup to resolve the address + filter_module: PESymbols.filter_modules_type = { + filename: {PESymbols.wanted_addresses: [address]} + } + + found_symbols = PESymbols.find_symbols( + context, config_path, filter_module, collected_modules + ) + + if not found_symbols or not found_symbols[filename]: + return renderers.NotAvailableValue(), renderers.NotAvailableValue() + + return filepath, found_symbols[filename][0][0] + + @staticmethod + def _get_exported_symbols( + context: interfaces.context.ContextInterface, + pe_table_name: str, + mod_name: str, + module_info: Tuple[str, int, int], + ) -> Optional[ExportSymbolFinder]: + """ + Attempts to locate symbols based on export analysis + + mod_name: lower case name of the module to resolve symbols in + module_info: (layer_name, module_start, module_size) of the module to examine + """ + + layer_name = module_info[0] + module_start = module_info[1] + + # we need a valid PE with an export table + pe_module = PESymbols._get_pefile_obj( + context, pe_table_name, layer_name, module_start + ) + if not pe_module: + return None + + pe_module.parse_data_directories( + directories=[pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"]] + ) + if not hasattr(pe_module, "DIRECTORY_ENTRY_EXPORT"): + return None + + return ExportSymbolFinder( + layer_name, mod_name, module_start, pe_module.DIRECTORY_ENTRY_EXPORT.symbols + ) + + @staticmethod + def _get_pdb_module( + context: interfaces.context.ContextInterface, + config_path: str, + mod_name: str, + module_info: Tuple[str, int, int], + ) -> Optional[PDBSymbolFinder]: + """ + Attempts to locate symbols based on PDB analysis + + mod_name: lower case name of the module to resolve symbols in + module_info: (layer_name, module_start, module_size) of the module to examine + """ + + mod_symbols = None + + layer_name, module_start, module_size = module_info + + # the PDB name of the kernel file is not consistent for an exe, for example, + # a `ntoskrnl.exe` can have an internal PDB name of any of the ones in the following list + # The code attempts to find all possible PDBs to ensure the best chance of recovery + if mod_name == PESymbols.os_module_name: + pdb_names = ["ntkrnlmp.pdb", "ntkrnlpa.pdb", "ntkrpamp.pdb", "ntoskrnl.pdb"] + + # for non-kernel files, replace the exe, sys, or dll extension with pdb + else: + mod_name = mod_name[:-3] + "pdb" + first_upper = mod_name[0].upper() + mod_name[1:] + pdb_names = [mod_name, first_upper] + + # loop through each PDB name (will be just one for all but the kernel) + for pdb_name in pdb_names: + try: + mod_symbols = pdbutil.PDBUtility.symbol_table_from_pdb( + context, + interfaces.configuration.path_join(config_path, mod_name), + layer_name, + pdb_name, + module_start, + module_size, + ) + + if mod_symbols: + break + + # this exception is expected when the PDB can't be found or downloaded + except exceptions.VolatilityException: + continue + + # this is not expected - it means pdbconv broke when parsing the PDB + except TypeError as e: + vollog.error( + f"Unable to parse PDB file for module {pdb_name} -> {e}. Please file a bug on the GitHub issue tracker." + ) + + # cannot do anything without the symbols + if not mod_symbols: + return None + + pdb_module = context.module( + mod_symbols, layer_name=layer_name, offset=module_start + ) + + return PDBSymbolFinder(layer_name, mod_name, module_start, pdb_module) + + @staticmethod + def _find_symbols_through_pdb( + context: interfaces.context.ContextInterface, + config_path: str, + module_instances: List[Tuple[str, int, int]], + mod_name: str, + ) -> Generator[PDBSymbolFinder, None, None]: + """ + Attempts to resolve the symbols in `wanted_symbols` through PDB analysis + """ + for module_info in module_instances: + mod_module = PESymbols._get_pdb_module( + context, config_path, mod_name, module_info + ) + if mod_module: + yield mod_module + + @staticmethod + def _find_symbols_through_exports( + context: interfaces.context.ContextInterface, + config_path: str, + module_instances: List[Tuple[str, int, int]], + mod_name: str, + ) -> Generator[ExportSymbolFinder, None, None]: + """ + Attempts to resolve the symbols in `wanted_symbols` through export analysis + """ + pe_table_name = intermed.IntermediateSymbolTable.create( + context, config_path, "windows", "pe", class_types=pe.class_types + ) + + # for each process layer and VAD, construct a PE and examine the export table + for module_info in module_instances: + exported_symbols = PESymbols._get_exported_symbols( + context, pe_table_name, mod_name, module_info + ) + if exported_symbols: + yield exported_symbols + + @staticmethod + def _get_symbol_value( + wanted_modules: PESymbolFinder.cached_value_dict, + mod_name: str, + symbol_resolver: PESymbolFinder, + ) -> Generator[Tuple[str, int], None, None]: + """ + Enumerates the symbols specified as wanted by the calling plugin + """ + wanted_symbols = wanted_modules[mod_name] + + if ( + PESymbols.wanted_names not in wanted_symbols + and PESymbols.wanted_addresses not in wanted_symbols + ): + vollog.warning( + f"Invalid `wanted_symbols` sent to `find_symbols` for module {mod_name}. addresses and names keys both misssing." + ) + return + + symbol_keys = [ + (PESymbols.wanted_names, "get_address_for_name"), + (PESymbols.wanted_addresses, "get_name_for_address"), + ] + + for symbol_key, symbol_getter in symbol_keys: + # address or name + if symbol_key in wanted_symbols: + # walk each wanted address or name + for wanted_value in wanted_symbols[symbol_key]: + symbol_value = symbol_resolver.__getattribute__(symbol_getter)( + wanted_value + ) + if symbol_value: + # yield out symbol name, symbol address + if symbol_key == PESymbols.wanted_names: + yield wanted_value, symbol_value # type: ignore + else: + yield symbol_value, wanted_value # type: ignore + + index = wanted_modules[mod_name][symbol_key].index(wanted_value) # type: ignore + + del wanted_modules[mod_name][symbol_key][index] + + # if all names or addresses from a module are found, delete the key + if not wanted_modules[mod_name][symbol_key]: + del wanted_modules[mod_name][symbol_key] + break + + @staticmethod + def _resolve_symbols_through_methods( + context: interfaces.context.ContextInterface, + config_path: str, + module_instances: List[Tuple[str, int, int]], + wanted_modules: PESymbolFinder.cached_value_dict, + mod_name: str, + ) -> Generator[Tuple[str, int], None, None]: + """ + Attempts to resolve every wanted symbol in `mod_name` + Every layer is enumerated for maximum chance of recovery + """ + symbol_resolving_methods = [ + PESymbols._find_symbols_through_pdb, + PESymbols._find_symbols_through_exports, + ] + + for method in symbol_resolving_methods: + for symbol_resolver in method( + context, config_path, module_instances, mod_name + ): + vollog.debug(f"Have resolver for method {method}") + yield from PESymbols._get_symbol_value( + wanted_modules, mod_name, symbol_resolver + ) + + if not wanted_modules[mod_name]: + break + + if not wanted_modules[mod_name]: + break + + @staticmethod + def find_symbols( + context: interfaces.context.ContextInterface, + config_path: str, + wanted_modules: PESymbolFinder.cached_value_dict, + collected_modules: Dict[str, List[Tuple[str, int, int]]], + ) -> found_symbols_type: + """ + Loops through each method of symbol analysis until each wanted symbol is found + Returns the resolved symbols as a dictionary that includes the name and runtime address + """ + found_symbols: PESymbols.found_symbols_type = {} + + for mod_name in wanted_modules: + if mod_name not in collected_modules: + continue + + module_instances = collected_modules[mod_name] + + # try to resolve the symbols for `mod_name` through each method (PDB and export table currently) + for symbol_name, address in PESymbols._resolve_symbols_through_methods( + context, config_path, module_instances, wanted_modules, mod_name + ): + if mod_name not in found_symbols: + found_symbols[mod_name] = [] + + found_symbols[mod_name].append((symbol_name, address)) + + # stop processing the layers (processes) if we found all the symbols for this module + if not wanted_modules[mod_name]: + break + + # stop processing this module if/when all symbols are found + if not wanted_modules[mod_name]: + del wanted_modules[mod_name] + break + + return found_symbols + + @staticmethod + def get_kernel_modules( + context: interfaces.context.ContextInterface, + layer_name: str, + symbol_table: str, + filter_modules: Optional[filter_modules_type], + ) -> Dict[str, List[Tuple[str, int, int]]]: + """ + Walks the kernel module list and finds the session layer, base, and size of each wanted module + """ + found_modules: Dict[str, List[Tuple[str, int, int]]] = {} + + if filter_modules: + # create a tuple of module names for use with `endswith` + filter_modules_check = tuple([key.lower() for key in filter_modules.keys()]) + else: + filter_modules_check = None + + session_layers = list( + modules.Modules.get_session_layers(context, layer_name, symbol_table) + ) + + # special handling for the kernel + gather_kernel = ( + filter_modules_check and PESymbols.os_module_name in filter_modules_check + ) + + for index, mod in enumerate( + modules.Modules.list_modules(context, layer_name, symbol_table) + ): + try: + mod_name = str(mod.BaseDllName.get_string().lower()) + except exceptions.InvalidAddressException: + continue + + # to analyze, it must either be the kernel or a wanted module + if not filter_modules_check or (gather_kernel and index == 0): + mod_name = PESymbols.os_module_name + elif filter_modules_check and not mod_name.endswith(filter_modules_check): + continue + + # we won't find symbol information if we can't analyze the module + session_layer_name = modules.Modules.find_session_layer( + context, session_layers, mod.DllBase + ) + if not session_layer_name: + continue + + if mod_name not in found_modules: + found_modules[mod_name] = [] + + found_modules[mod_name].append( + (session_layer_name, mod.DllBase, mod.SizeOfImage) + ) + + return found_modules + + @staticmethod + def get_process_modules( + context: interfaces.context.ContextInterface, + layer_name: str, + symbol_table: str, + filter_modules: Optional[filter_modules_type], + ) -> Dict[str, List[Tuple[str, int, int]]]: + """ + Walks the process list and each process' VAD to determine the base address and size of wanted modules + """ + proc_modules: Dict[str, List[Tuple[str, int, int]]] = {} + + if filter_modules: + # create a tuple of module names for use with `endswith` + filter_modules_check = tuple([key.lower() for key in filter_modules.keys()]) + else: + filter_modules_check = None + + for _, proc_layer_name, vads in vadinfo.VadInfo.get_all_vads_with_file_paths( + context, layer_name, symbol_table + ): + for vad_start, vad_size, filepath in vads: + filename = PESymbols.filename_for_path(filepath) + + if filter_modules_check and not filename.endswith(filter_modules_check): + continue + + # track each module along with the process layer and range to find it + if filename not in proc_modules: + proc_modules[filename] = [] + + proc_modules[filename].append((proc_layer_name, vad_start, vad_size)) + + return proc_modules + + def _generator(self) -> Generator[Tuple[int, Tuple[str, str, int]], None, None]: + kernel = self.context.modules[self.config["kernel"]] + + if self.config["symbol"]: + filter_module = { + self.config["module"].lower(): { + PESymbols.wanted_names: [self.config["symbol"]] + } + } + + elif self.config["address"]: + filter_module = { + self.config["module"].lower(): { + PESymbols.wanted_addresses: [self.config["address"]] + } + } + + else: + vollog.error("--address or --symbol must be specified") + return + + if self.config["source"] == "kernel": + module_resolver = self.get_kernel_modules + else: + module_resolver = self.get_process_modules + + collected_modules = module_resolver( + self.context, kernel.layer_name, kernel.symbol_table_name, filter_module + ) + + found_symbols = PESymbols.find_symbols( + self.context, self.config_path, filter_module, collected_modules + ) + + for module, symbols in found_symbols.items(): + for symbol, address in symbols: + yield (0, (module, symbol, format_hints.Hex(address))) + + def run(self) -> renderers.TreeGrid: + return renderers.TreeGrid( + [ + ("Module", str), + ("Symbol", str), + ("Address", format_hints.Hex), + ], + self._generator(), + ) diff --git a/volatility3/framework/plugins/windows/unhooked_system_calls.py b/volatility3/framework/plugins/windows/unhooked_system_calls.py new file mode 100644 index 000000000..0438bc9e3 --- /dev/null +++ b/volatility3/framework/plugins/windows/unhooked_system_calls.py @@ -0,0 +1,183 @@ +import logging + +from typing import Dict, Tuple, List, Generator + +from volatility3.framework import interfaces, exceptions +from volatility3.framework import renderers +from volatility3.framework.configuration import requirements +from volatility3.framework.objects import utility +from volatility3.plugins.windows import pslist, pe_symbols + +vollog = logging.getLogger(__name__) + + +class unhooked_system_calls(interfaces.plugins.PluginInterface): + """Looks for signs of Skeleton Key malware""" + + _required_framework_version = (2, 4, 0) + + system_calls = { + "ntdll.dll": { + pe_symbols.PESymbols.wanted_names: [ + "NtCreateThread", + "NtProtectVirtualMemory", + "NtReadVirtualMemory", + "NtOpenProcess", + "NtWriteFile", + "NtQueryVirtualMemory", + "NtAllocateVirtualMemory", + "NtWorkerFactoryWorkerReady", + "NtAcceptConnectPort", + "NtAddDriverEntry", + "NtAdjustPrivilegesToken", + "NtAlpcCreatePort", + "NtClose", + "NtCreateFile", + "NtCreateMutant", + "NtOpenFile", + "NtOpenIoCompletion", + "NtOpenJobObject", + "NtOpenKey", + "NtOpenKeyEx", + "NtOpenThread", + "NtOpenThreadToken", + "NtOpenThreadTokenEx", + "NtWriteVirtualMemory", + "NtTraceEvent", + "NtTranslateFilePath", + "NtUmsThreadYield", + "NtUnloadDriver", + "NtUnloadKey", + "NtUnloadKey2", + "NtUnloadKeyEx", + "NtCreateKey", + "NtCreateSection", + "NtDeleteKey", + "NtDeleteValueKey", + "NtDuplicateObject", + "NtQueryValueKey", + "NtReplaceKey", + "NtRequestWaitReplyPort", + "NtRestoreKey", + "NtSetContextThread", + "NtSetSecurityObject", + "NtSetValueKey", + "NtSystemDebugControl", + "NtTerminateProcess", + ] + } + } + + _code_bytes_type = Dict[str, Dict[str, Dict[bytes, List[Tuple[int, str]]]]] + + @classmethod + def get_requirements(cls) -> List: + # Since we're calling the plugin, make sure we have the plugin's requirements + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(2, 0, 0) + ), + requirements.PluginRequirement( + name="pe_symbols", plugin=pe_symbols.PESymbols, version=(1, 0, 0) + ), + ] + + def _gather_code_bytes( + self, + kernel: interfaces.context.ModuleInterface, + found_symbols: pe_symbols.PESymbols.found_symbols_type, + ) -> _code_bytes_type: + """ + Enumerates the desired DLLs and function implementations in each process + Groups based on unique implementations of each DLLs' functions + The purpose is to detect when a function has different implementations (code) + in different processes. + This very effectively detects code injection. + """ + code_bytes: unhooked_system_calls._code_bytes_type = {} + + procs = pslist.PsList.list_processes( + context=self.context, + layer_name=kernel.layer_name, + symbol_table=kernel.symbol_table_name, + ) + + for proc in procs: + try: + proc_id = proc.UniqueProcessId + proc_name = utility.array_to_string(proc.ImageFileName) + proc_layer_name = proc.add_process_layer() + except exceptions.InvalidAddressException: + continue + + for dll_name, functions in found_symbols.items(): + for func_name, func_addr in functions: + try: + fbytes = self.context.layers[proc_layer_name].read( + func_addr, 0x20 + ) + except exceptions.InvalidAddressException: + continue + + if dll_name not in code_bytes: + code_bytes[dll_name] = {} + + if func_name not in code_bytes[dll_name]: + code_bytes[dll_name][func_name] = {} + + if fbytes not in code_bytes[dll_name][func_name]: + code_bytes[dll_name][func_name][fbytes] = [] + + code_bytes[dll_name][func_name][fbytes].append((proc_id, proc_name)) + + return code_bytes + + def _generator(self) -> Generator[Tuple[int, Tuple[str, str, int]], None, None]: + kernel = self.context.modules[self.config["kernel"]] + + found_symbols = pe_symbols.PESymbols.addresses_for_process_symbols( + self.context, + self.config_path, + kernel.layer_name, + kernel.symbol_table_name, + unhooked_system_calls.system_calls, + ) + + # code_bytes[dll_name][func_name][func_bytes] + code_bytes = self._gather_code_bytes(kernel, found_symbols) + + for functions in code_bytes.values(): + for func_name, cbb in functions.items(): + cb = list(cbb.values()) + + # same implementation in all + if len(cb) == 1: + yield 0, (func_name, "", len(cb[0])) + else: + # find the processes that are hooked for reporting + max_idx = 0 if len(cb[0]) > len(cb[1]) else 1 + small_idx = (~max_idx) & 1 + + ps = [] + + for pid, pname in cb[small_idx]: + ps.append("{:d}:{}".format(pid, pname)) + + proc_names = ", ".join(ps) + + yield 0, (func_name, proc_names, len(cb[max_idx])) + + def run(self) -> renderers.TreeGrid: + return renderers.TreeGrid( + [ + ("Function", str), + ("Distinct Implementations", str), + ("Total Implementations", int), + ], + self._generator(), + ) diff --git a/volatility3/framework/plugins/windows/vadinfo.py b/volatility3/framework/plugins/windows/vadinfo.py index abc6142fe..97a2f2455 100644 --- a/volatility3/framework/plugins/windows/vadinfo.py +++ b/volatility3/framework/plugins/windows/vadinfo.py @@ -3,13 +3,13 @@ # import logging -from typing import Callable, List, Generator, Iterable, Type, Optional +from typing import Callable, List, Generator, Iterable, Type, Optional, Tuple -from volatility3.framework import renderers, interfaces, exceptions +from volatility3.framework import renderers, interfaces, exceptions, symbols from volatility3.framework.configuration import requirements from volatility3.framework.objects import utility from volatility3.framework.renderers import format_hints -from volatility3.plugins.windows import pslist +from volatility3.plugins.windows import pslist, pe_symbols vollog = logging.getLogger(__name__) @@ -37,7 +37,7 @@ class VadInfo(interfaces.plugins.PluginInterface): _version = (2, 0, 0) MAXSIZE_DEFAULT = 1024 * 1024 * 1024 # 1 Gb - def __init__(self, *args, **kwargs): + def __init__(self, *args, **kwargs): # type: ignore super().__init__(*args, **kwargs) self._protect_values = None @@ -107,6 +107,58 @@ class VadInfo(interfaces.plugins.PluginInterface): ) return values # type: ignore + @staticmethod + def get_proc_vads_with_file_paths( + proc: interfaces.objects.ObjectInterface, + ) -> pe_symbols.PESymbols.ranges_type: + """ + Returns a list of the process' vads that map a file + """ + vads = [] + + for vad in proc.get_vad_root().traverse(): + filepath = vad.get_file_name() + if not isinstance(filepath, str) or filepath.count("\\") == 0: + continue + + vads.append((vad.get_start(), vad.get_size(), filepath)) + + return vads + + @classmethod + def get_all_vads_with_file_paths( + cls, + context: interfaces.context.ContextInterface, + layer_name: str, + symbol_table_name: str, + ) -> Generator[ + Tuple[ + interfaces.objects.ObjectInterface, str, pe_symbols.PESymbols.ranges_type + ], + None, + None, + ]: + """ + Yields each set of vads for a process that have a file mapped, along with the process itself and its layer + """ + is_32bit_arch = not symbols.symbol_table_is_64bit(context, symbol_table_name) + + procs = pslist.PsList.list_processes( + context=context, + layer_name=layer_name, + symbol_table=symbol_table_name, + ) + + for proc in procs: + try: + proc_layer_name = proc.add_process_layer() + except exceptions.InvalidAddressException: + continue + + vads = cls.get_proc_vads_with_file_paths(proc) + + yield proc, proc_layer_name, vads + @classmethod def list_vads( cls, @@ -196,11 +248,33 @@ class VadInfo(interfaces.plugins.PluginInterface): return file_handle - def _generator(self, procs): + def _generator( + self, procs: List[interfaces.objects.ObjectInterface] + ) -> Generator[ + Tuple[ + int, + Tuple[ + int, + str, + format_hints.Hex, + format_hints.Hex, + format_hints.Hex, + str, + str, + int, + int, + format_hints.Hex, + str, + str, + ], + ], + None, + None, + ]: kernel = self.context.modules[self.config["kernel"]] kernel_layer = self.context.layers[kernel.layer_name] - def passthrough(_: interfaces.objects.ObjectInterface) -> bool: + def passthrough(x: interfaces.objects.ObjectInterface) -> bool: return False filter_func = passthrough @@ -250,7 +324,7 @@ class VadInfo(interfaces.plugins.PluginInterface): ), ) - def run(self): + def run(self) -> renderers.TreeGrid: kernel = self.context.modules[self.config["kernel"]] filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None)) From 3bb9264a5770d6828487129b6ebfec509ade8680 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 12:11:16 -0500 Subject: [PATCH 02/17] formatting --- volatility3/framework/plugins/windows/debugregisters.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/plugins/windows/debugregisters.py b/volatility3/framework/plugins/windows/debugregisters.py index 5f3c2e519..c70f922ae 100644 --- a/volatility3/framework/plugins/windows/debugregisters.py +++ b/volatility3/framework/plugins/windows/debugregisters.py @@ -37,7 +37,7 @@ class DebugRegisters(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="pe_symbols", component=pe_symbols.PESymbols, version=(1, 0, 0) ), - ] + ] def _get_debug_info( self, ethread: interfaces.objects.ObjectInterface From 30cb5bd3ab4bac9f3ebb224d2f37cb98d6960545 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 12:14:06 -0500 Subject: [PATCH 03/17] Formatting that my local black doesn't understand --- volatility3/framework/plugins/windows/vadinfo.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/volatility3/framework/plugins/windows/vadinfo.py b/volatility3/framework/plugins/windows/vadinfo.py index 97a2f2455..b3b7bf5fb 100644 --- a/volatility3/framework/plugins/windows/vadinfo.py +++ b/volatility3/framework/plugins/windows/vadinfo.py @@ -249,8 +249,7 @@ class VadInfo(interfaces.plugins.PluginInterface): return file_handle def _generator( - self, procs: List[interfaces.objects.ObjectInterface] - ) -> Generator[ + self, procs: List[interfaces.objects.ObjectInterface]) -> Generator[ Tuple[ int, Tuple[ From c223ac6e762d1072f04c21276645d7c2dda79dde Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 12:31:55 -0500 Subject: [PATCH 04/17] more black help --- volatility3/framework/plugins/windows/vadinfo.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/volatility3/framework/plugins/windows/vadinfo.py b/volatility3/framework/plugins/windows/vadinfo.py index b3b7bf5fb..655e54be6 100644 --- a/volatility3/framework/plugins/windows/vadinfo.py +++ b/volatility3/framework/plugins/windows/vadinfo.py @@ -141,8 +141,6 @@ class VadInfo(interfaces.plugins.PluginInterface): """ Yields each set of vads for a process that have a file mapped, along with the process itself and its layer """ - is_32bit_arch = not symbols.symbol_table_is_64bit(context, symbol_table_name) - procs = pslist.PsList.list_processes( context=context, layer_name=layer_name, From 7b48ee4be489b4667c57cca55da84abd3c1c3d12 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 12:36:13 -0500 Subject: [PATCH 05/17] more black help --- volatility3/framework/plugins/windows/vadinfo.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/volatility3/framework/plugins/windows/vadinfo.py b/volatility3/framework/plugins/windows/vadinfo.py index 655e54be6..594c0d67c 100644 --- a/volatility3/framework/plugins/windows/vadinfo.py +++ b/volatility3/framework/plugins/windows/vadinfo.py @@ -246,8 +246,7 @@ class VadInfo(interfaces.plugins.PluginInterface): return file_handle - def _generator( - self, procs: List[interfaces.objects.ObjectInterface]) -> Generator[ + def _generator(self, procs: List[interfaces.objects.ObjectInterface]) -> Generator[ Tuple[ int, Tuple[ From bcd93616c0628fdb991866290845bbb861889f8f Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 12:36:35 -0500 Subject: [PATCH 06/17] more black help --- volatility3/framework/plugins/windows/vadinfo.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/plugins/windows/vadinfo.py b/volatility3/framework/plugins/windows/vadinfo.py index 594c0d67c..e129ad4ef 100644 --- a/volatility3/framework/plugins/windows/vadinfo.py +++ b/volatility3/framework/plugins/windows/vadinfo.py @@ -5,7 +5,7 @@ import logging from typing import Callable, List, Generator, Iterable, Type, Optional, Tuple -from volatility3.framework import renderers, interfaces, exceptions, symbols +from volatility3.framework import renderers, interfaces, exceptions from volatility3.framework.configuration import requirements from volatility3.framework.objects import utility from volatility3.framework.renderers import format_hints From fd8777287736baee75e6f1a8843b471a47b6ced2 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 12:47:42 -0500 Subject: [PATCH 07/17] Move VAD enumeration into pe_symbols --- .../framework/plugins/windows/pe_symbols.py | 57 +++++++++++++++++-- .../framework/plugins/windows/vadinfo.py | 50 ---------------- 2 files changed, 52 insertions(+), 55 deletions(-) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index b7faeaf55..51558f71b 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -16,7 +16,7 @@ from volatility3.framework.renderers import format_hints from volatility3.framework.symbols import intermed from volatility3.framework.symbols.windows import pdbutil from volatility3.framework.symbols.windows.extensions import pe -from volatility3.plugins.windows import pslist, vadinfo, modules +from volatility3.plugins.windows import pslist, modules vollog = logging.getLogger(__name__) @@ -170,9 +170,6 @@ class PESymbols(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="pslist", component=pslist.PsList, version=(2, 0, 0) ), - requirements.VersionRequirement( - name="vadinfo", component=vadinfo.VadInfo, version=(2, 0, 0) - ), requirements.VersionRequirement( name="modules", component=modules.Modules, version=(2, 0, 0) ), @@ -648,6 +645,56 @@ class PESymbols(interfaces.plugins.PluginInterface): return found_modules + @staticmethod + def get_proc_vads_with_file_paths( + proc: interfaces.objects.ObjectInterface, + ) -> ranges_type: + """ + Returns a list of the process' vads that map a file + """ + vads = [] + + for vad in proc.get_vad_root().traverse(): + filepath = vad.get_file_name() + if not isinstance(filepath, str) or filepath.count("\\") == 0: + continue + + vads.append((vad.get_start(), vad.get_size(), filepath)) + + return vads + + @classmethod + def get_all_vads_with_file_paths( + cls, + context: interfaces.context.ContextInterface, + layer_name: str, + symbol_table_name: str, + ) -> Generator[ + Tuple[ + interfaces.objects.ObjectInterface, str, ranges_type + ], + None, + None, + ]: + """ + Yields each set of vads for a process that have a file mapped, along with the process itself and its layer + """ + procs = pslist.PsList.list_processes( + context=context, + layer_name=layer_name, + symbol_table=symbol_table_name, + ) + + for proc in procs: + try: + proc_layer_name = proc.add_process_layer() + except exceptions.InvalidAddressException: + continue + + vads = PESymbols.get_proc_vads_with_file_paths(proc) + + yield proc, proc_layer_name, vads + @staticmethod def get_process_modules( context: interfaces.context.ContextInterface, @@ -666,7 +713,7 @@ class PESymbols(interfaces.plugins.PluginInterface): else: filter_modules_check = None - for _, proc_layer_name, vads in vadinfo.VadInfo.get_all_vads_with_file_paths( + for _, proc_layer_name, vads in PESymbols.get_all_vads_with_file_paths( context, layer_name, symbol_table ): for vad_start, vad_size, filepath in vads: diff --git a/volatility3/framework/plugins/windows/vadinfo.py b/volatility3/framework/plugins/windows/vadinfo.py index e129ad4ef..974793a71 100644 --- a/volatility3/framework/plugins/windows/vadinfo.py +++ b/volatility3/framework/plugins/windows/vadinfo.py @@ -107,56 +107,6 @@ class VadInfo(interfaces.plugins.PluginInterface): ) return values # type: ignore - @staticmethod - def get_proc_vads_with_file_paths( - proc: interfaces.objects.ObjectInterface, - ) -> pe_symbols.PESymbols.ranges_type: - """ - Returns a list of the process' vads that map a file - """ - vads = [] - - for vad in proc.get_vad_root().traverse(): - filepath = vad.get_file_name() - if not isinstance(filepath, str) or filepath.count("\\") == 0: - continue - - vads.append((vad.get_start(), vad.get_size(), filepath)) - - return vads - - @classmethod - def get_all_vads_with_file_paths( - cls, - context: interfaces.context.ContextInterface, - layer_name: str, - symbol_table_name: str, - ) -> Generator[ - Tuple[ - interfaces.objects.ObjectInterface, str, pe_symbols.PESymbols.ranges_type - ], - None, - None, - ]: - """ - Yields each set of vads for a process that have a file mapped, along with the process itself and its layer - """ - procs = pslist.PsList.list_processes( - context=context, - layer_name=layer_name, - symbol_table=symbol_table_name, - ) - - for proc in procs: - try: - proc_layer_name = proc.add_process_layer() - except exceptions.InvalidAddressException: - continue - - vads = cls.get_proc_vads_with_file_paths(proc) - - yield proc, proc_layer_name, vads - @classmethod def list_vads( cls, From 61cf58d97794359e3e093a97f3b8d2562661aea4 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 12:49:01 -0500 Subject: [PATCH 08/17] black fixes --- volatility3/framework/plugins/windows/pe_symbols.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index 51558f71b..10e87c552 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -670,9 +670,7 @@ class PESymbols(interfaces.plugins.PluginInterface): layer_name: str, symbol_table_name: str, ) -> Generator[ - Tuple[ - interfaces.objects.ObjectInterface, str, ranges_type - ], + Tuple[interfaces.objects.ObjectInterface, str, ranges_type], None, None, ]: From 802fc024b5a7666f1e56a641af4fe219c856f6b4 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 6 Sep 2024 13:27:04 -0500 Subject: [PATCH 09/17] switch api place --- volatility3/framework/plugins/windows/debugregisters.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/volatility3/framework/plugins/windows/debugregisters.py b/volatility3/framework/plugins/windows/debugregisters.py index c70f922ae..931ccef37 100644 --- a/volatility3/framework/plugins/windows/debugregisters.py +++ b/volatility3/framework/plugins/windows/debugregisters.py @@ -9,7 +9,6 @@ from volatility3.framework.configuration import requirements from volatility3.framework.renderers import format_hints import volatility3.plugins.windows.pslist as pslist import volatility3.plugins.windows.threads as threads -import volatility3.plugins.windows.vadinfo as vadinfo import volatility3.plugins.windows.pe_symbols as pe_symbols vollog = logging.getLogger(__name__) @@ -31,9 +30,6 @@ class DebugRegisters(interfaces.plugins.PluginInterface): requirements.VersionRequirement( name="pslist", component=pslist.PsList, version=(2, 0, 0) ), - requirements.VersionRequirement( - name="vadinfo", component=vadinfo.VadInfo, version=(2, 0, 0) - ), requirements.VersionRequirement( name="pe_symbols", component=pe_symbols.PESymbols, version=(1, 0, 0) ), @@ -80,7 +76,7 @@ class DebugRegisters(interfaces.plugins.PluginInterface): if owner_proc.vol.offset in vads_cache: vads = vads_cache[owner_proc.vol.offset] else: - vads = vadinfo.VadInfo.get_proc_vads_with_file_paths(owner_proc) + vads = pe_symbols.PESymbols.get_proc_vads_with_file_paths(owner_proc) vads_cache[owner_proc.vol.offset] = vads # smear or terminated process From b7e604d63f667663f9a00415493ec4f8b12a5024 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Tue, 10 Sep 2024 14:50:40 -0500 Subject: [PATCH 10/17] Address feedback --- .../plugins/windows/debugregisters.py | 33 +- .../framework/plugins/windows/pe_symbols.py | 394 ++++++++++++++---- .../plugins/windows/unhooked_system_calls.py | 7 +- .../framework/plugins/windows/vadinfo.py | 4 +- 4 files changed, 331 insertions(+), 107 deletions(-) diff --git a/volatility3/framework/plugins/windows/debugregisters.py b/volatility3/framework/plugins/windows/debugregisters.py index 931ccef37..65b2e625b 100644 --- a/volatility3/framework/plugins/windows/debugregisters.py +++ b/volatility3/framework/plugins/windows/debugregisters.py @@ -1,3 +1,6 @@ +# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 + import logging from typing import Tuple, Optional, Generator, List, Dict @@ -35,11 +38,16 @@ class DebugRegisters(interfaces.plugins.PluginInterface): ), ] + @staticmethod def _get_debug_info( - self, ethread: interfaces.objects.ObjectInterface + ethread: interfaces.objects.ObjectInterface, ) -> Optional[Tuple[interfaces.objects.ObjectInterface, int, int, int, int, int]]: """ Gathers information related to the debug registers for the given thread + Args: + ethread: the thread (_ETHREAD) to examine + Returns: + Tuple[interfaces.objects.ObjectInterface, int, int, int, int, int]: The owner process of the thread and the values for dr7, dr0, dr1, dr2, dr3 """ try: dr7 = ethread.Tcb.TrapFrame.Dr7 @@ -68,23 +76,6 @@ class DebugRegisters(interfaces.plugins.PluginInterface): return owner_proc, dr7, dr0, dr1, dr2, dr3 - def _get_vads( - self, - vads_cache: Dict[int, List[Tuple[int, int, str]]], - owner_proc: interfaces.objects.ObjectInterface, - ) -> Optional[List[Tuple[int, int, str]]]: - if owner_proc.vol.offset in vads_cache: - vads = vads_cache[owner_proc.vol.offset] - else: - vads = pe_symbols.PESymbols.get_proc_vads_with_file_paths(owner_proc) - vads_cache[owner_proc.vol.offset] = vads - - # smear or terminated process - if len(vads) == 0: - return None - - return vads - def _generator( self, ) -> Generator[ @@ -115,7 +106,7 @@ class DebugRegisters(interfaces.plugins.PluginInterface): ]: kernel = self.context.modules[self.config["kernel"]] - vads_cache: Dict[int, List[Tuple[int, int, str]]] = {} + vads_cache: Dict[int, pe_symbols.ranges_type] = {} proc_modules = None @@ -133,7 +124,9 @@ class DebugRegisters(interfaces.plugins.PluginInterface): owner_proc, dr7, dr0, dr1, dr2, dr3 = debug_info - vads = self._get_vads(vads_cache, owner_proc) + vads = pe_symbols.PESymbols.get_vads_for_process_cache( + vads_cache, owner_proc + ) if not vads: continue diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index 10e87c552..c9785a1c9 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -1,9 +1,9 @@ # This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 - import io import logging +import ntpath from typing import Dict, Tuple, Optional, List, Generator, Union @@ -17,9 +17,37 @@ from volatility3.framework.symbols import intermed from volatility3.framework.symbols.windows import pdbutil from volatility3.framework.symbols.windows.extensions import pe from volatility3.plugins.windows import pslist, modules +from volatility3.framework.constants.windows import KERNEL_MODULE_NAMES vollog = logging.getLogger(__name__) +# keys for specifying wanted names and/or addresses +# used for consistent access between the API and plugins +wanted_names_identifier = "names" +wanted_addresses_identifier = "addresses" + +# how wanted modules/symbols are specified, such as: +# {"ntdll.dll" : {wanted_addresses : [42, 43, 43]}} +# {"ntdll.dll" : {wanted_names : ["NtCreateThread"]}} +filter_modules_type = Dict[str, Union[Dict[str, List[str]], Dict[str, List[int]]]] + +# holds resolved symbols +# {"ntdll.dll": [("Bob", 123), ("Alice", 456)]} +found_symbols_type = Dict[str, List[Tuple[str, int]]] + +# used to hold informatin about a range (VAD or kernel module) +# (start address, size, file path) +range_type = Tuple[int, int, str] +ranges_type = List[range_type] + +# collected_modules are modules and their symbols found when walking vads or kernel modules +# Tuple of (process or kernel layer name, range start, range size) +collected_module_instance = Tuple[str, int, int] +collected_modules_info = List[collected_module_instance] +collected_modules_type = Dict[str, collected_modules_info] + +PESymbolFinders = Union[interfaces.context.ModuleInterface, pefile.ExportDirData] + class PESymbolFinder: """ @@ -28,21 +56,19 @@ class PESymbolFinder: lookup symbols by name or address """ - cached_str = Union[str, None] - cached_str_dict = Dict[str, cached_str] + cached_str_dict = Dict[str, Optional[str]] - cached_int = Union[int, None] - cached_int_dict = Dict[str, cached_int] + cached_int_dict = Dict[str, Optional[int]] cached_value = Union[int, str, None] - cached_value_dict = Dict[str, Union[Dict[str, List[str]], Dict[str, List[int]]]] + cached_value_dict = Dict[str, Dict[str, List[str]] | Dict[str, List[int]]] def __init__( self, layer_name: str, mod_name: str, module_start: int, - symbol_module: Union[interfaces.context.ModuleInterface, pefile.ExportDirData], + symbol_module: PESymbolFinders, ): self._layer_name = layer_name self._mod_name = mod_name @@ -56,10 +82,25 @@ class PESymbolFinder: """ Maintain a cache for symbol lookups to avoid re-walking of PDB symbols or export tables within the same module for the same address in the same layer + + Args: + value: The value (address or name) being cached + + Returns: + str: The constructed cache key that includes the layer and module name """ return f"{self._layer_name}|{self._mod_name}|{value}" - def get_name_for_address(self, address: int) -> cached_str: + def get_name_for_address(self, address: int) -> Optional[str]: + """ + Returns the name for the given address within the particular layer and module + + Args: + address: the address to resolve within the module + + Returns: + str: the name of the symbol, if found + """ cached_key = self._get_cache_key(address) if cached_key not in self._name_cache: name = self._do_get_name(address) @@ -67,7 +108,16 @@ class PESymbolFinder: return self._name_cache[cached_key] - def get_address_for_name(self, name: str) -> cached_int: + def get_address_for_name(self, name: str) -> Optional[int]: + """ + Returns the name for the given address within the particular layer and module + + Args: + str: the name of the symbol to resolve + + Returns: + address: the address of the symbol, if found + """ cached_key = self._get_cache_key(name) if cached_key not in self._address_cache: address = self._do_get_address(name) @@ -75,10 +125,30 @@ class PESymbolFinder: return self._address_cache[cached_key] - def _do_get_name(self, address: int) -> cached_str: + def _do_get_name(self, address: int) -> Optional[str]: + """ + Returns the name for the given address within the particular layer and module. + This method must be overwritten by sub classes. + + Args: + address: the address to resolve within the module + + Returns: + str: the name of the symbol, if found + """ raise NotImplementedError("_do_get_name must be overwritten") - def _do_get_address(self, name: str) -> cached_int: + def _do_get_address(self, name: str) -> Optional[int]: + """ + Returns the name for the given address within the particular layer and module + This method must be overwritten by sub classes. + + Args: + str: the name of the symbol to resolve + + Returns: + address: the address of the symbol, if found + """ raise NotImplementedError("_do_get_address must be overwritten") @@ -87,13 +157,31 @@ class PDBSymbolFinder(PESymbolFinder): PESymbolFinder implementation for PDB modules """ - def _do_get_address(self, name: str) -> PESymbolFinder.cached_int: + def _do_get_address(self, name: str) -> Optional[int]: + """ + _do_get_address implementation for PDBSymbolFinder + + Args: + str: the name of the symbol to resolve + + Returns: + address: the address of the symbol, if found + """ try: return self._symbol_module.get_absolute_symbol_address(name) except exceptions.SymbolError: return None - def _do_get_name(self, address: int) -> PESymbolFinder.cached_str: + def _do_get_name(self, address: int) -> Optional[str]: + """ + _do_get_name implementation for PDBSymbolFinder + + Args: + address: the address to resolve within the module + + Returns: + str: the name of the symbol, if found + """ try: name = self._symbol_module.get_symbols_by_absolute_location(address)[0] return name.split(constants.BANG)[1] @@ -113,14 +201,32 @@ class ExportSymbolFinder(PESymbolFinder): except AttributeError: return None - def _do_get_name(self, address: int) -> PESymbolFinder.cached_str: + def _do_get_name(self, address: int) -> Optional[str]: + """ + _do_get_name implementation for ExportSymbolFinder + + Args: + address: the address to resolve within the module + + Returns: + str: the name of the symbol, if found + """ for export in self._symbol_module: if export.address + self._module_start == address: return self._get_name(export) return None - def _do_get_address(self, name: str) -> PESymbolFinder.cached_int: + def _do_get_address(self, name: str) -> Optional[int]: + """ + _do_get_address implementation for ExportSymbolFinder + Args: + str: the name of the symbol to resolve + + Returns: + address: the address of the symbol, if found + """ + for export in self._symbol_module: sym_name = self._get_name(export) if sym_name and sym_name == name: @@ -139,25 +245,6 @@ class PESymbols(interfaces.plugins.PluginInterface): # used for special handling of the kernel PDB file. See later notes os_module_name = "ntoskrnl.exe" - # keys for specifying wanted names and/or addresses - # used for consistent access between the API and plugins - wanted_names = "names" - wanted_addresses = "addresses" - - # how wanted modules/symbols are specified, such as: - # {"ntdll.dll" : {wanted_addresses : [42, 43, 43]}} - # {"ntdll.dll" : {wanted_names : ["NtCreateThread"]}} - filter_modules_type = Dict[str, Union[Dict[str, List[str]], Dict[str, List[int]]]] - - # holds resolved symbols - # {"ntdll.dll": [("Bob", 123), ("Alice", 456)]} - found_symbols_type = Dict[str, List[Tuple[str, int]]] - - # used to hold informatin about a range (VAD or kernel module) - # (start address, size, file path) - range_type = Tuple[int, int, str] - ranges_type = List[range_type] - @classmethod def get_requirements(cls) -> List: # Since we're calling the plugin, make sure we have the plugin's requirements @@ -187,13 +274,15 @@ class PESymbols(interfaces.plugins.PluginInterface): description='Module in which to resolve symbols. Use "ntoskrnl.exe" to resolve in the base kernel executable.', optional=False, ), - requirements.StringRequirement( - name="symbol", + requirements.ListRequirement( + name="symbols", + element_type=str, description="Symbol name to resolve", optional=True, ), - requirements.IntRequirement( - name="address", + requirements.ListRequirement( + name="addresses", + element_type=int, description="Address of symbol to resolve", optional=True, ), @@ -242,7 +331,15 @@ class PESymbols(interfaces.plugins.PluginInterface): ranges: ranges_type, address: int ) -> Optional[range_type]: """ - Helper for getting the range information for an address + Helper for getting the range information for an address. + Finds the range holding the `address` parameter + + Args: + address: the address to find the range for + + Returns: + Tuple[int, int, str]: The starting address, size, and file path of the range + """ for start, size, filepath in ranges: if start <= address < start + size: @@ -254,6 +351,13 @@ class PESymbols(interfaces.plugins.PluginInterface): def filepath_for_address(ranges: ranges_type, address: int) -> Optional[str]: """ Helper to get the file path for an address + + Args: + ranges: The set of VADs with mapped files to find the address + address: The address to find inside of the VADs set + + Returns: + str: The full path of the file, if found and present """ info = PESymbols.range_info_for_address(ranges, address) if info: @@ -264,9 +368,15 @@ class PESymbols(interfaces.plugins.PluginInterface): @staticmethod def filename_for_path(filepath: str) -> str: """ - Consistent way to get the filename + Consistent way to get the filename regardless of platform + + Args: + str: the file path from `filepath_for_address` + + Returns: + str: the bsae file name of the full path """ - return filepath.split("\\")[-1] + return ntpath.basename(filepath) @staticmethod def addresses_for_process_symbols( @@ -276,6 +386,18 @@ class PESymbols(interfaces.plugins.PluginInterface): symbol_table_name: str, symbols: filter_modules_type, ) -> found_symbols_type: + """ + Used to easily resolve the addresses of names inside of modules. + + See the usage of this function for system call resolution in unhooked_system_calls.py + for an easy to understand example. + + Args: + symbols: The dictionary of symbols requested by the caller + + Returns: + found_symbols_type: The dictionary of symbols that were resolved + """ collected_modules = PESymbols.get_process_modules( context, layer_name, symbol_table_name, symbols ) @@ -294,16 +416,22 @@ class PESymbols(interfaces.plugins.PluginInterface): def path_and_symbol_for_address( context: interfaces.context.ContextInterface, config_path: str, - collected_modules: Dict[str, List[Tuple[str, int, int]]], + collected_modules: collected_modules_type, ranges: ranges_type, address: int, ) -> Tuple[str, str]: """ Method for plugins to determine the file path and symbol name for a given address - collected_modules: return value from `get_kernel_modules` or `get_process_modules` - ranges: the memory ranges to examine in this layer. - address: address to resolve to its symbol name + See debugregisters.py for an example of how this function is used along with get_vads_for_process_cache + for resolving symbols in processes. + + Args: + collected_modules: return value from `get_kernel_modules` or `get_process_modules` + ranges: the memory ranges to examine in this layer. + address: address to resolve to its symbol name + Returns: + Tuple[str|renderers.NotApplicableValue|renderers.NotAvailableValue, str|renderers.NotApplicableValue|renderers.NotAvailableValue] """ if not address: @@ -317,15 +445,15 @@ class PESymbols(interfaces.plugins.PluginInterface): filename = PESymbols.filename_for_path(filepath).lower() # setup to resolve the address - filter_module: PESymbols.filter_modules_type = { - filename: {PESymbols.wanted_addresses: [address]} + filter_module: filter_modules_type = { + filename: {wanted_addresses_identifier: [address]} } found_symbols = PESymbols.find_symbols( context, config_path, filter_module, collected_modules ) - if not found_symbols or not found_symbols[filename]: + if not found_symbols or filename not in found_symbols: return renderers.NotAvailableValue(), renderers.NotAvailableValue() return filepath, found_symbols[filename][0][0] @@ -335,13 +463,18 @@ class PESymbols(interfaces.plugins.PluginInterface): context: interfaces.context.ContextInterface, pe_table_name: str, mod_name: str, - module_info: Tuple[str, int, int], + module_info: collected_module_instance, ) -> Optional[ExportSymbolFinder]: """ Attempts to locate symbols based on export analysis - mod_name: lower case name of the module to resolve symbols in - module_info: (layer_name, module_start, module_size) of the module to examine + Args: + mod_name: lower case name of the module to resolve symbols in + module_info: (layer_name, module_start, module_size) of the module to examine + + Returns: + Optional[ExportSymbolFinder]: If the export table can be resolved, then the ExportSymbolFinder + instance for it """ layer_name = module_info[0] @@ -361,7 +494,10 @@ class PESymbols(interfaces.plugins.PluginInterface): return None return ExportSymbolFinder( - layer_name, mod_name, module_start, pe_module.DIRECTORY_ENTRY_EXPORT.symbols + layer_name, + mod_name.lower(), + module_start, + pe_module.DIRECTORY_ENTRY_EXPORT.symbols, ) @staticmethod @@ -369,13 +505,17 @@ class PESymbols(interfaces.plugins.PluginInterface): context: interfaces.context.ContextInterface, config_path: str, mod_name: str, - module_info: Tuple[str, int, int], + module_info: collected_module_instance, ) -> Optional[PDBSymbolFinder]: """ - Attempts to locate symbols based on PDB analysis + Attempts to locate symbols based on PDB analysis through each layer where the mod_name module was found - mod_name: lower case name of the module to resolve symbols in - module_info: (layer_name, module_start, module_size) of the module to examine + Args: + mod_name: lower case name of the module to resolve symbols in + module_info: (layer_name, module_start, module_size) of the module to examine + + Returns: + Optional[PDBSymbolFinder]: If the export table can be resolved, then the ExportSymbolFinder """ mod_symbols = None @@ -386,15 +526,17 @@ class PESymbols(interfaces.plugins.PluginInterface): # a `ntoskrnl.exe` can have an internal PDB name of any of the ones in the following list # The code attempts to find all possible PDBs to ensure the best chance of recovery if mod_name == PESymbols.os_module_name: - pdb_names = ["ntkrnlmp.pdb", "ntkrnlpa.pdb", "ntkrpamp.pdb", "ntoskrnl.pdb"] + pdb_names = [fn + ".pdb" for fn in KERNEL_MODULE_NAMES] # for non-kernel files, replace the exe, sys, or dll extension with pdb else: + # in testing we found where some DLLs, such amsi.dll, have its PDB string as Amsi.dll + # in certain Windows versions mod_name = mod_name[:-3] + "pdb" first_upper = mod_name[0].upper() + mod_name[1:] pdb_names = [mod_name, first_upper] - # loop through each PDB name (will be just one for all but the kernel) + # loop through each PDB name (all the kernel names or the dll name as lower() + first char upper case) for pdb_name in pdb_names: try: mod_symbols = pdbutil.PDBUtility.symbol_table_from_pdb( @@ -433,11 +575,17 @@ class PESymbols(interfaces.plugins.PluginInterface): def _find_symbols_through_pdb( context: interfaces.context.ContextInterface, config_path: str, - module_instances: List[Tuple[str, int, int]], + module_instances: collected_modules_info, mod_name: str, ) -> Generator[PDBSymbolFinder, None, None]: """ - Attempts to resolve the symbols in `wanted_symbols` through PDB analysis + Attempts to resolve the symbols in `mod_name` through PDB analysis + + Args: + module_instances: the set of layers in which the module was found + mod_name: name of the module to resolve symbols in + Returns: + Generator[PDBSymbolFinder]: a PDBSymbolFinder instance for each layer in which the module was found """ for module_info in module_instances: mod_module = PESymbols._get_pdb_module( @@ -450,11 +598,17 @@ class PESymbols(interfaces.plugins.PluginInterface): def _find_symbols_through_exports( context: interfaces.context.ContextInterface, config_path: str, - module_instances: List[Tuple[str, int, int]], + module_instances: collected_modules_info, mod_name: str, ) -> Generator[ExportSymbolFinder, None, None]: """ - Attempts to resolve the symbols in `wanted_symbols` through export analysis + Attempts to resolve the symbols in `mod_name` through export analysis + + Args: + module_instances: the set of layers in which the module was found + mod_name: name of the module to resolve symbols in + Returns: + Generator[ExportSymbolFinder]: an ExportSymbolFinder instance for each layer in which the module was found """ pe_table_name = intermed.IntermediateSymbolTable.create( context, config_path, "windows", "pe", class_types=pe.class_types @@ -476,12 +630,21 @@ class PESymbols(interfaces.plugins.PluginInterface): ) -> Generator[Tuple[str, int], None, None]: """ Enumerates the symbols specified as wanted by the calling plugin + + removes entries from wanted_modules as they found to avoid PDB or export analysis after resolving all symbols + + Args: + wanted_modules: the dictionary of modules and symbols to resolve. Modified to remove symbols as they are resolved. + mod_name: the name of module to resolve symbols in + + Returns: + Tuple[str, int]: the name and address of resolved symbols """ wanted_symbols = wanted_modules[mod_name] if ( - PESymbols.wanted_names not in wanted_symbols - and PESymbols.wanted_addresses not in wanted_symbols + wanted_names_identifier not in wanted_symbols + and wanted_addresses_identifier not in wanted_symbols ): vollog.warning( f"Invalid `wanted_symbols` sent to `find_symbols` for module {mod_name}. addresses and names keys both misssing." @@ -489,8 +652,8 @@ class PESymbols(interfaces.plugins.PluginInterface): return symbol_keys = [ - (PESymbols.wanted_names, "get_address_for_name"), - (PESymbols.wanted_addresses, "get_name_for_address"), + (wanted_names_identifier, "get_address_for_name"), + (wanted_addresses_identifier, "get_name_for_address"), ] for symbol_key, symbol_getter in symbol_keys: @@ -503,7 +666,7 @@ class PESymbols(interfaces.plugins.PluginInterface): ) if symbol_value: # yield out symbol name, symbol address - if symbol_key == PESymbols.wanted_names: + if symbol_key == wanted_names_identifier: yield wanted_value, symbol_value # type: ignore else: yield symbol_value, wanted_value # type: ignore @@ -521,13 +684,20 @@ class PESymbols(interfaces.plugins.PluginInterface): def _resolve_symbols_through_methods( context: interfaces.context.ContextInterface, config_path: str, - module_instances: List[Tuple[str, int, int]], + module_instances: collected_modules_info, wanted_modules: PESymbolFinder.cached_value_dict, mod_name: str, ) -> Generator[Tuple[str, int], None, None]: """ Attempts to resolve every wanted symbol in `mod_name` Every layer is enumerated for maximum chance of recovery + + Args: + module_instances: the set of layers in which the module was found + wanted_modules: The symbols to resolve tied to their module names + mod_name: name of the module to resolve symbols in + Returns: + Generator[Tuple[str, int]]: resolved symbol names and addresses """ symbol_resolving_methods = [ PESymbols._find_symbols_through_pdb, @@ -554,13 +724,19 @@ class PESymbols(interfaces.plugins.PluginInterface): context: interfaces.context.ContextInterface, config_path: str, wanted_modules: PESymbolFinder.cached_value_dict, - collected_modules: Dict[str, List[Tuple[str, int, int]]], + collected_modules: collected_modules_type, ) -> found_symbols_type: """ Loops through each method of symbol analysis until each wanted symbol is found Returns the resolved symbols as a dictionary that includes the name and runtime address + + Args: + wanted_modules: the dictionary of modules and symbols to resolve. Modified to remove symbols as they are resolved. + collected_modules: return value from `get_kernel_modules` or `get_process_modules` + Returns: + found_symbols_type: The set of symbols resolved to their name and/or address """ - found_symbols: PESymbols.found_symbols_type = {} + found_symbols: found_symbols_type = {} for mod_name in wanted_modules: if mod_name not in collected_modules: @@ -594,11 +770,16 @@ class PESymbols(interfaces.plugins.PluginInterface): layer_name: str, symbol_table: str, filter_modules: Optional[filter_modules_type], - ) -> Dict[str, List[Tuple[str, int, int]]]: + ) -> collected_modules_type: """ Walks the kernel module list and finds the session layer, base, and size of each wanted module + + Args: + filter_modules: The modules to filter the gathering to. If left as None, all kernel modules are gathered. + Returns: + collected_modules_type: The collection of modules found with at least one layer present """ - found_modules: Dict[str, List[Tuple[str, int, int]]] = {} + found_modules: collected_modules_type = {} if filter_modules: # create a tuple of module names for use with `endswith` @@ -645,16 +826,55 @@ class PESymbols(interfaces.plugins.PluginInterface): return found_modules + @staticmethod + def get_vads_for_process_cache( + vads_cache: Dict[int, ranges_type], + owner_proc: interfaces.objects.ObjectInterface, + ) -> Optional[ranges_type]: + """ + Creates and utilizes a cache of a process' VADs for efficient lookups + + Returns the vad information of the VAD hosting the address, if found + + Args: + vads_cache: The existing cache of VADs + owner_proc: The process being inspected + Returns: + Optional[ranges_type]: The range holding the address, if found + """ + if owner_proc.vol.offset in vads_cache: + vads = vads_cache[owner_proc.vol.offset] + else: + vads = PESymbols.get_proc_vads_with_file_paths(owner_proc) + vads_cache[owner_proc.vol.offset] = vads + + # smear or terminated process + if len(vads) == 0: + return None + + return vads + @staticmethod def get_proc_vads_with_file_paths( proc: interfaces.objects.ObjectInterface, ) -> ranges_type: """ Returns a list of the process' vads that map a file - """ - vads = [] - for vad in proc.get_vad_root().traverse(): + Args: + proc: The process to gather the VADs for + + Returns: + ranges_type: The list of VADs for this process that map a file + """ + vads: ranges_type = [] + + try: + vad_root = proc.get_vad_root() + except exceptions.InvalidAddressException: + return vads + + for vad in vad_root.traverse(): filepath = vad.get_file_name() if not isinstance(filepath, str) or filepath.count("\\") == 0: continue @@ -676,6 +896,9 @@ class PESymbols(interfaces.plugins.PluginInterface): ]: """ Yields each set of vads for a process that have a file mapped, along with the process itself and its layer + + Args: + Generator[Tuple[interfaces.objects.ObjectInterface, str, ranges_type]]: Yields tuple of process objects, layers, and VADs mapping files """ procs = pslist.PsList.list_processes( context=context, @@ -689,7 +912,7 @@ class PESymbols(interfaces.plugins.PluginInterface): except exceptions.InvalidAddressException: continue - vads = PESymbols.get_proc_vads_with_file_paths(proc) + vads = cls.get_proc_vads_with_file_paths(proc) yield proc, proc_layer_name, vads @@ -699,11 +922,16 @@ class PESymbols(interfaces.plugins.PluginInterface): layer_name: str, symbol_table: str, filter_modules: Optional[filter_modules_type], - ) -> Dict[str, List[Tuple[str, int, int]]]: + ) -> collected_modules_type: """ Walks the process list and each process' VAD to determine the base address and size of wanted modules + + Args: + filter_modules: The modules to filter the gathering to. If left as None, all process modules are gathered. + Returns: + collected_modules_type: The collection of modules found with at least one layer present """ - proc_modules: Dict[str, List[Tuple[str, int, int]]] = {} + proc_modules: collected_modules_type = {} if filter_modules: # create a tuple of module names for use with `endswith` @@ -711,7 +939,7 @@ class PESymbols(interfaces.plugins.PluginInterface): else: filter_modules_check = None - for _, proc_layer_name, vads in PESymbols.get_all_vads_with_file_paths( + for _proc, proc_layer_name, vads in PESymbols.get_all_vads_with_file_paths( context, layer_name, symbol_table ): for vad_start, vad_size, filepath in vads: @@ -731,17 +959,17 @@ class PESymbols(interfaces.plugins.PluginInterface): def _generator(self) -> Generator[Tuple[int, Tuple[str, str, int]], None, None]: kernel = self.context.modules[self.config["kernel"]] - if self.config["symbol"]: + if self.config["symbols"]: filter_module = { self.config["module"].lower(): { - PESymbols.wanted_names: [self.config["symbol"]] + wanted_names_identifier: self.config["symbols"] } } - elif self.config["address"]: + elif self.config["addresses"]: filter_module = { self.config["module"].lower(): { - PESymbols.wanted_addresses: [self.config["address"]] + wanted_addresses_identifier: self.config["addresses"] } } diff --git a/volatility3/framework/plugins/windows/unhooked_system_calls.py b/volatility3/framework/plugins/windows/unhooked_system_calls.py index 0438bc9e3..68f4c4b80 100644 --- a/volatility3/framework/plugins/windows/unhooked_system_calls.py +++ b/volatility3/framework/plugins/windows/unhooked_system_calls.py @@ -1,3 +1,6 @@ +# This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 + import logging from typing import Dict, Tuple, List, Generator @@ -18,7 +21,7 @@ class unhooked_system_calls(interfaces.plugins.PluginInterface): system_calls = { "ntdll.dll": { - pe_symbols.PESymbols.wanted_names: [ + pe_symbols.wanted_names_identifier: [ "NtCreateThread", "NtProtectVirtualMemory", "NtReadVirtualMemory", @@ -90,7 +93,7 @@ class unhooked_system_calls(interfaces.plugins.PluginInterface): def _gather_code_bytes( self, kernel: interfaces.context.ModuleInterface, - found_symbols: pe_symbols.PESymbols.found_symbols_type, + found_symbols: pe_symbols.found_symbols_type, ) -> _code_bytes_type: """ Enumerates the desired DLLs and function implementations in each process diff --git a/volatility3/framework/plugins/windows/vadinfo.py b/volatility3/framework/plugins/windows/vadinfo.py index 974793a71..2c6ed4daf 100644 --- a/volatility3/framework/plugins/windows/vadinfo.py +++ b/volatility3/framework/plugins/windows/vadinfo.py @@ -9,7 +9,7 @@ from volatility3.framework import renderers, interfaces, exceptions from volatility3.framework.configuration import requirements from volatility3.framework.objects import utility from volatility3.framework.renderers import format_hints -from volatility3.plugins.windows import pslist, pe_symbols +from volatility3.plugins.windows import pslist vollog = logging.getLogger(__name__) @@ -37,7 +37,7 @@ class VadInfo(interfaces.plugins.PluginInterface): _version = (2, 0, 0) MAXSIZE_DEFAULT = 1024 * 1024 * 1024 # 1 Gb - def __init__(self, *args, **kwargs): # type: ignore + def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._protect_values = None From b40c20dd7fe0d1e348f94531c290a9aceee34dcb Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Tue, 10 Sep 2024 14:55:00 -0500 Subject: [PATCH 11/17] Revert back to union to avoid failed tests --- volatility3/framework/plugins/windows/pe_symbols.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index c9785a1c9..30e9b49d1 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -61,7 +61,7 @@ class PESymbolFinder: cached_int_dict = Dict[str, Optional[int]] cached_value = Union[int, str, None] - cached_value_dict = Dict[str, Dict[str, List[str]] | Dict[str, List[int]]] + cached_value_dict = Dict[str, Union[Dict[str, List[str]], Dict[str, List[int]]]] def __init__( self, From ba0c975e73207ee6555bd80067460ddbea6426a2 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Fri, 13 Sep 2024 18:50:47 -0500 Subject: [PATCH 12/17] Address all feedback --- .../framework/plugins/windows/pe_symbols.py | 126 +++++++++--------- 1 file changed, 65 insertions(+), 61 deletions(-) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index 30e9b49d1..85bfb572e 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -1,11 +1,12 @@ # This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +import copy import io import logging import ntpath -from typing import Dict, Tuple, Optional, List, Generator, Union +from typing import Dict, Tuple, Optional, List, Generator, Union, Callable import pefile @@ -29,11 +30,13 @@ wanted_addresses_identifier = "addresses" # how wanted modules/symbols are specified, such as: # {"ntdll.dll" : {wanted_addresses : [42, 43, 43]}} # {"ntdll.dll" : {wanted_names : ["NtCreateThread"]}} -filter_modules_type = Dict[str, Union[Dict[str, List[str]], Dict[str, List[int]]]] +filter_module_info = Union[Dict[str, List[str]], Dict[str, List[int]]] +filter_modules_type = Dict[str, filter_module_info] # holds resolved symbols # {"ntdll.dll": [("Bob", 123), ("Alice", 456)]} -found_symbols_type = Dict[str, List[Tuple[str, int]]] +found_symbols_module = List[Tuple[str, int]] +found_symbols_type = Dict[str, found_symbols_module] # used to hold informatin about a range (VAD or kernel module) # (start address, size, file path) @@ -61,7 +64,8 @@ class PESymbolFinder: cached_int_dict = Dict[str, Optional[int]] cached_value = Union[int, str, None] - cached_value_dict = Dict[str, Union[Dict[str, List[str]], Dict[str, List[int]]]] + cached_module_lists = Union[Dict[str, List[str]], Dict[str, List[int]]] + cached_value_dict = Dict[str, cached_module_lists] def __init__( self, @@ -402,11 +406,11 @@ class PESymbols(interfaces.plugins.PluginInterface): context, layer_name, symbol_table_name, symbols ) - found_symbols = PESymbols.find_symbols( + found_symbols, missing_symbols = PESymbols.find_symbols( context, config_path, symbols, collected_modules ) - for mod_name, unresolved_symbols in symbols.items(): + for mod_name, unresolved_symbols in missing_symbols.items(): for symbol in unresolved_symbols: vollog.debug(f"Unable to resolve symbol {symbol} in module {mod_name}") @@ -449,7 +453,7 @@ class PESymbols(interfaces.plugins.PluginInterface): filename: {wanted_addresses_identifier: [address]} } - found_symbols = PESymbols.find_symbols( + found_symbols, _missing_msybols = PESymbols.find_symbols( context, config_path, filter_module, collected_modules ) @@ -624,61 +628,46 @@ class PESymbols(interfaces.plugins.PluginInterface): @staticmethod def _get_symbol_value( - wanted_modules: PESymbolFinder.cached_value_dict, - mod_name: str, + wanted_symbols: filter_module_info, symbol_resolver: PESymbolFinder, - ) -> Generator[Tuple[str, int], None, None]: + ) -> Generator[Tuple[str, int, str, int], None, None]: """ Enumerates the symbols specified as wanted by the calling plugin - removes entries from wanted_modules as they found to avoid PDB or export analysis after resolving all symbols - Args: - wanted_modules: the dictionary of modules and symbols to resolve. Modified to remove symbols as they are resolved. - mod_name: the name of module to resolve symbols in + wanted_symbols: the set of symbols for a particular module + symbol_resolver: method in a layer to resolve the symbols Returns: - Tuple[str, int]: the name and address of resolved symbols + Tuple[str, int, str, int]: the index and value of the found symbol in the wanted list, and the name and address of resolved symbol """ - wanted_symbols = wanted_modules[mod_name] - if ( wanted_names_identifier not in wanted_symbols and wanted_addresses_identifier not in wanted_symbols ): vollog.warning( - f"Invalid `wanted_symbols` sent to `find_symbols` for module {mod_name}. addresses and names keys both misssing." + f"Invalid `wanted_symbols` sent to `find_symbols`. addresses and names keys both misssing." ) return - symbol_keys = [ - (wanted_names_identifier, "get_address_for_name"), - (wanted_addresses_identifier, "get_name_for_address"), + symbol_keys: List[Tuple[str, Callable]] = [ + (wanted_names_identifier, symbol_resolver.get_address_for_name), + (wanted_addresses_identifier, symbol_resolver.get_name_for_address), ] for symbol_key, symbol_getter in symbol_keys: # address or name if symbol_key in wanted_symbols: # walk each wanted address or name - for wanted_value in wanted_symbols[symbol_key]: - symbol_value = symbol_resolver.__getattribute__(symbol_getter)( - wanted_value - ) + for value_index, wanted_value in enumerate(wanted_symbols[symbol_key]): + symbol_value = symbol_getter(wanted_value) + if symbol_value: - # yield out symbol name, symbol address + # yield out deleteion key, deletion index, symbol name, symbol address if symbol_key == wanted_names_identifier: - yield wanted_value, symbol_value # type: ignore + yield symbol_key, value_index, wanted_value, symbol_value # type: ignore else: - yield symbol_value, wanted_value # type: ignore - - index = wanted_modules[mod_name][symbol_key].index(wanted_value) # type: ignore - - del wanted_modules[mod_name][symbol_key][index] - - # if all names or addresses from a module are found, delete the key - if not wanted_modules[mod_name][symbol_key]: - del wanted_modules[mod_name][symbol_key] - break + yield symbol_key, value_index, symbol_value, wanted_value # type: ignore @staticmethod def _resolve_symbols_through_methods( @@ -687,7 +676,7 @@ class PESymbols(interfaces.plugins.PluginInterface): module_instances: collected_modules_info, wanted_modules: PESymbolFinder.cached_value_dict, mod_name: str, - ) -> Generator[Tuple[str, int], None, None]: + ) -> Tuple[found_symbols_module, PESymbolFinder.cached_module_lists]: """ Attempts to resolve every wanted symbol in `mod_name` Every layer is enumerated for maximum chance of recovery @@ -697,35 +686,53 @@ class PESymbols(interfaces.plugins.PluginInterface): wanted_modules: The symbols to resolve tied to their module names mod_name: name of the module to resolve symbols in Returns: - Generator[Tuple[str, int]]: resolved symbol names and addresses + Tuple[found_symbols_module, PESymbolFinder.cached_module_lists]: The set of found symbols and the ones that could not be resolved """ symbol_resolving_methods = [ PESymbols._find_symbols_through_pdb, PESymbols._find_symbols_through_exports, ] + found: found_symbols_module = [] + + # the symbols wanted from this module by the caller + wanted = wanted_modules[mod_name] + + # make a copy to remove from inside this function for returning to the caller + remaining = copy.deepcopy(wanted) + for method in symbol_resolving_methods: + # every layer where this module was found through the given method for symbol_resolver in method( context, config_path, module_instances, mod_name ): vollog.debug(f"Have resolver for method {method}") - yield from PESymbols._get_symbol_value( - wanted_modules, mod_name, symbol_resolver - ) + for ( + symbol_key, + value_index, + symbol_name, + symbol_address, + ) in PESymbols._get_symbol_value(remaining, symbol_resolver): + found.append((symbol_name, symbol_address)) + del remaining[symbol_key][value_index] - if not wanted_modules[mod_name]: + # everything was resolved, stop this resolver + if not remaining: break - if not wanted_modules[mod_name]: + # stop all resolving + if not remaining: break + return found, remaining + @staticmethod def find_symbols( context: interfaces.context.ContextInterface, config_path: str, wanted_modules: PESymbolFinder.cached_value_dict, collected_modules: collected_modules_type, - ) -> found_symbols_type: + ) -> Tuple[found_symbols_type, PESymbolFinder.cached_value_dict]: """ Loops through each method of symbol analysis until each wanted symbol is found Returns the resolved symbols as a dictionary that includes the name and runtime address @@ -734,9 +741,10 @@ class PESymbols(interfaces.plugins.PluginInterface): wanted_modules: the dictionary of modules and symbols to resolve. Modified to remove symbols as they are resolved. collected_modules: return value from `get_kernel_modules` or `get_process_modules` Returns: - found_symbols_type: The set of symbols resolved to their name and/or address + Tuple[found_symbols_type, PESymbolFinder.cached_value_dict]: The set of found symbols but the ones that could not be resolved """ found_symbols: found_symbols_type = {} + missing_symbols: PESymbolFinder.cached_value_dict = {} for mod_name in wanted_modules: if mod_name not in collected_modules: @@ -745,24 +753,20 @@ class PESymbols(interfaces.plugins.PluginInterface): module_instances = collected_modules[mod_name] # try to resolve the symbols for `mod_name` through each method (PDB and export table currently) - for symbol_name, address in PESymbols._resolve_symbols_through_methods( + ( + found_in_module, + missing_in_module, + ) = PESymbols._resolve_symbols_through_methods( context, config_path, module_instances, wanted_modules, mod_name - ): - if mod_name not in found_symbols: - found_symbols[mod_name] = [] + ) - found_symbols[mod_name].append((symbol_name, address)) + if found_in_module: + found_symbols[mod_name] = found_in_module - # stop processing the layers (processes) if we found all the symbols for this module - if not wanted_modules[mod_name]: - break + if missing_in_module: + missing_symbols[mod_name] = missing_in_module - # stop processing this module if/when all symbols are found - if not wanted_modules[mod_name]: - del wanted_modules[mod_name] - break - - return found_symbols + return found_symbols, missing_symbols @staticmethod def get_kernel_modules( @@ -986,7 +990,7 @@ class PESymbols(interfaces.plugins.PluginInterface): self.context, kernel.layer_name, kernel.symbol_table_name, filter_module ) - found_symbols = PESymbols.find_symbols( + found_symbols, _missing_symbols = PESymbols.find_symbols( self.context, self.config_path, filter_module, collected_modules ) From 21d21cf4b8ec3aafa5e98c563b207b21c41d5c3f Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Sat, 14 Sep 2024 16:26:38 -0500 Subject: [PATCH 13/17] Break properly in all paths. Help callers to ensure always lower case module name matching. --- volatility3/framework/plugins/windows/pe_symbols.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index 85bfb572e..d0ddcac57 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -380,7 +380,7 @@ class PESymbols(interfaces.plugins.PluginInterface): Returns: str: the bsae file name of the full path """ - return ntpath.basename(filepath) + return ntpath.basename(filepath).lower() @staticmethod def addresses_for_process_symbols( @@ -717,11 +717,12 @@ class PESymbols(interfaces.plugins.PluginInterface): del remaining[symbol_key][value_index] # everything was resolved, stop this resolver - if not remaining: + if not remaining[symbol_key]: break # stop all resolving - if not remaining: + if not remaining[symbol_key]: + del remaining[symbol_key] break return found, remaining From 291bc878ad771388f3514b2c66d31b7325b2bc01 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Sat, 14 Sep 2024 16:31:33 -0500 Subject: [PATCH 14/17] Break in a cleaner flow --- volatility3/framework/plugins/windows/pe_symbols.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index d0ddcac57..44254513f 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -701,6 +701,8 @@ class PESymbols(interfaces.plugins.PluginInterface): # make a copy to remove from inside this function for returning to the caller remaining = copy.deepcopy(wanted) + done_processing = False + for method in symbol_resolving_methods: # every layer where this module was found through the given method for symbol_resolver in method( @@ -717,12 +719,14 @@ class PESymbols(interfaces.plugins.PluginInterface): del remaining[symbol_key][value_index] # everything was resolved, stop this resolver + # remove this key from the remaining symbols to resolve if not remaining[symbol_key]: + del remaining[symbol_key] + done_processing = True break # stop all resolving - if not remaining[symbol_key]: - del remaining[symbol_key] + if done_processing: break return found, remaining From 322f79fb5040e3e7ebdfa9de8c82a639729a39f2 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Sat, 14 Sep 2024 17:18:26 -0500 Subject: [PATCH 15/17] Bail as early as possible --- .../framework/plugins/windows/pe_symbols.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index 44254513f..0bf03e7d6 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -437,7 +437,6 @@ class PESymbols(interfaces.plugins.PluginInterface): Returns: Tuple[str|renderers.NotApplicableValue|renderers.NotAvailableValue, str|renderers.NotApplicableValue|renderers.NotAvailableValue] """ - if not address: return renderers.NotApplicableValue(), renderers.NotApplicableValue() @@ -458,7 +457,7 @@ class PESymbols(interfaces.plugins.PluginInterface): ) if not found_symbols or filename not in found_symbols: - return renderers.NotAvailableValue(), renderers.NotAvailableValue() + return filepath, renderers.NotAvailableValue() return filepath, found_symbols[filename][0][0] @@ -718,11 +717,14 @@ class PESymbols(interfaces.plugins.PluginInterface): found.append((symbol_name, symbol_address)) del remaining[symbol_key][value_index] - # everything was resolved, stop this resolver - # remove this key from the remaining symbols to resolve - if not remaining[symbol_key]: - del remaining[symbol_key] - done_processing = True + # everything was resolved, stop this resolver + # remove this key from the remaining symbols to resolve + if not remaining[symbol_key]: + del remaining[symbol_key] + done_processing = True + break + + if done_processing: break # stop all resolving @@ -885,6 +887,7 @@ class PESymbols(interfaces.plugins.PluginInterface): for vad in vad_root.traverse(): filepath = vad.get_file_name() + if not isinstance(filepath, str) or filepath.count("\\") == 0: continue From 79b8ff7d05b56316d90be682f56db22401f2c265 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Sun, 15 Sep 2024 18:29:05 -0500 Subject: [PATCH 16/17] Address final feedback --- .../plugins/windows/debugregisters.py | 23 ++++++++----------- .../framework/plugins/windows/pe_symbols.py | 10 ++++---- .../plugins/windows/unhooked_system_calls.py | 8 +++++++ 3 files changed, 22 insertions(+), 19 deletions(-) diff --git a/volatility3/framework/plugins/windows/debugregisters.py b/volatility3/framework/plugins/windows/debugregisters.py index 65b2e625b..945ba1df0 100644 --- a/volatility3/framework/plugins/windows/debugregisters.py +++ b/volatility3/framework/plugins/windows/debugregisters.py @@ -148,12 +148,7 @@ class DebugRegisters(interfaces.plugins.PluginInterface): file3, sym3 = path_and_symbol(vads, dr3) # if none map to an actual file VAD then bail - if not ( - isinstance(file0, str) - or isinstance(file1, str) - or isinstance(file2, str) - or isinstance(file3, str) - ): + if not (file0 or file1 or file2 or file3): continue process_name = owner_proc.ImageFileName.cast( @@ -173,17 +168,17 @@ class DebugRegisters(interfaces.plugins.PluginInterface): thread.Tcb.State, dr7, format_hints.Hex(dr0), - file0, - sym0, + file0 or renderers.NotApplicableValue(), + sym0 or renderers.NotApplicableValue(), format_hints.Hex(dr1), - file1, - sym1, + file1 or renderers.NotApplicableValue(), + sym1 or renderers.NotApplicableValue(), format_hints.Hex(dr2), - file2, - sym2, + file2 or renderers.NotApplicableValue(), + sym2 or renderers.NotApplicableValue(), format_hints.Hex(dr3), - file3, - sym3, + file3 or renderers.NotApplicableValue(), + sym3 or renderers.NotApplicableValue(), ), ) diff --git a/volatility3/framework/plugins/windows/pe_symbols.py b/volatility3/framework/plugins/windows/pe_symbols.py index 0bf03e7d6..955098d6b 100644 --- a/volatility3/framework/plugins/windows/pe_symbols.py +++ b/volatility3/framework/plugins/windows/pe_symbols.py @@ -423,7 +423,7 @@ class PESymbols(interfaces.plugins.PluginInterface): collected_modules: collected_modules_type, ranges: ranges_type, address: int, - ) -> Tuple[str, str]: + ) -> Tuple[Optional[str], Optional[str]]: """ Method for plugins to determine the file path and symbol name for a given address @@ -438,12 +438,12 @@ class PESymbols(interfaces.plugins.PluginInterface): Tuple[str|renderers.NotApplicableValue|renderers.NotAvailableValue, str|renderers.NotApplicableValue|renderers.NotAvailableValue] """ if not address: - return renderers.NotApplicableValue(), renderers.NotApplicableValue() + return None, None filepath = PESymbols.filepath_for_address(ranges, address) if not filepath: - return renderers.NotAvailableValue(), renderers.NotAvailableValue() + return None, None filename = PESymbols.filename_for_path(filepath).lower() @@ -452,12 +452,12 @@ class PESymbols(interfaces.plugins.PluginInterface): filename: {wanted_addresses_identifier: [address]} } - found_symbols, _missing_msybols = PESymbols.find_symbols( + found_symbols, _missing_symbols = PESymbols.find_symbols( context, config_path, filter_module, collected_modules ) if not found_symbols or filename not in found_symbols: - return filepath, renderers.NotAvailableValue() + return filepath, None return filepath, found_symbols[filename][0][0] diff --git a/volatility3/framework/plugins/windows/unhooked_system_calls.py b/volatility3/framework/plugins/windows/unhooked_system_calls.py index 68f4c4b80..c3d98254d 100644 --- a/volatility3/framework/plugins/windows/unhooked_system_calls.py +++ b/volatility3/framework/plugins/windows/unhooked_system_calls.py @@ -71,6 +71,13 @@ class unhooked_system_calls(interfaces.plugins.PluginInterface): } } + # This data structure is used to track unique implementations of functions across processes + # The outer dictionary holds the module name (e.g., ntdll.dll) + # The next dictionary holds the function names (NtTerminateProcess, NtSetValueKey, etc.) inside a module + # The innermost dictionary holds the unique implementation (bytes) of a function across processes + # Each implementation is tracked along with the process(es) that host it + # For systems without malware, all functions should have the same implementation + # When API hooking/module unhooking is done, the victim (infected) processes will have unique implementations _code_bytes_type = Dict[str, Dict[str, Dict[bytes, List[Tuple[int, str]]]]] @classmethod @@ -127,6 +134,7 @@ class unhooked_system_calls(interfaces.plugins.PluginInterface): except exceptions.InvalidAddressException: continue + # see the definition of _code_bytes_type for details of this data structure if dll_name not in code_bytes: code_bytes[dll_name] = {} From b788733683256e2cfd436750f616054f707252e9 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Sun, 15 Sep 2024 19:48:55 -0500 Subject: [PATCH 17/17] More comments on unhooked system calls --- .../plugins/windows/debugregisters.py | 4 ++++ .../plugins/windows/unhooked_system_calls.py | 18 ++++++++++++++++-- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/volatility3/framework/plugins/windows/debugregisters.py b/volatility3/framework/plugins/windows/debugregisters.py index 945ba1df0..57dd1822c 100644 --- a/volatility3/framework/plugins/windows/debugregisters.py +++ b/volatility3/framework/plugins/windows/debugregisters.py @@ -1,6 +1,10 @@ # This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# Full details on the techniques used in these plugins to detect EDR-evading malware +# can be found in our 20 page whitepaper submitted to DEFCON along with the presentation +# https://www.volexity.com/wp-content/uploads/2024/08/Defcon24_EDR_Evasion_Detection_White-Paper_Andrew-Case.pdf + import logging from typing import Tuple, Optional, Generator, List, Dict diff --git a/volatility3/framework/plugins/windows/unhooked_system_calls.py b/volatility3/framework/plugins/windows/unhooked_system_calls.py index c3d98254d..1a1e59940 100644 --- a/volatility3/framework/plugins/windows/unhooked_system_calls.py +++ b/volatility3/framework/plugins/windows/unhooked_system_calls.py @@ -1,6 +1,10 @@ # This file is Copyright 2024 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# Full details on the techniques used in these plugins to detect EDR-evading malware +# can be found in our 20 page whitepaper submitted to DEFCON along with the presentation +# https://www.volexity.com/wp-content/uploads/2024/08/Defcon24_EDR_Evasion_Detection_White-Paper_Andrew-Case.pdf + import logging from typing import Dict, Tuple, List, Generator @@ -162,20 +166,30 @@ class unhooked_system_calls(interfaces.plugins.PluginInterface): # code_bytes[dll_name][func_name][func_bytes] code_bytes = self._gather_code_bytes(kernel, found_symbols) + # walk the functions that were evaluated for functions in code_bytes.values(): + # cbb is the distinct groups of bytes (instructions) + # for this function across processes for func_name, cbb in functions.items(): + # the dict key here is the raw instructions, which is not helpful to look at + # the values are the list of tuples for the (proc_id, proc_name) pairs for this set of bytes (instructions) cb = list(cbb.values()) - # same implementation in all + # if all processes map to the same implementation, then no malware is present if len(cb) == 1: yield 0, (func_name, "", len(cb[0])) else: - # find the processes that are hooked for reporting + # if there are differing implementations then it means + # that malware has overwritten system call(s) in infected processes + # max_idx and small_idx find which implementation of a system call has the least processes + # as all observed malware and open source projects only infected a few targets, leaving the + # rest with the original EDR hooks in place max_idx = 0 if len(cb[0]) > len(cb[1]) else 1 small_idx = (~max_idx) & 1 ps = [] + # gather processes on small_idx since these are the malware infected ones for pid, pname in cb[small_idx]: ps.append("{:d}:{}".format(pid, pname))