diff --git a/test/test_impscan.py b/test/test_impscan.py new file mode 100644 index 0000000000..12a94c3ac2 --- /dev/null +++ b/test/test_impscan.py @@ -0,0 +1,116 @@ +# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# +"""Unit tests for the pure logic of the windows.malware.impscan plugin. + +These exercise the IAT-detection and vicinity-scan logic directly (no memory +image required) by disassembling hand-built instruction bytes with capstone. +""" + +import pytest + +from volatility3.framework import exceptions + +capstone = pytest.importorskip("capstone") + +from volatility3.plugins.windows.malware.impscan import ImpScan # noqa: E402 + + +def _disasm_one(code: bytes, address: int, mode): + md = capstone.Cs(capstone.CS_ARCH_X86, mode) + md.detail = True + return next(md.disasm(code, address)) + + +def test_iat_target_x86_call_absolute(): + # call dword ptr [0x401000] + inst = _disasm_one(b"\xff\x15\x00\x10\x40\x00", 0x1000, capstone.CS_MODE_32) + assert ImpScan._iat_target(inst, is_64bit=False) == 0x401000 + + +def test_iat_target_x86_jmp_absolute(): + # jmp dword ptr [0x402000] + inst = _disasm_one(b"\xff\x25\x00\x20\x40\x00", 0x1000, capstone.CS_MODE_32) + assert ImpScan._iat_target(inst, is_64bit=False) == 0x402000 + + +def test_iat_target_x64_call_rip_relative(): + # call qword ptr [rip + 0x989d] at 0x1000; target = 0x1000 + 6 + 0x989d + inst = _disasm_one(b"\xff\x15\x9d\x98\x00\x00", 0x1000, capstone.CS_MODE_64) + assert ImpScan._iat_target(inst, is_64bit=True) == 0x1000 + 6 + 0x989D + + +def test_iat_target_ignores_register_call(): + # call eax (register operand, not a memory/IAT reference) + inst = _disasm_one(b"\xff\xd0", 0x1000, capstone.CS_MODE_32) + assert ImpScan._iat_target(inst, is_64bit=False) is None + + +def test_iat_target_ignores_non_branch(): + inst = _disasm_one(b"\x90", 0x1000, capstone.CS_MODE_32) # nop + assert ImpScan._iat_target(inst, is_64bit=False) is None + + +def test_call_scan_resyncs_past_undecodable_bytes(): + # capstone stops at the first byte it cannot decode (0xffff is an invalid + # FF /7 encoding). call_scan must skip it and still find the real + # "call dword [0x1004]" that follows. + md = capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32) + md.detail = True + scanner = ImpScan.__new__(ImpScan) + data = b"\xff\xff" + b"\xff\x15\x04\x10\x00\x00" + results = list(scanner.call_scan(md.disasm, 0x1000, data, is_64bit=False)) + assert (0x1002, 0x1004) in results + + +def test_original_import_remaps_forwarded(): + scanner = ImpScan.__new__(ImpScan) + assert scanner._original_import("ntdll.dll", "RtlAllocateHeap") == ( + "kernel32.dll", + "HeapAlloc", + ) + + +def test_original_import_passes_through_normal(): + scanner = ImpScan.__new__(ImpScan) + assert scanner._original_import("ws2_32.dll", "connect") == ( + "ws2_32.dll", + "connect", + ) + + +class _FakeLayer: + """Minimal layer returning little-endian pointers from a backing dict.""" + + def __init__(self, pointers, ptr_size): + self._pointers = pointers + self._ptr_size = ptr_size + + def read(self, address, length): + if address not in self._pointers: + raise exceptions.InvalidAddressException("fake", address, "no mapping") + return self._pointers[address].to_bytes(length, "little") + + +def test_read_pointer_reads_and_handles_missing(): + layer = _FakeLayer({0x2000: 0xDEADBEEF}, 4) + assert ImpScan._read_pointer(layer, 0x2000, 4) == 0xDEADBEEF + assert ImpScan._read_pointer(layer, 0x9999, 4) is None + + +def test_vicinity_scan_recovers_adjacent_imports(): + # Two IAT entries 4 bytes apart both point at known APIs; only the first + # was found by the call scan, the second should be recovered by walking. + ptr_size = 4 + base, data_len = 0x400000, 0x1000 + api_a, api_b = 0x77000000, 0x77000010 + apis = {api_a: ("a.dll", "A"), api_b: ("b.dll", "B")} + pointers = {0x500000: api_a, 0x500004: api_b} + calls_imported = {0x500000: api_a} + + scanner = ImpScan.__new__(ImpScan) + layer = _FakeLayer(pointers, ptr_size) + scanner._vicinity_scan( + layer, calls_imported, apis, base, data_len, ptr_size, forward=True + ) + assert calls_imported.get(0x500004) == api_b diff --git a/volatility3/framework/plugins/windows/malware/impscan.py b/volatility3/framework/plugins/windows/malware/impscan.py new file mode 100644 index 0000000000..6260d2cdf2 --- /dev/null +++ b/volatility3/framework/plugins/windows/malware/impscan.py @@ -0,0 +1,379 @@ +# This file is Copyright 2025 Volatility Foundation and licensed under the Volatility Software License 1.0 +# which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 +# +"""A plugin that scans process or kernel memory for calls to imported functions. + +This is a port of the Volatility 2 ``impscan`` plugin (originally by Michael +Ligh). It disassembles an executable region, finds ``CALL``/``JMP`` instructions +that reference the import address table (IAT), and resolves each IAT entry to the +exported function it points at. The result reconstructs a process's imports even +when the PE headers have been damaged or the program was injected, which is why +it is useful for malware analysis. +""" + +import logging +from typing import Dict, Generator, List, Optional, Tuple + +from volatility3.framework import exceptions, interfaces, renderers, symbols +from volatility3.framework.configuration import requirements +from volatility3.framework.renderers import format_hints +from volatility3.framework.symbols import intermed +from volatility3.framework.symbols.windows.extensions import pe +from volatility3.plugins.windows import pe_symbols, pslist + +try: + import capstone + + has_capstone = True +except ImportError: + has_capstone = False + +try: + import pefile + + has_pefile = True +except ImportError: + has_pefile = False + +vollog = logging.getLogger(__name__) + +# How far the vicinity scan walks away from a confirmed IAT entry, and how many +# consecutive misses it tolerates before giving up (ported from Volatility 2). +VICINITY_MAX_ENTRIES = 0x2000 +VICINITY_THRESHOLD = 5 + + +class ImpScan(interfaces.plugins.PluginInterface): + """Scans for calls to imported functions.""" + + _required_framework_version = (2, 22, 0) + _version = (1, 0, 0) + + # Some Rtl* routines are forwarded to a kernel32 export; map them back so the + # output shows the name an analyst expects (ported from Volatility 2). + forwarded_imports = { + "RtlGetLastWin32Error": "kernel32.dll!GetLastError", + "RtlSetLastWin32Error": "kernel32.dll!SetLastError", + "RtlRestoreLastWin32Error": "kernel32.dll!SetLastError", + "RtlAllocateHeap": "kernel32.dll!HeapAlloc", + "RtlReAllocateHeap": "kernel32.dll!HeapReAlloc", + "RtlFreeHeap": "kernel32.dll!HeapFree", + "RtlEnterCriticalSection": "kernel32.dll!EnterCriticalSection", + "RtlLeaveCriticalSection": "kernel32.dll!LeaveCriticalSection", + "RtlDeleteCriticalSection": "kernel32.dll!DeleteCriticalSection", + "RtlZeroMemory": "kernel32.dll!ZeroMemory", + "RtlSizeHeap": "kernel32.dll!HeapSize", + "RtlUnwind": "kernel32.dll!RtlUnwind", + } + + @classmethod + def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: + return [ + requirements.ModuleRequirement( + name="kernel", + description="Windows kernel", + architectures=["Intel32", "Intel64"], + ), + requirements.VersionRequirement( + name="pslist", component=pslist.PsList, version=(3, 0, 0) + ), + requirements.VersionRequirement( + name="pe_symbols", component=pe_symbols.PESymbols, version=(3, 0, 0) + ), + requirements.ListRequirement( + name="pid", + element_type=int, + description="Process IDs to scan (default: every process)", + optional=True, + ), + requirements.IntRequirement( + name="base", + description="Base address in process memory to scan from " + "(default: the main module)", + optional=True, + ), + requirements.IntRequirement( + name="size", + description="Size of memory to scan from --base", + optional=True, + ), + ] + + def _original_import(self, mod_name: str, func_name: str) -> Tuple[str, str]: + """Revert a forwarded import to its original module and function name.""" + if func_name in self.forwarded_imports: + mod, func = self.forwarded_imports[func_name].split("!") + return mod, func + return mod_name, func_name + + @staticmethod + def _iat_target(inst: "capstone.CsInsn", is_64bit: bool) -> Optional[int]: + """Return the IAT location a ``CALL``/``JMP`` references, or ``None``. + + On x86 the import is an absolute memory operand, e.g. + ``CALL DWORD [0x401000]``. On x64 it is RIP-relative, e.g. + ``CALL QWORD [RIP+0x989d]``, whose target is computed from the + instruction's own address. + """ + if inst.mnemonic not in ("call", "jmp"): + return None + if len(inst.operands) != 1: + return None + operand = inst.operands[0] + if operand.type != capstone.CS_OP_MEM: + return None + if is_64bit: + if inst.reg_name(operand.mem.base) == "rip": + return inst.address + inst.size + operand.mem.disp + return None + # x86: a direct memory operand with no base or index register. + if operand.mem.base == 0 and operand.mem.index == 0: + return operand.mem.disp & 0xFFFFFFFF + return None + + def enum_apis( + self, + context: interfaces.context.ContextInterface, + pe_table_name: str, + layer_name: str, + modules: List[Tuple[int, str]], + ) -> Dict[int, Tuple[str, str]]: + """Map every exported function address to its ``(module, name)``. + + ``modules`` is a list of ``(base_address, module_name)`` for the loaded + DLLs (process scan) or drivers (kernel scan). + """ + apis: Dict[int, Tuple[str, str]] = {} + for base, mod_name in modules: + pe_module = pe_symbols.PESymbols.get_pefile_obj( + context, pe_table_name, layer_name, base + ) + if not pe_module: + continue + pe_module.parse_data_directories( + directories=[pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"]] + ) + if not hasattr(pe_module, "DIRECTORY_ENTRY_EXPORT"): + continue + for export in pe_module.DIRECTORY_ENTRY_EXPORT.symbols: + if export.address is None: + continue + try: + name = export.name.decode("ascii") + except AttributeError: + name = str(export.ordinal) + apis[base + export.address] = (mod_name, name) + return apis + + def call_scan( + self, + disasm_func, + base_address: int, + data: bytes, + is_64bit: bool, + ) -> Generator[Tuple[int, int], None, None]: + """Disassemble ``data`` and yield ``(instruction_address, iat_location)`` + for every call or jump that references an address inside the region. + + capstone stops at the first byte it cannot decode, so the scanned region + (a whole module image, headers and data included) is swept linearly, + skipping one byte and resynchronising whenever decoding stalls. + """ + end_address = base_address + len(data) + view = memoryview(data) + data_len = len(data) + offset = 0 + while offset < data_len: + produced = False + for inst in disasm_func(view[offset:], base_address + offset): + produced = True + offset = (inst.address - base_address) + inst.size + iat_loc = self._iat_target(inst, is_64bit) + if iat_loc is None: + continue + if base_address <= iat_loc <= end_address: + yield inst.address, iat_loc + if not produced: + offset += 1 + + def _vicinity_scan( + self, + proc_layer: interfaces.layers.DataLayerInterface, + calls_imported: Dict[int, int], + apis: Dict[int, Tuple[str, str]], + base_address: int, + data_len: int, + ptr_size: int, + forward: bool, + ) -> None: + """Walk forward/backward from the found IAT entries to recover imports + that are present in the table but never called from the scanned code.""" + sorted_iats = sorted(calls_imported.keys()) + if not sorted_iats: + return + + start_addr = sorted_iats[0] if forward else sorted_iats[-1] + threshold = VICINITY_THRESHOLD + i = 0 + while threshold and i < VICINITY_MAX_ENTRIES: + if forward: + next_addr = start_addr + (i * ptr_size) + else: + next_addr = start_addr - (i * ptr_size) + i += 1 + + call_dest = self._read_pointer(proc_layer, next_addr, ptr_size) + if call_dest is None or ( + base_address < call_dest < base_address + data_len + ): + threshold -= 1 + continue + + if call_dest in apis and next_addr not in calls_imported: + calls_imported[next_addr] = call_dest + threshold = VICINITY_THRESHOLD + else: + threshold -= 1 + + @staticmethod + def _read_pointer( + proc_layer: interfaces.layers.DataLayerInterface, + address: int, + ptr_size: int, + ) -> Optional[int]: + """Read a little-endian pointer of ``ptr_size`` bytes, or ``None``.""" + try: + data = proc_layer.read(address, ptr_size) + except exceptions.InvalidAddressException: + return None + return int.from_bytes(data, byteorder="little") + + def _generator( + self, + kernel: interfaces.context.ModuleInterface, + procs: Generator[interfaces.objects.ObjectInterface, None, None], + ) -> Generator[Tuple[int, Tuple], None, None]: + is_64bit = symbols.symbol_table_is_64bit( + context=self.context, symbol_table_name=kernel.symbol_table_name + ) + pe_table_name = intermed.IntermediateSymbolTable.create( + self.context, self.config_path, "windows", "pe", class_types=pe.class_types + ) + + for proc in procs: + pid = proc.UniqueProcessId + try: + proc_layer_name = proc.add_process_layer() + except exceptions.InvalidAddressException: + continue + proc_layer = self.context.layers[proc_layer_name] + + wow64 = bool(proc.get_is_wow64()) + disasm_bits = "intel" if (not is_64bit or wow64) else "intel64" + ptr_size = 4 if (not is_64bit or wow64) else 8 + + modules = [] + main_base = main_size = None + for entry in proc.load_order_modules(): + try: + name = entry.BaseDllName.get_string() + base = int(entry.DllBase) + except exceptions.InvalidAddressException: + continue + modules.append((base, name)) + if main_base is None: + main_base, main_size = base, int(entry.SizeOfImage) + + base_address = self.config.get("base", None) or main_base + size_to_read = self.config.get("size", None) or main_size + if base_address is None or not size_to_read: + continue + + try: + data = proc_layer.read(base_address, size_to_read, pad=True) + except exceptions.InvalidAddressException: + continue + + apis = self.enum_apis(self.context, pe_table_name, proc_layer_name, modules) + disasm_func = self._get_disasm_function(disasm_bits) + + calls_imported: Dict[int, int] = {} + for _inst_addr, iat_loc in self.call_scan( + disasm_func, base_address, data, ptr_size == 8 + ): + call_dest = self._read_pointer(proc_layer, iat_loc, ptr_size) + if call_dest is not None and call_dest in apis: + calls_imported[iat_loc] = call_dest + + self._vicinity_scan( + proc_layer, + calls_imported, + apis, + base_address, + len(data), + ptr_size, + forward=True, + ) + self._vicinity_scan( + proc_layer, + calls_imported, + apis, + base_address, + len(data), + ptr_size, + forward=False, + ) + + for iat, call in sorted(calls_imported.items()): + mod_name, func_name = self._original_import(*apis[call]) + yield ( + 0, + ( + pid, + format_hints.Hex(iat), + format_hints.Hex(call), + mod_name, + func_name, + ), + ) + + @staticmethod + def _get_disasm_function(architecture: str): + disasm_types = { + "intel": capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32), + "intel64": capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_64), + } + disasm = disasm_types[architecture] + disasm.detail = True + return disasm.disasm + + def run(self) -> renderers.TreeGrid: + if not has_capstone: + raise exceptions.PluginRequirementException( + "impscan requires capstone to disassemble process memory." + ) + if not has_pefile: + raise exceptions.PluginRequirementException( + "impscan requires pefile to parse module export tables." + ) + + kernel = self.context.modules[self.config["kernel"]] + filter_func = pslist.PsList.create_pid_filter(self.config.get("pid", None)) + + return renderers.TreeGrid( + [ + ("PID", int), + ("IAT", format_hints.Hex), + ("Call", format_hints.Hex), + ("Module", str), + ("Function", str), + ], + self._generator( + kernel, + pslist.PsList.list_processes( + context=self.context, + kernel_module_name=self.config["kernel"], + filter_func=filter_func, + ), + ), + )