diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/value_demos')
14 files changed, 1956 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/README.md b/debug/risc-v-gsp-rm-vm/harness/value_demos/README.md new file mode 100644 index 0000000..e1a450c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/README.md @@ -0,0 +1,83 @@ +# green-eye value-validation demos + +Runnable demonstrations that tie green-eye tool capabilities to **real** +Open-IOV scenarios (cited from `Handoff-vGPU-IO.md` and the project docs) and +measure the time-to-value against the static/Ghidra-only baseline. + +Each demo: + +* names the real scenario (handoff PART/addendum + date) and what it cost; +* prints the question, the green-eye commands used, the answer, the evidence + artifact and the elapsed wall-clock; +* writes a machine-readable `result.json` under `artifacts/<demo>/`; +* labels what is **measured on the analog** vs **projected for the real + case**; device-model-blocked scenarios say so explicitly. + +> green-eye is development-only tooling (docs/00); these demos never touch +> hardware, a guest or a kernel module. The real `.fwimage` cannot boot +> without the H2 device model (docs/06 §3), so every demo is honest about +> being a *shape* reproduction on the bundled `uproc_demo` firmware + native +> RV64 simulator / QEMU. + +## Run + +From the tool root (`open-iov/debug/risc-v-gsp-rm-vm`): + +```bash +make demo # once: build firmware/uproc_demo/build/uproc_demo.elf +make sdk # optional (s06): builds sdk/build/injected_demo.elf + +python3 harness/value_demos/run_all.py # all 10, results JSON +python3 harness/value_demos/run_all.py --only s01,s04,s05 # subset +python3 harness/value_demos/s01_reply_consumer.py # one demo, verbose + +VALUE_DEMO_BACKEND=qemu python3 harness/value_demos/run_all.py # force QEMU +``` + +Defaults: native backend (deterministic, in-process) for s01–s08; QEMU for +s09/s10 (with a native fallback if QEMU is absent). `VALUE_DEMO_BACKEND` +overrides all of them. No hardware is required. + +The aggregate lands in `artifacts/results.json`; each demo writes under +`artifacts/<name>/` (results, sessions, TSVs, reports). The runner enforces a +per-demo timeout and the full suite runs in ~18 s on the reference host +(budget: each demo < ~30 s, suite < ~3 min). + +## Demos + +| # | Script | Real scenario (source, date) | Static/Ghidra-only baseline | Analog answer | +|---|---|---|---|---| +| s01 | `s01_reply_consumer.py` | 0x59 RM-init chase; fn72/fn73 hidden error (PART 91 + ADDENDA 2–7, 2026-10-01) | cycles 2–13, ~5 h, 10+ supervised boots; reply was a generic `0x59` | hook dispatcher + watch `result`: every command consumed and its exact reply decoded (1.14 s) | +| s02 | `s02_counter_publication.py` | 20 GR per-channel counters; only 4/20 moved (PARTs 46–64, 2026-09-24…29) | ~5 days; slot layout re-derived per boot; ~2–3 s pin vs ~2 s deadline | 20/20 slots published, each store PC captured (1.34 s) | +| s03 | `s03_unknown_command.py` | unknown/new control discovery; injected slots 16–23 (docs/19, 2026-10-02) | read the indirect dispatch table per stripped image | 24 IDs swept + classified: 6 known, 8 injected-slot, 10 default (5.96 s) | +| s04 | `s04_run_until_function.py` | RM-client ladder reachability per rung (PART 91 ADDENDUM 9 cycles 18–25, 2026-10-01) | ~1 h 41 m / 8 supervised cycles | natural run / timeout / trigger / direct call in one session (0.24 s) | +| s05 | `s05_snapshot_diff.py` | fn72 identity fix + 1656-B static-config reply (PART 91 ADDENDUM 6/7, 2026-10-01) | malformed field suspected statically; a live boot per fix iteration | every changed byte attributed to a named schema field (0.50 s) | +| s06 | `s06_inject_service.py` | P6 service wiring / guest CE copies (PART 92 ADDENDA 5–11, 2026-10-02) | kernel rebuild + supervised cycle (~5–8 min) per rung | service blob loaded, registered, invoked both routes (0.63 s) | +| s07 | `s07_version_diff.py` | 570.144 vs 570.211 pairing disproof; semantic diff (PART 91 cycles 2–3, 2026-10-01) | one full boot per version + manual log compare | same stimulus on two builds; exactly one reply field moved (1.14 s) | +| s08 | `s08_mutation_triage.py` | error-path corpus / fail-closed rules (R5; docs/19 §3.4, docs/18 §2) | hand-derived asserts and NOCAT records across cycles | 7 seeded mutations classified ok/fault/no-progress (1.82 s) | +| s09 | `s09_grabbing_a_ghidra_address.py` | static/dynamic join; GPFIFO GET fault `0x40000` vs `0x101000` (PART 91 ADDENDUM 9 cycle 24) | hypothesis unproven until the decoded status-queue fault (8 cycles) | Ghidra addr + bias = runtime; hook hit + `greeneye_hits.tsv` (0.72 s) | +| s10 | `s10_full_session_evidence.py` | evidence pack for a milestone (docs/14 §6; cycles 11–13, 2026-10-01) | hashes/sessions/logs/compare gathered by hand per cycle | one run → analyze + compare + hits + manifest + M1–M6 (4.92 s) | + +Times are the measured green-eye path on the reference host (native for +s01–s08, QEMU for s09/s10); see `artifacts/results.json` for a fresh run. + +## Projected vs measured + +* **Measured:** every analog result, elapsed time, session size, trace + slowdown and slot/reply count in `result.json`. +* **Projected (labelled `projected` in each result):** the real-case time + saving. No demo claims a real-GSP result the simulator cannot reach. +* **Blocked (`blocked_on_device_model: true`):** all real-firmware claims + wait on the H2.2 device model; the workflow is demonstrated on the analog. + +## Tests + +```bash +python3 -m unittest tests.test_value_demos -v # 5 tests, ~5 s +``` + +The suite runs the fastest demos end-to-end as subprocesses, asserts their +answers and checks `run_all.py` emits `artifacts/results.json`. It skips +cleanly when QEMU or the demo firmware is absent. See +[`docs/25-VALUE-VALIDATION.md`](../../docs/25-VALUE-VALIDATION.md) for the +methodology, overhead analysis and scenario representativeness argument. diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/__init__.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/__init__.py new file mode 100644 index 0000000..87cabe7 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/__init__.py @@ -0,0 +1 @@ +"""green-eye value-validation demos (see docs/25-VALUE-VALIDATION.md).""" diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py new file mode 100644 index 0000000..f70ef1f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +"""Shared plumbing for the green-eye value-validation demos. + +Every demo maps one *real* Open-IOV scenario (cited from the handoffs) onto a +runnable demonstration on the bundled demo firmware / native RV64 simulator, +prints the question, the green-eye commands, the answer, the evidence artifact +and the measured wall-clock, and emits one machine-readable ``RESULT_JSON`` +line so ``run_all.py`` can aggregate. + +Stdlib only; no hardware; runs from the tool root +(``open-iov/debug/risc-v-gsp-rm-vm``). +""" +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.dirname(os.path.dirname(HERE)) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +ARTIFACT_ROOT = os.path.join(HERE, "artifacts") +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +SCHEMA_JSON = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "gsp_ctrl.schema.json") +SDK_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf") + +CTRL_MAGIC = 0x47535031 +CTRL_RPC_COUNT = 0x10 +CTRL_LAST_RPC = 0x18 +CTRL_MAILBOX = 0x20 +CTRL_ARG0 = 0x28 +CTRL_ARG1 = 0x30 +CTRL_ARG2 = 0x38 +CTRL_RESULT = 0x40 +CTRL_BUF = 0x48 +CTRL_SIZE = 0x190 +PING_XOR = 0xBEEF +INJECT_CMD_BASE = 16 +SERVICE_SLOT = 6 + +#: Per-demo wall-clock ceiling the runner enforces (subprocess timeout). +DEMO_TIMEOUT_S = 45.0 +#: Runner ceiling so the whole suite stays well under ~3 min. +SUITE_BUDGET_S = 180.0 + + +class DemoError(RuntimeError): + """A demo precondition failed in a way the caller should report, not mask.""" + + +def qemu_available() -> bool: + return bool(shutil.which("qemu-system-riscv64")) + + +def demo_available() -> bool: + return os.path.isfile(DEMO_ELF) + + +def choose_backend(preferred: str = "native") -> str: + """Environment override wins; QEMU is used only when preferred and present.""" + env = os.environ.get("VALUE_DEMO_BACKEND") + if env: + return env + if preferred == "qemu": + return "qemu" if qemu_available() else "native" + return "native" + + +def sha256_file(path: str) -> str: + digest = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +class Stopwatch: + def __init__(self) -> None: + self.t0 = time.monotonic() + self.elapsed = 0.0 + + def stop(self) -> float: + self.elapsed = time.monotonic() - self.t0 + return self.elapsed + + +def artifact_dir(name: str) -> str: + path = os.path.join(ARTIFACT_ROOT, name) + os.makedirs(path, exist_ok=True) + return path + + +def write_json(path: str, payload) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + fh.write("\n") + return path + + +def write_text(path: str, text: str) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +def load_schema(ge): + with open(SCHEMA_JSON, "r", encoding="utf-8") as fh: + return ge.schema_from_json(json.load(fh)) + + +def decode_ctrl(ge, data: bytes, schema=None) -> dict: + schema = schema or load_schema(ge) + out = schema.decode(data, base=0) + if isinstance(out.get("buf"), (bytes, bytearray)): + out["buf"] = bytes(out["buf"]).split(b"\x00", 1)[0].decode("utf-8", "replace") + return out + + +def short_hex(value) -> str: + if value is None: + return "None" + if isinstance(value, str): + return value + return f"{value:#x}" if isinstance(value, int) else str(value) + + +def render_banner(demo: str, question: str, scenario: str, baseline: str, commands) -> None: + print(f"=== {demo} ===") + print(f"real scenario : {scenario}") + print(f"baseline : {baseline}") + print(f"question : {question}") + print("commands :") + for command in commands: + print(f" - {command}") + + +def finish(result: dict, artifact: str, elapsed: float) -> dict: + """Persist the per-demo result and print the human + machine summary.""" + result["elapsed_s"] = round(elapsed, 4) + result["artifact"] = os.path.relpath(artifact, TOOL_ROOT) if artifact else None + result.setdefault("success", True) + if artifact: + write_json(os.path.join(artifact, "result.json"), result) + print(f"answer : {result.get('answer')}") + print(f"evidence : {result.get('artifact')}") + print(f"analog time : {elapsed:.3f} s ({result.get('backend')} backend)") + if result.get("real_baseline"): + print(f"real baseline : {result['real_baseline']}") + if result.get("projected"): + print(f"projected : {result['projected']}") + if result.get("blocked_on_device_model"): + print("blocked : real case blocked on H2 device model; workflow shown on analog") + if not result.get("success", True): + print(f"status : FAILED - {result.get('failure', 'unknown')}") + print("RESULT_JSON: " + json.dumps(result, sort_keys=True)) + sys.stdout.flush() + return result + + +def timed_demo(fn, artifact: str) -> dict: + watch = Stopwatch() + try: + result = fn() + except Exception as exc: # report, never hide + import traceback + + result = { + "success": False, + "failure": f"{type(exc).__name__}: {exc}", + "traceback": traceback.format_exc()[-2000:], + } + return finish(result, artifact, watch.stop()) + + +def scenario( + *, + ids, + baseline, + projection, + blocked=False, + device=None, +) -> dict: + return { + "citation_ids": ids, + "static_baseline": baseline, + "projection": projection, + "blocked_on_device_model": blocked, + "device_model_needed": device, + } diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/run_all.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/run_all.py new file mode 100644 index 0000000..808c227 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/run_all.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Run every value-validation demo and emit one machine-readable results JSON. + +Usage (from the tool root, ``open-iov/debug/risc-v-gsp-rm-vm``): + + python3 harness/value_demos/run_all.py + python3 harness/value_demos/run_all.py --only s01,s03,s05,s09 + VALUE_DEMO_BACKEND=qemu python3 harness/value_demos/run_all.py + +Each demo runs in its own subprocess with a hard timeout, so one wedged demo +cannot take the suite down. The aggregate lands in +``harness/value_demos/artifacts/results.json``. +""" +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.dirname(os.path.dirname(HERE)) +sys.path.insert(0, HERE) + +from _common import ARTIFACT_ROOT, SUITE_BUDGET_S, write_json # noqa: E402 + +DEMOS = [ + ("s01", "s01_reply_consumer.py"), + ("s02", "s02_counter_publication.py"), + ("s03", "s03_unknown_command.py"), + ("s04", "s04_run_until_function.py"), + ("s05", "s05_snapshot_diff.py"), + ("s06", "s06_inject_service.py"), + ("s07", "s07_version_diff.py"), + ("s08", "s08_mutation_triage.py"), + ("s09", "s09_grabbing_a_ghidra_address.py"), + ("s10", "s10_full_session_evidence.py"), +] + + +def _run_one(key: str, script: str, timeout: float) -> dict: + path = os.path.join(HERE, script) + start = time.monotonic() + try: + proc = subprocess.run( + [sys.executable, path], + cwd=TOOL_ROOT, + capture_output=True, + text=True, + timeout=timeout, + ) + stdout, stderr, returncode = proc.stdout, proc.stderr, proc.returncode + except subprocess.TimeoutExpired as exc: + stdout = (exc.stdout or b"").decode("utf-8", "replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "") + stderr = (exc.stderr or b"").decode("utf-8", "replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "") + returncode = -9 + elapsed = time.monotonic() - start + result = None + for line in reversed(stdout.splitlines()): + if line.startswith("RESULT_JSON: "): + try: + result = json.loads(line[len("RESULT_JSON: ") :]) + except json.JSONDecodeError: + result = None + break + if result is None: + result = { + "success": False, + "failure": f"no RESULT_JSON (exit {returncode})", + "stderr_tail": stderr[-800:], + } + result["demo_key"] = key + result["runner_elapsed_s"] = round(elapsed, 4) + result["exit_code"] = returncode + return result + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--only", default="", help="comma-separated demo keys (e.g. s01,s03)") + parser.add_argument("--timeout", type=float, default=45.0, help="per-demo timeout seconds") + parser.add_argument("--out", default=os.path.join(ARTIFACT_ROOT, "results.json")) + args = parser.parse_args() + + selected = [item for item in DEMOS if not args.only or item[0] in args.only.split(",")] + if args.only and not selected: + print(f"no demos match --only {args.only!r}", file=sys.stderr) + return 2 + + started = time.monotonic() + results = [] + for key, script in selected: + print(f"--- running {script} ---", flush=True) + result = _run_one(key, script, args.timeout) + results.append(result) + status = "PASS" if result.get("success") else "FAIL" + print(f"--- {key}: {status} ({result['runner_elapsed_s']:.2f}s) ---", flush=True) + if time.monotonic() - started > SUITE_BUDGET_S: + print(f"[run_all] suite budget {SUITE_BUDGET_S:.0f}s exceeded; continuing", flush=True) + + total = time.monotonic() - started + passed = [r for r in results if r.get("success")] + payload = { + "format": "greeneye-value-demos-results/1", + "tool": "harness/value_demos/run_all.py", + "backend_env": os.environ.get("VALUE_DEMO_BACKEND", "native (default)"), + "demos_selected": [k for k, _ in selected], + "demos_passed": [r["demo_key"] for r in passed], + "demos_failed": [r["demo_key"] for r in results if not r.get("success")], + "total_elapsed_s": round(total, 3), + "measured_elapsed_s": round(sum(r["runner_elapsed_s"] for r in results), 3), + "results": results, + } + write_json(args.out, payload) + print(f"\n{len(passed)}/{len(results)} demos passed in {total:.1f}s; results: {args.out}") + print(f"RESULT_JSON: {json.dumps({k: payload[k] for k in ('format', 'demos_passed', 'demos_failed', 'total_elapsed_s')}, sort_keys=True)}") + return 0 if len(passed) == len(results) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s01_reply_consumer.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s01_reply_consumer.py new file mode 100644 index 0000000..b86f272 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s01_reply_consumer.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +"""s01 - reply consumer: did the server consume the command, and what came back? + +Real scenario (Handoff-vGPU-IO.md, PART 91 + ADDENDA 2-7, 2026-10-01): +the physical backend booted the real GSP-RM and sent the pre-boot +``GSP_SET_SYSTEM_INFO`` (fn72) / ``SET_REGISTRY`` (fn73) RPCs; the only reply +the host could see was a generic ``GSP_INIT_DONE rpc_result=0x59`` +(``NV_ERR_OPERATING_SYSTEM``) wrapped by ``RmStartup``. Cycles 2-13 +(~11:40-16:52, ten supervised hardware cycles) were spent chasing which +collector step consumed which field because the RM log ring was never mapped +(LOGRM id bug) and the failure was un-journaled. Static/Ghidra-only work +(REPORT3-10) produced hypotheses (display, devid, chipset, ACPI) that needed a +live ``0x59``/success reply to accept or reject. + +Analog: hook the demo dispatcher, watch the control block's ``result`` field, +post known/unknown/fault commands and decode the exact reply per command. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def run() -> dict: + import greeneye as ge + + backend = C.choose_backend("native") + out = C.artifact_dir("s01_reply_consumer") + commands = [ + 'ge.launch(kernel=DEMO_ELF, backend="%s")' % backend, + 's.hook("rpc_dispatch", on_enter=record cmd/args)', + "s.watch(gsp_ctrl+0x40, size=8, kind='w') # result", + "s.memory.write_u64(ctrl+0x20, cmd); s.run(duration=...) # mailbox post", + "s.memory.read_u64(ctrl+0x40) # decode reply", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 91 (2026-10-01) cycles 2-3", + "PART 91 ADDENDUM 2-7 (2026-10-01): NOCAT, fn72/fn73, LOGRM id, identity", + ], + baseline=( + "cycles 2-13, ~5 h wall-clock and 10+ supervised hardware boots; " + "each candidate fix cost one full boot cycle (~5-6 min) and the " + "only visible reply was a generic 0x59" + ), + projection=( + "real GSP-RM reply consumption is blocked on the H2 device model; " + "on the analog the same observe->decode step is one hook + one " + "watchpoint per command" + ), + blocked=True, + device="H2.2 QEMU device model (RM message loop / real .fwimage)", + ) + C.render_banner( + "s01_reply_consumer", + "When the host posts a command, did the dispatcher consume it and what exact reply came back?", + "0x59 RM-init chase: generic reply hid which collector step consumed which field", + scenario["static_baseline"], + commands, + ) + + cases = [ + {"cmd": 1, "arg0": 0x1234, "label": "PING"}, + {"cmd": 7, "arg0": 0, "label": "unknown-id"}, + {"cmd": 6, "arg0": 0, "label": "FAULT"}, + ] + replies = [] + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s01") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + dispatches = [] + stores = [] + session.hook( + "rpc_dispatch", + on_enter=lambda hit: dispatches.append( + {"cmd": hit.arg(0), "arg0": hit.arg(1), "arg1": hit.arg(2), "arg2": hit.arg(3)} + ), + ) + session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w", callback=lambda hit: stores.append(hit.pc)) + session.run(duration=0.3) + for case in cases: + before_count = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + session.memory.write_u64(ctrl + C.CTRL_ARG0, case["arg0"]) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, case["cmd"]) + session.run(duration=0.25) + after_count = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + result = session.memory.read_u64(ctrl + C.CTRL_RESULT) + expected = { + 1: (0x1234 ^ C.PING_XOR), + 7: (~7) & 0xFFFFFFFFFFFFFFFF, + 6: None, + }[case["cmd"]] + replies.append( + { + "label": case["label"], + "cmd": case["cmd"], + "arg0": case["arg0"], + "consumed": after_count > before_count, + "result": result, + "expected": expected, + "result_ok": (expected is None) or (result == expected), + "rpc_count_delta": after_count - before_count, + } + ) + if case["cmd"] == 6: + # ebreak trap: the dispatcher's post-handler stores never run. + session.step(1) + known = replies[0] + unknown = replies[1] + fault = replies[2] + payload = { + "success": known["result_ok"] and unknown["result_ok"] and not fault["consumed"], + "backend": backend, + "firmware": os.path.relpath(C.DEMO_ELF, C.TOOL_ROOT), + "firmware_sha256": C.sha256_file(C.DEMO_ELF), + "dispatches": dispatches, + "reply_store_pcs": stores, + "replies": replies, + "question": "consumed and decoded every posted command", + "answer": ( + f"PING consumed, reply {known['result']:#x} == arg0 ^ 0xbeef " + f"(expected {known['expected']:#x}); unknown id returns ~cmd " + f"({unknown['result']:#x}); FAULT consumed={fault['consumed']} " + f"(trap before the reply store) - the exact first-error visibility " + "gap the 0x59 chase lacked" + ), + "success_criteria": { + "pass": "3/3 dispatches observed; PING/unknown replies match the ABI; FAULT traps before result", + "fail": "a posted command is not consumed or a reply disagrees with the ABI", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "reply.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s01_reply_consumer"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s02_counter_publication.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s02_counter_publication.py new file mode 100644 index 0000000..690f60f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s02_counter_publication.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""s02 - counter publication: which of N slots advanced, and when? + +Real scenario (Handoff-vGPU-IO.md PART 46-64, 2026-09-24..2026-09-29): +``cuCtxCreate`` waits on 20 GR per-channel completion counters. The host saw +the client's releases but could not translate the counter page; for many boots +only 4 of 20 client-visible slots ever moved (PART 49). The exact slot layout +(page VA ``0x20400f000``, slots at ``+0xec0``, 16-byte stride) was only proven +by a raw live page dump (PART 64, 2026-09-29 ~16:55), and the host raced a +~2-3 s pin path against the client's ~2 s GR-phase deadline. Static reading +could not tell which slots were published: the page's GPA changed every boot +and stale translations pointed at the previous client's page (PART 51). + +Analog: post 20 sequential commands and watch the demo's published result +field; prove every slot was observed, with the publish PC recorded per slot. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +SLOTS = 20 + + +def run() -> dict: + import greeneye as ge + + backend = C.choose_backend("native") + out = C.artifact_dir("s02_counter_publication") + commands = [ + 'ge.launch(kernel=DEMO_ELF, backend="%s")' % backend, + "s.watch(gsp_ctrl+0x40, size=8, kind='w', callback=record pc) # publication", + "for slot in 1..20: write mailbox cmd=5 (STATE); run(duration=0.05)", + "read gsp_ctrl.result after every slot; compare against the expected stage value", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 49/50 (2026-09-29): only 4/20 counters moved; all 20 reached 1 after the queue fix", + "PART 64 (2026-09-29 ~16:55): raw page dump proves 20 slots at 16-byte stride", + "PART 46/48 (2026-09-29): rescue/sweep latency and tracker waits", + ], + baseline=( + "PARTs 46->64: ~5 days of sessions; PART 49 'only 4 of 20 counters " + "ever moved'; each boot re-derived the slot layout because the " + "counter page GPA changed (PART 51/53)" + ), + projection=( + "on real silicon the 20 slots are per-channel memory pages; the " + "analog publishes 20 logical slots through the same watchpoint " + "shape, one store each" + ), + blocked=True, + device="H2.2 device model + real GSP-RM counter pages", + ) + C.render_banner( + "s02_counter_publication", + "Of the N client-visible slots, which ones advanced, and where was each published?", + "20 GR per-channel completion counters: 4/20 published for days until a live page dump", + scenario["static_baseline"], + commands, + ) + + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s02") + slots = [] + try: + ctrl = session.symbols.lookup("gsp_ctrl") + publishes = [] + session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w", callback=lambda hit: publishes.append(hit.pc)) + session.run(duration=0.25) + before = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + for slot in range(SLOTS): + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 5) # STATE + session.run(duration=0.05) + value = session.memory.read_u64(ctrl + C.CTRL_RESULT) + slots.append( + { + "slot": slot + 1, + "result": value, + "result_hex": f"{value:#x}", + "published": value in (0x10, 0x20, 0x30, 0x40), + } + ) + after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + published = [s for s in slots if s["published"]] + payload = { + "success": len(published) == SLOTS and (after - before) == SLOTS, + "backend": backend, + "slots_requested": SLOTS, + "slots_published": len(published), + "rpc_count_delta": after - before, + "watch_stores": len(publishes), + "distinct_values": sorted({s["result_hex"] for s in published}), + "per_slot": slots, + "publish_pcs": sorted(set(publishes)), + "question": f"did all {SLOTS} slots publish, and can each publication be attributed?", + "answer": ( + f"{len(published)}/{SLOTS} slots published ({after - before} dispatcher " + f"completions, {len(publishes)} stores to gsp_ctrl.result, values " + f"{sorted({s['result_hex'] for s in published})}); every store PC is captured, " + "exactly the evidence class that took a live 512-byte page dump on the real path" + ), + "success_criteria": { + "pass": f"{SLOTS}/{SLOTS} slots observed with rpc_count advancing once per slot", + "fail": "any slot does not publish or the count delta disagrees", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "counters.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s02_counter_publication"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s03_unknown_command.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s03_unknown_command.py new file mode 100644 index 0000000..162bc47 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s03_unknown_command.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""s03 - unknown command discovery: which IDs are handled, injected, or new? + +Real scenario (docs/19-CONTROL-DISCOVERY.md, measured 2026-10-02): a swept +input surface is how new/undocumented controls are found between firmware +versions. On the demo, IDs 16..23 were *discovered* to fall through the +switch into the injected-service handler table (16..21 alias the built-in +handlers, 22..23 are NULL slots returning ``~id``) - wiring a new image can +add silently. Statically this requires reading an indirect dispatch table +per version; behaviorally it is a bounded sweep. + +Analog: 24 command IDs, zero arguments, deterministic, fresh session per ID +(the same safety strategy as the real discovery tool), classified by hooks on +the dispatcher and every handler-table entry. +""" +from __future__ import annotations + +import json +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +HANDLERS = ("rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault") + + +def _classify(cmd: int, handler_hits, dispatch_seen, result, trap) -> str: + if 16 <= cmd <= 23: + return "injected-slot" + if 1 <= cmd <= 6: + return "known-handler" + if trap: + return "trap" + if result is not None and result == (~cmd) & 0xFFFFFFFFFFFFFFFF: + return "default-path" + return "no-progress" + + +def run() -> dict: + import greeneye as ge + + backend = C.choose_backend("native") + out = C.artifact_dir("s03_unknown_command") + commands = [ + 'for cmd in 1..24: ge.launch(backend="%s"); hook(rpc_dispatch + 6 handlers)' % backend, + "write mailbox cmd with zero args; run bounded; classify", + "record: outcome, handler alias, result, trap signal, rpc_count delta", + ] + scenario = C.scenario( + ids=[ + "docs/19-CONTROL-DISCOVERY.md (2026-10-02): 189-probe demo sweep in ~36 s", + "Handoff-vGPU-IO.md PART 91 (2026-10-01): new controls appear between images (0x59 era)", + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 4.1: static diff says code changed, not which input activates it", + ], + baseline=( + "static-only: read the indirect dispatcher table in Ghidra for every " + "image; on the real corpus the demo sweep took ~36 s for 189 probes " + "and still found an undocumented injected-slot class" + ), + projection=( + "the RM surface (control IDs by class, method opcodes) is gated on " + "H1.3/H2.2; the sweep/classification shape demonstrated here is " + "exactly what docs/19 section 5 designs for it" + ), + blocked=True, + device="H1.3/H2.2: RM dispatch table exposed to the simulator", + ) + C.render_banner( + "s03_unknown_command", + "Which command IDs are handled, which route to injected slots, and which are new/default?", + "unknown-control discovery between stripped firmware versions", + scenario["static_baseline"], + commands, + ) + + findings = [] + for cmd in range(1, 25): + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name=f"vd-s03-{cmd:02d}") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + dispatch = [] + handler_hits = [] + signals = [] + session.hook( + "rpc_dispatch", + on_enter=lambda hit, d=dispatch: d.append( + {"cmd": hit.arg(0), "arg0": hit.arg(1), "arg1": hit.arg(2), "arg2": hit.arg(3)} + ), + ) + for name in HANDLERS: + try: + session.hook(name, on_enter=lambda hit, n=name: handler_hits.append(n)) + except Exception: + pass + session.on_event("stop", lambda hit, sig=signals: sig.append(hit.stop.signal) if hit.stop else None) + session.run(duration=0.15) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, cmd) + session.run(duration=0.15) + result = session.memory.read_u64(ctrl + C.CTRL_RESULT) + trap = 5 in [s for s in signals if s] + outcome = _classify(cmd, handler_hits, bool(dispatch), result, trap) + findings.append( + { + "cmd": cmd, + "outcome": outcome, + "trap": trap, + "signals": [s for s in signals if s], + "handler_alias": handler_hits[0] if handler_hits else None, + "dispatch_seen": bool(dispatch), + "result": result, + "result_hex": f"{result:#x}", + "expected_default": f"{(~cmd) & 0xFFFFFFFFFFFFFFFF:#x}", + } + ) + finally: + session.close() + + by_class: dict[str, list[int]] = {} + for finding in findings: + by_class.setdefault(finding["outcome"], []).append(finding["cmd"]) + injected = by_class.get("injected-slot", []) + aliases = {f["cmd"]: f["handler_alias"] for f in findings if f["cmd"] in injected} + payload = { + "success": len(injected) == 8 and len(by_class.get("known-handler", [])) == 6, + "backend": backend, + "ids_swept": len(findings), + "classes": {key: {"count": len(v), "ids": v} for key, v in sorted(by_class.items())}, + "injected_aliases": aliases, + "findings": findings, + "question": "enumerate handled / injected / default command space in one bounded sweep", + "answer": ( + f"{len(by_class.get('known-handler', []))} known handlers, " + f"{len(injected)} injected-slot IDs ({injected}), aliases {aliases}, " + f"{len(by_class.get('default-path', []))} default-path IDs; " + "the same shape (sweep + dispatcher/handler hooks) is what found the " + "silent 16..23 wiring the static map did not highlight" + ), + "success_criteria": { + "pass": "IDs 1..6 classified known, 16..23 classified injected-slot, remaining default; cmd 6 traps", + "fail": "an ID is misclassified or a fresh-session probe cannot be replayed", + }, + "real_case": scenario, + } + os.makedirs(out, exist_ok=True) + with open(os.path.join(out, "findings.jsonl"), "w", encoding="utf-8") as fh: + for finding in findings: + fh.write(json.dumps(finding, sort_keys=True) + "\n") + lines = ["# control discovery (analog)", "", f"* backend: {backend}", f"* ids swept: {len(findings)}", ""] + for key, value in sorted(by_class.items()): + lines.append(f"* {key}: {value}") + lines.append("") + C.write_text(os.path.join(out, "control-discovery.md"), "\n".join(lines) + "\n") + C.write_json(os.path.join(out, "discovery.json"), payload) + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s03_unknown_command"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s04_run_until_function.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s04_run_until_function.py new file mode 100644 index 0000000..a74c83f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s04_run_until_function.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +"""s04 - run-until a function: is this code on the live path, and with what args? + +Real scenario (docs/19-CONTROL-DISCOVERY.md 4.4 and +Handoff-vGPU-IO.md PART 91 ADDENDUM 9, 2026-10-01 cycles 18-25): the +``gsp-phys-rm-call`` RM-client ladder asked, per cycle, whether a symbolized +function had executed under the current stimulus (VAS commit, token, BIND, +fetch, semaphore). Cycles 18->25 (~1 h 41 m, eight supervised boots) walked +the ladder one reachability question at a time; ``reach.py`` answers that +question in ~0.5 s per target, and a ``run_until`` timeout means "not executed +under this stimulus at this fidelity", never "unreachable". + +Analog: run to the self-test's ``rpc_sum``, show ``rpc_fault`` is *not* on the +natural path (timeout), then trigger it through the mailbox and reach it with +the dispatch arguments captured; finally call ``rpc_ping`` directly +(Frida NativeFunction) to show the unit-test route. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def run() -> dict: + import greeneye as ge + from greeneye.errors import RspTimeout + + backend = C.choose_backend("native") + out = C.artifact_dir("s04_run_until_function") + commands = [ + 's.run_until("rpc_sum", timeout=5.0) # natural path', + 's.run_until("rpc_fault", timeout=0.2) # not on natural path', + 'write mailbox cmd=6; s.run_until("rpc_fault") # trigger route', + 's.call("rpc_ping", (0x41,)) # Frida-like direct call', + ] + scenario = C.scenario( + ids=[ + "docs/19-CONTROL-DISCOVERY.md 4.4 (2026-10-02): reach.py measured ~0.5 s per target", + "Handoff-vGPU-IO.md PART 91 ADDENDUM 9 (2026-10-01): RM-client ladder cycles 18-25", + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 5: run_until timeouts are not unreachability", + ], + baseline=( + "cycles 18-25 took ~1 h 41 m of supervised hardware cycles (full " + "boot + RPC ladder per hypothesis) to answer the same class of " + "'did this function execute under this stimulus' question" + ), + projection=( + "real RM ladder reachability needs the live GSP-RM (H1.3/H2.2); " + "run_until/hook/call already work against the re-hosted firmware " + "shape and every probe spec emitted here replays" + ), + blocked=True, + device="H1.3 live RM method entry / H2.2 device model", + ) + C.render_banner( + "s04_run_until_function", + "Is a symbolized function on the live path under this stimulus, what args does it take, and can it be called directly?", + "RM-client ladder: reachability of each rung, one supervised boot per question", + scenario["static_baseline"], + commands, + ) + + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s04") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + dispatch_args = [] + session.hook( + "rpc_dispatch", + on_enter=lambda hit: dispatch_args.append( + {"cmd": hit.arg(0), "a0": hit.arg(1), "a1": hit.arg(2), "a2": hit.arg(3)} + ), + ) + natural_hit = session.run_until("rpc_sum", timeout=5.0) + natural = { + "target": "rpc_sum", + "method": "run", + "reachable": True, + "pc": f"{natural_hit.addr:#x}", + "dispatch_args": list(dispatch_args), + } + timeout_hit = None + try: + timeout_hit = session.run_until("rpc_fault", timeout=0.2) + timeout_seen = False + except RspTimeout: + timeout_seen = True + fault_before = { + "target": "rpc_fault", + "method": "run_until", + "reachable": timeout_hit is not None, + "timeout": timeout_seen, + "reading": "not executed under this stimulus at this fidelity (docs/14 section 5)", + } + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 6) + trigger_hit = session.run_until("rpc_fault", timeout=5.0) + triggered = { + "target": "rpc_fault", + "method": "trigger (mailbox cmd=6)", + "reachable": True, + "pc": f"{trigger_hit.addr:#x}", + "dispatch_args": list(dispatch_args[-1:]), + } + direct = session.call("rpc_ping", (0x41,)) + payload = { + "success": natural["reachable"] and fault_before["timeout"] and triggered["reachable"] + and direct == (0x41 ^ C.PING_XOR), + "backend": backend, + "natural_path": natural, + "before_trigger": fault_before, + "after_trigger": triggered, + "direct_call": {"target": "rpc_ping", "args": [0x41], "result": direct, "expected": 0x41 ^ C.PING_XOR}, + "question": "reachability and required arguments per symbolized function", + "answer": ( + f"rpc_sum reached on the self-test path (pc {natural['pc']}); rpc_fault " + f"times out (reachable=False) until mailbox cmd=6 triggers it " + f"(pc {triggered['pc']}, args {triggered['dispatch_args']}); direct call " + f"rpc_ping(0x41)={direct:#x} - three reachability answers, one session" + ), + "success_criteria": { + "pass": "natural target reached, un-triggered target times out, trigger route reaches it, direct call matches", + "fail": "a route reports reachable without a hit, or the direct call disagrees", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "reach.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s04_run_until_function"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s05_snapshot_diff.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s05_snapshot_diff.py new file mode 100644 index 0000000..fd89131 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s05_snapshot_diff.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +"""s05 - snapshot diff: which control-block fields did one command change? + +Real scenario (Handoff-vGPU-IO.md PART 91 ADDENDUM 6, 2026-10-01 cycle 12/13): +the fn72 ``GSP_SET_SYSTEM_INFO`` payload carried *unpacked* identity fields and +the reference sends raw config-space dwords; the server read the device id +back as zero and only a field-level comparison of the live payload/reply +could confirm the fix ("all zeroed fn72 fields cannot matter to this failure" +was settled by decoding, not by reading). Static review (REVIEW-FN72.md, +cycle 13's 1656-byte static-config reply) still needed a live before/after of +the control block to call the fix verified. + +Analog: snapshot the decoded control block before and after one command, +attribute every changed byte run to its named schema field, and show ECHO's +buffer side-effect separately. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def _fields_hit(schema, diffs) -> list[str]: + names = [] + for diff in diffs: + for field in schema.fields: + start = field.offset + end = field.offset + field.nbytes + if start <= diff.addr < end: + if field.name not in names: + names.append(field.name) + return names + + +def run() -> dict: + import greeneye as ge + from greeneye.memory import Memory + + backend = C.choose_backend("native") + out = C.artifact_dir("s05_snapshot_diff") + commands = [ + 's.snapshot(gsp_ctrl, 0x188, "before")', + "write mailbox cmd=2 (ECHO) with a known string; run", + 's.snapshot(gsp_ctrl, 0x188, "after")', + "Memory.diff(before, after); decode both via gsp_ctrl.schema.json", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 91 ADDENDUM 6 cycle 12 (2026-10-01): unpacked fn72 identity fields", + "PART 91 ADDENDUM 7 (2026-10-01): 1656-byte static-config reply decoded field-by-field", + "docs/06-ACCURACY.md M6: memory side-effect agreement / snapshot hashes", + ], + baseline=( + "the malformed identity value was suspected statically (REVIEW-FN72) " + "but accepting the fix required a live boot per iteration; ten NOCAT " + "cycles ran before the field-level root cause was visible" + ), + projection=( + "the same before/after diff runs against any firmware region once " + "the H2 device model exposes the real RM control pages" + ), + blocked=True, + device="H2.2 RM control/reply pages for the real GSP-RM", + ) + C.render_banner( + "s05_snapshot_diff", + "Which named control-block fields did this one command change?", + "fn72/static-config field-level validation (identity fix, cycle 12/13)", + scenario["static_baseline"], + commands, + ) + + schema = C.load_schema(ge) + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s05") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + session.run(duration=0.25) + before = session.snapshot(ctrl, C.CTRL_SIZE, "before-echo") + message = b"hello-green-eye" + # Locate the string literal in the loaded image so the stimulus is + # self-describing (no hardcoded address). + hits = session.memory.search(0x80000000, 0x2000, message) + literal = hits[0] if hits else 0 + session.memory.write_u64(ctrl + C.CTRL_ARG0, literal) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 2) + session.run(duration=0.25) + after = session.snapshot(ctrl, C.CTRL_SIZE, "after-echo") + + before_dec = C.decode_ctrl(ge, before, schema) + after_dec = C.decode_ctrl(ge, after, schema) + diffs = Memory.diff(before, after) + changed_fields = _fields_hit(schema, diffs) + changed_names = [name for name in changed_fields if before_dec.get(name) != after_dec.get(name)] + rendered = Memory.render_diff(diffs, symbols=session.symbols) + payload = { + "success": bool(changed_names) + and set(changed_names) >= {"rpc_count", "last_rpc", "result", "buf"} + and after_dec.get("result") == len(message), + "backend": backend, + "changed_field_names": changed_names, + "changed_byte_runs": len(diffs), + "before": {k: v for k, v in before_dec.items() if k in changed_names}, + "after": {k: v for k, v in after_dec.items() if k in changed_names}, + "rendered_diff": rendered, + "question": "attribute every changed byte in one region to a named field", + "answer": ( + f"ECHO changed {len(changed_names)} named fields {changed_names}; " + f"buf={after_dec.get('buf')!r} result={after_dec.get('result')} " + f"(len={len(message)}); {len(diffs)} byte runs exactly covered by the schema - " + "the field-level 'what did the reply actually consume' check" + ), + "success_criteria": { + "pass": "rpc_count/last_rpc/result/buf change, result equals the ECHO length, no unattributed run", + "fail": "a region changes with no named field, or the ECHO reply length disagrees", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "snapshot-diff.json"), payload) + C.write_text( + os.path.join(out, "snapshot-diff.md"), + "# snapshot diff (analog)\n\n" + f"* changed fields: {changed_names}\n" + f"* changed byte runs: {len(diffs)}\n\n" + "```\n" + rendered + "\n```\n", + ) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s05_snapshot_diff"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s06_inject_service.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s06_inject_service.py new file mode 100644 index 0000000..04dd001 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s06_inject_service.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +"""s06 - inject a service: add an RM engine without rebuilding the firmware. + +Real scenario (Handoff-vGPU-IO.md PART 92 ADDENDUM 5-11, 2026-10-02): the P6 +work wired guest-originated CE copies through the real engine; every change to +the service/wiring path cost a kernel rebuild plus a supervised physical boot +cycle (P6a/b/c, ~5-8 min each, one device-write experiment per cycle) and was +only accepted after guest smoke ``ALL OK 64/64``. docs/12-INJECTION-SDK.md +defines the in-simulator equivalent: load a position-independent service blob +and register it with the firmware's dispatch table (commands 16..23). + +Analog: load ``sdk/build/injected_demo.elf`` into slot 6, invoke it directly +(Frida NativeFunction) and through the firmware mailbox, and record the +``service_injected`` evidence event. No kernel rebuild, no hardware. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def run() -> dict: + import greeneye as ge + from greeneye.inject import invoke_remote, invoke_via_firmware, load_service + + backend = C.choose_backend("native") + out = C.artifact_dir("s06_inject_service") + commands = [ + 'load_service(s, "sdk/build/injected_demo.elf", slot=6)', + "invoke_remote(s, svc, cmd=1, a0=0x41) # debugger-side call", + "invoke_via_firmware(s, svc, cmd=1, a0=0x41) # mailbox cmd=16", + "read gsp_ctrl.result / handlers[6]", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 92 ADDENDUM 5-11 (2026-10-02): P6 service/wiring rungs", + "PART 92 ADDENDUM 6 (2026-10-02): p6run7 first real engine copy after service wiring", + "docs/12-INJECTION-SDK.md: blob/ABI contract for injecting RM services", + ], + baseline=( + "each P6 rung was a kernel rebuild plus a supervised physical cycle " + "(~5-8 min boots; P6a->P6c spanned ~02:20-08:30) with fail-safe " + "release discipline after every run" + ), + projection=( + "on real GSP-RM the registry is the RM service table (M5), gated on " + "H2.2; the load/register/invoke/observe shape is identical to the " + "demo handler-table slot" + ), + blocked=True, + device="H2.2 device model + M5 RM service registry", + ) + C.render_banner( + "s06_inject_service", + "Can a new service be loaded, registered and invoked without rebuilding the firmware?", + "guest-originated CE service wiring (P6a-P6c): kernel rebuild + supervised boot per rung", + scenario["static_baseline"], + commands, + ) + + if not os.path.isfile(C.SDK_ELF): + payload = { + "success": False, + "skipped": True, + "failure": "sdk/build/injected_demo.elf not built (run: make sdk)", + "backend": backend, + "real_case": scenario, + } + C.write_json(os.path.join(out, "inject.json"), payload) + return payload + + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, record=os.path.join(out, "session"), name="vd-s06") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + session.run(duration=0.25) + service = load_service(session, C.SDK_ELF, slot=C.SERVICE_SLOT) + remote_ping = invoke_remote(session, service, 1, 0x41) + remote_sum = invoke_remote(session, service, 2, 0x10, 0x20) + via_fw_ping = invoke_via_firmware(session, service, 1, 0x41) + via_fw_sum = invoke_via_firmware(session, service, 2, 0x10, 0x20) + handlers7 = session.memory.read_u64(ctrl + 0x150 + 8 * C.SERVICE_SLOT) + payload = { + "success": remote_ping == (0x41 ^ C.PING_XOR) and remote_sum == 0x30 + and via_fw_ping == remote_ping and via_fw_sum == remote_sum, + "backend": backend, + "service": service.summary(), + "service_sha256": C.sha256_file(C.SDK_ELF), + "handlers_slot6": f"{handlers7:#x}", + "remote": {"ping": remote_ping, "sum": remote_sum}, + "via_firmware": {"ping": via_fw_ping, "sum": via_fw_sum}, + "mailbox_command": C.INJECT_CMD_BASE + C.SERVICE_SLOT, + "session": os.path.relpath(os.path.join(out, "session"), C.TOOL_ROOT), + "question": "load/register/invoke a service blob end-to-end, both routes", + "answer": ( + f"service loaded at {service.load_addr:#x}, slot {service.slot} " + f"(handlers[{service.slot}]={handlers7:#x}); remote ping={remote_ping:#x} " + f"sum={remote_sum:#x}; via mailbox cmd={C.INJECT_CMD_BASE + C.SERVICE_SLOT} " + f"ping={via_fw_ping:#x} sum={via_fw_sum:#x} - both paths agree, one session" + ), + "success_criteria": { + "pass": "remote and firmware-mailbox invocations return the SDK ABI values and agree", + "fail": "a route disagrees, the slot is not registered, or the blob cannot load", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "inject.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s06_inject_service"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s07_version_diff.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s07_version_diff.py new file mode 100644 index 0000000..e4748d7 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s07_version_diff.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python3 +"""s07 - version diff: same stimulus against two firmware versions. + +Real scenario (Handoff-vGPU-IO.md PART 91, 2026-10-01 cycles 2-3): the staged +570.211 image and a verified 570.144 image both failed ``GSP_INIT_DONE`` with +``rpc_result=0x59``; disproving the image-pairing hypothesis cost one full +supervised boot cycle per version, and docs/14 section 4.1 exists because +"the code changed" and "the behaviour moved" must be separated before the +emulation is touched. Sorting a new image into the decision table +(static changed x hit/write profile changed) is the H4.1 triage. + +Analog: byte-patch the demo's PING reply constant (0xbeef -> 0xbef0) to make a +labelled "new version", run the *same* recorded stimulus on both, and diff the +sessions and the decoded control block. The answer is one live-path field. +""" +from __future__ import annotations + +import os +import shutil + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +#: addiw a1, a1, -0x111 (the 0xbeef PING key) -> -0x110 (0xbef0) +ADDW_OLD = bytes.fromhex("9b85f5ee") +ADDW_NEW = bytes.fromhex("9b8505ef") +STRING_OLD = b"hello-green-eye" +STRING_NEW = b"hello-green-eyf" + + +def _make_variant(src: str, dst: str) -> dict: + with open(src, "rb") as fh: + data = bytearray(fh.read()) + count = data.count(ADDW_OLD) + if count == 1: + offset = data.find(ADDW_OLD) + old_bytes = bytes(data[offset : offset + 4]) + data[offset : offset + 4] = ADDW_NEW + kind = "ping-key-0xbeef->0xbef0" + else: + count = data.count(STRING_OLD) + if count != 1: + raise C.DemoError(f"no unique patch site (addiw x{data.count(ADDW_OLD)}, string x{count})") + offset = data.find(STRING_OLD) + old_bytes = bytes(data[offset : offset + len(STRING_OLD)]) + data[offset : offset + len(STRING_OLD)] = STRING_NEW + kind = "echo-string" + with open(dst, "wb") as fh: + fh.write(data) + return { + "kind": kind, + "offset": offset, + "old_bytes": old_bytes.hex(), + "source_sha256": C.sha256_file(src), + "variant_sha256": C.sha256_file(dst), + } + + +def run() -> dict: + import greeneye as ge + from greeneye.analyze import compare_sessions + + backend = C.choose_backend("native") + out = C.artifact_dir("s07_version_diff") + va = os.path.join(out, "uproc_demo_vA.elf") + vb = os.path.join(out, "uproc_demo_vB.elf") + shutil.copyfile(C.DEMO_ELF, va) + patch = _make_variant(C.DEMO_ELF, vb) + + commands = [ + "byte-patch PING key 0xbeef -> 0xbef0 to create version B", + 'for v in (A,B): s = ge.launch(v, backend="%s", record=...)' % backend, + "identical stimulus: mailbox cmd=1 arg0=0x1234; snapshot+decode gsp_ctrl", + "compare_sessions(A, B, symbols) + decoded field diff", + ] + scenario = C.scenario( + ids=[ + "Handoff-vGPU-IO.md PART 91 (2026-10-01) cycles 2-3: 570.211 vs 570.144 pairing disproven", + "PART 91 ADDENDUM 6 (2026-10-01): LOGRM id + fn72 identity changed behaviour between cycles", + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 4.1: version-diff decision table", + ], + baseline=( + "one full supervised boot (~5-6 min) per version plus manual log " + "comparison; the pairing hypothesis took cycles 2 and 3 back-to-back" + ), + projection=( + "green-eye cannot boot the real image without the H2 device model; " + "the same same-stimulus/compare-sessions loop runs on re-hosted " + "firmware and the produced symbol/session deltas are the H4.1 artifact" + ), + blocked=True, + device="H2.2 device model for real .fwimage versions", + ) + C.render_banner( + "s07_version_diff", + "On the same stimulus, which live-path behaviour moved between two firmware versions?", + "570.211 vs 570.144 image pairing; semantic diff before touching the emulation", + scenario["static_baseline"], + commands, + ) + + schema = C.load_schema(ge) + results = {} + for label, path in (("A", va), ("B", vb)): + session = ge.launch(kernel=path, backend=backend, record=os.path.join(out, f"session-{label}"), name=f"vd-s07-{label}") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + dispatches = [] + session.hook( + "rpc_dispatch", + on_enter=lambda hit, d=dispatches: d.append({"cmd": hit.arg(0), "arg0": hit.arg(1)}), + ) + session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w") + session.run(duration=0.25) + session.memory.write_u64(ctrl + C.CTRL_ARG0, 0x1234) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 1) + session.run(duration=0.25) + decoded = C.decode_ctrl(ge, session.snapshot(ctrl, C.CTRL_SIZE, f"ctrl-{label}"), schema) + session.note("version", label=label, patch=patch["kind"], reply=decoded.get("result")) + results[label] = { + "path": os.path.relpath(path, C.TOOL_ROOT), + "sha256": C.sha256_file(path), + "dispatches": dispatches, + "reply": decoded.get("result"), + "rpc_count": decoded.get("rpc_count"), + "last_rpc": decoded.get("last_rpc"), + } + finally: + session.close() + + compare_md = compare_sessions( + os.path.join(out, "session-A"), os.path.join(out, "session-B"), symbols=session.symbols, max_rows=20 + ) + C.write_text(os.path.join(out, "compare-sessions.md"), compare_md) + moved = sorted( + key + for key in ("reply", "rpc_count", "last_rpc") + if results["A"][key] != results["B"][key] + ) + payload = { + "success": results["A"]["reply"] == 0xACDB and results["B"]["reply"] == 0xACC4 and moved == ["reply"], + "backend": backend, + "patch": patch, + "version_A": results["A"], + "version_B": results["B"], + "moved_fields": moved, + "decision_table_row": { + "static_code_changed": True, + "live_profile_changed": bool(moved), + "reading": "new logic on a live path - prioritise (docs/14 4.1)", + }, + "compare_report": os.path.relpath(os.path.join(out, "compare-sessions.md"), C.TOOL_ROOT), + "question": "separate 'the code changed' from 'the behaviour moved' on a fixed stimulus", + "answer": ( + f"version A replies {results['A']['reply']:#x}, version B replies {results['B']['reply']:#x}; " + f"moved fields {moved}; the patch site is {patch['kind']} at file offset {patch['offset']:#x} - " + "the decision-table row is 'code changed + profile changed -> prioritise'" + ), + "success_criteria": { + "pass": "exactly the patched reply field moves on the live path; both sessions replay from the same stimulus", + "fail": "unrelated fields move or a version cannot be replayed", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "version-diff.json"), payload) + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s07_version_diff"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s08_mutation_triage.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s08_mutation_triage.py new file mode 100644 index 0000000..8fcaed1 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s08_mutation_triage.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""s08 - mutation triage: mutate one field, classify the outcome. + +Real scenario (docs/18-HARNESS-AUTOMATION.md R5 and +Handoff-vGPU-IO.md PART 77 / docs/19 3.4, 2026-09-30..2026-10-02): the +error-path corpus (sentinel/boundary pointers on ECHO/SUM, invalid command +IDs) is how the emulation's fail-closed rules are derived. Statically the +demo's 4-of-24 outcomes were invisible; the discovery sweep found +``rpc_echo``/``rpc_sum`` fault on sentinel/boundary pointers while +``rpc_ping``/``rpc_state``/``rpc_fill`` stay on the handler path. + +Analog: a seeded mutation table over command/arguments, one fresh session per +attempt (the documented safety strategy), classified as ok / trap / fault / +no-progress from stop signals, ``rpc_count`` and the result field. +""" +from __future__ import annotations + +import json +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +FAULT_SIGNALS = {4, 7, 11} # SIGILL, SIGBUS, SIGSEGV +TRAP_SIGNAL = 5 # SIGTRAP (ebreak) + + +def _probe(ge, mutation: dict, backend: str) -> dict: + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s08") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + signals = [] + session.on_event("stop", lambda hit, sig=signals: sig.append(hit.stop.signal) if hit.stop else None) + session.run(duration=0.15) + literal = 0 + if mutation.get("arg0") == "hello": + hits = session.memory.search(0x80000000, 0x2000, b"hello-green-eye") + literal = hits[0] if hits else ctrl + C.CTRL_BUF + before = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + session.memory.write_u64(ctrl + C.CTRL_ARG0, literal if mutation.get("arg0") == "hello" else mutation.get("arg0", 0)) + session.memory.write_u64(ctrl + C.CTRL_ARG1, mutation.get("arg1", 0)) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, mutation["cmd"]) + # max_stops=1: a faulted target keeps re-faulting at the same PC, so + # stop at the first signal instead of spinning the fault loop. + session.run(duration=0.2, max_stops=1) + after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + if after == before and not any(signals): + session.run(duration=0.3, max_stops=1) + after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + result = session.memory.read_u64(ctrl + C.CTRL_RESULT) + if any(s == TRAP_SIGNAL for s in signals): + outcome = "trap" + elif any(s in FAULT_SIGNALS for s in signals): + outcome = "fault" + elif after > before: + outcome = "ok" + else: + outcome = "no-progress" + return { + "outcome": outcome, + "rpc_count_delta": after - before, + "result": result, + "result_hex": f"{result:#x}", + "signals": [s for s in signals if s], + "consumed": after > before, + } + finally: + session.close() + + +def run() -> dict: + import greeneye as ge + + out = C.artifact_dir("s08_mutation_triage") + backend = C.choose_backend("native") + mutations = [ + {"strategy": "arg-boundary", "cmd": 2, "arg0": 0, "arg1": 0, "expect": "fault"}, + {"strategy": "valid", "cmd": 2, "arg0": "hello", "arg1": 0, "expect": "ok"}, + {"strategy": "arg-boundary", "cmd": 2, "arg0": 0xFFFFFFFFFFFFFFFF, "arg1": 0, "expect": "fault"}, + {"strategy": "valid", "cmd": 4, "arg0": "hello", "arg1": 5, "expect": "ok"}, + {"strategy": "buffer-length", "cmd": 4, "arg0": "hello", "arg1": 0x8000000000000000, "expect": "no-progress"}, + {"strategy": "buffer-length", "cmd": 3, "arg0": 0x41, "arg1": 4096, "expect": "ok"}, + {"strategy": "cmd-invalid", "cmd": 7, "arg0": 0, "arg1": 0, "expect": "ok"}, + ] + commands = [ + "for each seeded mutation: fresh native session, settle, one mailbox write", + "run bounded; classify from stop signals + rpc_count + result", + "write findings.jsonl (deterministic order) for corpus diffing", + ] + scenario = C.scenario( + ids=[ + "docs/18-HARNESS-AUTOMATION.md 2 (R5): bounded seeded mutation, outcome classes", + "docs/19-CONTROL-DISCOVERY.md 3.4 (2026-10-02): sentinel/boundary pointer faults found by the sweep", + "Handoff-vGPU-IO.md PARTs 3-9 (2026-09-21..23): hand-derived assert/error paths", + ], + baseline=( + "statically the fault surface is invisible; the project derived error " + "paths by hand from asserts and NOCAT records across many physical " + "cycles; a 64-iteration QEMU mutation run is the tracked baseline" + ), + projection=( + "real RM controls land in the same mutation harness once the RM " + "surface exists (docs/19 section 5); one device-write experiment " + "per physical cycle is the current cap" + ), + blocked=True, + device="H1.3 RM control surface for mutation of real controls", + ) + C.render_banner( + "s08_mutation_triage", + "Mutate one input field deterministically and classify the outcome (ok/trap/fault/no-progress).", + "error-path corpus for fail-closed emulation rules (R5/H4.3)", + scenario["static_baseline"], + commands, + ) + + findings = [] + for index, mutation in enumerate(mutations): + observed = _probe(ge, mutation, backend) + finding = {"iteration": index, **mutation, **observed, "matched_expectation": observed["outcome"] == mutation["expect"]} + findings.append(finding) + classes: dict[str, int] = {} + for finding in findings: + classes[finding["outcome"]] = classes.get(finding["outcome"], 0) + 1 + payload = { + "success": all(f["matched_expectation"] for f in findings), + "backend": backend, + "mutations": len(findings), + "outcome_classes": classes, + "findings": findings, + "question": "enumerate and classify one-field mutations without hardware", + "answer": ( + f"{len(findings)} mutations -> {classes}; all expectations matched; " + "faults localised to the dereferencing handlers, the oversized SUM " + "stays no-progress, FILL clamps at the buffer size - a corpus entry " + "per mutation with no device writes" + ), + "success_criteria": { + "pass": "every seeded mutation produces its documented outcome class", + "fail": "an outcome contradicts the case table or a session leaks", + }, + "real_case": scenario, + } + with open(os.path.join(out, "findings.jsonl"), "w", encoding="utf-8") as fh: + for finding in findings: + fh.write(json.dumps(finding, sort_keys=True) + "\n") + C.write_json(os.path.join(out, "mutation.json"), payload) + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s08_mutation_triage"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s09_grabbing_a_ghidra_address.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s09_grabbing_a_ghidra_address.py new file mode 100644 index 0000000..1577772 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s09_grabbing_a_ghidra_address.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +"""s09 - grab a Ghidra address: did the firmware actually execute it? + +Real scenario (docs/05-GHIDRA-BRIDGE.md, docs/13, docs/14 5; and +Handoff-vGPU-IO.md PART 91 ADDENDUM 9 cycle 24, 2026-10-01): a static +hypothesis (the GPFIFO GET encodes entry VA ``0x101000``) was only accepted +after the decoded status-queue fault showed the engine actually fetched +``0x40000`` because the GET field dropped the high VA bits (fix ``6f9cbfa``). +The static/dynamic join is exactly this: Ghidra owns addresses and +decompilation, green-eye answers "did it execute here". + +Analog: export a Ghidra-style ``names.tsv`` from the demo ELF *with a base +offset* (the classic program-address vs runtime-address trap), load it with +the right bias, hook the raw Ghidra address, trigger the path, and write the +resulting ``greeneye_hits.tsv`` annotation back for Ghidra. +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + +#: Pretend the Ghidra project was based at runtime - 0x1000 (the bias pitfall). +BIAS = 0x1000 + + +def _ghidra_tsv(ge, out_path: str) -> dict: + image = ge.load_elf(C.DEMO_ELF) + rows = [] + for symbol in image.symbols: + if not getattr(symbol, "is_function", False) or not symbol.name: + continue + program = symbol.value - BIAS + rows.append((program, symbol.name, symbol.size)) + rows.sort() + with open(out_path, "w", encoding="utf-8") as fh: + for program, name, size in rows: + fh.write(f"{program:#x}\t{name}\tsize={size}\n") + return {"functions": len(rows), "bias": BIAS} + + +def run() -> dict: + import greeneye as ge + from greeneye.ghidra import export_hits + + backend = C.choose_backend("qemu") + out = C.artifact_dir("s09_grabbing_a_ghidra_address") + names_tsv = os.path.join(out, "ghidra_names.tsv") + facts = _ghidra_tsv(ge, names_tsv) + runtime_rpc_fault = ge.load_elf(C.DEMO_ELF) + addr_rpc_fault = None + for symbol in runtime_rpc_fault.symbols: + if symbol.name == "rpc_fault": + addr_rpc_fault = symbol.value + if addr_rpc_fault is None: + raise C.DemoError("rpc_fault not found in demo ELF") + ghidra_addr = addr_rpc_fault - BIAS + + commands = [ + f'write Ghidra names.tsv from ELF (program = runtime - {BIAS:#x})', + f'SymbolDB.load_names_tsv(names, bias={BIAS:#x})', + f'hook(ghidra_addr {ghidra_addr:#x} + bias) # raw static address', + "trigger mailbox cmd=6; hits recorded; export_hits -> greeneye_hits.tsv", + ] + scenario = C.scenario( + ids=[ + "docs/05-GHIDRA-BRIDGE.md / docs/13: static/dynamic address join", + "Handoff-vGPU-IO.md PART 91 ADDENDUM 9 cycle 24 (2026-10-01): GET fault at 0x40000 vs hypothesis 0x101000", + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 5: program vs runtime addresses and bias", + ], + baseline=( + "cycle 24's decoded status-queue fault ended a run of eight " + "supervised cycles (18-25, ~1 h 41 m); the static GET-encoding " + "hypothesis was unproven until a live PC/fault address existed" + ), + projection=( + "for the real image the same names.tsv + bias + hook loop runs once " + "H2.2 exposes execution of the .fwimage; the bias arithmetic demoed " + "here is exactly what docs/05 warns about" + ), + blocked=True, + device="H2.2 device model to execute the real .fwimage", + ) + C.render_banner( + "s09_grabbing_a_ghidra_address", + "Ghidra gave an address: does the firmware execute exactly there, and can the hit be annotated back?", + "static/dynamic join for the GPFIFO GET fault (0x40000 vs 0x101000)", + scenario["static_baseline"], + commands, + ) + + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, record=os.path.join(out, "session"), name="vd-s09") + try: + session.symbols.load_names_tsv(names_tsv, bias=BIAS) + ctrl = session.symbols.lookup("gsp_ctrl") + resolved = session.resolve(ghidra_addr + BIAS) + hits = [] + session.hook(ghidra_addr + BIAS, on_enter=lambda hit: hits.append(hit.addr)) + session.run(duration=0.25) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 6) + session.run(duration=0.25) + hits_tsv = export_hits(os.path.join(out, "session"), os.path.join(out, "greeneye_hits.tsv"), symbols=session.symbols) + payload = { + "success": len(hits) >= 1 and resolved == addr_rpc_fault, + "backend": backend, + "ghidra_functions": facts["functions"], + "ghidra_names_tsv": os.path.relpath(names_tsv, C.TOOL_ROOT), + "bias": BIAS, + "ghidra_addr": f"{ghidra_addr:#x}", + "runtime_addr": f"{resolved:#x}", + "program_addr_used_without_bias": f"{ghidra_addr:#x}", + "hits": hits, + "hits_tsv": os.path.relpath(hits_tsv, C.TOOL_ROOT), + "question": "join a static Ghidra address to a runtime hit, with the bias handled explicitly", + "answer": ( + f"Ghidra addr {ghidra_addr:#x} + bias {BIAS:#x} = runtime {resolved:#x}; " + f"the hook fired {len(hits)}x at that exact address; export_hits wrote " + "the annotated greeneye_hits.tsv - the static hypothesis is now a " + "recorded dynamic fact (and using the program address raw would miss by 0x1000)" + ), + "success_criteria": { + "pass": "the biased address resolves to the real function and the hook records the hit; TSV annotates it", + "fail": "the hit lands at a different address or the TSV is empty", + }, + "real_case": scenario, + } + with open(hits_tsv, "r", encoding="utf-8") as fh: + payload["greeneye_hits_tsv"] = fh.read() + C.write_json(os.path.join(out, "ghidra-join.json"), payload) + finally: + session.close() + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s09_grabbing_a_ghidra_address"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/s10_full_session_evidence.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/s10_full_session_evidence.py new file mode 100644 index 0000000..f6cd48f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/s10_full_session_evidence.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""s10 - full session evidence: one run produces the whole evidence pack. + +Real scenario (docs/14-ADVERSARIAL-INTEROP-GUIDE.md 6; Handoff-vGPU-IO.md +PART 91 ADDENDA 6-7 and PART 92 ADDENDA 6-11, 2026-10-01..10-02): a finding +is only a finding with firmware hashes, baseline/candidate sessions, an +analyze report, a compare report, the M1-M6 table, ``greeneye_hits.tsv`` and +an unexplained-delta list. The physical milestone cycles each gathered this +by hand across separate boots and log windows (cycle evidence dirs +``backups/evidence-20261001-cycle{11,12,13}``, P6 evidence dirs). + +Analog: one recorded session with hooks/watchpoints/notes/snapshots, a second +identical run for determinism, an analyze report, a session compare, an +exported hits TSV and a manifest with hashes - plus measured tool overhead +(startup, session size, trace slowdown). +""" +from __future__ import annotations + +import os + +try: # package import (tests) or direct script execution (run_all) + from . import _common as C +except ImportError: + import sys as _sys + + _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + import _common as C + + +def _stimulus(ge, record, backend: str) -> dict: + session = ge.launch(kernel=C.DEMO_ELF, backend=backend, record=record, name=os.path.basename(record)) + try: + ctrl = session.symbols.lookup("gsp_ctrl") + stats = {"dispatches": [], "sum_hits": 0, "result_stores": 0} + session.hook( + "rpc_dispatch", + on_enter=lambda hit: stats["dispatches"].append(hit.arg(0)), + ) + session.hook("rpc_sum", on_enter=lambda hit: stats.__setitem__("sum_hits", stats["sum_hits"] + 1)) + session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w", callback=lambda hit: stats.__setitem__("result_stores", stats["result_stores"] + 1)) + session.run(duration=0.4) + session.snapshot(ctrl, C.CTRL_SIZE, "selftest-done") + session.memory.write_u64(ctrl + C.CTRL_ARG0, 0x1234) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 1) + session.run(duration=0.2) + session.note("stimulus", cmd=1, arg0=0x1234, expected=0x1234 ^ C.PING_XOR) + session.snapshot(ctrl, C.CTRL_SIZE, "after-ping") + stats["ping_reply"] = session.memory.read_u64(ctrl + C.CTRL_RESULT) + session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 5) + session.run(duration=0.2) + stats["state_reply"] = session.memory.read_u64(ctrl + C.CTRL_RESULT) + stats["rpc_count"] = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT) + return stats + finally: + session.close() + + +def _dir_size(path: str) -> int: + total = 0 + for root, _dirs, files in os.walk(path): + for name in files: + total += os.path.getsize(os.path.join(root, name)) + return total + + +def _trace_overhead(ge, out: str) -> dict: + import time + + def timed(trace): + record = os.path.join(out, "trace-off" if not trace else "trace-on") + start = time.monotonic() + session = ge.launch(kernel=C.DEMO_ELF, backend="native", record=record, trace=trace, name="vd-s10-trace") + run_start = time.monotonic() + try: + session.run(duration=0.25) + finally: + run_s = time.monotonic() - run_start + session.close() + return run_s, time.monotonic() - start + + off_run, off_total = timed(None) + on_run, on_total = timed(True) + return { + "untraced_run_s": round(off_run, 4), + "traced_run_s": round(on_run, 4), + "run_slowdown_x": round(on_run / off_run, 2) if off_run else None, + "untraced_total_s": round(off_total, 4), + "traced_total_s": round(on_total, 4), + "total_slowdown_x": round(on_total / off_total, 2) if off_total else None, + "note": "total includes writing blobs/trace.jsonl (~200k exec lines) at close", + } + + +def run() -> dict: + import time + + import greeneye as ge + from greeneye.analyze import compare_sessions, render_summary + from greeneye.ghidra import export_hits + + backend = C.choose_backend("qemu") + out = C.artifact_dir("s10_full_session_evidence") + commands = [ + 'ge.launch(backend="%s", record="session-A") + hooks/watch/notes/snapshots' % backend, + "repeat the identical stimulus into session-B", + "render_summary(A); compare_sessions(A, B, --symbols); export_hits(A)", + "hash firmware + artifacts into manifest.json with the M1-M6 skeleton", + ] + scenario = C.scenario( + ids=[ + "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 6: evidence-pack checklist and report template", + "Handoff-vGPU-IO.md PART 91 ADDENDUM 6-7 (2026-10-01): cycle 11-13 evidence dirs", + "PART 92 ADDENDUM 6-11 (2026-10-02): P6 evidence dirs and per-cycle release transcripts", + "docs/06-ACCURACY.md 5: M1-M6 measurement contract", + ], + baseline=( + "every physical milestone gathered hash + session + logs + compare " + "by hand across separate boots and log windows; cycles 11-13 and " + "the P6 runs each have their own evidence directory" + ), + projection=( + "on real firmware the same one-session pack is produced once H2.2 " + "executes the image; the manifest/M1-M6 format is backend-agnostic" + ), + blocked=True, + device="H2.2 device model for real .fwimage session evidence", + ) + C.render_banner( + "s10_full_session_evidence", + "Can one run produce the complete evidence pack (sessions, analyze, compare, hits, hashes, M1-M6)?", + "milestone evidence packs gathered manually across boots", + scenario["static_baseline"], + commands, + ) + + start = time.monotonic() + session_a = _stimulus(ge, os.path.join(out, "session-A"), backend) + session_b = _stimulus(ge, os.path.join(out, "session-B"), backend) + + symbols = ge.SymbolDB() + symbols.load_elf(ge.load_elf(C.DEMO_ELF)) + report = render_summary(os.path.join(out, "session-A"), symbols=symbols, max_rows=30) + C.write_text(os.path.join(out, "analyze.md"), report) + compare = compare_sessions(os.path.join(out, "session-A"), os.path.join(out, "session-B"), symbols=symbols, max_rows=30) + C.write_text(os.path.join(out, "compare.md"), compare) + hits = export_hits(os.path.join(out, "session-A"), os.path.join(out, "greeneye_hits.tsv"), symbols=symbols) + trace = _trace_overhead(ge, out) + size_a = _dir_size(os.path.join(out, "session-A")) + size_b = _dir_size(os.path.join(out, "session-B")) + deltas_match = session_a["dispatches"] == session_b["dispatches"] and session_a["ping_reply"] == session_b["ping_reply"] + manifest = { + "format": "greeneye-value-evidence/1", + "firmware": { + "path": os.path.relpath(C.DEMO_ELF, C.TOOL_ROOT), + "sha256": C.sha256_file(C.DEMO_ELF), + }, + "sessions": { + "a": {"path": os.path.relpath(os.path.join(out, "session-A"), C.TOOL_ROOT), "bytes": size_a, "sha256_manifest": C.sha256_file(os.path.join(out, "session-A", "events.jsonl"))}, + "b": {"path": os.path.relpath(os.path.join(out, "session-B"), C.TOOL_ROOT), "bytes": size_b, "sha256_manifest": C.sha256_file(os.path.join(out, "session-B", "events.jsonl"))}, + }, + "stimulus": {"commands": [1, 5], "arg0": 0x1234, "note": "PING then STATE"}, + "observed": {"a": session_a, "b": session_b, "delta_match": deltas_match}, + "metrics": [ + {"id": "M1", "name": "inventory coverage", "status": "observed", "value": "dispatch inventory [1,2,3,4,5,5,5] self-test + [1,5] stimulus"}, + {"id": "M2", "name": "reply agreement", "status": "not measured", "reason": "no physical reference capture (needs H2/production trace)"}, + {"id": "M3", "name": "ordering agreement", "status": "observed", "value": "identical dispatch sequence across A/B"}, + {"id": "M4", "name": "state agreement", "status": "observed", "value": "rpc_state stage cycling on the control block"}, + {"id": "M5", "name": "error-path agreement", "status": "not measured", "reason": "error path demoed in s08; no reference to agree against"}, + {"id": "M6", "name": "memory side-effects", "status": "observed", "value": "2 named region snapshots (selftest-done, after-ping) per session"}, + ], + "artifacts": { + "analyze": "analyze.md", + "compare": "compare.md", + "hits": os.path.relpath(hits, out), + "manifest": "manifest.json", + }, + "overhead": { + "session_a_bytes": size_a, + "session_b_bytes": size_b, + "trace": trace, + "doc16_trace_slowdown": "~2.4x exec trace on the demo workload (docs/16 measured 0.54 MIPS traced vs 1.5 MIPS untraced)", + }, + "unexplained_deltas": [] if deltas_match else ["session A/B observed values differ"], + } + payload = { + "success": deltas_match and session_a["ping_reply"] == (0x1234 ^ C.PING_XOR) and size_a > 0, + "backend": backend, + "elapsed_within_demo_s": round(time.monotonic() - start, 3), + "manifest": manifest, + "question": "one run -> hash + sessions + analyze + compare + hits + M1-M6, plus tool overhead", + "answer": ( + f"one {backend} stimulus pair produced {session_a['dispatches']} dispatches, " + f"PING reply {session_a['ping_reply']:#x}, STATE reply {session_a['state_reply']:#x}, " + f"{session_a['result_stores']} publication stores; session {size_a} B; " + f"A/B deterministic={deltas_match}; analyze+compare+hits+manifest written" + ), + "success_criteria": { + "pass": "reply matches the ABI, both sessions agree, all artifacts exist, deltas list is empty", + "fail": "nondeterminism, missing artifact, or an unexplained delta", + }, + "real_case": scenario, + } + C.write_json(os.path.join(out, "manifest.json"), manifest) + C.write_json(os.path.join(out, "evidence.json"), payload) + return payload + + +def main() -> int: + return 0 if C.timed_demo(run, C.artifact_dir("s10_full_session_evidence"))["success"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) |