From fdd53711dc63321fd3032854e629908889e6fae4 Mon Sep 17 00:00:00 2001 From: alibhutto69 <92374187+thunderstornX@users.noreply.github.com> Date: Thu, 25 Jun 2026 15:17:41 +0500 Subject: [PATCH] Add impscan plugin Ports the Volatility 2 impscan plugin (by Michael Ligh) to Volatility 3. It disassembles a process's executable region, finds CALL/JMP instructions that reference the import address table, and resolves each IAT entry to the exported function it points at, reconstructing imports even when the PE headers are damaged or the code was injected. Export enumeration reuses the existing pe_symbols machinery, disassembly uses capstone, and a linear sweep with resynchronisation handles the non-code bytes in a mapped module image. Adds unit tests for the IAT-detection, resynchronisation and vicinity-scan logic. Closes #748 --- test/test_impscan.py | 116 ++++++ .../plugins/windows/malware/impscan.py | 379 ++++++++++++++++++ 2 files changed, 495 insertions(+) create mode 100644 test/test_impscan.py create mode 100644 volatility3/framework/plugins/windows/malware/impscan.py 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, + ), + ), + )