feat(emu): resume out of a parked syscall via stack unwind

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

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:

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}]")

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]: