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))