summaryrefslogtreecommitdiffstats
path: root/debug/risc-v-gsp-rm-vm/greeneye
diff options
context:
space:
mode:
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/greeneye')
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/__init__.py115
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/__main__.py3
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/analyze.py293
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/cli.py688
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/elf.py322
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/errors.py30
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/ghidra.py80
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/memory.py190
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/plugins.py121
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/probes.py166
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/record.py156
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/rsp.py583
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/session.py562
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/structs.py197
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/symbols.py313
-rw-r--r--debug/risc-v-gsp-rm-vm/greeneye/target.py223
16 files changed, 4042 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/__init__.py b/debug/risc-v-gsp-rm-vm/greeneye/__init__.py
new file mode 100644
index 0000000..8b1c1e5
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/__init__.py
@@ -0,0 +1,115 @@
+"""green-eye: Frida-like runtime instrumentation for GSP/uproc firmware in a
+RISC-V QEMU VM.
+
+green-eye talks to QEMU's gdbstub directly (no GDB dependency) and layers a
+scriptable instrumentation API on top: breakpoints, hooks, watchpoints,
+memory inspection, structure decoding, session recording and post-run analysis.
+
+Public API (stable-ish for v0.1):
+
+ import greeneye as ge
+ s = ge.launch(kernel="firmware/uproc_demo/build/uproc_demo.elf")
+ fn = s.symbols.lookup("rpc_sum")
+ h = s.hook(fn, on_enter=lambda hit: print(hit))
+ s.run(duration=2.0)
+ s.close()
+
+Everything is development tooling; see docs/00-OVERVIEW.md.
+"""
+from __future__ import annotations
+
+__version__ = "0.1.0"
+
+from .errors import ( # noqa: F401
+ GreenEyeError,
+ RspError,
+ RspTimeout,
+ RspProtocolError,
+ SymbolNotFound,
+ TargetError,
+ StructError,
+)
+from .rsp import RspClient, StopReply, Register # noqa: F401
+from .elf import FirmwareImage, Segment, ElfSymbol, load_elf # noqa: F401
+from .symbols import Symbol, SymbolDB # noqa: F401
+from .memory import Memory # noqa: F401
+from .structs import Field, Schema, decode, schema_from_json # noqa: F401
+from .record import Recorder, NullRecorder, load_events # noqa: F401
+from .session import Session, Hit, HookHandle, Breakpoint, Watchpoint # noqa: F401
+from .target import QemuTarget # noqa: F401
+
+__all__ = [
+ "attach",
+ "launch",
+ "Session",
+ "Hit",
+ "HookHandle",
+ "Breakpoint",
+ "Watchpoint",
+ "QemuTarget",
+ "RspClient",
+ "SymbolDB",
+ "Symbol",
+ "Memory",
+ "Recorder",
+]
+
+
+def attach(
+ host: str = "127.0.0.1",
+ port: int = 4444,
+ *,
+ timeout: float = 10.0,
+ record: str | None = None,
+ symbols: SymbolDB | None = None,
+ name: str = "attach",
+) -> Session:
+ """Attach to an already-running QEMU gdbstub and return a Session."""
+ client = RspClient(host, port, timeout=timeout)
+ client.connect()
+ client.sync()
+ return Session(client=client, symbols=symbols, record=record, name=name)
+
+
+def launch(
+ *,
+ kernel: str | None = None,
+ machine: str = "virt",
+ memory_mb: int = 256,
+ gdb_port: int = 0,
+ bios: str = "none",
+ extra_args: list[str] | None = None,
+ timeout: float = 10.0,
+ record: str | None = None,
+ symbols: SymbolDB | None = None,
+ name: str = "session",
+ load_symbols: bool = True,
+ keep_alive: bool = False,
+) -> Session:
+ """Launch QEMU (paused), attach, and return a ready Session.
+
+ The target starts paused at its reset vector; use Session.load_image() to
+ load an ELF (which also sets PC), or QemuTarget.run_kernel() style helper
+ launch with kernel= for the fast `-kernel` path.
+ """
+ target = QemuTarget(
+ machine=machine,
+ memory_mb=memory_mb,
+ gdb_port=gdb_port,
+ bios=bios,
+ extra_args=extra_args,
+ kernel=kernel,
+ )
+ if keep_alive:
+ target.keep_alive = True
+ target.start(paused=True)
+ client = RspClient(target.host, target.port, timeout=timeout)
+ client.connect()
+ client.sync()
+
+ symdb = symbols or SymbolDB()
+ sess = Session(client=client, target=target, symbols=symdb, record=record, name=name)
+ if kernel:
+ image = load_elf(kernel)
+ sess.load_image(image, load_symbols=load_symbols)
+ return sess
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/__main__.py b/debug/risc-v-gsp-rm-vm/greeneye/__main__.py
new file mode 100644
index 0000000..eb53e2f
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/__main__.py
@@ -0,0 +1,3 @@
+from .cli import main
+
+raise SystemExit(main())
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/analyze.py b/debug/risc-v-gsp-rm-vm/greeneye/analyze.py
new file mode 100644
index 0000000..07c24eb
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/analyze.py
@@ -0,0 +1,293 @@
+"""Post-run analysis: turn a session directory into a symbol-annotated report.
+
+The report is the "complete map" half of the deliverable: hits by function,
+watch hits, memory writes, coverage and snapshot inventory, ready for
+offline comparison between runs (bare-metal vs VFIO, old firmware vs new).
+"""
+from __future__ import annotations
+
+import os
+import re
+from collections import Counter, defaultdict
+
+from .record import load_events, load_meta
+
+_ACCESS_RE = re.compile(
+ r"Invalid (read|write) at addr (0x[0-9a-fA-F]+), size (\d+), region '([^']*)', reason: (\w+)"
+)
+_INTERRUPT_RE = re.compile(
+ r"riscv_cpu_do_interrupt: hart:(\d+), async:(\d+), cause:([0-9a-fA-F]+), "
+ r"epc:(0x[0-9a-fA-F]+), tval:(0x[0-9a-fA-F]+), desc=([\w_]+)"
+)
+
+
+def parse_qemu_faults(log_text: str) -> list[dict]:
+ """Parse QEMU ``-d guest_errors,unimp,int`` logs into fault records.
+
+ Each record pairs the most recent invalid access (if any) with the
+ exception it raised, which is exactly the "first divergence" report a
+ firmware probe needs.
+ """
+ faults: list[dict] = []
+ pending_access: dict | None = None
+ for line in log_text.splitlines():
+ access = _ACCESS_RE.search(line)
+ if access:
+ pending_access = {
+ "rw": access.group(1),
+ "access_addr": int(access.group(2), 16),
+ "size": int(access.group(3)),
+ "region": access.group(4),
+ "reason": access.group(5),
+ }
+ continue
+ interrupt = _INTERRUPT_RE.search(line)
+ if interrupt:
+ faults.append(
+ {
+ "hart": int(interrupt.group(1)),
+ "async": int(interrupt.group(2)),
+ "cause": int(interrupt.group(3), 16),
+ "epc": int(interrupt.group(4), 16),
+ "tval": int(interrupt.group(5), 16),
+ "desc": interrupt.group(6),
+ "access": pending_access,
+ }
+ )
+ pending_access = None
+ return faults
+
+
+_RISCV_CAUSES = {
+ 0: "instruction address misaligned",
+ 1: "instruction access fault",
+ 2: "illegal instruction",
+ 3: "breakpoint",
+ 4: "load address misaligned",
+ 5: "load access fault",
+ 6: "store address misaligned",
+ 7: "store access fault",
+}
+
+
+def describe_fault(fault: dict, bias: int = 0) -> str:
+ epc = fault["epc"]
+ fw_pc = epc - bias
+ cause = fault["cause"]
+ desc = _RISCV_CAUSES.get(cause & 0xFF, fault.get("desc", "exception"))
+ lines = [
+ f"first exception: {fault.get('desc', '?')} (cause {cause:#x}: {desc})",
+ f" epc: {epc:#x}" + (f" (firmware vaddr {fw_pc:#x})" if bias else ""),
+ f" tval: {fault.get('tval', 0):#x}",
+ ]
+ access = fault.get("access")
+ if access:
+ lines.insert(
+ 1,
+ f" access: {access['rw']} at {access['access_addr']:#x} "
+ f"({access['size']} bytes, reason={access['reason']})",
+ )
+ return "\n".join(lines)
+
+
+def render_summary(session_dir: str, symbols=None, max_rows: int = 40) -> str:
+ meta = load_meta(session_dir)
+ events = load_events(session_dir)
+ return render_summary_from(meta, events, symbols=symbols, max_rows=max_rows)
+
+
+def render_summary_from(meta: dict, events: list[dict], symbols=None, max_rows: int = 40) -> str:
+ lines: list[str] = []
+ lines.append(f"# green-eye session summary: {meta.get('name', '?')}")
+ lines.append("")
+ lines.append(f"* format: `{meta.get('format', '?')}` green-eye {meta.get('greeneye', '?')}")
+ lines.append(f"* directory: `{meta.get('dir', '.')}`")
+ lines.append(f"* events: {len(events)}")
+ counts = Counter(event.get("type", "?") for event in events)
+ lines.append("* event counts: " + ", ".join(f"{k}={v}" for k, v in sorted(counts.items())))
+ counts_meta = meta.get("counts")
+ if counts_meta:
+ lines.append("* recorder counts: " + ", ".join(f"{k}={v}" for k, v in sorted(counts_meta.items())))
+ lines.append("")
+
+ def annotate(addr):
+ if not isinstance(addr, int):
+ return ""
+ if symbols is not None:
+ return symbols.annotate(addr)
+ return f"{addr:#x}"
+
+ # Breakpoint hits
+ bp_hits = [e for e in events if e.get("type") == "bp_hit"]
+ if bp_hits:
+ by_site: Counter = Counter()
+ by_symbol: Counter = Counter()
+ for event in bp_hits:
+ key = event.get("symbol") or f"{event.get('pc', 0):#x}"
+ by_site[(event.get("pc"), key)] += 1
+ by_symbol[key] += 1
+ lines.append("## Breakpoint hits")
+ lines.append("")
+ lines.append("| hits | pc | symbol |")
+ lines.append("|---:|---|---|")
+ for (pc, key), count in by_site.most_common(max_rows):
+ lines.append(f"| {count} | `{pc:#x}` | `{key}` |")
+ lines.append("")
+
+ # Watchpoint hits
+ watch_hits = [e for e in events if e.get("type") == "watch_hit"]
+ if watch_hits:
+ by_addr: Counter = Counter()
+ for event in watch_hits:
+ by_addr[(event.get("addr"), event.get("kind"))] += 1
+ lines.append("## Watchpoint hits")
+ lines.append("")
+ lines.append("| hits | address | kind | symbol |")
+ lines.append("|---:|---|---|---|")
+ for (addr, kind), count in by_addr.most_common(max_rows):
+ sym = symbols.resolve(addr) if symbols else None
+ lines.append(f"| {count} | `{addr:#x}` | {kind} | `{sym.name if sym else ''}` |")
+ lines.append("")
+
+ # Memory writes
+ writes = [e for e in events if e.get("type") == "mem_write"]
+ if writes:
+ regions = Counter()
+ for event in writes:
+ regions[(event.get("addr", 0) >> 12) << 12] += 1
+ lines.append("## Memory writes")
+ lines.append("")
+ lines.append(f"* writes: {len(writes)}, distinct 4 KiB pages: {len(regions)}")
+ lines.append("")
+ lines.append("| page | writes | symbol |")
+ lines.append("|---|---:|---|")
+ for page, count in regions.most_common(max_rows):
+ sym = symbols.resolve(page) if symbols else None
+ lines.append(f"| `{page:#x}` | {count} | `{sym.name if sym else ''}` |")
+ lines.append("")
+
+ # Coverage
+ steps = [e for e in events if e.get("type") == "step"]
+ if steps:
+ pcs = {(e.get("pc")) for e in steps}
+ lines.append("## Stepped coverage")
+ lines.append("")
+ lines.append(f"* steps: {len(steps)}, unique PCs: {len(pcs)}")
+ by_sym = Counter()
+ for pc in pcs:
+ sym = symbols.resolve(pc) if symbols else None
+ by_sym[sym.name if sym else f"{pc:#x}"] += 1
+ lines.append("")
+ lines.append("| PCs | symbol |")
+ lines.append("|---:|---|")
+ for name, count in by_sym.most_common(max_rows):
+ lines.append(f"| {count} | `{name}` |")
+ lines.append("")
+
+ # Snapshots
+ snapshots = meta.get("snapshots") or [e for e in events if e.get("type") == "snapshot_saved"]
+ if snapshots:
+ lines.append("## Snapshots")
+ lines.append("")
+ lines.append("| label | address | size | path | sha256 |")
+ lines.append("|---|---|---:|---|---|")
+ for snap in snapshots[:max_rows]:
+ lines.append(
+ f"| {snap.get('label')} | `{snap.get('addr', 0):#x}` | {snap.get('size', 0)} | "
+ f"`{snap.get('path')}` | `{(snap.get('sha256') or '')[:16]}` |"
+ )
+ lines.append("")
+
+ # Timeline excerpt
+ notable = [e for e in events if e.get("type") in ("bp_hit", "leave", "watch_hit", "mem_write", "note", "exit")]
+ if notable:
+ lines.append("## Timeline (first hits)")
+ lines.append("")
+ for event in notable[:max_rows]:
+ t = event.get("t", 0)
+ kind = event.get("type")
+ if kind in ("bp_hit", "leave"):
+ site = event.get("symbol") or f"{event.get('pc', 0):#x}"
+ extra = ""
+ regs = event.get("regs") or {}
+ if "a0" in regs:
+ extra = f" a0={regs['a0']:#x}"
+ if kind == "leave" and "retval" in event:
+ extra += f" -> {event['retval']:#x}"
+ lines.append(f"* `{t:8.3f}s` {kind}: {site}{extra}")
+ elif kind == "watch_hit":
+ lines.append(f"* `{t:8.3f}s` watch: {annotate(event.get('addr'))} ({event.get('kind')})")
+ elif kind == "mem_write":
+ lines.append(
+ f"* `{t:8.3f}s` write: {annotate(event.get('addr'))} "
+ f"{event.get('size')}B {event.get('data', '')[:32]}"
+ )
+ elif kind == "note":
+ lines.append(f"* `{t:8.3f}s` note: {event.get('text')}")
+ elif kind == "exit":
+ lines.append(f"* `{t:8.3f}s` exit status={event.get('status')}")
+ lines.append("")
+
+ # Symbol-annotation table for automation
+ lines.append("## Annotated sites")
+ lines.append("")
+ sites = Counter()
+ for event in events:
+ for key in ("pc", "addr"):
+ value = event.get(key)
+ if isinstance(value, int) and value and event.get("type") in ("bp_hit", "watch_hit", "mem_write"):
+ sites[value] += 1
+ lines.append("| address | count | annotation |")
+ lines.append("|---|---:|---|")
+ for addr, count in sites.most_common(max_rows):
+ lines.append(f"| `{addr:#x}` | {count} | `{annotate(addr)}` |")
+ lines.append("")
+ return "\n".join(lines) + "\n"
+
+
+def compare_sessions(a_dir: str, b_dir: str, symbols=None, max_rows: int = 40) -> str:
+ """Diff two recorded sessions at the symbolic event level (A/B tooling)."""
+ a_meta, b_meta = load_meta(a_dir), load_meta(b_dir)
+ a_events, b_events = load_events(a_dir), load_events(b_dir)
+ lines = [f"# green-eye session comparison", "", f"* A: {a_meta.get('name')} ({a_dir})", f"* B: {b_meta.get('name')} ({b_dir})", ""]
+
+ def profile(events):
+ bp = Counter()
+ watch = Counter()
+ writes = Counter()
+ for event in events:
+ if event.get("type") == "bp_hit":
+ bp[event.get("symbol") or f"{event.get('pc', 0):#x}"] += 1
+ elif event.get("type") == "watch_hit":
+ watch[(event.get("addr"), event.get("kind"))] += 1
+ elif event.get("type") == "mem_write":
+ writes[(event.get("addr", 0) >> 12) << 12] += 1
+ return bp, watch, writes
+
+ a_bp, a_watch, a_writes = profile(a_events)
+ b_bp, b_watch, b_writes = profile(b_events)
+
+ lines.append("## Breakpoint profile delta (A vs B)")
+ lines.append("")
+ lines.append("| symbol | A hits | B hits | delta |")
+ lines.append("|---|---:|---:|---:|")
+ for name in sorted(set(a_bp) | set(b_bp), key=lambda k: -(a_bp.get(k, 0) + b_bp.get(k, 0)))[:max_rows]:
+ lines.append(f"| `{name}` | {a_bp.get(name, 0)} | {b_bp.get(name, 0)} | {b_bp.get(name, 0) - a_bp.get(name, 0):+d} |")
+ lines.append("")
+
+ lines.append("## Watchpoint delta")
+ lines.append("")
+ lines.append("| address | kind | A | B |")
+ lines.append("|---|---|---:|---:|")
+ for key in sorted(set(a_watch) | set(b_watch), key=lambda k: -(a_watch.get(k, 0) + b_watch.get(k, 0)))[:max_rows]:
+ lines.append(f"| `{key[0]:#x}` | {key[1]} | {a_watch.get(key, 0)} | {b_watch.get(key, 0)} |")
+ lines.append("")
+
+ lines.append("## Written pages delta")
+ lines.append("")
+ lines.append("| page | A writes | B writes |")
+ lines.append("|---|---:|---:|")
+ for page in sorted(set(a_writes) | set(b_writes), key=lambda k: -(a_writes.get(k, 0) + b_writes.get(k, 0)))[:max_rows]:
+ lines.append(f"| `{page:#x}` | {a_writes.get(page, 0)} | {b_writes.get(page, 0)} |")
+ lines.append("")
+ return "\n".join(lines) + "\n"
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/cli.py b/debug/risc-v-gsp-rm-vm/greeneye/cli.py
new file mode 100644
index 0000000..5bfcee2
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/cli.py
@@ -0,0 +1,688 @@
+"""green-eye command line interface.
+
+Commands:
+ info environment/tooling check
+ demo boot the bundled uproc demo and run a canned instrumentation pass
+ run run a Python instrumentation script against a target
+ attach interactive REPL against a QEMU gdbstub
+ probe run an arbitrary firmware (default: the real GSP fwimage) for N
+ instructions and report the first divergence
+ record boot a firmware and record a memory-activity session
+ analyze render a recorded session to a symbol-annotated report
+ compare A/B two recorded sessions at the symbolic level
+ symbols build/inspect a green-eye symbol database from Ghidra exports
+ plugins list discoverable plugins
+"""
+from __future__ import annotations
+
+import argparse
+import cmd
+import os
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+import greeneye as ge
+from . import analyze as _analyze
+from . import ghidra as _ghidra
+from .elf import load_elf
+from .errors import GreenEyeError
+from .plugins import PluginManager
+from .structs import schema_from_json
+from .symbols import SymbolDB
+from .target import QemuTarget
+
+TOOL_ROOT = Path(__file__).resolve().parent.parent
+DUMPS = TOOL_ROOT / "dumps"
+DEMO_ELF = TOOL_ROOT / "firmware" / "uproc_demo" / "build" / "uproc_demo.elf"
+DEMO_SCHEMA = TOOL_ROOT / "firmware" / "uproc_demo" / "gsp_ctrl.schema.json"
+DEFAULT_FWIMAGE = TOOL_ROOT.parents[1] / "firmware" / "57X-GSP" / "570.144" / "gsp-570.144.fwimage.bin"
+
+RPC_HANDLERS = ["rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault"]
+
+
+# ---------------------------------------------------------------------------
+# helpers
+# ---------------------------------------------------------------------------
+
+
+def _ensure_dumps_env() -> None:
+ os.environ.setdefault("GREENEYE_DUMPS", str(DUMPS))
+
+
+def _build_demo(quiet: bool = False) -> Path:
+ import subprocess
+
+ fw_dir = TOOL_ROOT / "firmware" / "uproc_demo"
+ result = subprocess.run(
+ ["make", "-C", str(fw_dir)],
+ capture_output=quiet,
+ text=True,
+ )
+ if result.returncode != 0:
+ output = (result.stdout or "") + (result.stderr or "")
+ raise GreenEyeError(f"demo firmware build failed:\n{output}")
+ if not DEMO_ELF.exists():
+ raise GreenEyeError(f"demo build produced no {DEMO_ELF}")
+ return DEMO_ELF
+
+
+def _load_schema(path: Path):
+ import json
+
+ with open(path, "r", encoding="utf-8") as fh:
+ return schema_from_json(json.load(fh))
+
+
+def _start_session(
+ *,
+ kernel: str | None,
+ port: int | None,
+ record,
+ name: str,
+ plugins: list[str] | None = None,
+):
+ if port:
+ db = SymbolDB()
+ if kernel:
+ db.load_elf(kernel)
+ session = ge.attach(port=port, record=record, symbols=db, name=name)
+ else:
+ session = ge.launch(kernel=kernel, record=record, name=name)
+ manager = PluginManager(session)
+ manager.load_all(plugins)
+ return session, manager
+
+
+def _print_session_dir(session) -> None:
+ directory = getattr(session.recorder, "dir", None)
+ if directory:
+ print(f"[green-eye] session: {directory}")
+
+
+# ---------------------------------------------------------------------------
+# commands
+# ---------------------------------------------------------------------------
+
+
+def cmd_info(args) -> int:
+ from .target import DEFAULT_QEMU
+
+ print(f"green-eye {ge.__version__}")
+ print(f"tool root: {TOOL_ROOT}")
+ qemu = os.environ.get("GREENEYE_QEMU", DEFAULT_QEMU)
+ import shutil
+
+ resolved = shutil.which(qemu) if os.path.sep not in qemu else (qemu if os.path.exists(qemu) else None)
+ print(f"qemu: {resolved or 'NOT FOUND (install qemu-system-misc)'}")
+ if resolved:
+ import subprocess
+
+ version = subprocess.run([resolved, "--version"], capture_output=True, text=True).stdout.strip()
+ print(f"qemu version: {version}")
+ for label, path in [
+ ("demo firmware", DEMO_ELF),
+ ("demo schema", DEMO_SCHEMA),
+ ("default fwimage", DEFAULT_FWIMAGE),
+ ]:
+ print(f"{label}: {path} ({'present' if Path(path).exists() else 'missing'})")
+ cuda_names = Path(TOOL_ROOT).parents[1] / "userspace" / "cuda" / "570.144" / "annotations" / "names.tsv"
+ print(f"CUDA Ghidra names.tsv: {cuda_names} ({'present' if cuda_names.exists() else 'missing'})")
+ plugins_dir = TOOL_ROOT / "plugins"
+ plugins = sorted(p.name for p in plugins_dir.glob("*.py") if not p.name.startswith("_"))
+ print(f"bundled plugins: {', '.join(plugins) or 'none'}")
+ return 0
+
+
+def cmd_demo(args) -> int:
+ _ensure_dumps_env()
+ elf = args.kernel or (_build_demo() if not args.no_build else DEMO_ELF)
+ if not Path(elf).exists():
+ raise GreenEyeError(f"demo firmware not built: {elf}")
+ schema = _load_schema(DEMO_SCHEMA)
+ record = True if args.record is None else args.record
+ session, _manager = _start_session(kernel=str(elf), port=None, record=record, name="demo")
+ hits: list[str] = []
+ try:
+ ctrl = session.symbols.lookup("gsp_ctrl")
+ for name in RPC_HANDLERS:
+ sym = session.symbols.get(name)
+ if sym is not None:
+ session.hook(
+ sym.addr,
+ on_enter=lambda hit: hits.append(f"enter {hit.label()} a0={hit.arg(0):#x}"),
+ on_leave=lambda hit: hits.append(f"leave {hit.label()} -> {hit.retval:#x}"),
+ )
+ session.watch(ctrl + 16, size=8, kind="w", listener="rpc_count")
+ print(f"[demo] ctrl={ctrl:#x} schema={schema.name} size={schema.size:#x}")
+ print("[demo] running the firmware self-test ...")
+ session.run(duration=args.duration)
+ banner = session.target.read_uart() if session.target else ""
+ print("[demo] uart:", banner.strip().splitlines()[0] if banner.strip() else "<empty>")
+ before = session.snapshot(ctrl, schema.size, "ctrl-after-selftest")
+
+ print("[demo] host stimulus: PING")
+ session.memory.write_u64(ctrl + 32, 1)
+ session.memory.write_u64(ctrl + 40, 0x1234)
+ session.run(duration=0.5)
+ after = session.snapshot(ctrl, schema.size, "ctrl-after-ping")
+
+ values = schema.decode(after, base=ctrl)
+ print("[demo] control block after stimulus:")
+ from .structs import render
+
+ print(render(schema, values))
+ diffs = ge.Memory.diff(before, after, base=ctrl)
+ if diffs:
+ print("[demo] snapshot diff (self-test -> ping):")
+ print(ge.Memory.render_diff(diffs, session.symbols, max_lines=16))
+
+ print("[demo] host stimulus: FAULT (ebreak) to demonstrate trap capture")
+ session.memory.write_u64(ctrl + 32, 6)
+ session.run(duration=1.0, max_stops=1)
+ pc = session.pc
+ sym = session.symbols.resolve(pc)
+ print(f"[demo] stopped at {pc:#x} {sym.name if sym else '<unknown>'} (ebreak)")
+ session.step() # step past the trap
+ print("[demo] stepped past the trap; hits:")
+ for line in hits[:24]:
+ print(" ", line)
+ print(f"[demo] total hook lines: {len(hits)}")
+ _print_session_dir(session)
+ finally:
+ session.close()
+ return 0
+
+
+def cmd_run(args) -> int:
+ _ensure_dumps_env()
+ script = Path(args.script)
+ if not script.exists():
+ raise GreenEyeError(f"script not found: {script}")
+ kernel = args.kernel
+ if not kernel and not args.port:
+ kernel = str(_build_demo())
+ session, manager = _start_session(
+ kernel=kernel, port=args.port, record=args.record or True, name=script.stem, plugins=args.plugin
+ )
+ namespace = {
+ "ge": ge,
+ "session": session,
+ "s": session,
+ "memory": session.memory,
+ "symbols": session.symbols,
+ "recorder": session.recorder,
+ "greeneye": ge,
+ }
+ try:
+ source = script.read_text(encoding="utf-8")
+ exec(compile(source, str(script), "exec"), namespace)
+ return 0
+ except SystemExit as exc:
+ return int(exc.code or 0)
+ finally:
+ session.close()
+
+
+# -- interactive REPL -------------------------------------------------------
+
+
+class GreenEyeRepl(cmd.Cmd):
+ intro = "green-eye interactive instrumentation. Type 'help' for commands."
+ prompt = "greeneye> "
+
+ def __init__(self, session, manager):
+ super().__init__()
+ self.session = session
+ self.manager = manager
+ self._listener_hooks = []
+
+ # utility
+ def _print_hit(self, hit):
+ print(" " + str(hit))
+
+ def _say(self, text):
+ print(text)
+
+ # commands
+ def do_regs(self, arg):
+ """regs -- dump all registers."""
+ regs = self.session.read_regs()
+ for name, value in regs.items():
+ print(f" {name:<5} = {value:#018x}")
+
+ def do_reg(self, arg):
+ """reg <name> [value] -- read or write one register."""
+ parts = arg.split()
+ if not parts:
+ return self._say("usage: reg <name> [value]")
+ if len(parts) == 1:
+ print(f" {parts[0]} = {self.session.client.read_reg(parts[0]):#x}")
+ else:
+ self.session.client.write_reg(parts[0], int(parts[1], 0))
+ self._say("ok")
+
+ def do_x(self, arg):
+ """x <addr> [len] -- hexdump memory (annotated)."""
+ parts = arg.split()
+ if not parts:
+ return self._say("usage: x <addr> [len]")
+ addr = self.session.resolve(parts[0])
+ length = int(parts[1], 0) if len(parts) > 1 else 64
+ print(self.session.memory.hexdump(addr, length))
+
+ def do_mem(self, arg):
+ """mem <addr> <len> -- raw memory read."""
+ addr, length = arg.split()
+ print(self.session.memory.read(self.session.resolve(addr), int(length, 0)).hex())
+
+ def do_w(self, arg):
+ """w <addr> <hexbytes> -- write memory."""
+ addr, data = arg.split()
+ self.session.memory.write(self.session.resolve(addr), bytes.fromhex(data))
+ self._say("ok")
+
+ def do_wstr(self, arg):
+ """wstr <addr> <text> -- write a string plus NUL."""
+ addr, _, text = arg.partition(" ")
+ self.session.memory.write(self.session.resolve(addr), text.encode() + b"\x00")
+ self._say("ok")
+
+ def do_bp(self, arg):
+ """bp <addr|symbol> [sw|hw] -- set a breakpoint that prints on hit."""
+ parts = arg.split()
+ if not parts:
+ return self._say("usage: bp <addr|symbol> [sw|hw]")
+ kind = parts[1] if len(parts) > 1 else "sw"
+ handle = self.session.hook(parts[0], on_enter=self._print_hit, kind=kind, listener="repl")
+ self._listener_hooks.append(handle)
+ self._say(f"breakpoint @ {handle.addr:#x} ({handle.label})")
+
+ def do_hook(self, arg):
+ """hook <addr|symbol> -- alias for bp with enter/leave printing."""
+ parts = arg.split()
+ if not parts:
+ return self._say("usage: hook <addr|symbol>")
+ handle = self.session.hook(
+ parts[0], on_enter=self._print_hit, on_leave=self._print_hit, listener="repl"
+ )
+ self._listener_hooks.append(handle)
+ self._say(f"hooked {handle.label}")
+
+ def do_watch(self, arg):
+ """watch <addr|symbol> [size] [w|r|a] -- set a watchpoint."""
+ parts = arg.split()
+ if not parts:
+ return self._say("usage: watch <addr|symbol> [size] [w|r|a]")
+ size = int(parts[1], 0) if len(parts) > 1 else 8
+ kind = parts[2] if len(parts) > 2 else "w"
+ wp = self.session.watch(parts[0], size=size, kind=kind, callback=self._print_hit, listener="repl")
+ self._say(f"watch {kind}@{wp.addr:#x} size={size}")
+
+ def do_s(self, arg):
+ """s [n] -- step n instructions (default 1)."""
+ count = int(arg) if arg.strip() else 1
+ for hit in self.session.step(count):
+ self._print_hit(hit)
+
+ def do_c(self, arg):
+ """c [seconds] -- continue and service events for N seconds (default 1)."""
+ seconds = float(arg) if arg.strip() else 1.0
+ stops = self.session.run(duration=seconds)
+ self._say(f"resumed for {seconds}s, serviced {stops} stops")
+
+ def do_events(self, arg):
+ """events [n] -- show the last n recorded events (default 20)."""
+ from .record import load_events
+
+ directory = getattr(self.session.recorder, "dir", None)
+ if not directory:
+ return self._say("this session is not recording")
+ events = load_events(directory)
+ for event in events[-int(arg or 20) :]:
+ print(" " + str(event))
+
+ def do_snap(self, arg):
+ """snap <label> <addr> <len> -- snapshot a memory region into the session."""
+ parts = arg.split()
+ if len(parts) < 3:
+ return self._say("usage: snap <label> <addr> <len>")
+ data = self.session.memory.snapshot(self.session.resolve(parts[1]), int(parts[2], 0), name=parts[0])
+ self._say(f"snapshot {parts[0]}: {len(data)} bytes")
+
+ def do_sym(self, arg):
+ """sym <addr|name> -- resolve a symbol or an address."""
+ target = arg.strip()
+ if not target:
+ return self._say("usage: sym <addr|name>")
+ try:
+ print(" " + self.session.symbols.annotate(self.session.resolve(target)))
+ except GreenEyeError as exc:
+ self._say(f"error: {exc}")
+
+ def do_note(self, arg):
+ """note <text> -- add a note to the session recording."""
+ self.session.note(arg)
+ self._say("noted")
+
+ def do_quit(self, arg):
+ """quit -- close the session and exit."""
+ return True
+
+ do_EOF = do_quit
+
+ def onecmd(self, line):
+ name = line.split()[0] if line.split() else ""
+ if name in self.manager.api.commands:
+ fn, _help = self.manager.api.commands[name]
+ try:
+ fn(self.session, line.split()[1:])
+ except Exception as exc:
+ self._say(f"plugin error: {exc}")
+ return False
+ if name and name not in self.get_names() and not hasattr(self, f"do_{name}"):
+ return super().onecmd(line)
+ return super().onecmd(line)
+
+ def get_names(self):
+ names = super().get_names()
+ names += [f"do_{cmd}" for cmd in self.manager.api.commands]
+ return names
+
+ def default(self, line):
+ self._say(f"unknown command: {line.split()[0] if line.split() else line!r}")
+
+
+def cmd_attach(args) -> int:
+ _ensure_dumps_env()
+ kernel = args.kernel
+ if kernel:
+ kernel = str(Path(kernel))
+ elif not args.port:
+ kernel = str(_build_demo())
+ session, manager = _start_session(
+ kernel=kernel, port=args.port, record=args.record, name="repl", plugins=args.plugin
+ )
+ try:
+ GreenEyeRepl(session, manager).cmdloop()
+ finally:
+ session.close()
+ return 0
+
+
+def cmd_launch(args) -> int:
+ """Start a target, print the gdbstub endpoint, and keep QEMU alive."""
+ _ensure_dumps_env()
+ kernel = args.kernel or str(_build_demo())
+ if not Path(kernel).exists():
+ raise GreenEyeError(f"kernel not found: {kernel}")
+ session = ge.launch(kernel=str(kernel), record=args.record, name="launch", keep_alive=True)
+ port = session.target.port
+ print(f"[green-eye] target running: {kernel}")
+ print(f"[green-eye] gdbstub: 127.0.0.1:{port}")
+ print(f"[green-eye] attach: ./greeneye.sh attach --port {port} --kernel {kernel}")
+ print("[green-eye] Ctrl-C stops waiting; QEMU keeps running for attach.")
+ try:
+ while True:
+ time.sleep(1)
+ except KeyboardInterrupt:
+ print("[green-eye] detached (QEMU still running)")
+ return 0
+
+
+def cmd_record(args) -> int:
+ _ensure_dumps_env()
+ elf = args.kernel or str(_build_demo())
+ session, _manager = _start_session(kernel=str(elf), port=None, record=args.record or True, name=args.name)
+ try:
+ for spec in args.watch or []:
+ addr_or_sym, size = (spec.split(":", 1) + ["8"])[:2]
+ session.watch(addr_or_sym, size=int(size, 0))
+ for spec in args.snapshot or []:
+ addr_or_sym, size = spec.split(":", 1)
+ session.snapshot(session.resolve(addr_or_sym), int(size, 0), name=addr_or_sym)
+ session.run(duration=args.duration)
+ print(f"[record] serviced run for {args.duration}s")
+ _print_session_dir(session)
+ finally:
+ session.close()
+ return 0
+
+
+def cmd_probe(args) -> int:
+ _ensure_dumps_env()
+ elf = Path(args.elf)
+ if not elf.exists():
+ raise GreenEyeError(f"firmware not found: {elf}")
+ image = load_elf(str(elf))
+ print(f"[probe] {image.summary()}")
+ relocate = not args.no_relocate
+ bias = (0x80000000 - image.min_vaddr) if relocate else 0
+ kernel_path = str(elf)
+ if relocate:
+ from .elf import write_flat_kernel
+
+ out = Path(tempfile.mkdtemp(prefix="greeneye-probe-")) / (elf.name + ".flat.bin")
+ kernel_path, entry_addr = write_flat_kernel(str(elf), str(out), load_addr=0x80000000)
+ size = Path(kernel_path).stat().st_size
+ print(
+ f"[probe] flat kernel {kernel_path} ({size} bytes); "
+ f"load 0x80000000 entry {entry_addr:#x} (original {image.entry:#x})"
+ )
+ target = QemuTarget(
+ kernel=kernel_path,
+ dlog="guest_errors,unimp,int",
+ memory_mb=max(256, (image.max_vaddr - image.min_vaddr) // (1024 * 1024) + 64),
+ )
+ target.start(paused=True)
+ from .rsp import RspClient
+
+ client = RspClient(target.host, target.port)
+ client.connect()
+ client.sync()
+ symbols = SymbolDB()
+ if args.symbols:
+ symbols.load_json(args.symbols)
+ session = ge.Session(
+ client=client,
+ target=target,
+ symbols=symbols,
+ record=args.record or True,
+ name=f"probe-{elf.stem}",
+ meta={"firmware": str(elf), "entry": image.entry, "bias": bias, "kernel": kernel_path},
+ )
+ session.load_image(image, bias=bias)
+ try:
+ if args.instructions:
+ print(f"[probe] stepping {args.instructions} instructions ...")
+ session.step(args.instructions)
+ else:
+ print(f"[probe] running for {args.duration}s ...")
+ session.run(duration=args.duration)
+ pc = session.pc
+ sym = session.symbols.resolve(pc)
+ print(f"[probe] stopped at {pc:#x} {sym.name if sym else '<unknown>'}")
+ from .analyze import describe_fault, parse_qemu_faults
+
+ qlog = target.read_qemu_log()
+ faults = parse_qemu_faults(qlog)
+ if faults:
+ print("[probe] " + describe_fault(faults[0], bias=bias))
+ print(f"[probe] total exceptions: {len(faults)}")
+ else:
+ print("[probe] no exception recorded (execution stayed in mapped memory)")
+ tail = qlog.strip().splitlines()
+ if tail:
+ print("[probe] qemu log tail:")
+ for line in tail[-6:]:
+ print(" ", line)
+ _print_session_dir(session)
+ finally:
+ session.close()
+ return 0
+
+
+def cmd_analyze(args) -> int:
+ symbols = None
+ if args.symbols:
+ symbols = SymbolDB()
+ symbols.load_json(args.symbols)
+ report = _analyze.render_summary(args.session, symbols=symbols)
+ if args.out:
+ Path(args.out).write_text(report, encoding="utf-8")
+ print(f"wrote {args.out}")
+ else:
+ print(report)
+ return 0
+
+
+def cmd_compare(args) -> int:
+ symbols = SymbolDB() if args.symbols else None
+ report = _analyze.compare_sessions(args.a, args.b, symbols=symbols)
+ print(report)
+ return 0
+
+
+def cmd_symbols(args) -> int:
+ if args.symbols_command == "export":
+ db = _ghidra.build_symbol_db(
+ names_tsv=args.names,
+ index_tsv=args.index,
+ exports_tsv=args.exports,
+ callgraph_tsv=args.callgraph,
+ elf=args.elf,
+ bias=args.bias,
+ )
+ db.save_json(args.out, meta={"bias": args.bias})
+ print(f"wrote {args.out}: {len(db)} symbols")
+ return 0
+ if args.symbols_command == "resolve":
+ db = SymbolDB()
+ if args.db:
+ db.load_json(args.db)
+ print(db.annotate(int(args.addr, 0)))
+ return 0
+ return 1
+
+
+def cmd_plugins(args) -> int:
+ plugins_dir = TOOL_ROOT / "plugins"
+ print("bundled plugins:")
+ for path in sorted(plugins_dir.glob("*.py")):
+ if path.name.startswith("_"):
+ continue
+ print(f" {path.name}")
+ env = os.environ.get("GREENEYE_PLUGINS")
+ if env:
+ print(f"GREENEYE_PLUGINS: {env}")
+ return 0
+
+
+# ---------------------------------------------------------------------------
+# parser
+# ---------------------------------------------------------------------------
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(prog="greeneye", description=__doc__.splitlines()[0])
+ parser.add_argument("--version", action="version", version=f"green-eye {ge.__version__}")
+ sub = parser.add_subparsers(dest="command", required=True)
+
+ sub.add_parser("info", help="environment/tooling check").set_defaults(fn=cmd_info)
+
+ demo = sub.add_parser("demo", help="boot the demo uproc and instrument it")
+ demo.add_argument("--kernel", help="firmware ELF (default: built demo)")
+ demo.add_argument("--no-build", action="store_true", help="do not rebuild the demo")
+ demo.add_argument("--record", nargs="?", const=True, default=None, help="record to a path (or default dumps/)")
+ demo.add_argument("--duration", type=float, default=1.0)
+ demo.set_defaults(fn=cmd_demo)
+
+ run = sub.add_parser("run", help="run a Python instrumentation script")
+ run.add_argument("script")
+ run.add_argument("--kernel")
+ run.add_argument("--port", type=int)
+ run.add_argument("--record", nargs="?", const=True, default=None)
+ run.add_argument("--plugin", action="append", default=[])
+ run.set_defaults(fn=cmd_run)
+
+ attach = sub.add_parser("attach", help="interactive REPL over a gdbstub (launches if no --port)")
+ attach.add_argument("--port", type=int, help="attach to an existing gdbstub instead of launching")
+ attach.add_argument("--kernel", help="firmware ELF (default: built demo)")
+ attach.add_argument("--record", nargs="?", const=True, default=None)
+ attach.add_argument("--plugin", action="append", default=[])
+ attach.set_defaults(fn=cmd_attach)
+
+ launch = sub.add_parser("launch", help="start a target and keep it alive for attach")
+ launch.add_argument("--kernel", help="firmware ELF (default: built demo)")
+ launch.add_argument("--record", nargs="?", const=True, default=None)
+ launch.set_defaults(fn=cmd_launch)
+
+ probe = sub.add_parser("probe", help="run a firmware image and report first divergence")
+ probe.add_argument("elf", nargs="?", default=str(DEFAULT_FWIMAGE))
+ probe.add_argument("--no-relocate", action="store_true")
+ probe.add_argument("--instructions", type=int, default=0)
+ probe.add_argument("--duration", type=float, default=1.0)
+ probe.add_argument("--symbols", help="green-eye symbol DB JSON to annotate the stop")
+ probe.add_argument("--record", nargs="?", const=True, default=None)
+ probe.set_defaults(fn=cmd_probe)
+
+ record = sub.add_parser("record", help="boot and record a memory-activity session")
+ record.add_argument("--kernel")
+ record.add_argument("--name", default="record")
+ record.add_argument("--duration", type=float, default=2.0)
+ record.add_argument("--watch", action="append", help="addr|symbol[:size]")
+ record.add_argument("--snapshot", action="append", help="addr|symbol:len")
+ record.add_argument("--record", nargs="?", const=True, default=None)
+ record.set_defaults(fn=cmd_record)
+
+ analyze = sub.add_parser("analyze", help="render a recorded session report")
+ analyze.add_argument("session")
+ analyze.add_argument("--symbols")
+ analyze.add_argument("--out")
+ analyze.set_defaults(fn=cmd_analyze)
+
+ compare = sub.add_parser("compare", help="A/B two recorded sessions")
+ compare.add_argument("a")
+ compare.add_argument("b")
+ compare.add_argument("--symbols", action="store_true")
+ compare.set_defaults(fn=cmd_compare)
+
+ symbols = sub.add_parser("symbols", help="build/inspect a symbol database")
+ symbols_sub = symbols.add_subparsers(dest="symbols_command", required=True)
+ exp = symbols_sub.add_parser("export")
+ exp.add_argument("--names")
+ exp.add_argument("--index")
+ exp.add_argument("--exports")
+ exp.add_argument("--callgraph")
+ exp.add_argument("--elf")
+ exp.add_argument("--bias", type=lambda v: int(v, 0), default=0)
+ exp.add_argument("-o", "--out", required=True)
+ exp.set_defaults(fn=cmd_symbols)
+ res = symbols_sub.add_parser("resolve")
+ res.add_argument("addr")
+ res.add_argument("--db")
+ res.set_defaults(fn=cmd_symbols)
+
+ plugins = sub.add_parser("plugins", help="list discoverable plugins")
+ plugins.set_defaults(fn=cmd_plugins)
+
+ return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+ _ensure_dumps_env()
+ parser = build_parser()
+ args = parser.parse_args(argv)
+ try:
+ return args.fn(args)
+ except GreenEyeError as exc:
+ print(f"green-eye error: {exc}", file=sys.stderr)
+ return 2
+ except KeyboardInterrupt:
+ print("interrupted", file=sys.stderr)
+ return 130
+
+
+if __name__ == "__main__": # pragma: no cover
+ raise SystemExit(main())
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/elf.py b/debug/risc-v-gsp-rm-vm/greeneye/elf.py
new file mode 100644
index 0000000..abfccd9
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/elf.py
@@ -0,0 +1,322 @@
+"""Minimal ELF64/ELF32 reader (stdlib only).
+
+Needed for three things:
+ * symbol resolution for the demo uproc firmware (breakpoints by name),
+ * program-header extraction so the real GSP ``.fwimage`` can be placed into
+ QEMU RAM (and relocated when a probe needs it),
+ * entry-point detection for the first-fault firmware probe.
+
+This is not a full ELF library; it parses headers, PT_LOAD segments and
+``.symtab`` symbols, which is all green-eye needs.
+"""
+from __future__ import annotations
+
+import struct
+from dataclasses import dataclass, field as _field
+
+from .errors import GreenEyeError
+
+EM_RISCV = 243
+PT_LOAD = 1
+SHT_SYMTAB = 2
+SHT_STRTAB = 3
+STT_FUNC = 2
+STT_OBJECT = 1
+STT_NOTYPE = 0
+
+
+class ElfError(GreenEyeError):
+ """Malformed or unsupported ELF file."""
+
+
+@dataclass
+class Segment:
+ p_type: int
+ p_flags: int
+ p_offset: int
+ p_vaddr: int
+ p_paddr: int
+ p_filesz: int
+ p_memsz: int
+ p_align: int
+
+ @property
+ def is_load(self) -> bool:
+ return self.p_type == PT_LOAD
+
+ @property
+ def writable(self) -> bool:
+ return bool(self.p_flags & 0x2)
+
+ @property
+ def executable(self) -> bool:
+ return bool(self.p_flags & 0x1)
+
+
+@dataclass
+class ElfSymbol:
+ name: str
+ value: int
+ size: int
+ bind: int
+ type: int
+
+ @property
+ def is_function(self) -> bool:
+ return self.type == STT_FUNC
+
+ @property
+ def is_object(self) -> bool:
+ return self.type == STT_OBJECT
+
+
+@dataclass
+class FirmwareImage:
+ path: str
+ is64: bool
+ e_machine: int
+ entry: int
+ segments: list[Segment] = _field(default_factory=list)
+ symbols: list[ElfSymbol] = _field(default_factory=list)
+ _blob: bytes | None = None
+
+ @property
+ def is_riscv(self) -> bool:
+ return self.e_machine == EM_RISCV
+
+ @property
+ def load_segments(self) -> list[Segment]:
+ return [s for s in self.segments if s.is_load]
+
+ @property
+ def min_vaddr(self) -> int:
+ loads = self.load_segments
+ return min((s.p_vaddr for s in loads), default=0)
+
+ @property
+ def max_vaddr(self) -> int:
+ loads = self.load_segments
+ return max((s.p_vaddr + s.p_memsz for s in loads), default=0)
+
+ def read(self) -> bytes:
+ if self._blob is None:
+ with open(self.path, "rb") as fh:
+ self._blob = fh.read()
+ return self._blob
+
+ def segment_bytes(self, seg: Segment) -> bytes:
+ return self.read()[seg.p_offset : seg.p_offset + seg.p_filesz]
+
+ def flat_load_plan(self, bias: int = 0) -> list[tuple[int, bytes]]:
+ """Return ``(load_addr, data)`` pairs for every PT_LOAD segment.
+
+ ``bias`` is added to each segment's virtual address (use it to place a
+ firmware linked for GPU physical addresses into QEMU RAM).
+ """
+ plan: list[tuple[int, bytes]] = []
+ for seg in self.load_segments:
+ data = bytearray(self.segment_bytes(seg))
+ if len(data) < seg.p_memsz:
+ data += b"\x00" * (seg.p_memsz - len(data))
+ plan.append((seg.p_vaddr + bias, bytes(data)))
+ return plan
+
+ def relocated_entry(self, bias: int = 0) -> int:
+ return self.entry + bias
+
+ def summary(self) -> str:
+ return (
+ f"{self.path}: ELF{'64' if self.is64 else '32'} "
+ f"machine={self.e_machine} entry={self.entry:#x} "
+ f"segments={len(self.segments)} loads={len(self.load_segments)} "
+ f"symbols={len(self.symbols)} "
+ f"vaddr=[{self.min_vaddr:#x},{self.max_vaddr:#x})"
+ )
+
+
+def load_elf(path: str) -> FirmwareImage:
+ with open(path, "rb") as fh:
+ data = fh.read()
+ return parse_elf(data, path=path)
+
+
+def parse_elf(data: bytes, path: str = "<memory>") -> FirmwareImage:
+ if len(data) < 52 or data[:4] != b"\x7fELF":
+ raise ElfError(f"{path}: not an ELF file")
+ ei_class = data[4]
+ if ei_class == 2:
+ is64 = True
+ hdr_fmt = "<HHIQQQIHHHHHH"
+ elif ei_class == 1:
+ is64 = False
+ hdr_fmt = "<HHIIIIIHHHHHH"
+ else:
+ raise ElfError(f"{path}: unknown EI_CLASS {ei_class}")
+ hdr_size = struct.calcsize(hdr_fmt)
+ fields = struct.unpack_from(hdr_fmt, data, 16)
+ (
+ _e_type,
+ e_machine,
+ _e_version,
+ e_entry,
+ e_phoff,
+ e_shoff,
+ _e_flags,
+ _e_ehsize,
+ e_phentsize,
+ e_phnum,
+ e_shentsize,
+ e_shnum,
+ e_shstrndx,
+ ) = fields
+
+ image = FirmwareImage(path=path, is64=is64, e_machine=e_machine, entry=e_entry)
+
+ if is64:
+ ph_fmt = "<IIQQQQQQ"
+ sh_fmt = "<IIQQQQIIQQ"
+ sym_fmt = "<IBBHQQ"
+ else:
+ ph_fmt = "<IIIIIIII"
+ sh_fmt = "<IIIIIIIIII"
+ sym_fmt = "<IIIBBH"
+
+ ph_size = struct.calcsize(ph_fmt)
+ for i in range(e_phnum):
+ base = e_phoff + i * e_phentsize
+ if base + ph_size > len(data):
+ break
+ p = struct.unpack_from(ph_fmt, data, base)
+ image.segments.append(Segment(*p[:8]))
+
+ # Sections -> symbols
+ sh_size = struct.calcsize(sh_fmt)
+ sections: list[tuple[int, int, int, int]] = [] # type, offset, size, link
+ for i in range(e_shnum):
+ base = e_shoff + i * e_shentsize
+ if base + sh_size > len(data):
+ break
+ sh = struct.unpack_from(sh_fmt, data, base)
+ sh_type, sh_offset, sh_size_val, sh_link = sh[1], sh[4], sh[5], sh[6]
+ sections.append((sh_type, sh_offset, sh_size_val, sh_link))
+
+ for sh_type, sh_offset, sh_size_val, sh_link in sections:
+ if sh_type != SHT_SYMTAB:
+ continue
+ strtab_off, strtab_size = 0, 0
+ if 0 <= sh_link < len(sections):
+ _, strtab_off, strtab_size, _ = sections[sh_link]
+ strtab = data[strtab_off : strtab_off + strtab_size]
+ count = sh_size_val // struct.calcsize(sym_fmt)
+ for i in range(count):
+ base = sh_offset + i * struct.calcsize(sym_fmt)
+ if base + struct.calcsize(sym_fmt) > len(data):
+ break
+ st = struct.unpack_from(sym_fmt, data, base)
+ if is64:
+ st_name, st_info, _st_other, _st_shndx, st_value, st_size_val = st
+ else:
+ st_name, st_value, st_size_val, st_info, _st_other, _st_shndx = st
+ if st_name == 0 or st_value == 0:
+ continue
+ name = _read_cstr(strtab, st_name)
+ if not name:
+ continue
+ image.symbols.append(
+ ElfSymbol(
+ name=name,
+ value=st_value,
+ size=st_size_val,
+ bind=(st_info >> 4) & 0xF,
+ type=st_info & 0xF,
+ )
+ )
+ return image
+
+
+def _read_cstr(blob: bytes, offset: int) -> str:
+ if offset >= len(blob):
+ return ""
+ end = blob.find(b"\x00", offset)
+ if end == -1:
+ end = len(blob)
+ return blob[offset:end].decode("utf-8", "replace")
+
+
+def write_relocated_elf(path: str, out_path: str, bias: int) -> str:
+ """Write a copy of an ELF64 whose PT_LOADs and entry are shifted by bias.
+
+ ``p_paddr`` is forced to ``p_vaddr + bias`` so the layout matches the
+ virtual addresses. Useful for formats/tools that honour e_entry.
+ """
+ with open(path, "rb") as fh:
+ data = bytearray(fh.read())
+ if len(data) < 64 or data[:4] != b"\x7fELF" or data[4] != 2:
+ raise ElfError(f"{path}: relocation requires a 64-bit ELF")
+ e_entry = struct.unpack_from("<Q", data, 24)[0]
+ e_phoff = struct.unpack_from("<Q", data, 32)[0]
+ e_phentsize = struct.unpack_from("<H", data, 54)[0]
+ e_phnum = struct.unpack_from("<H", data, 56)[0]
+ struct.pack_into("<Q", data, 24, e_entry + bias)
+ for i in range(e_phnum):
+ base = e_phoff + i * e_phentsize
+ if base + 56 > len(data):
+ break
+ p_type = struct.unpack_from("<I", data, base)[0]
+ if p_type != PT_LOAD:
+ continue
+ p_vaddr = struct.unpack_from("<Q", data, base + 16)[0]
+ struct.pack_into("<Q", data, base + 16, p_vaddr + bias)
+ struct.pack_into("<Q", data, base + 24, p_vaddr + bias)
+ with open(out_path, "wb") as fh:
+ fh.write(data)
+ return out_path
+
+
+def _trampoline(entry_offset: int) -> bytes:
+ """RV64 code at offset 0 that jumps to ``PC + entry_offset``.
+
+ QEMU's RISC-V ``-kernel`` raw/ELF path starts the CPU at the RAM base for
+ images whose e_entry it does not honour; this 8-byte stub (auipc t0, hi;
+ jalr x0, t0, lo) makes the firmware entry reachable from the base without
+ patching the image itself.
+ """
+ if not -0x80000000 <= entry_offset < 0x80000000:
+ raise ElfError("entry offset out of trampoline range")
+ hi20 = ((entry_offset + 0x800) >> 12) & 0xFFFFF
+ lo12 = entry_offset & 0xFFF
+ auipc = (hi20 << 12) | (5 << 7) | 0x17 # auipc t0, hi20
+ jalr = (lo12 << 20) | (5 << 15) | 0x67 # jalr x0, t0, lo12
+ return struct.pack("<II", auipc, jalr)
+
+
+def write_flat_kernel(
+ path: str,
+ out_path: str,
+ *,
+ load_addr: int = 0x80000000,
+ trampoline: bool = True,
+) -> tuple[str, int]:
+ """Flatten an ELF's PT_LOADs into a raw image QEMU can boot as `-kernel`.
+
+ Returns ``(out_path, entry_addr)`` where ``entry_addr = load_addr +
+ (e_entry - min_vaddr)``. When the entry is not at offset 0 a tiny
+ PC-relative trampoline is written at offset 0 (QEMU starts at the load
+ base for raw kernels). ``load_addr`` is not encoded anywhere except in
+ the returned entry address: all code is position-independent relative to
+ the image base, exactly as it executes on the device.
+ """
+ image = load_elf(path)
+ span = image.max_vaddr - image.min_vaddr
+ flat = bytearray(span)
+ for seg in image.load_segments:
+ start = seg.p_vaddr - image.min_vaddr
+ end = start + seg.p_filesz
+ flat[start:end] = image.segment_bytes(seg)
+ entry_offset = image.entry - image.min_vaddr
+ if trampoline and entry_offset != 0:
+ stub = _trampoline(entry_offset)
+ flat[0 : len(stub)] = stub
+ with open(out_path, "wb") as fh:
+ fh.write(flat)
+ return out_path, load_addr + entry_offset
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/errors.py b/debug/risc-v-gsp-rm-vm/greeneye/errors.py
new file mode 100644
index 0000000..fab5874
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/errors.py
@@ -0,0 +1,30 @@
+"""green-eye error types. Keep them few and meaningful."""
+from __future__ import annotations
+
+
+class GreenEyeError(Exception):
+ """Base class for all green-eye errors."""
+
+
+class RspError(GreenEyeError):
+ """Base class for gdbstub RSP transport errors."""
+
+
+class RspTimeout(RspError):
+ """The stub did not answer within the timeout."""
+
+
+class RspProtocolError(RspError):
+ """The stub answered with a malformed packet or an error reply."""
+
+
+class TargetError(GreenEyeError):
+ """QEMU target lifecycle/configuration error."""
+
+
+class SymbolNotFound(GreenEyeError):
+ """A requested symbol is not in the symbol database."""
+
+
+class StructError(GreenEyeError):
+ """Structure schema/decode error."""
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py b/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py
new file mode 100644
index 0000000..6986d74
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/ghidra.py
@@ -0,0 +1,80 @@
+"""Ghidra bridge: symbols in, hit annotations out.
+
+Two flows:
+
+*out of Ghidra*: export function names/addresses/sizes to a green-eye JSON
+symbol database (from ``names.tsv``, ``decompiled_index.tsv``, exports or ELF);
+*hits*: turn a recorded session into ``greeneye_hits.tsv``
+(``addr<TAB>hits<TAB>symbol``) that ``MarkGreenEyeHits.java`` can annotate in
+the Ghidra project.
+
+This is the static/dynamic join: Ghidra owns the address space and
+decompilation, green-eye owns what actually executed.
+"""
+from __future__ import annotations
+
+import os
+from collections import Counter
+
+from .record import load_events
+from .symbols import SymbolDB
+
+
+def build_symbol_db(
+ *,
+ names_tsv: str | None = None,
+ index_tsv: str | None = None,
+ exports_tsv: str | None = None,
+ callgraph_tsv: str | None = None,
+ elf: str | None = None,
+ bias: int = 0,
+) -> SymbolDB:
+ db = SymbolDB()
+ if elf:
+ db.load_elf(elf, bias=0) # ELF symbol values are already runtime addresses
+ if names_tsv:
+ db.load_names_tsv(names_tsv, bias=bias)
+ if index_tsv:
+ db.load_index_tsv(index_tsv, bias=bias)
+ if exports_tsv:
+ db.load_index_tsv(exports_tsv, bias=bias)
+ if callgraph_tsv:
+ db.load_callgraph(callgraph_tsv)
+ return db
+
+
+def export_hits(session_dir: str, out_tsv: str, symbols: SymbolDB | None = None) -> str:
+ events = load_events(session_dir)
+ hits: Counter = Counter()
+ for event in events:
+ if event.get("type") == "bp_hit":
+ hits[event.get("pc")] += 1
+ elif event.get("type") == "watch_hit":
+ hits[event.get("addr")] += 1
+ with open(out_tsv, "w", encoding="utf-8") as fh:
+ fh.write("addr\thits\tsymbol\tannotation\n")
+ for addr, count in sorted(hits.items(), key=lambda kv: -kv[1]):
+ if not isinstance(addr, int):
+ continue
+ sym = symbols.resolve(addr) if symbols else None
+ name = sym.name if sym else ""
+ annotation = symbols.annotate(addr) if symbols else f"{addr:#x}"
+ fh.write(f"{addr:#x}\t{count}\t{name}\t{annotation}\n")
+ return os.path.abspath(out_tsv)
+
+
+def corpus_location(symbol, corpus_root: str) -> str | None:
+ """Return ``<bucket>:<line>`` for a symbol with Ghidra index extras."""
+ bucket = symbol.extras.get("bucket")
+ line = symbol.extras.get("line")
+ if not bucket:
+ return None
+ return os.path.join(corpus_root, bucket) + (f":{line}" if line else "")
+
+
+def render_ghidra_command(project_dir: str, project_name: str, program: str, hits_tsv: str) -> str:
+ return (
+ "analyzeHeadless {proj} {name} -process {program} -noanalysis "
+ "-scriptPath <open-iov>/debug/risc-v-gsp-rm-vm/ghidra "
+ "-postScript MarkGreenEyeHits.java {hits}"
+ ).format(proj=project_dir, name=project_name, program=program, hits=hits_tsv)
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/memory.py b/debug/risc-v-gsp-rm-vm/greeneye/memory.py
new file mode 100644
index 0000000..23d9ca0
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/memory.py
@@ -0,0 +1,190 @@
+"""Memory inspection helpers: read/write, hexdump, search, snapshots, diffs."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .errors import GreenEyeError
+
+
+@dataclass
+class Diff:
+ addr: int
+ old: bytes
+ new: bytes
+
+ @property
+ def size(self) -> int:
+ return len(self.old)
+
+
+class Memory:
+ """A convenience facade over an RSP client (and optional symbol DB)."""
+
+ def __init__(self, client, symbols=None, recorder=None):
+ self.client = client
+ self.symbols = symbols
+ self.recorder = recorder
+
+ # -- basic access ------------------------------------------------------
+
+ def read(self, addr: int, size: int) -> bytes:
+ data = self.client.read_mem(addr, size)
+ if self.recorder is not None:
+ self.recorder.event("mem_read", addr=addr, size=size)
+ return data
+
+ def write(self, addr: int, data: bytes | int | str) -> None:
+ if isinstance(data, int):
+ data = bytes([data & 0xFF])
+ elif isinstance(data, str):
+ data = data.encode("utf-8")
+ self.client.write_mem(addr, data)
+ if self.recorder is not None:
+ self.recorder.event("mem_write", addr=addr, size=len(data), data=data.hex())
+
+ def read_u8(self, addr: int) -> int:
+ return self.read(addr, 1)[0]
+
+ def read_u16(self, addr: int) -> int:
+ return int.from_bytes(self.read(addr, 2), "little")
+
+ def read_u32(self, addr: int) -> int:
+ return int.from_bytes(self.read(addr, 4), "little")
+
+ def read_u64(self, addr: int) -> int:
+ return int.from_bytes(self.read(addr, 8), "little")
+
+ def write_u32(self, addr: int, value: int) -> None:
+ self.write(addr, value.to_bytes(4, "little"))
+
+ def write_u64(self, addr: int, value: int) -> None:
+ self.write(addr, value.to_bytes(8, "little"))
+
+ def read_cstring(self, addr: int, max_len: int = 256) -> str:
+ return self.read(addr, max_len).split(b"\x00", 1)[0].decode("utf-8", "replace")
+
+ def read_ptr(self, addr: int, width: int = 8) -> int:
+ return int.from_bytes(self.read(addr, width), "little")
+
+ # -- display -----------------------------------------------------------
+
+ def hexdump(self, addr: int, size: int, width: int = 16, annotate: bool = True) -> str:
+ data = self.read(addr, size)
+ lines: list[str] = []
+ if annotate and self.symbols is not None:
+ sym = self.symbols.resolve(addr)
+ if sym is not None:
+ lines.append(f"; {sym.label(addr)} (source={sym.source or '?'})")
+ for offset in range(0, len(data), width):
+ chunk = data[offset : offset + width]
+ hex_part = " ".join(f"{b:02x}" for b in chunk)
+ ascii_part = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
+ lines.append(f"{addr + offset:08x} {hex_part:<{width * 3 - 1}} |{ascii_part}|")
+ return "\n".join(lines)
+
+ def hexdump_annotated(
+ self,
+ addr: int,
+ size: int,
+ struct_schema=None,
+ width: int = 16,
+ ) -> str:
+ """Hexdump plus a decoded structure overlay when a schema is given."""
+ lines = [self.hexdump(addr, size, width=width, annotate=True)]
+ if struct_schema is not None:
+ from .structs import decode, render
+
+ values = decode(struct_schema, self.read(addr, struct_schema.size), base=addr)
+ lines.append("")
+ lines.append(render(struct_schema, values))
+ return "\n".join(lines)
+
+ # -- search ------------------------------------------------------------
+
+ def search(
+ self,
+ addr: int,
+ size: int,
+ pattern: bytes | int,
+ *,
+ mask: bytes | None = None,
+ max_hits: int = 64,
+ chunk: int = 0x10000,
+ ) -> list[int]:
+ if isinstance(pattern, int):
+ pattern = bytes([pattern & 0xFF])
+ if not pattern:
+ raise GreenEyeError("empty search pattern")
+ if mask is not None and len(mask) != len(pattern):
+ raise GreenEyeError("mask length must match pattern length")
+ hits: list[int] = []
+ offset = 0
+ overlap = len(pattern) - 1
+ carry = b""
+ while offset < size and len(hits) < max_hits:
+ want = min(chunk, size - offset)
+ blob = carry + self.read(addr + offset - len(carry), want + len(carry)) if carry else self.read(addr + offset, want)
+ base = addr + offset - len(carry)
+ for index in range(0, len(blob) - len(pattern) + 1):
+ candidate = blob[index : index + len(pattern)]
+ if mask is None:
+ if candidate == pattern:
+ hits.append(base + index)
+ else:
+ if all((candidate[i] & mask[i]) == (pattern[i] & mask[i]) for i in range(len(pattern))):
+ hits.append(base + index)
+ if len(hits) >= max_hits:
+ break
+ carry = blob[-overlap:] if overlap else b""
+ offset += want
+ return hits
+
+ def search_u32(self, addr: int, size: int, value: int, **kw) -> list[int]:
+ return self.search(addr, size, value.to_bytes(4, "little"), **kw)
+
+ def search_string(self, addr: int, size: int, text: str, **kw) -> list[int]:
+ return self.search(addr, size, text.encode("utf-8"), **kw)
+
+ # -- capture / diff ----------------------------------------------------
+
+ def snapshot(self, addr: int, size: int, name: str = "region") -> bytes:
+ data = self.client.read_mem(addr, size)
+ if self.recorder is not None:
+ path = self.recorder.snapshot(name, addr, data)
+ self.recorder.event("snapshot", label=name, addr=addr, size=size, path=path)
+ return data
+
+ @staticmethod
+ def diff(old: bytes, new: bytes, base: int = 0) -> list[Diff]:
+ length = min(len(old), len(new))
+ changed = [i for i in range(length) if old[i] != new[i]]
+ if not changed:
+ return []
+ # coalesce adjacent bytes into runs
+ runs: list[Diff] = []
+ start = changed[0]
+ prev = changed[0]
+ for index in changed[1:]:
+ if index == prev + 1:
+ prev = index
+ continue
+ runs.append(Diff(base + start, old[start : prev + 1], new[start : prev + 1]))
+ start = prev = index
+ runs.append(Diff(base + start, old[start : prev + 1], new[start : prev + 1]))
+ return runs
+
+ @staticmethod
+ def render_diff(diffs: list[Diff], symbols=None, max_lines: int = 64) -> str:
+ lines: list[str] = []
+ for diff in diffs[:max_lines]:
+ old = " ".join(f"{b:02x}" for b in diff.old[:8])
+ new = " ".join(f"{b:02x}" for b in diff.new[:8])
+ label = ""
+ if symbols is not None:
+ sym = symbols.resolve(diff.addr)
+ if sym is not None:
+ label = f" ; {sym.label(diff.addr)}"
+ lines.append(f"{diff.addr:08x} {old:<24} -> {new:<24}{label}")
+ if len(diffs) > max_lines:
+ lines.append(f"... {len(diffs) - max_lines} more changed runs")
+ return "\n".join(lines)
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/plugins.py b/debug/risc-v-gsp-rm-vm/greeneye/plugins.py
new file mode 100644
index 0000000..af4545f
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/plugins.py
@@ -0,0 +1,121 @@
+"""green-eye plugin system.
+
+A plugin is a normal Python module that exposes::
+
+ def register(ge):
+ ge.log("my plugin loaded")
+ ge.add_command("mycmd", run_mycmd, help="…")
+ ge.hook("rpc_sum", on_enter=lambda hit: ge.log(f"sum({hit.arg(0)})"))
+
+Plugins can extend green-eye with:
+
+* new CLI commands (including an MCP server later - roadmap);
+* runtime RE tools (extra watchpoint strategies, structure probes, coverage);
+* virtualisation-environment updates (e.g. record/replay harnesses);
+* AI-harness integration and reporting.
+
+Discovery order:
+
+1. explicit ``--plugin PATH`` arguments;
+2. colon-separated paths in ``GREENEYE_PLUGINS``;
+3. ``<tool>/plugins/*.py`` (the bundled examples).
+"""
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+from typing import Callable
+
+
+class GreenEyeAPI:
+ """Handle passed to ``register(ge)``."""
+
+ def __init__(self, session, cli=None):
+ self.session = session
+ self.cli = cli
+ self.commands: dict[str, tuple[Callable, str]] = {}
+ self.loaded: list[str] = []
+
+ # -- convenience --------------------------------------------------------
+
+ @property
+ def memory(self):
+ return self.session.memory
+
+ @property
+ def symbols(self):
+ return self.session.symbols
+
+ @property
+ def recorder(self):
+ return self.session.recorder
+
+ def log(self, message: str, **fields) -> None:
+ self.session.recorder.event("plugin", message=message, **fields)
+
+ def hook(self, target, on_enter=None, on_leave=None, **kw):
+ return self.session.hook(target, on_enter=on_enter, on_leave=on_leave, **kw)
+
+ def watch(self, target, size: int = 8, kind: str = "w", callback=None, **kw):
+ return self.session.watch(target, size=size, kind=kind, callback=callback, **kw)
+
+ def on_event(self, kind: str, callback) -> None:
+ self.session.on_event(kind, callback)
+
+ def resolve(self, target):
+ return self.session.resolve(target)
+
+ def add_command(self, name: str, fn: Callable, help: str = "") -> None:
+ self.commands[name] = (fn, help)
+
+ def run_command(self, name: str, args: list[str]) -> int:
+ fn, _help = self.commands[name]
+ return int(fn(self.session, args) or 0)
+
+
+class PluginManager:
+ def __init__(self, session, cli=None):
+ self.session = session
+ self.api = GreenEyeAPI(session, cli=cli)
+ self.loaded: list[str] = []
+
+ def default_paths(self) -> list[str]:
+ here = os.path.dirname(os.path.abspath(__file__))
+ bundled = os.path.join(os.path.dirname(here), "plugins")
+ paths = []
+ env = os.environ.get("GREENEYE_PLUGINS", "")
+ paths += [p for p in env.split(os.pathsep) if p]
+ if os.path.isdir(bundled):
+ paths.append(bundled)
+ return paths
+
+ def load_all(self, explicit: list[str] | None = None) -> list[str]:
+ for path in (explicit or []) + self.default_paths():
+ if os.path.isdir(path):
+ for entry in sorted(os.listdir(path)):
+ if entry.endswith(".py") and not entry.startswith("_"):
+ self._load_file(os.path.join(path, entry))
+ elif path.endswith(".py") and os.path.isfile(path):
+ self._load_file(path)
+ return self.loaded
+
+ def _load_file(self, path: str) -> None:
+ name = os.path.splitext(os.path.basename(path))[0]
+ mod_name = f"greeneye_plugin_{name}"
+ try:
+ spec = importlib.util.spec_from_file_location(mod_name, path)
+ if spec is None or spec.loader is None:
+ return
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[mod_name] = module
+ spec.loader.exec_module(module)
+ register = getattr(module, "register", None)
+ if register is None:
+ self.session.recorder.event("plugin_error", plugin=name, error="no register(ge)")
+ return
+ register(self.api)
+ self.loaded.append(name)
+ self.session.recorder.event("plugin_loaded", plugin=name)
+ except Exception as exc: # plugins must never break the session
+ self.session.recorder.event("plugin_error", plugin=name, error=repr(exc))
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/probes.py b/debug/risc-v-gsp-rm-vm/greeneye/probes.py
new file mode 100644
index 0000000..5837bab
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/probes.py
@@ -0,0 +1,166 @@
+"""Declarative probe specifications (the Frida-script-like layer).
+
+A probe spec is JSON so it can be generated by another tool (or an AI
+harness) and replayed unchanged:
+
+ {
+ "probes": [
+ {
+ "name": "sum-entry",
+ "symbol": "rpc_sum",
+ "when": "a1 > 0",
+ "on_enter": {
+ "log": ["a0", "a1", "a2"],
+ "dump": [{"reg": "a0", "len": 32, "label": "sum-arg"}],
+ "note": "sum called"
+ },
+ "on_leave": {"log": ["retval"]}
+ }
+ ]
+ }
+
+Conditions are evaluated over a register dictionary (``a0``..``a7``, ``pc``,
+``ra``, or raw ``x10`` names). The evaluator is deliberately tiny: it only
+accepts comparisons/boolean arithmetic over those names and integer
+literals, never arbitrary Python.
+"""
+from __future__ import annotations
+
+import json
+import re
+
+from .errors import GreenEyeError
+
+_ALLOWED_CHARS = re.compile(r"^[\s0-9a-fA-FxX<>=!()+\-*/%&|^~]*$")
+_NAME_RE = re.compile(r"\b[A-Za-z_][A-Za-z0-9_]*\b")
+_ALLOWED_NAMES = re.compile(
+ r"^(retval|pc|ra|sp|gp|tp|fp|and|or|not|True|False|None"
+ r"|[a-z][0-9]?|x[0-9]{1,2}|[stga][0-9]{1,2})$"
+)
+
+
+def load_spec(path: str) -> dict:
+ with open(path, "r", encoding="utf-8") as fh:
+ spec = json.load(fh)
+ if "probes" not in spec or not isinstance(spec["probes"], list):
+ raise GreenEyeError("probe spec must be an object with a 'probes' list")
+ return spec
+
+
+def compile_condition(expr: str | None):
+ if not expr:
+ return None
+ for name in _NAME_RE.findall(expr):
+ if not _ALLOWED_NAMES.match(name):
+ raise GreenEyeError(f"probe condition references disallowed name {name!r}")
+ if not _ALLOWED_CHARS.match(_NAME_RE.sub("0", expr)):
+ raise GreenEyeError(f"probe condition contains disallowed characters: {expr!r}")
+ code = compile(expr, "<probe-when>", "eval")
+
+ def check(regs: dict) -> bool:
+ scope = {
+ **{k: v for k, v in regs.items() if isinstance(v, int)},
+ "retval": regs.get("a0", 0),
+ }
+ try:
+ return bool(eval(code, {"__builtins__": {}}, scope)) # noqa: S307 - whitelisted above
+ except Exception:
+ return False
+
+ return check
+
+
+def install(session, spec: dict) -> list:
+ """Install every probe in the spec; returns the HookHandles."""
+ handles = []
+ for entry in spec["probes"]:
+ target = entry.get("symbol") or entry.get("addr")
+ if target is None:
+ raise GreenEyeError(f"probe {entry.get('name', '?')}: needs 'symbol' or 'addr'")
+ when = compile_condition(entry.get("when"))
+ name = entry.get("name", str(target))
+ on_enter_actions = entry.get("on_enter")
+ on_leave_actions = entry.get("on_leave")
+
+ def make_enter(name=name, when=when, actions=on_enter_actions):
+ def callback(hit):
+ if when is not None and not when(hit.regs):
+ return
+ _run_actions(hit, name, actions)
+
+ return callback
+
+ def make_leave(name=name, actions=on_leave_actions):
+ def callback(hit):
+ _run_actions(hit, name, actions, is_leave=True)
+
+ return callback
+
+ handle = session.hook(
+ target,
+ on_enter=make_enter() if on_enter_actions else None,
+ on_leave=make_leave() if on_leave_actions else None,
+ listener=f"probe:{name}",
+ )
+ session.note(f"probe installed: {name}", target=str(target))
+ handles.append(handle)
+ return handles
+
+
+def _run_actions(hit, probe_name: str, actions: dict | None, is_leave: bool = False) -> None:
+ session = hit.session
+ if actions is None:
+ return
+ values = {}
+ for expr in actions.get("log", []):
+ values[expr] = _read_value(hit, expr)
+ if values:
+ session.recorder.event(
+ "probe_log",
+ probe=probe_name,
+ phase="leave" if is_leave else "enter",
+ addr=hit.addr,
+ symbol=hit.symbol.name if hit.symbol else None,
+ values=values,
+ )
+ for dump in actions.get("dump", []):
+ addr = _read_value(hit, dump.get("reg", "a0"))
+ if isinstance(addr, str):
+ continue
+ length = int(dump.get("len", 64))
+ label = dump.get("label", f"{probe_name}-dump")
+ try:
+ data = session.memory.read(addr & 0xFFFFFFFFFFFFFFFF, length)
+ except Exception as exc:
+ session.recorder.event("probe_dump_error", probe=probe_name, addr=addr, error=str(exc))
+ continue
+ path = session.recorder.blob(f"{label}-{addr:#x}.bin", data)
+ session.recorder.event(
+ "probe_dump",
+ probe=probe_name,
+ name=label,
+ addr=addr,
+ size=len(data),
+ path=path,
+ preview=data[:32].hex(),
+ )
+ for snapshot in actions.get("snapshot", []):
+ addr = _read_value(hit, snapshot.get("reg", "a0"))
+ if isinstance(addr, str):
+ continue
+ session.snapshot(addr & 0xFFFFFFFFFFFFFFFF, int(snapshot.get("len", 4096)), name=snapshot.get("label", probe_name))
+ if actions.get("note"):
+ session.note(actions["note"], probe=probe_name, addr=hit.addr)
+
+
+def _read_value(hit, expr: str):
+ if expr == "retval":
+ return hit.retval if hit.retval is not None else hit.regs.get("a0", 0)
+ if expr in hit.regs:
+ return hit.regs[expr]
+ if expr.startswith("0x"):
+ return int(expr, 16)
+ try:
+ return int(expr, 0)
+ except ValueError:
+ return f"<unknown:{expr}>"
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/record.py b/debug/risc-v-gsp-rm-vm/greeneye/record.py
new file mode 100644
index 0000000..bf70e58
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/record.py
@@ -0,0 +1,156 @@
+"""Session recording: events.jsonl + snapshots + summary.
+
+A recorded session is the durable artifact green-eye produces: a complete,
+symbol-annotatable map of what the firmware did (hook hits, watch hits,
+memory writes, snapshots, coverage steps) that can be replayed offline,
+diffed between runs (bare-metal vs VFIO-mdev, old GSP vs new GSP) and
+annotated back into Ghidra.
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+import zlib
+
+FORMAT = "greeneye-session/1"
+
+
+class NullRecorder:
+ dir: str | None = None
+
+ def event(self, event_type: str, **fields) -> None:
+ pass
+
+ def snapshot(self, label: str, addr: int, data: bytes) -> None:
+ pass
+
+ def blob(self, name: str, data: bytes) -> None:
+ pass
+
+ def close(self) -> None:
+ pass
+
+
+class Recorder:
+ def __init__(self, root: str, *, name: str = "session", meta: dict | None = None):
+ self.dir = root
+ os.makedirs(self.dir, exist_ok=True)
+ self.name = name
+ 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._t0 = time.monotonic()
+ self._counts: dict[str, int] = {}
+ self._snapshots: list[dict] = []
+ self._closed = False
+
+ @classmethod
+ def new(cls, *, base: str | None = None, name: str = "session", meta: dict | None = None) -> "Recorder":
+ base = base or os.environ.get("GREENEYE_DUMPS") or os.path.join(os.getcwd(), "dumps")
+ stamp = time.strftime("%Y%m%d-%H%M%S")
+ return cls(os.path.join(base, f"{stamp}-{name}"), name=name, meta=meta)
+
+ # -- writing -----------------------------------------------------------
+
+ def event(self, event_type: str, **fields) -> None:
+ if self._closed:
+ return
+ record = {"t": round(time.monotonic() - self._t0, 6), "type": event_type}
+ for key, value in fields.items():
+ record[key] = _jsonable(value)
+ self._fh.write(json.dumps(record, sort_keys=True) + "\n")
+ self._counts[event_type] = self._counts.get(event_type, 0) + 1
+
+ def snapshot(self, label: str, addr: int, data: bytes) -> str:
+ rel = os.path.join("snapshots", f"{label}-{addr:#x}-{len(data)}B.bin")
+ path = os.path.join(self.dir, rel)
+ os.makedirs(os.path.dirname(path), exist_ok=True)
+ with open(path, "wb") as fh:
+ fh.write(zlib.compress(data))
+ info = {"label": label, "addr": addr, "size": len(data), "path": rel, "sha256": _sha(data)}
+ self._snapshots.append(info)
+ self.event("snapshot_saved", **info)
+ return rel
+
+ def blob(self, name: str, data: bytes) -> str:
+ rel = os.path.join("blobs", name)
+ path = os.path.join(self.dir, rel)
+ os.makedirs(os.path.dirname(path), exist_ok=True)
+ with open(path, "wb") as fh:
+ fh.write(data)
+ self.event("blob_saved", path=rel, size=len(data))
+ return rel
+
+ # -- closing -----------------------------------------------------------
+
+ def close(self, write_summary: bool = True) -> None:
+ if self._closed:
+ return
+ self.event("session_end")
+ self._fh.close()
+ self._closed = True
+ self.meta["counts"] = self._counts
+ self.meta["snapshots"] = self._snapshots
+ with open(os.path.join(self.dir, "session.json"), "w", encoding="utf-8") as fh:
+ json.dump(self.meta, fh, indent=1, sort_keys=True)
+ if write_summary:
+ try:
+ from .analyze import render_summary
+
+ report = render_summary(self.dir)
+ with open(os.path.join(self.dir, "summary.md"), "w", encoding="utf-8") as fh:
+ fh.write(report)
+ except Exception as exc: # pragma: no cover - never break close
+ with open(os.path.join(self.dir, "summary-error.txt"), "w", encoding="utf-8") as fh:
+ fh.write(f"summary failed: {exc}\n")
+
+ def __enter__(self) -> "Recorder":
+ return self
+
+ def __exit__(self, *exc) -> None:
+ self.close()
+
+
+def load_events(session_dir: str) -> list[dict]:
+ path = session_dir
+ if os.path.isdir(session_dir):
+ path = os.path.join(session_dir, "events.jsonl")
+ events: list[dict] = []
+ with open(path, "r", encoding="utf-8") as fh:
+ for line in fh:
+ line = line.strip()
+ if line:
+ events.append(json.loads(line))
+ return events
+
+
+def load_meta(session_dir: str) -> dict:
+ with open(os.path.join(session_dir, "session.json"), "r", encoding="utf-8") as fh:
+ return json.load(fh)
+
+
+def _jsonable(value):
+ if isinstance(value, bytes):
+ return value.hex()
+ if isinstance(value, (set, tuple)):
+ return list(value)
+ if isinstance(value, dict):
+ return {str(k): _jsonable(v) for k, v in value.items()}
+ return value
+
+
+def _sha(data: bytes) -> str:
+ import hashlib
+
+ return hashlib.sha256(data).hexdigest()
+
+
+def _version() -> str:
+ try:
+ from . import __version__
+
+ return __version__
+ except Exception: # pragma: no cover
+ return "unknown"
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/rsp.py b/debug/risc-v-gsp-rm-vm/greeneye/rsp.py
new file mode 100644
index 0000000..0263c63
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/rsp.py
@@ -0,0 +1,583 @@
+"""Pure-Python GDB Remote Serial Protocol (RSP) client for QEMU's gdbstub.
+
+green-eye deliberately avoids GDB: the whole instrumentation stack is a Python
+library, so sessions, probes and plugins are scriptable end to end. This
+module implements the RSP subset QEMU needs: packet framing/acks, no-ack mode,
+registers (with layout read from ``target.xml``), memory, software/hardware
+breakpoints, watchpoints, continue/step/interrupt and monitor commands.
+
+Tested against QEMU 8.2 (``qemu-system-riscv64``).
+"""
+from __future__ import annotations
+
+import socket
+import time
+from dataclasses import dataclass, field as _field
+
+from .errors import RspProtocolError, RspTimeout
+
+# ---------------------------------------------------------------------------
+# Packet framing
+# ---------------------------------------------------------------------------
+
+
+def checksum(data: bytes) -> int:
+ """RSP 8-bit checksum: sum of bytes mod 256."""
+ return sum(data) & 0xFF
+
+
+def encode_packet(data: bytes) -> bytes:
+ """Frame a payload as ``$<data>#<checksum>``."""
+ return b"$" + data + b"#" + f"{checksum(data):02x}".encode("ascii")
+
+
+def _hex_value(payload: str) -> int:
+ try:
+ return int(payload, 16)
+ except ValueError as exc: # pragma: no cover - defensive
+ raise RspProtocolError(f"malformed hex value: {payload!r}") from exc
+
+
+# ---------------------------------------------------------------------------
+# Stop replies
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class StopReply:
+ """A parsed stop reply (``T``/``S``/``W``/``X`` packet)."""
+
+ raw: str
+ signal: int | None = None
+ thread: str | None = None
+ regs: dict[int, int] = _field(default_factory=dict)
+ watch_addr: int | None = None
+ watch_kind: str | None = None
+ exit_status: int | None = None
+
+ @property
+ def is_exit(self) -> bool:
+ return self.raw[:1] in ("W", "X")
+
+ @property
+ def is_stop(self) -> bool:
+ return self.raw[:1] in ("T", "S")
+
+ def __str__(self) -> str: # pragma: no cover - cosmetic
+ if self.is_exit:
+ return f"exit(status={self.exit_status})"
+ bits = [f"signal={self.signal}"]
+ if self.watch_addr is not None:
+ bits.append(f"{self.watch_kind}@{self.watch_addr:#x}")
+ if self.thread:
+ bits.append(f"thread={self.thread}")
+ return "stop(" + ", ".join(bits) + ")"
+
+
+def parse_stop_reply(raw: str) -> StopReply:
+ """Parse a stop packet body (without the leading channel)."""
+ reply = StopReply(raw=raw)
+ if not raw:
+ return reply
+ head = raw[0]
+ if head in ("T", "S"):
+ if len(raw) >= 3:
+ try:
+ reply.signal = int(raw[1:3], 16)
+ except ValueError:
+ pass
+ rest = raw[3:]
+ for item in rest.split(";"):
+ if not item or ":" not in item:
+ continue
+ key, value = item.split(":", 1)
+ if key == "thread":
+ reply.thread = value
+ elif key in ("watch", "rwatch", "awatch"):
+ reply.watch_kind = key
+ reply.watch_addr = _hex_value(value)
+ else:
+ try:
+ reply.regs[int(key, 16)] = _hex_value(value)
+ except ValueError:
+ continue
+ elif head in ("W", "X"):
+ if len(raw) >= 3:
+ try:
+ reply.exit_status = int(raw[1:3], 16)
+ except ValueError:
+ pass
+ return reply
+
+
+# ---------------------------------------------------------------------------
+# Register layout
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class Register:
+ number: int
+ name: str
+ bitsize: int
+ offset: int = 0
+
+ @property
+ def nbytes(self) -> int:
+ return (self.bitsize + 7) // 8
+
+
+#: Fallback layout for RV64 when target.xml is unavailable: x0..x31 + pc.
+RV64_FALLBACK_REGS = [
+ Register(n, name, 64, n * 8)
+ for n, name in enumerate(
+ ["x0", "ra", "sp", "gp", "tp", "t0", "t1", "t2"]
+ + [f"s{i}" for i in range(0, 2)]
+ + [f"a{i}" for i in range(0, 8)]
+ + [f"s{i}" for i in range(2, 12)]
+ + [f"t{i}" for i in range(3, 7)]
+ + ["pc"]
+ )
+]
+
+
+# ---------------------------------------------------------------------------
+# Client
+# ---------------------------------------------------------------------------
+
+
+class RspClient:
+ """A blocking RSP client over TCP."""
+
+ def __init__(self, host: str = "127.0.0.1", port: int = 4444, timeout: float = 10.0):
+ self.host = host
+ self.port = port
+ self.timeout = timeout
+ self.sock: socket.socket | None = None
+ self.no_ack = False
+ self.packet_size = 0x1000
+ self.registers: list[Register] = []
+ self._by_name: dict[str, Register] = {}
+ self._by_number: dict[int, Register] = {}
+ self.features: str = ""
+
+ # -- lifecycle ---------------------------------------------------------
+
+ def connect(self, retries: int = 1, retry_delay: float = 0.1) -> None:
+ last: Exception | None = None
+ for _ in range(max(1, retries)):
+ try:
+ self.sock = socket.create_connection((self.host, self.port), timeout=self.timeout)
+ self.sock.settimeout(self.timeout)
+ return
+ except OSError as exc:
+ last = exc
+ time.sleep(retry_delay)
+ raise RspTimeout(f"cannot connect to gdbstub {self.host}:{self.port}: {last}")
+
+ def close(self) -> None:
+ if self.sock is not None:
+ try:
+ self.sock.close()
+ except OSError:
+ pass
+ self.sock = None
+
+ @property
+ def connected(self) -> bool:
+ return self.sock is not None
+
+ def __enter__(self) -> "RspClient":
+ if not self.connected:
+ self.connect()
+ return self
+
+ def __exit__(self, *exc) -> None:
+ self.close()
+
+ # -- low level I/O -----------------------------------------------------
+
+ def _send_raw(self, data: bytes) -> None:
+ if self.sock is None:
+ raise RspTimeout("not connected")
+ self.sock.sendall(data)
+
+ def _read_exact(self, count: int, deadline: float) -> bytes:
+ if self.sock is None:
+ raise RspTimeout("not connected")
+ chunks = bytearray()
+ while len(chunks) < count:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise RspTimeout(f"timeout waiting for {count} bytes")
+ self.sock.settimeout(remaining)
+ try:
+ part = self.sock.recv(count - len(chunks))
+ except socket.timeout as exc:
+ raise RspTimeout(f"timeout waiting for {count} bytes") from exc
+ if not part:
+ raise RspProtocolError("gdbstub closed the connection")
+ chunks += part
+ return bytes(chunks)
+
+ def _send_packet(self, payload: bytes) -> None:
+ packet = encode_packet(payload)
+ self._send_raw(packet)
+ if self.no_ack:
+ return
+ deadline = time.monotonic() + self.timeout
+ while True:
+ ack = self._read_exact(1, deadline)
+ if ack == b"+":
+ return
+ if ack == b"-":
+ self._send_raw(packet)
+ continue
+ # Some stubs answer with a stop packet instead of an ack when a
+ # run-control command stops immediately; push it back for the
+ # packet reader by buffering is complex, so treat as protocol
+ # error only if it is not a start-of-packet marker.
+ if ack == b"$":
+ raise RspProtocolError("unexpected packet while waiting for ack")
+
+ def _read_packet(self, timeout: float | None = None) -> str:
+ eff = self.timeout if timeout is None else timeout
+ deadline = time.monotonic() + eff
+ # Skip stale acks and interrupt notifications until a packet starts.
+ while True:
+ byte = self._read_exact(1, deadline)
+ if byte == b"$":
+ break
+ if byte == b"+":
+ continue
+ if byte == b"\x03":
+ return "S02" # async interrupt notification
+ body = bytearray()
+ while True:
+ byte = self._read_exact(1, deadline)
+ if byte == b"#":
+ break
+ body += byte
+ got = self._read_exact(2, deadline)
+ try:
+ expected = int(got.decode("ascii"), 16)
+ except ValueError as exc:
+ raise RspProtocolError(f"malformed checksum bytes {got!r}") from exc
+ if checksum(bytes(body)) != expected:
+ if not self.no_ack:
+ self._send_raw(b"-")
+ raise RspProtocolError(f"bad checksum for {bytes(body)!r}")
+ if not self.no_ack:
+ self._send_raw(b"+")
+ return bytes(body).decode("latin1")
+
+ def command(self, payload: str | bytes, timeout: float | None = None) -> str:
+ """Send one packet and return the reply payload."""
+ data = payload.encode("ascii") if isinstance(payload, str) else payload
+ self._send_packet(data)
+ return self._read_packet(timeout)
+
+ # -- handshake ---------------------------------------------------------
+
+ def sync(self) -> None:
+ """Negotiate features and register layout; consume the initial stop."""
+ # QEMU acknowledges QStartNoAckMode with an EMPTY packet ($#00), then
+ # switches to no-ack mode once the client acks that reply.
+ try:
+ self._send_packet(b"QStartNoAckMode")
+ reply = self._read_packet()
+ if reply in ("", "OK"):
+ self.no_ack = True
+ except RspProtocolError:
+ pass # older stub: stay in ack mode
+
+ self.packet_size = 0x1000
+ try:
+ self.features = self.command("qSupported:multiprocess+;swbreak+;hwbreak+;qXfer:features:read+")
+ for token in self.features.split(";"):
+ if token.startswith("PacketSize="):
+ try:
+ self.packet_size = int(token.split("=", 1)[1], 16)
+ except ValueError:
+ pass
+ except RspProtocolError:
+ self.features = ""
+
+ # Consume the initial stop reply (target is paused).
+ try:
+ self.command("?")
+ except RspProtocolError:
+ pass
+ # Select CPU 0 for register access in system mode.
+ for selection in ("Hg0", "Hc-1"):
+ try:
+ self.command(selection)
+ except RspProtocolError:
+ pass
+ self._load_registers()
+
+ def _load_registers(self) -> None:
+ self.registers = []
+ try:
+ xml = self._read_target_xml()
+ except RspProtocolError:
+ xml = ""
+ if xml:
+ self.registers = _parse_target_xml(xml)
+ if not self.registers:
+ self.registers = list(RV64_FALLBACK_REGS)
+ self._by_name = {r.name: r for r in self.registers}
+ self._by_number = {r.number: r for r in self.registers}
+
+ def _read_target_xml(self, depth: int = 0) -> str:
+ """Fetch target.xml and recursively inline its xi:include files."""
+ return self._collect_annex("target.xml", depth)
+
+ def _collect_annex(self, annex: str, depth: int) -> str:
+ import re
+
+ try:
+ xml = self._read_xfer("features", annex)
+ except RspProtocolError:
+ return ""
+ if depth >= 4:
+ return xml
+ parts = [xml]
+ for href in re.findall(r'xi:include[^>]*href="([^"]+)"', xml):
+ if href.startswith(("http:", "https:", "file:")):
+ continue
+ child = self._collect_annex(href, depth + 1)
+ if child:
+ parts.append(child)
+ return "\n".join(parts)
+
+ def _read_xfer(self, kind: str, annex: str) -> str:
+ data = bytearray()
+ offset = 0
+ while True:
+ reply = self.command(f"qXfer:{kind}:read:{annex}:{offset:x},fff")
+ if reply.startswith("E"):
+ raise RspProtocolError(f"qXfer {kind}:{annex} failed: {reply}")
+ if not reply or reply[0] not in ("l", "m"):
+ raise RspProtocolError(f"qXfer {kind}:{annex} malformed: {reply[:16]!r}")
+ chunk = reply[1:].encode("latin1")
+ data += chunk
+ offset += len(chunk)
+ if reply[0] == "l":
+ break
+ return data.decode("latin1", "replace")
+
+ # -- registers ---------------------------------------------------------
+
+ def read_registers(self) -> dict[str, int]:
+ payload = self.command("g")
+ raw = _unhex(payload)
+ out: dict[str, int] = {}
+ for reg in self.registers:
+ start = reg.offset
+ chunk = raw[start : start + reg.nbytes]
+ if len(chunk) < reg.nbytes:
+ break
+ out[reg.name] = int.from_bytes(chunk, "little", signed=False)
+ return out
+
+ def read_reg(self, name_or_number: str | int) -> int:
+ reg = self._reg(name_or_number)
+ reply = self.command(f"p{reg.number:x}")
+ if reply.startswith("E"):
+ raise RspProtocolError(f"read register {reg.name} failed: {reply}")
+ return int.from_bytes(_unhex(reply), "little", signed=False)
+
+ def write_reg(self, name_or_number: str | int, value: int) -> None:
+ reg = self._reg(name_or_number)
+ payload = value.to_bytes(reg.nbytes, "little", signed=False).hex()
+ reply = self.command(f"P{reg.number:x}={payload}")
+ if reply != "OK":
+ raise RspProtocolError(f"write register {reg.name} failed: {reply}")
+
+ def _reg(self, name_or_number: str | int) -> Register:
+ if isinstance(name_or_number, int):
+ if name_or_number in self._by_number:
+ return self._by_number[name_or_number]
+ # allow ABI-numbered convenience: 0..31 map to x0..x31
+ if name_or_number < len(self.registers):
+ return self.registers[name_or_number]
+ raise RspProtocolError(f"unknown register number {name_or_number}")
+ if name_or_number in self._by_name:
+ return self._by_name[name_or_number]
+ aliases = {"pc": "pc", "a0": "a0", "return": "ra"}
+ name_or_number = aliases.get(name_or_number, name_or_number)
+ if name_or_number in self._by_name:
+ return self._by_name[name_or_number]
+ raise RspProtocolError(f"unknown register {name_or_number!r}")
+
+ @property
+ def pc(self) -> int:
+ return self.read_reg("pc")
+
+ # -- memory ------------------------------------------------------------
+
+ def _mem_chunk(self) -> int:
+ # Each data byte costs two hex chars; leave room for the header.
+ return max(64, (self.packet_size - 128) // 2)
+
+ def read_mem(self, addr: int, size: int, chunk: int = 0) -> bytes:
+ if chunk <= 0:
+ chunk = self._mem_chunk()
+ out = bytearray()
+ while len(out) < size:
+ want = min(chunk, size - len(out))
+ here = addr + len(out)
+ reply = self.command(f"m{here:x},{want:x}")
+ if reply.startswith("E"):
+ raise RspProtocolError(f"memory read {here:#x}+{want:#x} failed: {reply}")
+ data = _unhex(reply)
+ if not data:
+ raise RspProtocolError(f"short memory read at {here:#x}")
+ out += data
+ return bytes(out)
+
+ def write_mem(self, addr: int, data: bytes, chunk: int = 0) -> None:
+ if chunk <= 0:
+ chunk = self._mem_chunk()
+ view = memoryview(data)
+ offset = 0
+ while offset < len(view):
+ part = bytes(view[offset : offset + chunk])
+ here = addr + offset
+ reply = self.command(f"M{here:x},{len(part):x}:{part.hex()}")
+ if reply != "OK":
+ raise RspProtocolError(f"memory write {here:#x} failed: {reply}")
+ offset += len(part)
+
+ def read_u32(self, addr: int) -> int:
+ return int.from_bytes(self.read_mem(addr, 4), "little")
+
+ def read_u64(self, addr: int) -> int:
+ return int.from_bytes(self.read_mem(addr, 8), "little")
+
+ def write_u32(self, addr: int, value: int) -> None:
+ self.write_mem(addr, value.to_bytes(4, "little"))
+
+ def write_u64(self, addr: int, value: int) -> None:
+ self.write_mem(addr, value.to_bytes(8, "little"))
+
+ def read_cstring(self, addr: int, max_len: int = 256) -> str:
+ raw = self.read_mem(addr, max_len)
+ return raw.split(b"\x00", 1)[0].decode("utf-8", "replace")
+
+ # -- breakpoints / watchpoints ----------------------------------------
+
+ def set_sw_break(self, addr: int, kind: int = 4) -> None:
+ reply = self.command(f"Z0,{addr:x},{kind:x}")
+ if reply != "OK":
+ raise RspProtocolError(f"set software breakpoint {addr:#x} failed: {reply}")
+
+ def clear_sw_break(self, addr: int, kind: int = 4) -> None:
+ reply = self.command(f"z0,{addr:x},{kind:x}")
+ if reply != "OK":
+ raise RspProtocolError(f"clear software breakpoint {addr:#x} failed: {reply}")
+
+ def set_hw_break(self, addr: int, kind: int = 4) -> None:
+ reply = self.command(f"Z1,{addr:x},{kind:x}")
+ if reply != "OK":
+ raise RspProtocolError(f"set hardware breakpoint {addr:#x} failed: {reply}")
+
+ def clear_hw_break(self, addr: int, kind: int = 4) -> None:
+ reply = self.command(f"z1,{addr:x},{kind:x}")
+ if reply != "OK":
+ raise RspProtocolError(f"clear hardware breakpoint {addr:#x} failed: {reply}")
+
+ _WATCH_KINDS = {"w": 2, "r": 3, "a": 4}
+
+ def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
+ code = self._WATCH_KINDS[kind]
+ reply = self.command(f"Z{code},{addr:x},{size:x}")
+ if reply != "OK":
+ raise RspProtocolError(f"set watchpoint {kind}@{addr:#x} failed: {reply}")
+
+ def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
+ code = self._WATCH_KINDS[kind]
+ reply = self.command(f"z{code},{addr:x},{size:x}")
+ if reply != "OK":
+ raise RspProtocolError(f"clear watchpoint {kind}@{addr:#x} failed: {reply}")
+
+ # -- run control -------------------------------------------------------
+
+ def continue_(self, timeout: float | None = None) -> StopReply:
+ self._send_packet(b"c")
+ raw = self._read_packet(timeout)
+ return parse_stop_reply(raw)
+
+ def step(self, timeout: float | None = None) -> StopReply:
+ self._send_packet(b"s")
+ raw = self._read_packet(timeout)
+ return parse_stop_reply(raw)
+
+ def interrupt(self, timeout: float | None = None) -> StopReply:
+ self._send_raw(b"\x03")
+ raw = self._read_packet(timeout)
+ return parse_stop_reply(raw)
+
+ # -- monitor -----------------------------------------------------------
+
+ def monitor(self, cmd: str) -> str:
+ payload = ("qRcmd," + cmd.encode("ascii").hex()).encode("ascii")
+ self._send_packet(payload)
+ out = bytearray()
+ while True:
+ reply = self._read_packet()
+ if reply == "OK":
+ break
+ if reply.startswith("E"):
+ raise RspProtocolError(f"monitor {cmd!r} failed: {reply}")
+ if reply.startswith("O"):
+ out += bytes.fromhex(reply[1:])
+ else: # pragma: no cover - defensive
+ raise RspProtocolError(f"unexpected monitor reply: {reply[:32]!r}")
+ return out.decode("utf-8", "replace")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _unhex(payload: str) -> bytes:
+ if len(payload) % 2:
+ payload = payload[:-1] # tolerate odd trailing nibble
+ try:
+ return bytes.fromhex(payload)
+ except ValueError as exc:
+ raise RspProtocolError(f"malformed hex payload: {payload[:32]!r}") from exc
+
+
+_RE_REG = None
+
+
+def _parse_target_xml(xml: str) -> list[Register]:
+ import re
+
+ global _RE_REG
+ if _RE_REG is None:
+ _RE_REG = re.compile(r"<reg\b[^>]*?/?>", re.IGNORECASE)
+ attrs_re = re.compile(r'(\w+)\s*=\s*"([^"]*)"')
+ regs: list[Register] = []
+ next_number = 0
+ offset = 0
+ for match in _RE_REG.finditer(xml):
+ attrs = {k: v for k, v in attrs_re.findall(match.group(0))}
+ name = attrs.get("name")
+ if not name:
+ continue
+ bitsize = int(attrs.get("bitsize", "64"))
+ number = int(attrs["regnum"], 0) if "regnum" in attrs else next_number
+ if "offset" in attrs:
+ try:
+ offset = int(attrs["offset"], 0)
+ except ValueError:
+ pass
+ regs.append(Register(number=number, name=name, bitsize=bitsize, offset=offset))
+ offset += (bitsize + 7) // 8
+ next_number = number + 1
+ return regs
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/session.py b/debug/risc-v-gsp-rm-vm/greeneye/session.py
new file mode 100644
index 0000000..7ad6444
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/session.py
@@ -0,0 +1,562 @@
+"""green-eye instrumentation session: hooks, breakpoints, watchpoints, events.
+
+This is the Frida-like core. It is deliberately transport-agnostic; the RSP
+client is the default backend, and a Session can also be constructed around a
+plugin-supplied backend later.
+"""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field as _field
+from typing import Callable
+
+from .errors import GreenEyeError, RspTimeout, SymbolNotFound
+from .memory import Memory
+from .record import NullRecorder, Recorder
+from .rsp import RspClient, StopReply
+from .symbols import Symbol, SymbolDB
+
+
+@dataclass
+class Hit:
+ """Everything known about one stop event."""
+
+ kind: str # breakpoint | leave | watchpoint | step | fault | exit | stop
+ addr: int
+ regs: dict = _field(default_factory=dict)
+ symbol: Symbol | None = None
+ retval: int | None = None
+ watch: "Watchpoint | None" = None
+ stop: StopReply | None = None
+ session: "Session | None" = None
+ data: dict = _field(default_factory=dict)
+
+ @property
+ def pc(self) -> int:
+ return self.regs.get("pc", self.addr)
+
+ def arg(self, index: int) -> int:
+ """RISC-V ABI argument a0..a7 -> x10..x17."""
+ return self.regs.get(f"a{index}", 0)
+
+ def label(self) -> str:
+ if self.symbol is not None:
+ return self.symbol.label(self.addr)
+ return f"{self.addr:#x}"
+
+ def __str__(self) -> str: # pragma: no cover - cosmetic
+ base = f"[{self.kind}] {self.label()}"
+ if self.kind in ("breakpoint", "leave"):
+ base += f" a0={self.arg(0):#x} a1={self.arg(1):#x}"
+ if self.kind == "leave" and self.retval is not None:
+ base += f" -> {self.retval:#x}"
+ if self.kind == "watchpoint" and self.watch is not None:
+ base += f" {self.watch.kind}@{self.watch.addr:#x}"
+ return base
+
+
+@dataclass
+class Breakpoint:
+ addr: int
+ kind: str = "sw" # sw | hw
+ temporary: bool = False
+ purpose: str = "user" # user | leave
+ on_enter: Callable[[Hit], None] | None = None
+ on_leave: Callable[[Hit], None] | None = None
+ symbol: Symbol | None = None
+ attached: bool = True
+ hits: int = 0
+ listener: str | None = None
+
+ @property
+ def label(self) -> str:
+ return self.symbol.label(self.addr) if self.symbol else f"{self.addr:#x}"
+
+
+@dataclass
+class Watchpoint:
+ addr: int
+ size: int = 8
+ kind: str = "w" # w | r | a
+ callback: Callable[[Hit], None] | None = None
+ symbol: Symbol | None = None
+ hits: int = 0
+ listener: str | None = None
+
+ @property
+ def label(self) -> str:
+ return self.symbol.label(self.addr) if self.symbol else f"{self.addr:#x}"
+
+
+class HookHandle:
+ def __init__(self, session: "Session", breakpoint: Breakpoint):
+ self._session = session
+ self._bp = breakpoint
+
+ @property
+ def addr(self) -> int:
+ return self._bp.addr
+
+ @property
+ def hits(self) -> int:
+ return self._bp.hits
+
+ @property
+ def label(self) -> str:
+ return self._bp.label
+
+ def detach(self) -> None:
+ self._session.remove_breakpoint(self._bp)
+
+
+class Session:
+ def __init__(
+ self,
+ *,
+ client: RspClient,
+ target=None,
+ symbols: SymbolDB | None = None,
+ record: str | bool | None = None,
+ name: str = "session",
+ meta: dict | None = None,
+ ):
+ self.client = client
+ self.target = target
+ self.symbols = symbols or SymbolDB()
+ self.name = name
+ self.image = None
+ self._owns_target = target is not None
+ if isinstance(record, str):
+ self.recorder = Recorder(record, name=name, meta=meta or {})
+ elif record:
+ self.recorder = Recorder.new(name=name, meta=meta or {})
+ else:
+ self.recorder = NullRecorder()
+ self.memory = Memory(client, symbols=self.symbols, recorder=self.recorder)
+ self._bps: dict[int, Breakpoint] = {}
+ self._bp_seq = 0
+ self._bps_by_addr: dict[int, list[Breakpoint]] = {}
+ self._watches: dict[tuple[int, int, str], Watchpoint] = {}
+ self._watches_by_addr: dict[int, Watchpoint] = {}
+ self._leave_pending: dict[int, list[Breakpoint]] = {}
+ self._stopped = True
+ self._event_hooks: list[tuple[str, Callable[[Hit], None]]] = []
+ self.recorder.event("session_start", name=name)
+
+ # -- convenience proxies ----------------------------------------------
+
+ def load_image(
+ self,
+ image,
+ *,
+ write_segments: bool = False,
+ set_pc: bool = False,
+ bias: int = 0,
+ load_symbols: bool = True,
+ source: str = "elf",
+ ):
+ """Register a firmware image with the session.
+
+ ``write_segments`` copies every PT_LOAD through the RSP client (for
+ targets started without their loader devices); ``set_pc`` points the
+ CPU at the (relocated) entry. With QEMU's ``-kernel`` fast path both
+ are already handled by QEMU, so the default only loads symbols.
+ """
+ self.image = image
+ if load_symbols:
+ self.symbols.load_elf(image, bias=bias, source=source)
+ if write_segments:
+ for addr, data in image.flat_load_plan(bias=bias):
+ self.client.write_mem(addr, data)
+ if set_pc:
+ self.client.write_reg("pc", image.relocated_entry(bias=bias))
+ self.recorder.event(
+ "image_loaded",
+ path=getattr(image, "path", None),
+ entry=image.entry,
+ bias=bias,
+ write_segments=write_segments,
+ )
+ return image
+
+ @property
+ def pc(self) -> int:
+ return self.client.pc
+
+ def read_regs(self) -> dict[str, int]:
+ return self.client.read_registers()
+
+ def snapshot(self, addr: int, size: int, name: str = "region") -> bytes:
+ return self.memory.snapshot(addr, size, name=name)
+
+ def note(self, text: str, **fields) -> None:
+ self.recorder.event("note", text=text, **fields)
+
+ # -- symbol helpers ----------------------------------------------------
+
+ def resolve(self, target: int | str) -> int:
+ if isinstance(target, int):
+ return target
+ if isinstance(target, str):
+ if target.lower().startswith("0x"):
+ return int(target, 16)
+ sym = self.symbols.get(target)
+ if sym is None:
+ raise SymbolNotFound(f"symbol {target!r} not found in symbol DB")
+ return sym.addr
+ raise GreenEyeError(f"cannot resolve target {target!r}")
+
+ def symbol_at(self, addr: int) -> Symbol | None:
+ return self.symbols.resolve(addr)
+
+ # -- breakpoints / hooks ----------------------------------------------
+
+ def hook(
+ self,
+ target: int | str,
+ on_enter: Callable[[Hit], None] | None = None,
+ on_leave: Callable[[Hit], None] | None = None,
+ *,
+ kind: str = "sw",
+ temporary: bool = False,
+ symbol: Symbol | None = None,
+ listener: str | None = None,
+ ) -> HookHandle:
+ addr = self.resolve(target)
+ sym = symbol or self.symbols.resolve(addr)
+ bp = Breakpoint(
+ addr=addr,
+ kind=kind,
+ temporary=temporary,
+ on_enter=on_enter,
+ on_leave=on_leave,
+ symbol=sym,
+ listener=listener,
+ )
+ self.add_breakpoint(bp)
+ self.recorder.event(
+ "hook",
+ addr=addr,
+ symbol=sym.name if sym else None,
+ kind=kind,
+ on_leave=bool(on_leave),
+ listener=listener,
+ )
+ return HookHandle(self, bp)
+
+ def breakpoint(self, target: int | str, callback: Callable[[Hit], None] | None = None, **kw) -> Breakpoint:
+ addr = self.resolve(target)
+ bp = Breakpoint(addr=addr, on_enter=callback, symbol=self.symbols.resolve(addr), **kw)
+ self.add_breakpoint(bp)
+ return bp
+
+ def add_breakpoint(self, bp: Breakpoint) -> None:
+ self._bp_seq += 1
+ self._bps[self._bp_seq] = bp
+ self._bps_by_addr.setdefault(bp.addr, []).append(bp)
+ if bp.kind == "hw":
+ self.client.set_hw_break(bp.addr)
+ else:
+ self.client.set_sw_break(bp.addr)
+
+ def remove_breakpoint(self, bp: Breakpoint) -> None:
+ for key, candidate in list(self._bps.items()):
+ if candidate is bp:
+ del self._bps[key]
+ lst = self._bps_by_addr.get(bp.addr, [])
+ if bp in lst:
+ lst.remove(bp)
+ if not lst:
+ self._bps_by_addr.pop(bp.addr, None)
+ self._leave_pending.pop(bp.addr, None)
+ try:
+ if bp.kind == "hw":
+ self.client.clear_hw_break(bp.addr)
+ else:
+ self.client.clear_sw_break(bp.addr)
+ except Exception:
+ pass
+ bp.attached = False
+
+ def on_event(self, kind: str, callback: Callable[[Hit], None]) -> None:
+ self._event_hooks.append((kind, callback))
+
+ # -- watchpoints -------------------------------------------------------
+
+ def watch(
+ self,
+ target: int | str,
+ size: int = 8,
+ kind: str = "w",
+ callback: Callable[[Hit], None] | None = None,
+ listener: str | None = None,
+ ) -> Watchpoint:
+ addr = self.resolve(target)
+ wp = Watchpoint(
+ addr=addr,
+ size=size,
+ kind=kind,
+ callback=callback,
+ symbol=self.symbols.resolve(addr),
+ listener=listener,
+ )
+ key = (addr, size, kind)
+ if key in self._watches:
+ raise GreenEyeError(f"watchpoint {kind}@{addr:#x}/{size} already exists")
+ self._watches[key] = wp
+ self._watches_by_addr[addr] = wp
+ self.client.set_watch(addr, size=size, kind=kind)
+ self.recorder.event(
+ "watch",
+ addr=addr,
+ size=size,
+ kind=kind,
+ symbol=wp.symbol.name if wp.symbol else None,
+ listener=listener,
+ )
+ return wp
+
+ def unwatch(self, wp: Watchpoint) -> None:
+ key = (wp.addr, wp.size, wp.kind)
+ self._watches.pop(key, None)
+ self._watches_by_addr.pop(wp.addr, None)
+ try:
+ self.client.clear_watch(wp.addr, size=wp.size, kind=wp.kind)
+ except Exception:
+ pass
+
+ # -- execution ---------------------------------------------------------
+
+ def step(self, count: int = 1) -> list[Hit]:
+ hits: list[Hit] = []
+ for _ in range(count):
+ stop = self.client.step()
+ hit = self._dispatch_stop(stop, from_step=True)
+ if hit is not None:
+ hits.append(hit)
+ if stop.is_exit:
+ break
+ return hits
+
+ def run(
+ self,
+ duration: float | None = None,
+ max_stops: int | None = None,
+ timeout: float | None = None,
+ ) -> int:
+ """Resume and service events until the deadline/budget/exit.
+
+ ``duration`` (or ``timeout`` when duration is None) is a wall-clock
+ budget: when it expires the target is interrupted once, the resulting
+ stop is dispatched, and the call returns. Returns the number of stops
+ serviced.
+ """
+ limit = duration if duration is not None else timeout
+ deadline = time.monotonic() + limit if limit is not None else None
+ stops = 0
+ while True:
+ if deadline is not None:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ break # budget spent while stopped: leave the target stopped
+ remaining = max(0.001, remaining)
+ else:
+ remaining = None
+ interrupted = False
+ try:
+ stop = self.client.continue_(timeout=remaining)
+ except RspTimeout:
+ if deadline is None:
+ raise
+ stop = self.client.interrupt()
+ interrupted = True
+ stops += 1
+ self._dispatch_stop(stop)
+ if stop.is_exit:
+ break
+ if max_stops is not None and stops >= max_stops:
+ break
+ if interrupted or (deadline is not None and time.monotonic() >= deadline):
+ break
+ return stops
+
+ def stop(self) -> Hit | None:
+ stop = self.client.interrupt()
+ return self._dispatch_stop(stop)
+
+ # -- stop dispatch -----------------------------------------------------
+
+ def _dispatch_stop(self, stop: StopReply, from_step: bool = False) -> Hit | None:
+ if stop.is_exit:
+ hit = Hit(kind="exit", addr=0, stop=stop, session=self)
+ self.recorder.event("exit", status=stop.exit_status)
+ self._fire_event_hooks(hit)
+ return hit
+
+ regs = self.client.read_registers()
+ pc = regs.get("pc", 0)
+ sym = self.symbols.resolve(pc)
+
+ if stop.watch_addr is not None:
+ wp = self._watches_by_addr.get(stop.watch_addr)
+ hit = Hit(
+ kind="watchpoint",
+ addr=stop.watch_addr,
+ regs=regs,
+ symbol=self.symbols.resolve(stop.watch_addr),
+ watch=wp,
+ stop=stop,
+ session=self,
+ )
+ if wp is not None:
+ wp.hits += 1
+ if wp.callback is not None:
+ wp.callback(hit)
+ self.recorder.event(
+ "watch_hit",
+ addr=stop.watch_addr,
+ kind=wp.kind if wp else stop.watch_kind,
+ size=wp.size if wp else None,
+ symbol=hit.symbol.name if hit.symbol else None,
+ pc=pc,
+ regs=_json_regs(regs),
+ )
+ self._fire_event_hooks(hit)
+ # QEMU stops *at* the instruction that performed the watched
+ # access. Step it once with the watchpoint temporarily cleared so
+ # resuming does not re-trigger on the same store forever.
+ try:
+ if wp is not None:
+ self.client.clear_watch(wp.addr, size=wp.size, kind=wp.kind)
+ self.client.step()
+ except RspTimeout:
+ pass
+ finally:
+ if wp is not None:
+ try:
+ self.client.set_watch(wp.addr, size=wp.size, kind=wp.kind)
+ except Exception:
+ pass
+ return hit
+
+ hit = Hit(
+ kind="step" if from_step else "stop",
+ addr=pc,
+ regs=regs,
+ symbol=sym,
+ stop=stop,
+ session=self,
+ )
+
+ bps = list(self._bps_by_addr.get(pc, []))
+ leave_queue = self._leave_pending.get(pc, [])
+ if not bps and not leave_queue:
+ self.recorder.event(
+ "stop",
+ pc=pc,
+ symbol=sym.name if sym else None,
+ signal=stop.signal,
+ reason="step" if from_step else "halt",
+ )
+ self._fire_event_hooks(hit)
+ return hit
+
+ hit.kind = "breakpoint"
+ self.recorder.event(
+ "bp_hit",
+ pc=pc,
+ addr=pc,
+ symbol=sym.name if sym else None,
+ signal=stop.signal,
+ regs=_json_regs(regs),
+ )
+ for bp in bps:
+ bp.hits += 1
+ if bp.on_enter is not None:
+ bp.on_enter(hit)
+ if bp.on_leave is not None and not bp.temporary:
+ self._arm_leave(regs.get("ra", 0), bp)
+ if bp.temporary:
+ self.remove_breakpoint(bp)
+
+ # Return hooks: pop one pending hook for this return address.
+ if leave_queue:
+ bp = leave_queue.pop()
+ if not leave_queue:
+ self._leave_pending.pop(pc, None)
+ for candidate in list(self._bps_by_addr.get(pc, [])):
+ if candidate.purpose == "leave":
+ self.remove_breakpoint(candidate)
+ hit.kind = "leave"
+ hit.retval = regs.get("a0", 0)
+ if bp.symbol is not None:
+ hit.symbol = bp.symbol
+ hit.addr = bp.addr
+ if bp.on_leave is not None:
+ bp.on_leave(hit)
+
+ # Software breakpoints need a single-step over them before resuming.
+ self._step_over(pc)
+ self._fire_event_hooks(hit)
+ return hit
+
+ def _arm_leave(self, retaddr: int, bp: Breakpoint) -> None:
+ if not retaddr:
+ return
+ queue = self._leave_pending.get(retaddr)
+ if queue is None:
+ queue = []
+ self._leave_pending[retaddr] = queue
+ temp = Breakpoint(
+ addr=retaddr,
+ kind="sw",
+ temporary=True,
+ purpose="leave",
+ symbol=self.symbols.resolve(retaddr),
+ )
+ self.add_breakpoint(temp)
+ queue.append(bp)
+
+ def _step_over(self, pc: int) -> None:
+ sw = [bp for bp in self._bps_by_addr.get(pc, []) if bp.kind == "sw"]
+ if not sw:
+ return
+ for bp in sw:
+ self.client.clear_sw_break(pc)
+ self.client.step()
+ for bp in sw:
+ if bp.attached:
+ self.client.set_sw_break(pc)
+
+ def _fire_event_hooks(self, hit: Hit) -> None:
+ for kind, callback in self._event_hooks:
+ if kind in (hit.kind, "*"):
+ callback(hit)
+
+ # -- lifecycle ---------------------------------------------------------
+
+ def close(self, kill_target: bool | None = None) -> None:
+ try:
+ for bp in list(self._bps.values()):
+ self.remove_breakpoint(bp)
+ for wp in list(self._watches.values()):
+ self.unwatch(wp)
+ except Exception:
+ pass
+ self.client.close()
+ if kill_target is None:
+ kill_target = self._owns_target
+ if kill_target and self.target is not None and not getattr(self.target, "keep_alive", False):
+ self.target.kill()
+ self.recorder.close()
+ self._owns_target = False
+
+ def __enter__(self) -> "Session":
+ return self
+
+ def __exit__(self, *exc) -> None:
+ self.close()
+
+
+def _json_regs(regs: dict[str, int]) -> dict[str, int]:
+ return {name: value for name, value in regs.items() if isinstance(value, int)}
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/structs.py b/debug/risc-v-gsp-rm-vm/greeneye/structs.py
new file mode 100644
index 0000000..3081bc9
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/structs.py
@@ -0,0 +1,197 @@
+"""Declarative structure decoding for firmware memory.
+
+A schema is plain data (JSON-friendly), so the same description can come from
+the repo, a Ghidra export, or a plugin:
+
+ {
+ "name": "gsp_ctrl",
+ "fields": [
+ {"name": "magic", "offset": 0, "fmt": "u32"},
+ {"name": "state", "offset": 8, "fmt": "u32", "enum": {"0": "IDLE", "1": "RUN"}},
+ {"name": "buf", "offset": 64, "fmt": "bytes", "size": 256},
+ {"name": "name", "offset": 64, "fmt": "cstr", "size": 32}
+ ]
+ }
+
+Supported formats: u8/u16/u32/u64, s8/s16/s32/s64, f32/f64, bool, bytes,
+cstr, ptr (u64 by default; override with ``size``).
+"""
+from __future__ import annotations
+
+import struct as _struct
+from dataclasses import dataclass, field as _field
+
+from .errors import StructError
+
+_INT_FORMATS = {
+ "u8": ("B", 1),
+ "u16": ("H", 2),
+ "u32": ("I", 4),
+ "u64": ("Q", 8),
+ "s8": ("b", 1),
+ "s16": ("h", 2),
+ "s32": ("i", 4),
+ "s64": ("q", 8),
+ "bool": ("B", 1),
+}
+_FLOAT_FORMATS = {"f32": ("f", 4), "f64": ("d", 8)}
+
+
+@dataclass
+class Field:
+ name: str
+ offset: int
+ fmt: str
+ size: int = 0
+ enum: dict | None = None
+ note: str | None = None
+ count: int = 1
+
+ @property
+ def nbytes(self) -> int:
+ if self.fmt in _INT_FORMATS:
+ return _INT_FORMATS[self.fmt][1]
+ if self.fmt in _FLOAT_FORMATS:
+ return _FLOAT_FORMATS[self.fmt][1]
+ if self.fmt in ("bytes", "cstr"):
+ return self.size
+ if self.fmt == "ptr":
+ return self.size or 8
+ if self.fmt == "u64x":
+ return 8
+ raise StructError(f"field {self.name!r}: unknown format {self.fmt!r}")
+
+
+@dataclass
+class Schema:
+ name: str
+ fields: list[Field] = _field(default_factory=list)
+ size: int = 0
+
+ def __post_init__(self) -> None:
+ computed = max((f.offset + f.nbytes for f in self.fields), default=0)
+ if not self.size:
+ self.size = computed
+ elif self.size < computed:
+ raise StructError(
+ f"schema {self.name!r}: declared size {self.size:#x} < fields end {computed:#x}"
+ )
+ self.validate()
+
+ def validate(self) -> None:
+ seen: set[str] = set()
+ for field in self.fields:
+ if field.name in seen:
+ raise StructError(f"schema {self.name!r}: duplicate field {field.name!r}")
+ seen.add(field.name)
+ if field.offset < 0:
+ raise StructError(f"schema {self.name!r}: negative offset for {field.name!r}")
+ if field.nbytes <= 0:
+ raise StructError(f"schema {self.name!r}: zero-size field {field.name!r}")
+
+ def decode(self, data: bytes, base: int = 0) -> dict:
+ if len(data) < self.size:
+ raise StructError(
+ f"schema {self.name!r}: need {self.size} bytes, got {len(data)}"
+ )
+ out: dict = {}
+ for field in self.fields:
+ out[field.name] = _decode_field(field, data, base)
+ return out
+
+
+def _decode_field(field: Field, data: bytes, base: int):
+ start = field.offset
+ if field.fmt in _INT_FORMATS:
+ code, width = _INT_FORMATS[field.fmt]
+ value = _struct.unpack_from("<" + code, data, start)[0]
+ if field.fmt == "bool":
+ value = bool(value)
+ return value
+ if field.fmt in _FLOAT_FORMATS:
+ code, _width = _FLOAT_FORMATS[field.fmt]
+ return _struct.unpack_from("<" + code, data, start)[0]
+ if field.fmt == "ptr":
+ return int.from_bytes(data[start : start + field.nbytes], "little")
+ if field.fmt == "bytes":
+ return bytes(data[start : start + field.size])
+ if field.fmt == "cstr":
+ raw = data[start : start + field.size]
+ return raw.split(b"\x00", 1)[0].decode("utf-8", "replace")
+ raise StructError(f"unknown format {field.fmt!r}")
+
+
+def decode(schema: Schema, data: bytes, base: int = 0) -> dict:
+ return schema.decode(data, base=base)
+
+
+def schema_from_json(payload: dict) -> Schema:
+ if not isinstance(payload, dict) or "fields" not in payload:
+ raise StructError("schema JSON must be an object with a 'fields' list")
+ fields = []
+ for item in payload["fields"]:
+ try:
+ fields.append(
+ Field(
+ name=item["name"],
+ offset=int(item["offset"]),
+ fmt=item.get("fmt", "bytes"),
+ size=int(item.get("size", 0)),
+ enum=item.get("enum"),
+ note=item.get("note"),
+ count=int(item.get("count", 1)),
+ )
+ )
+ except KeyError as exc:
+ raise StructError(f"schema field missing key: {exc}") from exc
+ return Schema(name=payload.get("name", "schema"), fields=fields, size=int(payload.get("size", 0)))
+
+
+def schema_to_json(schema: Schema) -> dict:
+ return {
+ "name": schema.name,
+ "size": schema.size,
+ "fields": [
+ {
+ k: v
+ for k, v in {
+ "name": f.name,
+ "offset": f.offset,
+ "fmt": f.fmt,
+ "size": f.size or None,
+ "enum": f.enum,
+ "note": f.note,
+ }.items()
+ if v is not None
+ }
+ for f in schema.fields
+ ],
+ }
+
+
+def render(schema: Schema, values: dict, indent: str = " ") -> str:
+ lines = [f"{schema.name} ({{base}}+{schema.size:#x} bytes)"]
+ width = max((len(f.name) for f in schema.fields), default=0)
+ for field in schema.fields:
+ value = values.get(field.name)
+ lines.append(f"{indent}{field.name:<{width}} @+0x{field.offset:03x} = {_format_value(field, value)}")
+ if field.note:
+ lines.append(f"{indent}{'':<{width}} ; {field.note}")
+ return "\n".join(lines)
+
+
+def _format_value(field: Field, value) -> str:
+ if isinstance(value, bool):
+ return str(value)
+ if isinstance(value, float):
+ return f"{value:g}"
+ if isinstance(value, int):
+ text = f"{value:#x}" if not field.enum else f"{value} ({field.enum.get(str(value), '?')})"
+ if field.enum and field.fmt not in ("bool",):
+ text = f"{value:#x} ({field.enum.get(str(value), field.enum.get(value, '?'))})"
+ return text
+ if isinstance(value, (bytes, bytearray)):
+ head = bytes(value[:16]).hex(" ")
+ suffix = " ..." if len(value) > 16 else ""
+ return f"[{len(value)}B] {head}{suffix}"
+ return repr(value)
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/symbols.py b/debug/risc-v-gsp-rm-vm/greeneye/symbols.py
new file mode 100644
index 0000000..7b2e9dd
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/symbols.py
@@ -0,0 +1,313 @@
+"""Symbol resolution and address translation for green-eye.
+
+Inputs supported (all optional, merged into one database):
+
+* ELF ``.symtab`` symbols (demo firmware);
+* the project's Ghidra exports:
+ - ``names.tsv`` ``<file_vaddr>\t<name>[\t<source>]`` (comments with #),
+ - ``decompiled_index.tsv`` / bucket TSVs
+ ``<program_addr>\t<file_addr>\t<size>\t<name>\t<bucket>\t<line>\t<status>``,
+ - ``exports.tsv`` ``addr\tfile_addr\tsize\tname``,
+* green-eye native JSON (``greeneye-symbols.json``).
+
+Address conventions (matching the project docs):
+ * Ghidra program address = file vaddr + image base (``0x100000`` for libcuda);
+ * runtime address = file vaddr + runtime load bias;
+ * for firmware images linked at their load address the bias is 0.
+
+``SymbolDB`` stores runtime addresses by default; each loader takes a ``bias``
+so a relocated firmware or a different image base can be normalised.
+"""
+from __future__ import annotations
+
+import bisect
+import json
+import os
+from dataclasses import dataclass, field as _field
+
+from .errors import SymbolNotFound
+
+
+@dataclass
+class Symbol:
+ addr: int
+ name: str
+ size: int = 0
+ source: str = ""
+ subsystem: str | None = None
+ extras: dict = _field(default_factory=dict)
+
+ @property
+ def end(self) -> int:
+ return self.addr + self.size if self.size else self.addr
+
+ def contains(self, addr: int) -> bool:
+ if self.size:
+ return self.addr <= addr < self.addr + self.size
+ return addr == self.addr
+
+ def label(self, addr: int | None = None) -> str:
+ if addr is None or addr == self.addr or not self.size:
+ return self.name
+ return f"{self.name}+{addr - self.addr:#x}"
+
+ def to_json(self) -> dict:
+ out = {"addr": self.addr, "name": self.name}
+ if self.size:
+ out["size"] = self.size
+ if self.source:
+ out["source"] = self.source
+ if self.subsystem:
+ out["subsystem"] = self.subsystem
+ if self.extras:
+ out.update(self.extras)
+ return out
+
+
+class SymbolDB:
+ def __init__(self) -> None:
+ self._symbols: list[Symbol] = []
+ self._sorted = False
+ self._by_name: dict[str, Symbol] = {}
+ self._addr_index: list[int] = []
+
+ # -- loading -----------------------------------------------------------
+
+ def add(self, symbol: Symbol) -> None:
+ self._symbols.append(symbol)
+ self._sorted = False
+ existing = self._by_name.get(symbol.name)
+ if existing is None or (symbol.source and not existing.source):
+ self._by_name[symbol.name] = symbol
+
+ def load_elf(self, image, bias: int = 0, source: str = "elf", only_code: bool = True) -> int:
+ """Load ``.symtab`` symbols from a FirmwareImage/ELF path."""
+ from .elf import ElfSymbol, load_elf
+
+ if isinstance(image, str):
+ image = load_elf(image)
+ count = 0
+ for sym in image.symbols: # type: ignore[union-attr]
+ if only_code and not (sym.is_function or sym.is_object):
+ continue
+ self.add(
+ Symbol(
+ addr=sym.value + bias,
+ name=sym.name,
+ size=sym.size,
+ source=source,
+ extras={"type": "func" if sym.is_function else "object"},
+ )
+ )
+ count += 1
+ return count
+
+ def load_names_tsv(self, path: str, bias: int = 0) -> int:
+ """Load the project's Ghidra ``names.tsv`` (file vaddrs)."""
+ count = 0
+ with open(path, "r", encoding="utf-8", errors="replace") as fh:
+ for line in fh:
+ line = line.rstrip("\n")
+ if not line or line.startswith("#"):
+ continue
+ cols = line.split("\t")
+ if len(cols) < 2:
+ continue
+ try:
+ addr = int(cols[0], 16)
+ except ValueError:
+ continue
+ self.add(
+ Symbol(addr=addr + bias, name=cols[1], source=cols[2] if len(cols) > 2 else "ghidra-names")
+ )
+ count += 1
+ return count
+
+ def load_index_tsv(self, path: str, bias: int = 0) -> int:
+ """Load ``decompiled_index.tsv``/bucket TSVs/exports.tsv heuristically."""
+ count = 0
+ with open(path, "r", encoding="utf-8", errors="replace") as fh:
+ for line in fh:
+ line = line.rstrip("\n")
+ if not line or line.startswith("#"):
+ continue
+ cols = line.split("\t")
+ if cols[0] in ("addr", "caller"):
+ continue
+ if len(cols) >= 7:
+ program, file_addr, size, name, bucket, lineno, status = cols[:7]
+ addr = _parse_hex(file_addr)
+ extras = {"program_addr": _parse_hex(program), "bucket": bucket, "line": lineno, "status": status}
+ elif len(cols) >= 4:
+ program, file_addr, size, name = cols[:4]
+ addr = _parse_hex(file_addr)
+ extras = {"program_addr": _parse_hex(program)}
+ elif len(cols) >= 2:
+ addr = _parse_hex(cols[0])
+ name = cols[1]
+ size = "0"
+ extras = {}
+ else:
+ continue
+ if addr is None or not name:
+ continue
+ try:
+ size_val = int(size, 0)
+ except ValueError:
+ size_val = 0
+ self.add(
+ Symbol(addr=addr + bias, name=name, size=size_val, source="ghidra-index", extras=extras)
+ )
+ count += 1
+ return count
+
+ def load_callgraph(self, path: str) -> int:
+ """Attach caller/callee lists from ``callgraph.tsv`` (program addrs)."""
+ by_program: dict[int, Symbol] = {}
+ for sym in self._symbols:
+ prog = sym.extras.get("program_addr")
+ if prog is not None:
+ by_program[int(prog)] = sym
+ if not by_program:
+ return 0
+ for sym in self._symbols:
+ sym.extras.setdefault("callers", [])
+ sym.extras.setdefault("callees", [])
+ count = 0
+ with open(path, "r", encoding="utf-8", errors="replace") as fh:
+ for line in fh:
+ line = line.rstrip("\n")
+ if not line or line.startswith("#") or line.startswith("caller"):
+ continue
+ cols = line.split("\t")
+ if len(cols) < 2:
+ continue
+ caller, callee = _parse_hex(cols[0]), _parse_hex(cols[1])
+ if caller is None or callee is None:
+ continue
+ if caller in by_program:
+ by_program[caller].extras["callees"].append(callee)
+ count += 1
+ if callee in by_program:
+ by_program[callee].extras["callers"].append(caller)
+ return count
+
+ def load_json(self, path: str) -> int:
+ with open(path, "r", encoding="utf-8") as fh:
+ payload = json.load(fh)
+ symbols = payload.get("symbols", payload if isinstance(payload, list) else [])
+ count = 0
+ for item in symbols:
+ try:
+ addr = int(item["addr"], 0) if isinstance(item["addr"], str) else int(item["addr"])
+ except (KeyError, TypeError, ValueError):
+ continue
+ extras = {k: v for k, v in item.items() if k not in ("addr", "name", "size", "source", "subsystem")}
+ self.add(
+ Symbol(
+ addr=addr,
+ name=item.get("name", f"sub_{addr:x}"),
+ size=int(item.get("size", 0)),
+ source=item.get("source", "json"),
+ subsystem=item.get("subsystem"),
+ extras=extras,
+ )
+ )
+ count += 1
+ return count
+
+ def save_json(self, path: str, meta: dict | None = None) -> None:
+ self._ensure_sorted()
+ payload = {
+ "format": "greeneye-symbols/1",
+ "count": len(self._symbols),
+ "meta": meta or {},
+ "symbols": [s.to_json() for s in self._symbols],
+ }
+ with open(path, "w", encoding="utf-8") as fh:
+ json.dump(payload, fh, indent=1, sort_keys=True)
+
+ # -- queries -----------------------------------------------------------
+
+ def _ensure_sorted(self) -> None:
+ if self._sorted:
+ return
+ self._symbols.sort(key=lambda s: (s.addr, -s.size))
+ self._addr_index = [s.addr for s in self._symbols]
+ self._sorted = True
+
+ def __len__(self) -> int:
+ return len(self._symbols)
+
+ def __iter__(self):
+ return iter(self._symbols)
+
+ def get(self, name: str) -> Symbol | None:
+ return self._by_name.get(name)
+
+ def lookup(self, name: str) -> int:
+ sym = self._by_name.get(name)
+ if sym is None:
+ raise SymbolNotFound(f"symbol {name!r} not found")
+ return sym.addr
+
+ def resolve(self, addr: int, max_delta: int = 0x10000) -> Symbol | None:
+ """Resolve an address to a containing or nearest preceding symbol."""
+ self._ensure_sorted()
+ if not self._symbols:
+ return None
+ pos = bisect.bisect_right(self._addr_index, addr) - 1
+ if pos < 0:
+ return None
+ candidate = self._symbols[pos]
+ if candidate.contains(addr):
+ return candidate
+ if candidate.size and addr < candidate.addr + candidate.size:
+ return candidate
+ # nearest-within-delta fallback for zero-size symbols (Ghidra names)
+ if not candidate.size and 0 <= addr - candidate.addr <= max_delta:
+ return candidate
+ return None
+
+ def annotate(self, addr: int, max_delta: int = 0x10000) -> str:
+ sym = self.resolve(addr, max_delta=max_delta)
+ if sym is None:
+ return f"{addr:#x} <unknown>"
+ return f"{addr:#x} {sym.label(addr)}"
+
+ def nearest(self, addr: int, max_delta: int = 0x10000) -> Symbol | None:
+ return self.resolve(addr, max_delta=max_delta)
+
+ def names(self) -> list[str]:
+ return sorted(self._by_name)
+
+ def by_subsystem(self, subsystem: str) -> list[Symbol]:
+ return [s for s in self._symbols if s.subsystem == subsystem]
+
+
+def _parse_hex(value: str) -> int | None:
+ value = value.strip()
+ if not value:
+ return None
+ try:
+ return int(value, 16)
+ except ValueError:
+ try:
+ return int(value, 0)
+ except ValueError:
+ return None
+
+
+def translation_for(
+ *,
+ file_vaddr: int,
+ program_addr: int | None = None,
+ runtime_addr: int | None = None,
+) -> dict:
+ """Explain the three address spaces for a debug session (docs aid)."""
+ out = {"file_vaddr": file_vaddr}
+ if program_addr is not None:
+ out["image_base"] = program_addr - file_vaddr
+ if runtime_addr is not None:
+ out["load_bias"] = runtime_addr - file_vaddr
+ return out
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/target.py b/debug/risc-v-gsp-rm-vm/greeneye/target.py
new file mode 100644
index 0000000..efcb1d1
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/target.py
@@ -0,0 +1,223 @@
+"""QEMU target lifecycle for green-eye.
+
+Owns the ``qemu-system-riscv64`` process, the gdbstub endpoint, the serial
+capture and the QEMU trace log. Both firmware paths are supported:
+
+* ``kernel=`` fast path - QEMU loads an ELF linked for RAM and sets the PC;
+* ``segments=`` probe path - each PT_LOAD segment is placed with a loader
+ device (raw), so a firmware linked for GPU physical addresses can be
+ relocated into RAM and observed.
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import signal
+import socket
+import subprocess
+import tempfile
+import time
+
+from .errors import TargetError
+
+DEFAULT_QEMU = "qemu-system-riscv64"
+
+
+def find_qemu() -> str:
+ env = os.environ.get("GREENEYE_QEMU")
+ if env:
+ return env
+ found = shutil.which(DEFAULT_QEMU)
+ if not found:
+ raise TargetError(
+ f"{DEFAULT_QEMU} not found. Install it (Ubuntu: qemu-system-misc) "
+ "or set GREENEYE_QEMU to the binary."
+ )
+ return found
+
+
+def free_port() -> int:
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+ sock.bind(("127.0.0.1", 0))
+ return sock.getsockname()[1]
+
+
+class QemuTarget:
+ def __init__(
+ self,
+ *,
+ machine: str = "virt",
+ memory_mb: int = 256,
+ gdb_port: int = 0,
+ bios: str = "none",
+ extra_args: list[str] | None = None,
+ qemu: str | None = None,
+ kernel: str | None = None,
+ segments: list[tuple[int, bytes]] | None = None,
+ dlog: str | None = None,
+ uart_log: str | None = None,
+ qemu_log: str | None = None,
+ workdir: str | None = None,
+ keep_alive: bool = False,
+ ):
+ self.qemu = qemu or find_qemu()
+ self.machine = machine
+ self.memory_mb = memory_mb
+ self.host = "127.0.0.1"
+ self.port = gdb_port or free_port()
+ self.bios = bios
+ self.extra_args = list(extra_args or [])
+ self.kernel = os.path.abspath(kernel) if kernel else None
+ self.segments = list(segments or [])
+ self.dlog = dlog
+ self.workdir = workdir or tempfile.mkdtemp(prefix="greeneye-qemu-")
+ os.makedirs(self.workdir, exist_ok=True)
+ self.uart_log = uart_log or os.path.join(self.workdir, "uart.log")
+ self.qemu_log = qemu_log or os.path.join(self.workdir, "qemu.log")
+ self.keep_alive = keep_alive
+ self.process: subprocess.Popen | None = None
+ self._segment_files: list[str] = []
+ self._cmd: list[str] = []
+
+ # -- lifecycle ---------------------------------------------------------
+
+ def _build_command(self, paused: bool) -> list[str]:
+ cmd = [
+ self.qemu,
+ "-machine",
+ self.machine,
+ "-m",
+ f"{self.memory_mb}M",
+ "-display",
+ "none",
+ "-monitor",
+ "none",
+ "-serial",
+ f"file:{self.uart_log}",
+ "-gdb",
+ f"tcp:{self.host}:{self.port}",
+ ]
+ if self.bios:
+ cmd += ["-bios", self.bios]
+ if paused:
+ cmd += ["-S"]
+ if self.kernel:
+ cmd += ["-kernel", self.kernel]
+ for index, (addr, blob) in enumerate(self.segments):
+ seg_path = os.path.join(self.workdir, f"segment-{index:02d}-{addr:#x}.bin")
+ with open(seg_path, "wb") as fh:
+ fh.write(blob)
+ self._segment_files.append(seg_path)
+ cmd += ["-device", f"loader,file={seg_path},addr={addr:#x},force-raw=on"]
+ if self.dlog:
+ cmd += ["-d", self.dlog, "-D", self.qemu_log]
+ cmd += self.extra_args
+ return cmd
+
+ def start(self, paused: bool = True, timeout: float = 15.0) -> "QemuTarget":
+ self._cmd = self._build_command(paused)
+ try:
+ self.process = subprocess.Popen(
+ self._cmd,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.PIPE,
+ cwd=self.workdir,
+ start_new_session=True,
+ )
+ except OSError as exc:
+ raise TargetError(f"cannot start QEMU: {exc}") from exc
+ try:
+ self.wait_for_port(timeout=timeout)
+ except TargetError:
+ self.kill()
+ raise
+ return self
+
+ def wait_for_port(self, timeout: float = 15.0) -> None:
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ if self.process is not None and self.process.poll() is not None:
+ stderr = b""
+ if self.process.stderr is not None:
+ try:
+ stderr = self.process.stderr.read() or b""
+ except OSError:
+ pass
+ raise TargetError(
+ f"QEMU exited early rc={self.process.returncode}: "
+ f"{stderr.decode('utf-8', 'replace')[-2000:]}"
+ )
+ try:
+ with socket.create_connection((self.host, self.port), timeout=0.25):
+ return
+ except OSError:
+ time.sleep(0.05)
+ raise TargetError(f"gdbstub port {self.port} not ready after {timeout}s")
+
+ def alive(self) -> bool:
+ return self.process is not None and self.process.poll() is None
+
+ def kill(self, timeout: float = 5.0) -> None:
+ if self.process is None:
+ return
+ if self.process.poll() is None:
+ try:
+ os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
+ except (ProcessLookupError, PermissionError):
+ try:
+ self.process.terminate()
+ except ProcessLookupError:
+ pass
+ try:
+ self.process.wait(timeout=timeout)
+ except subprocess.TimeoutExpired:
+ try:
+ os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
+ except (ProcessLookupError, PermissionError):
+ try:
+ self.process.kill()
+ except ProcessLookupError:
+ pass
+ self.process.wait(timeout=timeout)
+ if self.process.stderr is not None:
+ try:
+ self.process.stderr.close()
+ except OSError:
+ pass
+ self.process = None
+
+ # -- observation -------------------------------------------------------
+
+ def read_uart(self) -> str:
+ try:
+ with open(self.uart_log, "rb") as fh:
+ return fh.read().decode("utf-8", "replace")
+ except FileNotFoundError:
+ return ""
+
+ def wait_for_uart(self, marker: str, timeout: float = 5.0, interval: float = 0.05) -> bool:
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ if marker in self.read_uart():
+ return True
+ time.sleep(interval)
+ return False
+
+ def read_qemu_log(self) -> str:
+ try:
+ with open(self.qemu_log, "rb") as fh:
+ return fh.read().decode("utf-8", "replace")
+ except FileNotFoundError:
+ return ""
+
+ @property
+ def command_line(self) -> list[str]:
+ return list(self._cmd)
+
+ def shell_command(self) -> str:
+ import shlex
+
+ return " ".join(shlex.quote(part) for part in self._cmd)
+
+ def __repr__(self) -> str: # pragma: no cover - cosmetic
+ return f"QemuTarget(port={self.port}, alive={self.alive()})"
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-12 08:57:23 +0000