diff options
| author | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-10-02 08:36:35 -0400 |
|---|---|---|
| committer | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-10-02 08:36:35 -0400 |
| commit | ba58dcdc5ad6ca021c723d279f33e3f14e1184df (patch) | |
| tree | 47e43a2dad9a2d4bb40da46243654bff8b1302de /debug/risc-v-gsp-rm-vm/greeneye/probes.py | |
| parent | fe485080e159c983242d699f4a584659df159818 (diff) | |
| download | open-iov-ba58dcdc5ad6ca021c723d279f33e3f14e1184df.tar.gz open-iov-ba58dcdc5ad6ca021c723d279f33e3f14e1184df.zip | |
debug: add green-eye GSP/uproc RISC-V VM runtime analysis tooling
Adds open-iov/debug/risc-v-gsp-rm-vm: a GDB-free instrumentation stack that boots uproc/GSP firmware in qemu-system-riscv64 and provides Frida-like hooks, breakpoints, watchpoints, memory dumps, structure decoding, snapshot diffs, recorded sessions, declarative probes, a plugin API and a Ghidra symbol/hit bridge.
Verified: 25 tests green (including a live QEMU smoke test); the demo uproc boots with correct hook/leave/watch/schema/snapshot/fault evidence; the optional TCG trace plugin builds and recorded 15M trace lines in 3 s; the TU104 570.144 fwimage probe reports its first exception (load access fault at firmware vaddr 0x4009014 reading 0x2000000000000A00).
Documentation: lean tool docs under debug/risc-v-gsp-rm-vm/docs/00-10, a new docs/05-operations/03-risc-v-gsp-rm-vm.md debugging document, and '# TOOL IN DEVELOPMENT - NOT READY YET!' markers in docs/README.md, the onboarding tooling table and the debugging workflows doc.
Vendors qemu-plugin.h from QEMU v8.2.2 (GPL-2.0-or-later), with its license header intact, so the optional TCG plugin builds without a QEMU source tree.
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/greeneye/probes.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/probes.py | 166 |
1 files changed, 166 insertions, 0 deletions
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}>" |
