From d3324b7523f2ff26deb5be87c38a9a51e7b0fdc3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ricardo=20J=2E=20Rodr=C3=ADguez?= Date: Sun, 28 Jun 2026 17:08:56 +0200 Subject: [PATCH] feat(emu): resume out of a parked syscall via stack unwind A slice is frequently captured with its thread blocked in a library/kernel call (a Sleep, a wait, a recv), so the captured PC sits in ntdll/kernel. Stepping forward from there hits a syscall the emulator can't service, so the captured state was hard to drive. Add `MSLEmulator.resume_from_syscall()`: scan the stack from SP for the nearest return address that points into the program image (executable, captured memory), then set PC/SP so emulation continues in the caller as if the call had returned. The program image is identified by its `.exe` module; `--image-range` overrides it. `--pop-bytes N` additionally discards stdcall argument bytes the callee's `ret N` would have cleaned up (e.g. 4 for Sleep). Supporting helpers: `find_caller_frame()`, `in_system_call()`, `main_image()`, `module_at()`, and a public `sp_name`. Exposed on the CLI as `memslicer-emu --resume-from-syscall/-R`. Adds engine + CLI tests (skip-walking system/junk frames, the SP fixup, pop-bytes, the no-caller case, and CLI output). README + CHANGELOG updated. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 8 ++ README.md | 24 ++++++ src/memslicer/cli_emu.py | 58 ++++++++++++- src/memslicer/emu/engine.py | 114 +++++++++++++++++++++++- tests/test_emu.py | 167 +++++++++++++++++++++++++++++++++++- 5 files changed, 368 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index df13041..5697d9b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,14 @@ All notable changes to this project will be documented in this file. from the Thread Context, and single-steps execution. Install with the `emu` extra (`pip install memslicer[emu]`). Supports x86/x86_64/ARM/ARM64, and reverse execution (step back) via a CPU-context + memory-write journal. +- `memslicer-emu --resume-from-syscall` (`-R`): when a slice is captured + parked in a blocking library/syscall (e.g. a `Sleep`), unwind out of it by + finding the caller's return address into the program image on the stack and + continuing there as if the call had returned — stepping forward would + otherwise hit a syscall the emulator can't service. `--pop-bytes` discards + stdcall argument cleanup and `--image-range` overrides image auto-detection. + Exposed in the library as `MSLEmulator.resume_from_syscall()` / + `find_caller_frame()` / `in_system_call()` / `main_image()`. - New `memslicer-symbex` tool (and `memslicer.symbex` library) that loads a slice into [angr](https://angr.io) — captured memory and registers become a `SimState` at the captured PC — for symbolic execution / exploration. Behind diff --git a/README.md b/README.md index cb2074c..0dbf51c 100644 --- a/README.md +++ b/README.md @@ -343,6 +343,30 @@ emu.written_ranges() # coalesced dirtied byte ranges emu.dump_written("out/") # write each range out (executed ones = the payload) ``` +**Parked in a syscall.** A process is often captured blocked in a library/kernel +call (a `Sleep`, a wait, a `recv`), so the captured PC sits in `ntdll`/`kernel` +and stepping forward would hit a syscall the emulator can't service. Unwind back +to your own code with `--resume-from-syscall` (`-R`): it scans the stack for the +caller's return address into the program image and continues there as if the +call had returned. + +```bash +memslicer-emu parked.msl -R --until 0xb1070 -r +# [resume-from-syscall] pc = 0x7c90e514 (ntdll.dll+0x...) +# caller return @ 0xb3275 (sample.exe) [stack 0x..., depth 3] +# resumed: pc -> 0xb3275, esp -> 0x... +``` + +For a stdcall API that cleans up its own arguments (e.g. `Sleep`'s `ret 4`), add +`--pop-bytes N` so the caller's stack is left balanced. Override the auto-detected +image with `--image-range LO:HI`. As a library: + +```python +emu.in_system_call() # True — PC is outside the program image +frame = emu.resume_from_syscall(pop_bytes=4) # find caller, reposition pc/sp +emu.pc == frame.return_addr # now back in the image; keep stepping +``` + A slice can capture **more than one thread** (one Thread Context block each). By default the emulator seeds from the Current thread, but any captured thread can be selected: diff --git a/src/memslicer/cli_emu.py b/src/memslicer/cli_emu.py index 16d0143..45e86d3 100644 --- a/src/memslicer/cli_emu.py +++ b/src/memslicer/cli_emu.py @@ -6,6 +6,12 @@ then execution is stepped forward. Use ``--list-threads`` to see what was captured. +A slice is often captured parked in a blocking library/syscall (e.g. a Sleep), +so its PC sits in ntdll/kernel and stepping forward would hit a syscall the +emulator can't service. ``--resume-from-syscall`` unwinds that: it finds the +caller's return address on the stack and continues in the program image as if +the call had returned. + Requires the ``emu`` extra:: pip install memslicer[emu] """ from __future__ import annotations @@ -22,6 +28,25 @@ def _parse_addr(value: str | None) -> int | None: return int(value, 0) +def _parse_range(value: str | None) -> "tuple[int, int] | None": + if not value: + return None + lo, sep, hi = value.partition(":") + if not sep or not hi: + raise click.ClickException( + "--image-range expects LO:HI (e.g. 0xb0000:0xd5000)") + return int(lo, 0), int(hi, 0) + + +def _module_loc(image, addr: int) -> str | None: + """Format ``addr`` as ``module+offset`` if it falls inside a captured + module, else None.""" + for m in image.modules: + if m.base <= addr < m.base + m.size: + return f"{m.name}+{addr - m.base:#x}" + return None + + def _print_summary(image) -> None: click.echo(f"arch : {image.arch.name}") click.echo(f"os : {image.os.name}") @@ -70,6 +95,18 @@ def _trace(emu: MSLEmulator, results) -> None: help="Step until this address (hex or decimal)") @click.option("--pc", "pc_override", default=None, help="Override the start program counter") +@click.option("-R", "--resume-from-syscall", "resume_syscall", is_flag=True, + help="Unwind out of the library/syscall the slice is parked in: " + "find the caller's return address on the stack and continue " + "in the program image as if the call had returned") +@click.option("--pop-bytes", type=int, default=0, + help="With -R, also discard this many stdcall argument bytes " + "(the callee's 'ret N' cleanup, e.g. 4 for Sleep)") +@click.option("--image-range", default=None, + help="With -R, target return addresses in LO:HI instead of the " + "auto-detected program image (e.g. 0xb0000:0xd5000)") +@click.option("--unwind-depth", type=int, default=256, + help="With -R, max stack slots to scan for the return address") @click.option("--max-steps", type=int, default=100000, help="Safety cap when using --until") @click.option("-b", "--back", type=int, default=0, @@ -84,7 +121,8 @@ def _trace(emu: MSLEmulator, results) -> None: type=click.Path(file_okay=False), help="After stepping, write dirtied memory ranges to this dir " "(recovers unpacked/decoded payloads)") -def main(dump, steps, until_addr, pc_override, show_regs, max_steps, back, +def main(dump, steps, until_addr, pc_override, resume_syscall, pop_bytes, + image_range, unwind_depth, show_regs, max_steps, back, thread_id, list_threads, dump_written_dir): """Emulate the MSL slice DUMP.""" image = load_slice(dump) @@ -105,6 +143,24 @@ def main(dump, steps, until_addr, pc_override, show_regs, max_steps, back, if pc is not None: emu.pc = pc + if resume_syscall: + click.echo("") + here = _module_loc(image, emu.pc) + click.echo(f"[resume-from-syscall] pc = {emu.pc:#x}" + + (f" ({here})" if here else "")) + frame = emu.resume_from_syscall( + image_range=_parse_range(image_range), + max_depth=unwind_depth, pop_bytes=pop_bytes) + if frame is None: + raise click.ClickException( + "no return address into the program image found on the stack " + "(try --image-range LO:HI or a larger --unwind-depth)") + where = f" ({frame.module})" if frame.module else "" + click.echo(f" caller return @ {frame.return_addr:#x}{where}" + f" [stack {frame.sp_slot:#x}, depth {frame.depth}]") + click.echo(f" resumed: pc -> {emu.pc:#x}, " + f"{emu.sp_name} -> {emu.read_reg(emu.sp_name):#x}") + if steps > 0 or until_addr is not None: click.echo("") click.echo(f"[start pc = {emu.pc:#x}]") diff --git a/src/memslicer/emu/engine.py b/src/memslicer/emu/engine.py index de6bfa9..4233f45 100644 --- a/src/memslicer/emu/engine.py +++ b/src/memslicer/emu/engine.py @@ -8,7 +8,8 @@ from dataclasses import dataclass from memslicer.msl.constants import ArchType, OSType -from memslicer.emu.loader import EmuThread, SliceImage, load_slice +from memslicer.emu.loader import EmuModule, EmuThread, SliceImage, load_slice +from memslicer.utils.protection import PROT_X _UC_PAGE = 0x1000 @@ -32,6 +33,19 @@ def __str__(self) -> str: return text if self.ok else f"{text} ! {self.error}" +@dataclass +class CallerFrame: + """A return address recovered from the stack. + + ``sp_slot`` is the stack address that holds it, ``return_addr`` the caller's + next instruction, ``depth`` how many pointer-sized slots above SP it was + found, and ``module`` the owning module's name (if known).""" + sp_slot: int + return_addr: int + depth: int + module: str | None = None + + # arch -> (uc_arch, uc_mode, cs_arch, cs_mode, bits, const_module, reg_prefix, # pc_name, sp_name, gpr_names) def _arch_table(): @@ -270,6 +284,12 @@ def switch_thread(self, thread: "int | EmuThread | None") -> "EmuThread | None": # -- registers / memory -------------------------------------------------- + @property + def sp_name(self) -> str: + """Name of this architecture's stack-pointer register (``esp``/``rsp``/ + ``sp``).""" + return self._sp_name + @property def pc(self) -> int: return self.uc.reg_read(self._reg_const(self._pc_name)) @@ -332,6 +352,98 @@ def peb_address(self) -> int | None: except self._U.UcError: return None + # -- modules / call frames ---------------------------------------------- + + def module_at(self, addr: int) -> "EmuModule | None": + """Return the loaded module whose image range covers *addr*, or None.""" + for m in self.image.modules: + if m.base <= addr < m.base + m.size: + return m + return None + + def main_image(self) -> "EmuModule | None": + """Best guess at the analyzed program's own image: the ``.exe`` module + (lowest base if several), else the lowest-based module. None when the + slice captured no module list.""" + mods = self.image.modules + if not mods: + return None + exes = [m for m in mods if m.name.lower().endswith(".exe")] + return min(exes or mods, key=lambda m: m.base) + + def _addr_executable(self, addr: int) -> bool: + """True if *addr* lies in a captured, execute-protected region, i.e. a + plausible return address we could actually resume into.""" + for r in self.image.regions: + if r.contains(addr): + if not (r.protection & PROT_X): + return False + return (addr & ~(r.page_size - 1)) in r.pages + return False + + def in_system_call(self) -> bool | None: + """True if PC sits outside the program image (in a system DLL / kernel + thunk) -- the slice is parked in a library call. None when the image + can't be identified (no module list).""" + m = self.main_image() + if m is None: + return None + return not (m.base <= self.pc < m.base + m.size) + + def find_caller_frame(self, image_range: "tuple[int, int] | None" = None, + max_depth: int = 256) -> "CallerFrame | None": + """Walk the stack upward from SP and return the nearest return address + that points into the program image (executable, captured memory). + + When a slice is captured parked in a blocking call (e.g. a Sleep), the + library/kernel frames sit on top of the stack and the first stack slot + pointing back into the analyzed image is the instruction right after the + ``call`` that entered the library. *image_range* overrides the + ``(lo, hi)`` target range (default: :meth:`main_image`'s range). Returns + None if no such slot is found within *max_depth* pointer-sized slots.""" + if image_range is not None: + lo, hi = image_range + else: + m = self.main_image() + if m is None: + return None + lo, hi = m.base, m.base + m.size + sp = self.read_reg(self._sp_name) + ptr = self.bits // 8 + for depth in range(max_depth): + slot = sp + depth * ptr + try: + val = int.from_bytes(self.read_mem(slot, ptr), "little") + except self._U.UcError: + break + if lo <= val < hi and self._addr_executable(val): + mod = self.module_at(val) + return CallerFrame(sp_slot=slot, return_addr=val, depth=depth, + module=mod.name if mod else None) + return None + + def resume_from_syscall(self, image_range: "tuple[int, int] | None" = None, + max_depth: int = 256, pop_bytes: int = 0 + ) -> "CallerFrame | None": + """Unwind out of a library/syscall the slice is parked in: find the + caller's return address on the stack (:meth:`find_caller_frame`) and set + PC/SP so emulation continues in the program image as if the call had + returned. + + SP is moved just past the return-address slot (a plain ``ret``); pass + *pop_bytes* to additionally discard the stdcall argument bytes the + callee's ``ret N`` would have cleaned up (e.g. 4 for ``Sleep``). Returns + the :class:`CallerFrame` used, or None if no caller return address into + the image was found. The reverse-execution history is dropped: the + unwind is not a reversible single step.""" + frame = self.find_caller_frame(image_range=image_range, max_depth=max_depth) + if frame is None: + return None + self.pc = frame.return_addr + self.write_reg(self._sp_name, frame.sp_slot + self.bits // 8 + pop_bytes) + self._history = [] + return frame + # -- execution ----------------------------------------------------------- def _disasm_at(self, pc: int) -> tuple[int, str, str]: diff --git a/tests/test_emu.py b/tests/test_emu.py index fb4f86c..e3e4b5a 100644 --- a/tests/test_emu.py +++ b/tests/test_emu.py @@ -4,7 +4,8 @@ from memslicer.emu.loader import load_slice from memslicer.msl.writer import MSLWriter from memslicer.msl.types import ( - FileHeader, ProcessIdentity, MemoryRegion, ThreadContext, ThreadRegister, + FileHeader, ModuleEntry, ProcessIdentity, MemoryRegion, ThreadContext, + ThreadRegister, ) from memslicer.msl.constants import ( ArchType, CapBit, CompAlgo, OSType, PageState, RegionType, @@ -533,3 +534,167 @@ def test_emulator_x86_without_fs_base_still_runs(tmp_path): assert emu.step().ok # no regression +# ---- resume from a library/syscall the slice is parked in -------------------- + +SYS_VA = 0x70000000 # stand-in for ntdll.dll (where the parked PC sits) + + +def _write_parked_slice(path, *, rsp=STACK_VA + 0x100, stack_layout=None): + """A slice captured 'parked in a system call': the Current thread's PC is + inside a system module (SYS_VA, ~ntdll), and the stack carries a system + return address, some junk, then a real return address into the program + image (``demo.exe`` at CODE_VA). *stack_layout* maps a stack offset (from + STACK_VA) to an 8-byte value; defaults to that 3-slot layout.""" + if stack_layout is None: + stack_layout = { + 0x100: SYS_VA + 4, # nested system frame -> skipped (not image) + 0x108: 0x12345, # junk, not executable -> skipped + 0x110: CODE_VA, # return into the image -> the caller + } + stack = bytearray(b"\x00" * PS) + for off, val in stack_layout.items(): + stack[off:off + 8] = val.to_bytes(8, "little") + code_page = CODE + b"\x90" * (PS - len(CODE)) + cap = ((1 << CapBit.MemoryRegions) | (1 << CapBit.ProcessIdentity) + | (1 << CapBit.ThreadContexts)) + hdr = FileHeader(os_type=OSType.Windows, arch_type=ArchType.x86_64, + pid=964, cap_bitmap=cap) + with open(path, "wb") as f: + w = MSLWriter(f, hdr, CompAlgo.NONE) + w.write_process_identity(ProcessIdentity(exe_path="C:\\demo.exe")) # block 0 + w.write_module_list([ # block 1 (per spec) + ModuleEntry(base_addr=CODE_VA, module_size=PS, path="C:\\demo.exe"), + ModuleEntry(base_addr=SYS_VA, module_size=PS, + path="C:\\Windows\\System32\\ntdll.dll"), + ]) + w.write_memory_region(MemoryRegion( # program image (r-x) + base_addr=CODE_VA, region_size=PS, protection=0b101, + region_type=RegionType.Image, page_size=PS, + page_states=[PageState.CAPTURED], page_data_chunks=[code_page])) + w.write_memory_region(MemoryRegion( # ntdll (r-x), parked PC + base_addr=SYS_VA, region_size=PS, protection=0b101, + region_type=RegionType.Image, page_size=PS, + page_states=[PageState.CAPTURED], page_data_chunks=[b"\x90" * PS])) + w.write_memory_region(MemoryRegion( # stack (rw-) + base_addr=STACK_VA, region_size=PS, protection=0b011, + region_type=RegionType.Stack, page_size=PS, + page_states=[PageState.CAPTURED], page_data_chunks=[bytes(stack)])) + w.write_thread_context(ThreadContext( + thread_id=964, flags=THREAD_FLAG_CURRENT, state=ThreadState.Stopped, + name="main", registers=[ + ThreadRegister("rip", SYS_VA.to_bytes(8, "little"), REG_FLAG_PC), + ThreadRegister("rsp", rsp.to_bytes(8, "little"), REG_FLAG_SP), + ])) + w.finalize() + + +def test_emulator_identifies_image_and_system_call(tmp_path): + pytest.importorskip("unicorn") + pytest.importorskip("capstone") + from memslicer.emu.engine import MSLEmulator + + p = tmp_path / "parked.msl" + _write_parked_slice(p) + emu = MSLEmulator(load_slice(str(p))) + + assert emu.main_image().name == "demo.exe" # the .exe, not ntdll + assert emu.module_at(SYS_VA).name == "ntdll.dll" + assert emu.pc == SYS_VA + assert emu.in_system_call() is True # parked outside the image + + +def test_emulator_find_caller_frame_skips_system_and_junk(tmp_path): + pytest.importorskip("unicorn") + pytest.importorskip("capstone") + from memslicer.emu.engine import MSLEmulator + + p = tmp_path / "parked.msl" + _write_parked_slice(p) + emu = MSLEmulator(load_slice(str(p))) + + frame = emu.find_caller_frame() + assert frame is not None + assert frame.return_addr == CODE_VA # skipped ntdll + junk + assert frame.sp_slot == STACK_VA + 0x110 + assert frame.depth == 2 + assert frame.module == "demo.exe" + + +def test_emulator_resume_from_syscall(tmp_path): + pytest.importorskip("unicorn") + pytest.importorskip("capstone") + from memslicer.emu.engine import MSLEmulator + + p = tmp_path / "parked.msl" + _write_parked_slice(p) + emu = MSLEmulator(load_slice(str(p))) + + frame = emu.resume_from_syscall() + assert frame is not None + assert emu.pc == CODE_VA # back in the image + assert emu.in_system_call() is False + # plain 'ret': SP moved just past the return-address slot + assert emu.read_reg("rsp") == STACK_VA + 0x110 + 8 + assert not emu.can_step_back() # unwind dropped history + + emu.step() # mov rax, 1 (really runs) + assert emu.read_reg("rax") == 1 + + +def test_emulator_resume_from_syscall_pops_stdcall_args(tmp_path): + pytest.importorskip("unicorn") + pytest.importorskip("capstone") + from memslicer.emu.engine import MSLEmulator + + p = tmp_path / "parked.msl" + _write_parked_slice(p) + emu = MSLEmulator(load_slice(str(p))) + + emu.resume_from_syscall(pop_bytes=4) # e.g. Sleep's 'ret 4' + assert emu.pc == CODE_VA + assert emu.read_reg("rsp") == STACK_VA + 0x110 + 8 + 4 + + +def test_emulator_resume_from_syscall_no_caller(tmp_path): + pytest.importorskip("unicorn") + pytest.importorskip("capstone") + from memslicer.emu.engine import MSLEmulator + + # A stack with no return address into the image -> nothing to resume to. + p = tmp_path / "noframe.msl" + _write_parked_slice(p, stack_layout={0x100: SYS_VA + 4, 0x108: 0x12345}) + emu = MSLEmulator(load_slice(str(p))) + assert emu.find_caller_frame() is None + assert emu.resume_from_syscall() is None + + +def test_cli_resume_from_syscall(tmp_path): + pytest.importorskip("unicorn") + pytest.importorskip("capstone") + from click.testing import CliRunner + from memslicer.cli_emu import main + + p = tmp_path / "parked.msl" + _write_parked_slice(p) + res = CliRunner().invoke(main, [str(p), "-R", "-s", "1", "-r"]) + assert res.exit_code == 0, res.output + assert "resume-from-syscall" in res.output + assert "ntdll.dll" in res.output # reports where it was parked + assert f"caller return @ {CODE_VA:#x}" in res.output + assert "demo.exe" in res.output + assert "rsp" in res.output + + +def test_cli_resume_from_syscall_image_range(tmp_path): + pytest.importorskip("unicorn") + pytest.importorskip("capstone") + from click.testing import CliRunner + from memslicer.cli_emu import main + + p = tmp_path / "parked.msl" + _write_parked_slice(p) + # Explicit image range instead of auto-detection. + res = CliRunner().invoke( + main, [str(p), "-R", "--image-range", f"{CODE_VA:#x}:{CODE_VA + PS:#x}"]) + assert res.exit_code == 0, res.output + assert f"caller return @ {CODE_VA:#x}" in res.output