diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/greeneye/session.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/session.py | 113 |
1 files changed, 113 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/session.py b/debug/risc-v-gsp-rm-vm/greeneye/session.py index 7ad6444..fa54f77 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/session.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/session.py @@ -6,6 +6,7 @@ plugin-supplied backend later. """ from __future__ import annotations +import re import time from dataclasses import dataclass, field as _field from typing import Callable @@ -16,6 +17,15 @@ from .record import NullRecorder, Recorder from .rsp import RspClient, StopReply from .symbols import Symbol, SymbolDB +#: Registers saved and restored around ``Session.call`` (RISC-V ABI + pc). +RESTORE_REGS = ( + ["ra", "sp", "gp", "tp", "t0", "t1", "t2", "s0", "s1"] + + [f"a{i}" for i in range(8)] + + [f"s{i}" for i in range(2, 12)] + + [f"t{i}" for i in range(3, 7)] + + ["pc"] +) + @dataclass class Hit: @@ -195,11 +205,21 @@ class Session: # -- symbol helpers ---------------------------------------------------- def resolve(self, target: int | str) -> int: + """Resolve an address, a symbol name, or a ``symbol+0xNN`` expression.""" if isinstance(target, int): return target if isinstance(target, str): if target.lower().startswith("0x"): return int(target, 16) + expression = re.fullmatch( + r"([A-Za-z_][A-Za-z0-9_.]*)\s*([+-])\s*(0x[0-9a-fA-F]+|\d+)", target.strip() + ) + if expression: + base = self.symbols.get(expression.group(1)) + if base is None: + raise SymbolNotFound(f"symbol {expression.group(1)!r} not found in symbol DB") + delta = int(expression.group(3), 0) + return base.addr + delta if expression.group(2) == "+" else base.addr - delta sym = self.symbols.get(target) if sym is None: raise SymbolNotFound(f"symbol {target!r} not found in symbol DB") @@ -338,6 +358,99 @@ class Session: break return hits + # -- Frida-like remote call / run-until ------------------------------- + + def _scratch_addr(self, override: int | None = None) -> int: + if override: + return override + if self.target is not None and getattr(self.target, "memory_mb", 0): + base = getattr(self.target, "ram_base", 0x80000000) + return base + self.target.memory_mb * 1024 * 1024 - 0x1000 + if self.image is not None and getattr(self.image, "max_vaddr", 0): + return self.image.max_vaddr + 0x10000 + return 0x801FF000 + + def run_until(self, target: int | str, timeout: float = 5.0) -> Hit: + """Run until execution reaches ``target`` (Frida's run-to-cursor). + + Sets a temporary breakpoint, services intervening stops, and returns + the Hit for the target. Raises ``RspTimeout`` if not reached. + """ + addr = self.resolve(target) + if self.pc == addr: + return Hit(kind="stop", addr=addr, regs=self.read_regs(), symbol=self.symbols.resolve(addr), session=self) + temp = Breakpoint(addr=addr, kind="sw", temporary=True, on_enter=None) + self.add_breakpoint(temp) + deadline = time.monotonic() + timeout + try: + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RspTimeout(f"run_until {addr:#x}: not reached in {timeout}s") + try: + stop = self.client.continue_(timeout=remaining) + except RspTimeout: + self.client.interrupt() + raise RspTimeout(f"run_until {addr:#x}: not reached in {timeout}s") + hit = self._dispatch_stop(stop) + if hit is not None and hit.addr == addr and hit.kind == "breakpoint": + return hit + if stop.is_exit: + raise GreenEyeError(f"run_until {addr:#x}: target exited") + finally: + self.remove_breakpoint(temp) + + def call(self, target: int | str, args: tuple | list = (), timeout: float = 5.0, scratch: int | None = None) -> int: + """Call a function in the target from the debugger (Frida NativeFunction). + + Writes an ``ebreak`` trampoline into a scratch page, sets ``ra`` to it, + loads ``a0..a7`` and a private stack, jumps to ``target``, services + intervening stops (hooks/watchpoints included), and returns ``a0``. + All caller-saved registers and ``pc`` are restored afterwards. + """ + addr = self.resolve(target) + scratch = self._scratch_addr(scratch) + stack = scratch + 0x800 + saved = self.read_regs() + temp = Breakpoint(addr=scratch, kind="sw", temporary=True, purpose="user") + self.add_breakpoint(temp) + try: + for name in ("a0", "a1", "a2", "a3", "a4", "a5", "a6", "a7"): + self.client.write_reg(name, 0) + for index, value in enumerate(args[:8]): + self.client.write_reg(f"a{index}", int(value) & 0xFFFF_FFFF_FFFF_FFFF) + self.client.write_reg("ra", scratch) + self.client.write_reg("sp", stack) + self.client.write_reg("pc", addr) + self.recorder.event("call_enter", addr=addr, args=list(args), scratch=scratch) + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RspTimeout(f"call {addr:#x}: no return in {timeout}s") + try: + stop = self.client.continue_(timeout=remaining) + except RspTimeout: + self.client.interrupt() + raise RspTimeout(f"call {addr:#x}: no return in {timeout}s") + hit = self._dispatch_stop(stop) + pc = self.client.pc + if pc == scratch or (hit is not None and hit.addr == scratch): + retval = self.client.read_reg("a0") + self.recorder.event("call_return", addr=addr, retval=retval) + return retval + if stop.is_exit: + raise GreenEyeError(f"call {addr:#x}: target exited") + finally: + if temp.attached: + self.remove_breakpoint(temp) + for name in RESTORE_REGS: + if name in saved: + try: + self.client.write_reg(name, saved[name]) + except Exception: + pass + def run( self, duration: float | None = None, |