summaryrefslogtreecommitdiffstats
path: root/debug/risc-v-gsp-rm-vm/greeneye/probes.py
diff options
context:
space:
mode:
authorArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-02 08:36:35 -0400
committerArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-02 08:36:35 -0400
commitba58dcdc5ad6ca021c723d279f33e3f14e1184df (patch)
tree47e43a2dad9a2d4bb40da46243654bff8b1302de /debug/risc-v-gsp-rm-vm/greeneye/probes.py
parentfe485080e159c983242d699f4a584659df159818 (diff)
downloadopen-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.py166
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}>"