diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/greeneye')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/analyze.py | 28 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/cli.py | 96 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/gdbridge.py | 220 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/inject.py | 129 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/record.py | 2 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/session.py | 113 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/target.py | 2 |
7 files changed, 589 insertions, 1 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/analyze.py b/debug/risc-v-gsp-rm-vm/greeneye/analyze.py index 07c24eb..66429e9 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/analyze.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/analyze.py @@ -110,6 +110,34 @@ def render_summary_from(meta: dict, events: list[dict], symbols=None, max_rows: lines.append("* recorder counts: " + ", ".join(f"{k}={v}" for k, v in sorted(counts_meta.items()))) lines.append("") + # QEMU exception report (probe sessions) + fault_count = meta.get("qemu_fault_count") + first = meta.get("qemu_first_fault") + if fault_count or first: + lines.append("## QEMU exceptions") + lines.append("") + lines.append(f"* total exceptions: {fault_count if fault_count is not None else '?'}") + if meta.get("qemu_log_blob"): + lines.append(f"* qemu log blob: `{meta['qemu_log_blob']}`") + if first: + access = first.get("access") or {} + lines.append("") + lines.append("| field | value |") + lines.append("|---|---|") + lines.append(f"| descriptor | `{first.get('desc', '?')}` |") + lines.append(f"| cause | `{first.get('cause', 0):#x}` ({_RISCV_CAUSES.get(first.get('cause', 0) & 0xFF, '?')}) |") + epc = first.get("epc", 0) + bias = meta.get("bias") or 0 + epc_cell = f"`{epc:#x}`" + (f" (firmware vaddr `{epc - bias:#x}`)" if bias else "") + lines.append(f"| epc | {epc_cell} |") + lines.append(f"| tval | `{first.get('tval', 0):#x}` |") + if access: + lines.append( + f"| first access | `{access.get('rw')}` at `{access.get('access_addr', 0):#x}` " + f"({access.get('size')} bytes, reason `{access.get('reason')}`) |" + ) + lines.append("") + def annotate(addr): if not isinstance(addr, int): return "" diff --git a/debug/risc-v-gsp-rm-vm/greeneye/cli.py b/debug/risc-v-gsp-rm-vm/greeneye/cli.py index 5bfcee2..94102a4 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/cli.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/cli.py @@ -332,6 +332,15 @@ class GreenEyeRepl(cmd.Cmd): stops = self.session.run(duration=seconds) self._say(f"resumed for {seconds}s, serviced {stops} stops") + def do_call(self, arg): + """call <addr|symbol> [a0 [a1 ...]] -- remote function call; prints the result.""" + parts = arg.split() + if not parts: + return self._say("usage: call <addr|symbol> [args...]") + values = [int(part, 0) for part in parts[1:]] + retval = self.session.call(parts[0], values) + self._say(f" -> {retval:#x}") + def do_events(self, arg): """events [n] -- show the last n recorded events (default 20).""" from .record import load_events @@ -512,6 +521,12 @@ def cmd_probe(args) -> int: print(f"[probe] total exceptions: {len(faults)}") else: print("[probe] no exception recorded (execution stayed in mapped memory)") + recorder = session.recorder + if hasattr(recorder, "meta"): + recorder.meta["qemu_fault_count"] = len(faults) + recorder.meta["qemu_first_fault"] = faults[0] if faults else None + if qlog: + recorder.meta["qemu_log_blob"] = recorder.blob("qemu.log", qlog.encode("utf-8")) tail = qlog.strip().splitlines() if tail: print("[probe] qemu log tail:") @@ -523,6 +538,64 @@ def cmd_probe(args) -> int: return 0 +def cmd_inject(args) -> int: + _ensure_dumps_env() + from .inject import invoke_remote, invoke_via_firmware, load_service + from .target import free_port # noqa: F401 (kept for symmetry with launch) + + service_path = Path(args.service) + if not service_path.exists(): + raise GreenEyeError( + f"service not found: {service_path} (build it with `make sdk`)" + ) + kernel = args.kernel or str(_build_demo()) + session, _manager = _start_session( + kernel=str(kernel), port=args.port, record=args.record or True, name="inject", plugins=args.plugin + ) + try: + service = load_service(session, str(service_path), load_addr=args.load_addr, slot=args.slot) + print(f"[inject] {service.summary()}") + remote = invoke_remote(session, service, args.cmd, args.arg0, args.arg1) + print(f"[inject] direct call dispatch(cmd={args.cmd}, a0={args.arg0:#x}, a1={args.arg1:#x}) -> {remote:#x}") + if not args.no_firmware: + session.run(duration=args.duration) + fw = invoke_via_firmware(session, service, args.cmd, args.arg0, args.arg1) + print(f"[inject] firmware call mailbox={16 + service.slot} -> result {fw:#x}") + _print_session_dir(session) + finally: + session.close() + return 0 + + +def cmd_ghidra(args) -> int: + """Serve a live session to GhidraScripts over the JSON-lines bridge.""" + _ensure_dumps_env() + from .gdbridge import GhidraBridge + + kernel = args.kernel + if kernel: + kernel = str(Path(kernel)) + elif not args.target_port: + kernel = str(_build_demo()) + session, _manager = _start_session( + kernel=kernel, port=args.target_port, record=args.record or True, name="ghidra", plugins=args.plugin + ) + bridge = GhidraBridge(session, port=args.bridge_port) + port = bridge.start() + print(f"[green-eye] ghidra bridge on 127.0.0.1:{port}") + print(f"[green-eye] session: {getattr(session.recorder, 'dir', None)}") + print("[green-eye] in Ghidra: analyzeHeadless ... -postScript GreenEyeControl.java " + str(port)) + print("[green-eye] Ctrl-C to stop the bridge") + try: + bridge.serve_forever() + except KeyboardInterrupt: + pass + finally: + bridge.stop() + session.close() + return 0 + + def cmd_analyze(args) -> int: symbols = None if args.symbols: @@ -627,6 +700,21 @@ def build_parser() -> argparse.ArgumentParser: probe.add_argument("--record", nargs="?", const=True, default=None) probe.set_defaults(fn=cmd_probe) + inject = sub.add_parser("inject", help="load and invoke an injected RV64 service (Frida-like)") + inject.add_argument("--service", default=str(TOOL_ROOT / "sdk" / "build" / "injected_demo.elf")) + inject.add_argument("--kernel") + inject.add_argument("--port", type=int) + inject.add_argument("--load-addr", type=lambda v: int(v, 0)) + inject.add_argument("--slot", type=int, default=6, help="demo handler slot 0..7 (default 6)") + inject.add_argument("--cmd", type=lambda v: int(v, 0), default=1) + inject.add_argument("--arg0", type=lambda v: int(v, 0), default=0x41) + inject.add_argument("--arg1", type=lambda v: int(v, 0), default=0x42) + inject.add_argument("--duration", type=float, default=0.3, help="firmware settle time before the mailbox call") + inject.add_argument("--no-firmware", action="store_true", help="only make the direct debugger-side call") + inject.add_argument("--record", nargs="?", const=True, default=None) + inject.add_argument("--plugin", action="append", default=[]) + inject.set_defaults(fn=cmd_inject) + record = sub.add_parser("record", help="boot and record a memory-activity session") record.add_argument("--kernel") record.add_argument("--name", default="record") @@ -636,6 +724,14 @@ def build_parser() -> argparse.ArgumentParser: record.add_argument("--record", nargs="?", const=True, default=None) record.set_defaults(fn=cmd_record) + ghidra = sub.add_parser("ghidra", help="serve a live session to GhidraScripts (JSON-lines bridge)") + ghidra.add_argument("--kernel") + ghidra.add_argument("--target-port", type=int, help="attach to an existing gdbstub instead of launching") + ghidra.add_argument("--bridge-port", type=int, default=7000) + ghidra.add_argument("--record", nargs="?", const=True, default=None) + ghidra.add_argument("--plugin", action="append", default=[]) + ghidra.set_defaults(fn=cmd_ghidra) + analyze = sub.add_parser("analyze", help="render a recorded session report") analyze.add_argument("session") analyze.add_argument("--symbols") diff --git a/debug/risc-v-gsp-rm-vm/greeneye/gdbridge.py b/debug/risc-v-gsp-rm-vm/greeneye/gdbridge.py new file mode 100644 index 0000000..cc06e43 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/greeneye/gdbridge.py @@ -0,0 +1,220 @@ +"""Ghidra control bridge: let GhidraScripts drive a green-eye session. + +A tiny JSON-lines server over localhost. Each request is one line:: + + {"id": 1, "cmd": "run_until", "args": {"target": "0x800000ce", "timeout": 5}} + +and each response is one line:: + + {"id": 1, "ok": true, "result": {"addr": 2147483854}} + +This is the transport behind ``ghidra/GreenEyeControl.java``: a Ghidra user +can "execute until here", set/clear breakpoints, read memory and export hit +lists without leaving the decompiler. See ``docs/13-GHIDRA-CONTROL.md``. +""" +from __future__ import annotations + +import json +import socket +import threading + +from .errors import GreenEyeError + + +class GhidraBridge: + def __init__(self, session, host: str = "127.0.0.1", port: int = 0): + self.session = session + self.host = host + self.port = port + self._server: socket.socket | None = None + self._thread: threading.Thread | None = None + self._clients: list[threading.Thread] = [] + self._stop = threading.Event() + + # -- lifecycle --------------------------------------------------------- + + def start(self) -> int: + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind((self.host, self.port)) + server.listen(4) + self.port = server.getsockname()[1] + self._server = server + self._thread = threading.Thread(target=self._accept_loop, name="greeneye-ghidra", daemon=True) + self._thread.start() + return self.port + + def stop(self) -> None: + self._stop.set() + if self._server is not None: + try: + self._server.close() + except OSError: + pass + self._server = None + + def serve_forever(self) -> None: + if self._thread is None: + self.start() + try: + while not self._stop.is_set() and (self._thread is not None and self._thread.is_alive()): + self._thread.join(timeout=0.5) + except KeyboardInterrupt: + pass + finally: + self.stop() + + def _accept_loop(self) -> None: + assert self._server is not None + while not self._stop.is_set(): + try: + client, _addr = self._server.accept() + except OSError: + return + handler = threading.Thread(target=self._client_loop, args=(client,), daemon=True) + handler.start() + self._clients.append(handler) + + def _client_loop(self, client: socket.socket) -> None: + with client: + stream = client.makefile("rwb") + for line in stream: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError as exc: + response = {"id": None, "ok": False, "error": f"bad json: {exc}"} + else: + response = self._dispatch(request) + stream.write((json.dumps(response) + "\n").encode("utf-8")) + stream.flush() + if response.get("close"): + self.stop() + return + + # -- command dispatch -------------------------------------------------- + + def _dispatch(self, request: dict) -> dict: + request_id = request.get("id") + cmd = request.get("cmd", "") + args = request.get("args") or {} + try: + handler = getattr(self, f"cmd_{cmd}", None) + if handler is None: + raise GreenEyeError(f"unknown bridge command {cmd!r}") + result = handler(**args) + response = {"id": request_id, "ok": True, "result": result} + if cmd == "close": + response["close"] = True + return response + except Exception as exc: # never kill the bridge on a bad request + return {"id": request_id, "ok": False, "error": str(exc)} + + # -- commands ---------------------------------------------------------- + + def cmd_ping(self) -> str: + return "pong" + + def cmd_info(self) -> dict: + return { + "name": self.session.name, + "pc": self.session.pc, + "symbols": len(self.session.symbols), + "recording": getattr(self.session.recorder, "dir", None), + } + + def cmd_resolve(self, name: str) -> int: + return self.session.resolve(name) + + def cmd_bp_set(self, target, kind: str = "sw") -> int: + handle = self.session.hook(target, kind=kind, listener="ghidra") + return handle.addr + + def cmd_bp_clear(self, addr) -> bool: + addr = self.session.resolve(addr) if isinstance(addr, str) else int(addr) + for _key, bp in list(self.session._bps.items()): # noqa: SLF001 - bridge is a privileged client + if bp.addr == addr: + self.session.remove_breakpoint(bp) + return True + + def cmd_bp_list(self) -> list[dict]: + return [ + {"addr": bp.addr, "label": bp.label, "hits": bp.hits, "kind": bp.kind} + for bp in self.session._bps.values() # noqa: SLF001 + ] + + def cmd_watch_set(self, target, size: int = 8, kind: str = "w") -> int: + wp = self.session.watch(target, size=size, kind=kind, listener="ghidra") + return wp.addr + + def cmd_run_until(self, target, timeout: float = 5.0) -> dict: + hit = self.session.run_until(target, timeout=timeout) + return {"addr": hit.addr, "symbol": hit.symbol.name if hit.symbol else None, "kind": hit.kind} + + def cmd_run(self, duration: float = 1.0, max_stops: int | None = None) -> int: + return self.session.run(duration=duration, max_stops=max_stops) + + def cmd_step(self, count: int = 1) -> list[int]: + return [hit.addr for hit in self.session.step(count)] + + def cmd_read_mem(self, addr, size: int = 64) -> str: + real = self.session.resolve(addr) if isinstance(addr, str) else int(addr) + return self.session.memory.read(real, size).hex() + + def cmd_write_mem(self, addr, hex: str) -> bool: + real = self.session.resolve(addr) if isinstance(addr, str) else int(addr) + self.session.memory.write(real, bytes.fromhex(hex)) + return True + + def cmd_read_reg(self, name: str) -> int: + return self.session.client.read_reg(name) + + def cmd_write_reg(self, name: str, value) -> bool: + self.session.client.write_reg(name, int(value)) + return True + + def cmd_call(self, target, args: list | None = None, timeout: float = 5.0) -> int: + return self.session.call(target, tuple(args or ()), timeout=timeout) + + def cmd_snapshot(self, addr, size: int, label: str = "ghidra") -> str: + real = self.session.resolve(addr) if isinstance(addr, str) else int(addr) + self.session.memory.snapshot(real, size, name=label) + return label + + def cmd_note(self, text: str) -> bool: + self.session.note(text) + return True + + def cmd_hits(self) -> list[dict]: + from .record import load_events + + directory = getattr(self.session.recorder, "dir", None) + if not directory: + return [] + counts: dict[int, int] = {} + for event in load_events(directory): + if event.get("type") == "bp_hit": + addr = event.get("pc") + if isinstance(addr, int): + counts[addr] = counts.get(addr, 0) + 1 + return [ + { + "addr": addr, + "hits": hits, + "symbol": (self.session.symbols.resolve(addr).name if self.session.symbols.resolve(addr) else None), + } + for addr, hits in sorted(counts.items(), key=lambda item: -item[1]) + ] + + def cmd_export_hits(self, path: str) -> str: + from .ghidra import export_hits + + directory = getattr(self.session.recorder, "dir", None) + if not directory: + raise GreenEyeError("session is not recording; hits cannot be exported") + return export_hits(directory, path, symbols=self.session.symbols) + + def cmd_close(self) -> bool: + return True diff --git a/debug/risc-v-gsp-rm-vm/greeneye/inject.py b/debug/risc-v-gsp-rm-vm/greeneye/inject.py new file mode 100644 index 0000000..f9605fc --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/greeneye/inject.py @@ -0,0 +1,129 @@ +"""Frida-like service injection for green-eye. + +Loads a position-independent RV64 service blob into a live target, registers +its dispatch entry with the firmware (on the demo uproc: a handler-table slot +that commands 16..23 dispatch to), and invokes it either directly through +``Session.call`` (a debugger-side remote function call) or through the +firmware's own mailbox dispatcher. + +On a future GSP-RM simulator the same shape becomes "inject an RM engine and +register it with the simulated RM core"; see ``docs/12-INJECTION-SDK.md`` and +``GREEN-EYE-ROADMAP.md``. +""" +from __future__ import annotations + +from dataclasses import dataclass + +from .elf import load_elf +from .errors import GreenEyeError + +#: Symbol the injector calls in a service blob. +ENTRY_SYMBOL = "ge_demo_service_dispatch" + +#: Demo uproc control-block field offsets (see firmware/uproc_demo). +CTRL_HANDLERS = 0x150 +CTRL_MAILBOX = 0x20 +CTRL_ARG0 = 0x28 +CTRL_ARG1 = 0x30 +CTRL_ARG2 = 0x38 +CTRL_RESULT = 0x40 +INJECT_CMD_BASE = 16 + + +@dataclass +class InjectedService: + path: str + name: str + load_addr: int + entry_addr: int + dispatch_offset: int + slot: int + size: int + + def summary(self) -> str: + return ( + f"{self.name}: {self.size} bytes at {self.load_addr:#x}, " + f"entry {self.entry_addr:#x} (offset {self.dispatch_offset:#x}), slot {self.slot}" + ) + + +def load_service( + session, + service_elf: str, + *, + load_addr: int | None = None, + slot: int = 6, + entry_symbol: str = ENTRY_SYMBOL, +) -> InjectedService: + """Load a service blob and register it with the demo firmware.""" + image = load_elf(service_elf) + entry = None + for symbol in image.symbols: + if symbol.name == entry_symbol: + entry = symbol + break + if entry is None: + available = ", ".join(sorted(s.name for s in image.symbols if s.is_function)) or "<none>" + raise GreenEyeError(f"service {service_elf}: symbol {entry_symbol!r} not found (functions: {available})") + if not 0 <= slot <= 7: + raise GreenEyeError(f"slot must be 0..7 (got {slot})") + if load_addr is None: + base = getattr(session.target, "ram_base", 0x80000000) if session.target else 0x80000000 + size = getattr(session.target, "memory_mb", 256) * 1024 * 1024 if session.target else 256 * 1024 * 1024 + load_addr = base + 0x01000000 + if load_addr + image.max_vaddr + 0x1000 >= base + size: + load_addr = base + max(0x1000, size - image.max_vaddr - 0x2000) + if getattr(session.target, "memory_mb", 0): + ram_end = getattr(session.target, "ram_base", 0x80000000) + session.target.memory_mb * 1024 * 1024 + if load_addr + image.max_vaddr > ram_end: + raise GreenEyeError( + f"service does not fit: load {load_addr:#x} + {image.max_vaddr:#x} exceeds RAM end {ram_end:#x}" + ) + + total = 0 + for segment in image.load_segments: + data = image.segment_bytes(segment) + if data: + session.memory.write(load_addr + segment.p_vaddr, data) + total += len(data) + entry_addr = load_addr + entry.value + + ctrl = session.symbols.get("gsp_ctrl") + if ctrl is not None: + session.memory.write_u64(ctrl.addr + CTRL_HANDLERS + slot * 8, entry_addr) + + service = InjectedService( + path=service_elf, + name=service_elf.rsplit("/", 1)[-1], + load_addr=load_addr, + entry_addr=entry_addr, + dispatch_offset=entry.value, + slot=slot, + size=total, + ) + session.recorder.event( + "service_injected", + path=service.path, + name=service.name, + load_addr=load_addr, + entry_addr=entry_addr, + slot=slot, + size=total, + ) + return service + + +def invoke_remote(session, service: InjectedService, cmd: int, a0: int = 0, a1: int = 0, timeout: float = 3.0) -> int: + """Call the service directly from the debugger (Frida NativeFunction).""" + return session.call(service.entry_addr, (cmd, a0, a1), timeout=timeout) + + +def invoke_via_firmware(session, service: InjectedService, cmd: int, a0: int = 0, a1: int = 0, settle: float = 0.25) -> int: + """Invoke the service through the firmware's mailbox dispatcher.""" + ctrl = session.symbols.lookup("gsp_ctrl") + session.memory.write_u64(ctrl + CTRL_ARG0, cmd) + session.memory.write_u64(ctrl + CTRL_ARG1, a0) + session.memory.write_u64(ctrl + CTRL_ARG2, a1) + session.memory.write_u64(ctrl + CTRL_MAILBOX, INJECT_CMD_BASE + service.slot) + session.run(duration=settle) + return session.memory.read_u64(ctrl + CTRL_RESULT) diff --git a/debug/risc-v-gsp-rm-vm/greeneye/record.py b/debug/risc-v-gsp-rm-vm/greeneye/record.py index bf70e58..0b025b5 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/record.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/record.py @@ -40,7 +40,7 @@ class Recorder: self.meta = dict(meta or {}) self.meta.update({"format": FORMAT, "name": name, "created": time.time(), "greeneye": _version()}) self._events_path = os.path.join(self.dir, "events.jsonl") - self._fh = open(self._events_path, "w", encoding="utf-8") + self._fh = open(self._events_path, "w", encoding="utf-8", buffering=1) self._t0 = time.monotonic() self._counts: dict[str, int] = {} self._snapshots: list[dict] = [] 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, diff --git a/debug/risc-v-gsp-rm-vm/greeneye/target.py b/debug/risc-v-gsp-rm-vm/greeneye/target.py index efcb1d1..273f216 100644 --- a/debug/risc-v-gsp-rm-vm/greeneye/target.py +++ b/debug/risc-v-gsp-rm-vm/greeneye/target.py @@ -59,10 +59,12 @@ class QemuTarget: qemu_log: str | None = None, workdir: str | None = None, keep_alive: bool = False, + ram_base: int = 0x80000000, ): self.qemu = qemu or find_qemu() self.machine = machine self.memory_mb = memory_mb + self.ram_base = ram_base self.host = "127.0.0.1" self.port = gdb_port or free_port() self.bios = bios |