diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/contracts.py | 396 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/launch_audit.py | 691 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/log_decode.py | 660 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/rehost.py | 428 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/rm_capture.py | 1251 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/run_contracts.py | 131 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/server_snapshot.py | 698 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/state_decode.py | 1082 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/state_seed.py | 990 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/valprobe.py | 651 |
10 files changed, 6978 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/contracts.py b/debug/risc-v-gsp-rm-vm/harness/contracts.py new file mode 100644 index 0000000..ec387fe --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/contracts.py @@ -0,0 +1,396 @@ +#!/usr/bin/env python3 +"""Contract registry: every solved wall becomes a re-runnable regression. + +Why this exists +--------------- +The GR campaign's dominant cost is wall-hell: one blocking firmware problem +after another, each triaged by supervised hardware cycles. Several wall +classes now have offline, deterministic tools (docs/45-48), but their value is +only durable if the *answer* survives as a re-runnable check. This registry +pins the proven entry points of those tools to one explicit expectation each +and re-runs them: fast contracts run in the main suite, slow ones on demand. + +It wraps existing entry points as-is (no measurement logic is reimplemented) +and every contract names its wall class from ``docs/50-WALL-WORKFLOW.md``, so +per-domain coverage is visible - including the classes with no offline +contract by design (engine execution awaits supervised captures). + +Offline only: never touches a GPU, kernel module, mdev, guest or network. +""" +from __future__ import annotations + +import os +import time +from dataclasses import dataclass +from typing import Any, Callable + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +from harness.rehost import DEFAULT_IMAGE, DEMO_ELF # noqa: E402 + +#: Schema tag of a registry run summary. +RUN_FORMAT = "greeneye-contract-run/1" + +FAST = "fast" +SLOW = "slow" + +#: Wall classes from docs/50 section 2. A contract names exactly one. +WALL_CLASSES = { + "a": "firmware boot/status", + "b": "RM server behaviour", + "c": "engine memory contract", + "d": "engine execution/dispatch", + "e": "guest-driver/protocol interface", + "f": "version skew/portability", +} + + +class ContractError(Exception): + """A contract record or registry request is invalid.""" + + +class ContractSkip(Exception): + """A contract precondition is missing at run time (not a pass/fail).""" + + +@dataclass(frozen=True) +class Contract: + """One re-runnable regression over an existing tool entry point.""" + + name: str + wall_class: str # key into WALL_CLASSES (docs/50 section 2) + doc: str # where the wall and the tool are documented + entry: str # exact wrapped entry point, e.g. "sim.gmmu_walk:run_demo()" + speed: str # FAST runs in the suite; SLOW is --all only + expect: str # the pass condition, in words + run: Callable[[], dict] # no-arg callable returning the tool's report + check: Callable[[dict], tuple[bool, str]] # (passed, detail) + requires: tuple[str, ...] = () # paths that must exist; otherwise skip + + @property + def domain(self) -> str: + return WALL_CLASSES.get(self.wall_class, self.wall_class) + + +@dataclass(frozen=True) +class ContractResult: + """Outcome of one contract run; ``status`` is pass | fail | skip.""" + + name: str + wall_class: str + speed: str + doc: str + entry: str + expect: str + status: str + detail: str + seconds: float + + @property + def domain(self) -> str: + return WALL_CLASSES.get(self.wall_class, self.wall_class) + + @property + def passed(self) -> bool: + return self.status == "pass" + + def to_dict(self) -> dict: + return { + "name": self.name, + "wall_class": self.wall_class, + "domain": self.domain, + "speed": self.speed, + "doc": self.doc, + "entry": self.entry, + "expect": self.expect, + "status": self.status, + "detail": self.detail, + "seconds": round(self.seconds, 4), + } + + +# --------------------------------------------------------------------------- +# Runners: thin wrappers over the proven tool entry points. +# +# Every wrapper only imports, calls and (where the demo writes artifacts) +# suppresses output paths. The expectation lives in the checker next to it. +# --------------------------------------------------------------------------- + +def _run_fecs_matrix() -> dict: + from harness.valprobe import fecs_spec, run_spec + + return run_spec(spec=fecs_spec(), record=False, name="contract-fecs-matrix") + + +def _check_fecs_matrix(report: dict) -> tuple[bool, str]: + variants = report.get("variants", []) + evaluated = report.get("evaluated") + failed = report.get("failed") + ok = ( + report.get("format") == "greeneye-valprobe-matrix/1" + and len(variants) == 15 + and evaluated == 15 + and failed == 0 + and all(row.get("pass") is True for row in variants) + ) + return ok, f"{evaluated}/{len(variants)} PROMOTE_CTX variants labelled, {failed} failed" + + +def _run_fecs_determinism() -> dict: + from harness.valprobe import run_demo + + return run_demo() + + +def _check_fecs_determinism(report: dict) -> tuple[bool, str]: + ok = ( + report.get("format") == "greeneye-valprobe-demo/1" + and report.get("ok") is True + and report.get("deterministic") is True + and report.get("failed") == 0 + ) + return ok, ( + f"evaluated={report.get('evaluated')} failed={report.get('failed')} " + f"deterministic={report.get('deterministic')}" + ) + + +def _run_gmmu_demo() -> dict: + from sim.gmmu_walk import run_demo + + return run_demo() + + +def _check_gmmu_demo(report: dict) -> tuple[bool, str]: + rows = report.get("rows", []) + matched = sum(1 for row in rows if row.get("match") is True) + ok = report.get("format") == "greeneye-gmmu-walk-demo/1" and report.get("ok") is True \ + and matched == len(rows) == 5 + return ok, f"{matched}/{len(rows)} VAs walk to the captured DMA addresses" + + +def _run_launch_audit_demo() -> dict: + from harness.launch_audit import run_demo + + return run_demo() + + +def _check_launch_audit_demo(report: dict) -> tuple[bool, str]: + from harness.launch_audit import VERDICT_OK, VERDICT_PA_MISMATCH, VERDICT_UNMAPPED + + legs = report.get("captured_channel", {}).get("contract", []) + ok_legs = sum(1 for leg in legs if leg.get("verdict") == VERDICT_OK) + classes = { + (scenario.get("report", {}).get("contract") or [{}])[0].get("verdict") + for scenario in report.get("qmd_hypotheses", []) + } + ok = ( + report.get("format") == "greeneye-launch-audit-demo/1" + and report.get("ok") is True + and ok_legs == len(legs) == 5 + and {VERDICT_PA_MISMATCH, VERDICT_UNMAPPED} <= classes + ) + return ok, ( + f"captured channel {ok_legs}/{len(legs)} legs ok; " + f"QMD verdict classes {sorted(c for c in classes if c)}" + ) + + +def _run_rehost_demo() -> dict: + from harness.rehost import run_demo + + return run_demo() + + +def _check_rehost_demo(report: dict) -> tuple[bool, str]: + first = report.get("first", {}) + second = report.get("second", {}) + agree = ( + first.get("status") == second.get("status") + and first.get("retval") == second.get("retval") + ) + ok = ( + report.get("format") == "greeneye-rehost-demo/1" + and report.get("ok") is True + and first.get("status") == "return" + and first.get("retval") == report.get("expected") + ) + return ok, ( + f"status={first.get('status')} retval={first.get('retval')} " + f"(expected {report.get('expected')}); two runs agree={agree}" + ) + + +def _run_rm_capture_demo() -> dict: + from harness.rm_capture import run_demo + + return run_demo() + + +def _check_rm_capture_demo(report: dict) -> tuple[bool, str]: + summary = report.get("summary", {}) + ok = report.get("format") == "greeneye-rm-rpc-capture-demo/1" and report.get("ok") is True + return ok, ( + f"records {report.get('records_a')}/{report.get('records_b')}; " + f"stimulus messages {report.get('messages')}; " + f"agree={summary.get('agree')} shape_agree={summary.get('shape_agree')} " + f"a_only={summary.get('a_only')} b_only={summary.get('b_only')}" + ) + + +# --------------------------------------------------------------------------- +# Registry +# --------------------------------------------------------------------------- + +def _registry() -> list[Contract]: + return [ + Contract( + name="fecs-predicate-matrix", + wall_class="b", + doc="docs/47, docs/50", + entry="harness.valprobe:run_spec(fecs_spec())", + speed=FAST, + expect="15/15 PROMOTE_CTX predicate variants reach their labelled outcome (failed == 0)", + requires=(DEFAULT_IMAGE,), + run=_run_fecs_matrix, + check=_check_fecs_matrix, + ), + Contract( + name="fecs-predicate-determinism", + wall_class="b", + doc="docs/47, docs/25, docs/50", + entry="harness.valprobe:run_demo()", + speed=SLOW, + expect="the bundled matrix runs twice with identical verdicts (ok and deterministic)", + requires=(DEFAULT_IMAGE,), + run=_run_fecs_determinism, + check=_check_fecs_determinism, + ), + Contract( + name="gmmu-golden-vector", + wall_class="c", + doc="docs/48, docs/50", + entry="sim.gmmu_walk:run_demo()", + speed=FAST, + expect="the captured channel-PDB vector walks 5/5 VAs to the logged DMA addresses", + run=_run_gmmu_demo, + check=_check_gmmu_demo, + ), + Contract( + name="launch-audit-captured-contract", + wall_class="c", + doc="docs/48, docs/50", + entry="harness.launch_audit:run_demo()", + speed=FAST, + expect="captured contract all ok; QMD hypotheses classify pa_mismatch and unmapped", + run=_run_launch_audit_demo, + check=_check_launch_audit_demo, + ), + Contract( + name="rehost-demo", + wall_class="b", + doc="docs/46, docs/25, docs/50", + entry="harness.rehost:run_demo()", + speed=FAST, + expect="the demo function returns the expected sum twice (ok == true)", + requires=(DEMO_ELF,), + run=_run_rehost_demo, + check=_check_rehost_demo, + ), + Contract( + name="rm-capture-decode-demo", + wall_class="e", + doc="docs/45, docs/25, docs/50", + entry="harness.rm_capture:run_demo()", + speed=FAST, + expect="synthetic capture decodes, compares and yields a stimulus (ok == true)", + run=_run_rm_capture_demo, + check=_check_rm_capture_demo, + ), + ] + + +def all_contracts() -> list[Contract]: + """Every contract in the registry, fast and slow.""" + return _registry() + + +def fast_contracts() -> list[Contract]: + """Contracts the main suite runs.""" + return [c for c in all_contracts() if c.speed == FAST] + + +def slow_contracts() -> list[Contract]: + """Contracts only ``run_contracts.py --all`` runs.""" + return [c for c in all_contracts() if c.speed == SLOW] + + +def _result(contract: Contract, status: str, detail: str, seconds: float) -> ContractResult: + return ContractResult( + name=contract.name, + wall_class=contract.wall_class, + speed=contract.speed, + doc=contract.doc, + entry=contract.entry, + expect=contract.expect, + status=status, + detail=detail, + seconds=seconds, + ) + + +def run_contract(contract: Contract) -> ContractResult: + """Run one contract, classifying every outcome without raising.""" + missing = [path for path in contract.requires if not os.path.exists(path)] + if missing: + return _result(contract, "skip", "prerequisite missing: " + ", ".join(missing), 0.0) + started = time.monotonic() + try: + report = contract.run() + except ContractSkip as exc: + return _result(contract, "skip", str(exc), time.monotonic() - started) + except Exception as exc: # noqa: BLE001 - a raising entry point is a failed contract + return _result( + contract, "fail", f"{type(exc).__name__}: {exc}", time.monotonic() - started + ) + elapsed = time.monotonic() - started + try: + passed, detail = contract.check(report) + except Exception as exc: # noqa: BLE001 - a broken checker is a failed contract + return _result(contract, "fail", f"checker error: {type(exc).__name__}: {exc}", elapsed) + return _result(contract, "pass" if passed else "fail", str(detail), elapsed) + + +def run_contracts(contracts: list[Contract]) -> list[ContractResult]: + """Run the given contracts in order; never raises for a contract outcome.""" + return [run_contract(contract) for contract in contracts] + + +def summarize(results: list[ContractResult]) -> dict: + """Machine-readable run summary with per-wall-class coverage.""" + counts = {"pass": 0, "fail": 0, "skip": 0} + domains: dict[str, dict[str, Any]] = { + key: { + "wall_class": key, + "domain": label, + "total": 0, + "pass": 0, + "fail": 0, + "skip": 0, + } + for key, label in WALL_CLASSES.items() + } + for result in results: + if result.status not in counts: + raise ContractError(f"unknown contract status {result.status!r}") + counts[result.status] += 1 + bucket = domains[result.wall_class] + bucket["total"] += 1 + bucket[result.status] += 1 + return { + "format": RUN_FORMAT, + "ok": counts["fail"] == 0, + "counts": counts, + "domains": [domains[key] for key in sorted(domains)], + "contracts": [result.to_dict() for result in results], + } diff --git a/debug/risc-v-gsp-rm-vm/harness/launch_audit.py b/debug/risc-v-gsp-rm-vm/harness/launch_audit.py new file mode 100644 index 0000000..ba7a623 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/launch_audit.py @@ -0,0 +1,691 @@ +#!/usr/bin/env python3 +"""Launch memory-contract audit: where would the engine read/write? + +A forwarded submission goes silent when one of the engine-visible memory legs +is wrong: the QMD destination the front-end copies to, the release semaphore +the engine writes, the ring the front-end fetches, the parameter/program +pages the QMD points at. The host's bookkeeping (shadows, guest pages) is not +what the engine sees -- the *page tables* are. This harness walks each +address of a submission's memory contract through a captured/constructed +page-table view (``sim.gmmu_walk``) and reports, per leg, the PA the engine +would use, its access flags, and any divergence from what the host believes. + +It answers the class of question behind "the front-end never wrote the QMD"/ +"the release never fired": *is the mapping the engine uses the mapping we are +checking?* -- offline, deterministically, without hardware. + +``check`` closes the capture loop: run it against a partial or empty +``memory[]`` view and it reports, per contract leg, the FIRST table page the +walk needed but the view does not contain (with the level/index path), plus a +consolidated ``needed_pages[]`` list. Dump exactly those pages, re-run, and +repeat -- one capture iteration per table level. + +Schemas: + +* ``greeneye-launch-audit-spec/1`` -- inputs (page tables + contract). +* ``greeneye-launch-audit/1`` -- the verdict report. +* ``greeneye-launch-audit-check/1`` -- the dump-completeness report. + +Honesty: a verdict is only as good as the captured table view in the spec. +``capture-hints`` prints what to capture for a real audit and never executes +anything. Fidelity is the address view (L1); no timing claim. +""" +from __future__ import annotations + +import argparse +import json +import os +import struct +import sys +from typing import Any + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from sim.gmmu_pte import KIND_VER2, LEVELS, SUPPORTED_KINDS # noqa: E402 +from sim.gmmu_walk import ( # noqa: E402 + FAULT_TABLE_MISSING, + MemoryView, + WalkError, + channel_pdb_fixture, + encode_dual_slot, + encode_pde_pointer, + encode_pte, + walk, +) + +SPEC_FORMAT = "greeneye-launch-audit-spec/1" +AUDIT_FORMAT = "greeneye-launch-audit/1" +CHECK_FORMAT = "greeneye-launch-audit-check/1" +DEMO_FORMAT = "greeneye-launch-audit-demo/1" + +VERDICT_OK = "ok" +VERDICT_UNMAPPED = "unmapped" +VERDICT_PA_MISMATCH = "pa_mismatch" +VERDICT_ACCESS_DENIED = "access_denied" + +#: Walk depth -> display label (root first, leaf page table last). Mirrors +#: the ``LEVELS`` geometry in ``sim.gmmu_pte``; kept here as display names so +#: check reports read like the capture hints (root/pde2/pde1/pde0/pte). +LEVEL_LABELS = ("root", "pde2", "pde1", "pde0", "pte") + + +class AuditError(ValueError): + """A spec or request the tool cannot honour.""" + + +def _parse_int(value: Any) -> int: + if isinstance(value, bool): + raise AuditError(f"bad integer {value!r}") + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value, 0) + except ValueError as exc: + raise AuditError(f"bad integer {value!r}") from exc + raise AuditError(f"bad integer {value!r}") + + +def _parse_hex_bytes(text: str) -> bytes: + data = "".join(str(text).split()) + if len(data) % 2: + raise AuditError("odd hex length") + try: + return bytes.fromhex(data) + except ValueError as exc: + raise AuditError(f"bad hex payload: {exc}") from exc + + +def load_memory(items) -> MemoryView: + mem = MemoryView() + for item in items or (): + if "pa" not in item or "hex" not in item: + raise AuditError(f"memory item needs pa and hex: {item!r}") + mem.add(_parse_int(item["pa"]), _parse_hex_bytes(item["hex"])) + return mem + + +def audit(spec: dict) -> dict: + """Run one audit spec; return a ``greeneye-launch-audit/1`` report.""" + fmt = spec.get("format", SPEC_FORMAT) + if fmt != SPEC_FORMAT: + raise AuditError(f"unsupported spec format {fmt!r} (want {SPEC_FORMAT})") + if "pdb_pa" not in spec or "contract" not in spec: + raise AuditError("spec needs pdb_pa and contract") + kind = spec.get("gmmu", KIND_VER2) + if kind not in SUPPORTED_KINDS: + raise AuditError(f"unsupported gmmu kind {kind!r} (support: {SUPPORTED_KINDS})") + page_mode = spec.get("page_mode", "auto") + root_pa = _parse_int(spec["pdb_pa"]) + mem = load_memory(spec.get("memory")) + + rows = [] + for item in spec["contract"]: + label = item.get("label") or "unnamed" + if "va" not in item: + raise AuditError(f"contract item needs va: {item!r}") + va = _parse_int(item["va"]) + access = item.get("access", "read") + report = walk(va, root_pa=root_pa, mem=mem, kind=kind, page_mode=page_mode) + row: dict[str, Any] = { + "label": label, + "va": f"{va:#x}", + "access": access, + "note": item.get("note"), + "verdict": None, + "walk": report, + } + if report["status"] != "ok": + row["verdict"] = VERDICT_UNMAPPED + row["detail"] = report.get("fault_message") + else: + expect_pa = item.get("expect_pa") + if access == "write" and not report.get("writable", False): + row["verdict"] = VERDICT_ACCESS_DENIED + row["detail"] = "PTE is read-only but the engine must write here" + elif expect_pa is not None and _parse_int(expect_pa) != _parse_int(report["pa"]): + row["verdict"] = VERDICT_PA_MISMATCH + row["detail"] = (f"engine walk -> {report['pa']}; host believes " + f"{_parse_int(expect_pa):#x}") + else: + row["verdict"] = VERDICT_OK + rows.append(row) + + summary = {v: sum(1 for r in rows if r["verdict"] == v) + for v in (VERDICT_OK, VERDICT_UNMAPPED, VERDICT_PA_MISMATCH, + VERDICT_ACCESS_DENIED)} + report = { + "format": AUDIT_FORMAT, + "name": spec.get("name"), + "firmware": spec.get("firmware"), + "gmmu": kind, + "page_mode": page_mode, + "pdb_pa": f"{root_pa:#x}", + "contract": rows, + "summary": summary, + "ok": summary[VERDICT_UNMAPPED] == 0 and summary[VERDICT_PA_MISMATCH] == 0 + and summary[VERDICT_ACCESS_DENIED] == 0, + "limits": [ + "Address view only (L1): a clean audit proves the engine CAN reach these PAs, not that it runs.", + "Verdicts depend on the captured table view in the spec; an incomplete view reports unmapped, not unreachability.", + ], + } + return report + + +# --------------------------------------------------------------------------- +# Check: which table pages does the walk still need? +# --------------------------------------------------------------------------- + +def _path_from_steps(steps: list[dict]) -> list[dict]: + """The completed walk hops, in walk order, as check ``path`` elements.""" + return [ + { + "level": LEVEL_LABELS[step["depth"]], + "depth": step["depth"], + "table_pa": step["table_pa"], + "index": step["index"], + "entry_pa": step["entry_pa"], + "present": True, + } + for step in steps + ] + + +def _missing_table(report: dict, root_pa: int, va: int) -> dict: + """Describe the first table page a ``table_missing`` walk needed. + + ``report["steps"]`` holds the levels that completed before the failing + read; the missing table is the page that read was addressing. The + returned dict carries the page base (``pa`` and ``table_pa``), the level + label, the walk depth, the index inside the table and the exact entry + address. ``report`` must be a ``table_missing`` fault. + """ + depth = len(report["steps"]) + if depth == 0: + table_pa = root_pa + else: + prev = report["steps"][-1] + if prev.get("role") == "dual_pde": + half = prev.get("page_mode_used") + table_pa = (prev["big"]["address"] if half == "64k" + else prev["small"]["address"]) + else: + table_pa = prev["address"] + if depth not in range(len(LEVEL_LABELS)): + # Cannot happen for a walker fault, but never guess a geometry. + return {"pa": f"{table_pa:#x}", "table_pa": f"{table_pa:#x}", + "depth": depth, "level": f"depth{depth}", "index": None, + "entry_pa": None, "present": False} + if depth == 4: + prev = report["steps"][-1] + shift = 16 if prev.get("page_mode_used") == "64k" else 12 + index = (va >> shift) & 0x1FF + entry_size = 8 + else: + geo = LEVELS[depth] + index = (va >> geo["va_bits"][1]) & ((1 << geo["index_bits"]) - 1) + entry_size = geo["entry_size"] + return { + "pa": f"{table_pa:#x}", + "table_pa": f"{table_pa:#x}", + "depth": depth, + "level": LEVEL_LABELS[depth], + "index": index, + "entry_pa": f"{table_pa + index * entry_size:#x}", + "present": False, + } + + +def check(spec: dict) -> dict: + """Report the table pages a partial ``memory[]`` view still needs. + + Runs the same walks as :func:`audit` but, instead of verdicts, reports per + contract leg the FIRST table page the walk could not read (the page the + dump is missing) with the level/index path that reached it, and a + consolidated ``needed_pages`` list. Iterate: run ``check`` with a partial + or empty dump, dump exactly the named pages, repeat until ``complete`` is + true, then run :func:`audit` for verdicts. A leg whose walk faults for a + non-dump reason (invalid aperture/PTE, unsupported encoding) is reported + as ``fault`` and needs no more pages. + + An empty spec memory is valid: every leg then reports the root page as + its first needed page. + """ + fmt = spec.get("format", SPEC_FORMAT) + if fmt != SPEC_FORMAT: + raise AuditError(f"unsupported spec format {fmt!r} (want {SPEC_FORMAT})") + if "pdb_pa" not in spec or "contract" not in spec: + raise AuditError("spec needs pdb_pa and contract") + kind = spec.get("gmmu", KIND_VER2) + if kind not in SUPPORTED_KINDS: + raise AuditError(f"unsupported gmmu kind {kind!r} (support: {SUPPORTED_KINDS})") + page_mode = spec.get("page_mode", "auto") + root_pa = _parse_int(spec["pdb_pa"]) + mem = load_memory(spec.get("memory")) + + rows = [] + needed: dict[int, dict[str, Any]] = {} + for item in spec["contract"]: + label = item.get("label") or "unnamed" + if "va" not in item: + raise AuditError(f"contract item needs va: {item!r}") + va = _parse_int(item["va"]) + report = walk(va, root_pa=root_pa, mem=mem, kind=kind, page_mode=page_mode) + row: dict[str, Any] = { + "label": label, + "va": f"{va:#x}", + "access": item.get("access", "read"), + "expect_pa": None, + "status": None, + "resolved_pa": None, + "writable": None, + "first_missing": None, + "fault": report.get("fault"), + "detail": report.get("fault_message"), + "path": _path_from_steps(report["steps"]), + } + if item.get("expect_pa") is not None: + row["expect_pa"] = f"{_parse_int(item['expect_pa']):#x}" + if report["status"] == "ok": + row["status"] = "complete" + row["resolved_pa"] = report["pa"] + row["writable"] = report.get("writable") + row["fault"] = None + row["detail"] = None + elif report.get("fault") == FAULT_TABLE_MISSING: + missing = _missing_table(report, root_pa, va) + row["status"] = "missing" + row["first_missing"] = missing + row["path"].append(missing) + entry = needed.setdefault(int(missing["pa"], 16), + {"levels": set(), "legs": []}) + entry["levels"].add(missing["level"]) + entry["legs"].append(label) + else: + row["status"] = "fault" + rows.append(row) + + needed_pages = [ + {"pa": f"{pa:#x}", "levels": sorted(value["levels"]), + "legs": sorted(value["legs"])} + for pa, value in sorted(needed.items()) + ] + return { + "format": CHECK_FORMAT, + "name": spec.get("name"), + "firmware": spec.get("firmware"), + "gmmu": kind, + "page_mode": page_mode, + "pdb_pa": f"{root_pa:#x}", + "memory_pages": len(mem.spans()), + "contract": rows, + "needed_pages": needed_pages, + "complete": all(row["status"] != "missing" for row in rows), + "limits": [ + "Dump-completeness view only: `complete` means every table page the walk needed is in the spec, not that any mapping is correct.", + "A `fault` leg needs no more pages: the walk read its entries and refused them (invalid aperture/PTE or unsupported encoding).", + "`needed_pages` is the FIRST missing page per leg; dumping it can reveal the next one, so iterate.", + ], + } + + +# --------------------------------------------------------------------------- +# Demo: the captured channel-PDB contract + the QMD-destination hypotheses +# --------------------------------------------------------------------------- + +def _tree_for_va(root_pa: int, target_va: int, target_pa: int | None, + *, tables_base: int = 0x50000000) -> MemoryView: + """Build a minimal VER2 tree mapping one VA to one PA (None = unmapped).""" + va = target_va + i0 = (va >> 47) & 0x3 + i1 = (va >> 38) & 0x1FF + i2 = (va >> 29) & 0x1FF + i3 = (va >> 21) & 0xFF + i4 = (va >> 12) & 0x1FF + pd2 = tables_base + 0x1000 + pd1 = tables_base + 0x2000 + pd0 = tables_base + 0x3000 + pt = tables_base + 0x4000 + mem = MemoryView() + root = bytearray(0x1000) + struct.pack_into("<Q", root, i0 * 8, encode_pde_pointer(pd2)) + mem.add(root_pa, bytes(root)) + page = bytearray(0x1000) + struct.pack_into("<Q", page, i1 * 8, encode_pde_pointer(pd1)) + mem.add(pd2, bytes(page)) + page = bytearray(0x1000) + struct.pack_into("<Q", page, i2 * 8, encode_pde_pointer(pd0)) + mem.add(pd1, bytes(page)) + page = bytearray(0x1000) + struct.pack_into("<16s", page, i3 * 16, encode_dual_slot(None, pt)) + mem.add(pd0, bytes(page)) + page = bytearray(0x1000) + if target_pa is not None: + struct.pack_into("<Q", page, i4 * 8, encode_pte(target_pa)) + mem.add(pt, bytes(page)) + return mem + + +def run_demo(out: str | None = None) -> dict: + """Audit the captured channel contract and the fork classes. + + Expected verdicts (all pinned by tests): + + * captured channel contract -- all 5 low-window legs ``ok``; + * QMD destination (VA ``0x204608000``): engine resolves to the host shadow + while the host checks the guest page -> ``pa_mismatch``; bookkeeping + trusting the shadow -> ``ok``; unmapped leaf -> ``unmapped``; + * engine-write through shadow (the next wall class: CUDA output buffers at + high VAs, where Tier-1 refuses 0x1f and the leg resolves through the + guest scratch PT pool): the write leg resolves to a shadow PA while the + host bookkeeping still expects the guest PA -> ``pa_mismatch``; when the + bookkeeping is aligned to the engine-visible shadow PA -> ``ok``. + The shadow/guest PAs in this scenario are synthetic (documented in + docs/48), not m7 captures. + """ + # 1) Captured channel PDB contract (validated environment probe). + root_pa, mem = channel_pdb_fixture() + captured = { + "format": SPEC_FORMAT, + "name": "captured-channel-pdb", + "firmware": "570.144", + "gmmu": KIND_VER2, + "page_mode": "4k", + "pdb_pa": f"{root_pa:#x}", + "memory": [{"pa": f"{pa:#x}", "hex": b.hex()} for pa, b in _memory_items(mem)], + "contract": [ + {"label": "gpfifo-ring", "va": "0x100000", "access": "read", + "expect_pa": "0x2045d9000", "note": "front-end GPFIFO fetch"}, + {"label": "push-buffer", "va": "0x101000", "access": "write", + "expect_pa": "0x121bd9000"}, + {"label": "copy-src", "va": "0x102000", "access": "read", + "expect_pa": "0x152115000"}, + {"label": "copy-dst", "va": "0x103000", "access": "write", + "expect_pa": "0x15a5f5000"}, + {"label": "semaphore", "va": "0x104000", "access": "write", + "expect_pa": "0x10cf27000"}, + ], + } + captured_report = audit(captured) + + # 2) QMD-destination hypotheses, same VA (0x204608000): + # A: engine resolves to the host shadow (0x15e8fa000) -- host checks + # the guest page (0x204608000) -> pa_mismatch (the "wrong page" bug). + # B: no mapping / invalid leaf -> unmapped (the "copy dropped" bug). + qmd_va = 0x204608000 + shadow_pa = 0x15E8FA000 + scenarios = [] + for label, target, expect in ( + ("qmd-dest-shadow-trusted", shadow_pa, shadow_pa), + ("qmd-dest-host-checks-guest", shadow_pa, 0x204608000), + ("qmd-dest-unmapped", None, None), + ): + root = 0x60000000 + tree = _tree_for_va(root, qmd_va, target) + spec = { + "format": SPEC_FORMAT, "name": label, "firmware": "570.144", + "gmmu": KIND_VER2, "page_mode": "4k", "pdb_pa": f"{root:#x}", + "memory": [{"pa": f"{pa:#x}", "hex": b.hex()} for pa, b in _memory_items(tree)], + "contract": [{"label": "inline-qmd-destination", "va": f"{qmd_va:#x}", + "access": "write", "expect_pa": f"{expect:#x}" if expect else None, + "note": "front-end copies the inline QMD here"}], + } + report = audit(spec) + scenarios.append({"name": label, "report": report}) + + # 3) Engine-write through shadow (the next wall class). A high guest VA + # is refused by Tier-1 and resolved through the guest scratch PT pool + # to a shadow page; the host bookkeeping for the output may still track + # the guest page. Same walk, two bookkeeping states: + # A: host expects the guest PA -> pa_mismatch (the write is invisible). + # B: bookkeeping aligned to the engine-visible shadow PA -> ok. + shadow_va = 0x210000 + shadow_target = 0x13D2A0000 + guest_pa = 0x2045C0000 + shadow_scenarios = [] + for label, expect, want in ( + ("engine-write-shadow-host-checks-guest", guest_pa, VERDICT_PA_MISMATCH), + ("engine-write-shadow-bookkeeping-aligned", shadow_target, VERDICT_OK), + ): + root = 0x70000000 + tree = _tree_for_va(root, shadow_va, shadow_target) + spec = { + "format": SPEC_FORMAT, "name": label, "firmware": "570.144", + "gmmu": KIND_VER2, "page_mode": "4k", "pdb_pa": f"{root:#x}", + "memory": [{"pa": f"{pa:#x}", "hex": b.hex()} for pa, b in _memory_items(tree)], + "contract": [{"label": "high-va-output", "va": f"{shadow_va:#x}", + "access": "write", "expect_pa": f"{expect:#x}", + "note": "Tier-2 shadow alias for a high-VA write leg"}], + } + report = audit(spec) + shadow_scenarios.append({"name": label, "expected_verdict": want, + "report": report}) + + demo = { + "format": DEMO_FORMAT, + "ok": (captured_report["ok"] + and scenarios[0]["report"]["ok"] + and scenarios[1]["report"]["contract"][0]["verdict"] == VERDICT_PA_MISMATCH + and scenarios[2]["report"]["contract"][0]["verdict"] == VERDICT_UNMAPPED + and all(s["report"]["contract"][0]["verdict"] == s["expected_verdict"] + for s in shadow_scenarios)), + "captured_channel": captured_report, + "qmd_hypotheses": scenarios, + "engine_write_shadow": shadow_scenarios, + } + if out: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, "launch-audit-demo.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(demo, fh, indent=2, sort_keys=True) + fh.write("\n") + demo["artifacts"] = {"demo_json": os.path.abspath(path)} + return demo + + +def _memory_items(mem: MemoryView): + return [(int(item["pa"], 16), mem.read(int(item["pa"], 16), item["size"])) + for item in mem.spans()] + + +CAPTURE_HINTS = """\ +CAPTURE HINTS (print-only; nothing is executed by this tool) +============================================================ +For a real launch audit you need the *engine's* table view, not the host's +bookkeeping. Capture, for one failing submission: + +1. The channel's page-directory base PA (PDB) -- from the channel's INSTANCE + dump field, or the SET_PAGE_DIRECTORY / GPU_PDB_ADDRESS your stack programs. +2. The table pages reachable from that root for the contract VAs: root, PDE2, + PDE1, PDE0 (dual slot) and the leaf page table(s). A raw PA->bytes dump of + those 4 KB pages is enough; keep them byte-exact (no pretty-printing). +3. The submitted stream's own pointers: GPFIFO ring, inline-QMD destination + (SET_INLINE_QMD_ADDRESS_A/B), the QMD's parameter/program/release-slot + addresses, and the channel report semaphore. +4. Your shadow/guest bookkeeping map for those same VAs (what you currently + check after a timeout). +5. Put 1-4 into a `greeneye-launch-audit-spec/1` JSON (see docs/48). + +ITERATE WITH `check` (one capture round per table level): + # start with only pdb_pa + contract and `"memory": []` (or a partial dump) + python3 harness/launch_audit.py check --spec m7-launch.json + # -> prints the FIRST missing table page per leg + `needed_pages[]` + # dump exactly those pages, append them to memory[], re-run check + # when check says complete: + python3 harness/launch_audit.py run --spec m7-launch.json + +CURRENT LIVE WALL (m7, 2026-10-06) -- exact dump list: + Host-side chain (the validated 5/5 fixture; dump all five raw pages): + root 0x12f16e000 + pd2 0x12f16f000 + pd1 0x11065e000 + pd0 0x110663000 (dual slot pd0[0].small -> pt) + pt 0x15908d000 (pt entries 256..260 = the low-window legs) + Forward-window PT pages (Tier-1 DMA_UPDATE_PDE_2 windows 4128 / 4131): + 0x1b8706000 (pdeIndex 4128) + 0x1197a6000 (pdeIndex 4131) + Geometry rule: pdeIndex = (pd1_idx << 8) | pd0_idx, so the forward windows + land at pd1 index 16. Their pd1[16] and pd0-level chain entries are + written by the SERVER on DMA_UPDATE_PDE_2 acceptance and are in no host + log -- include them (and any page they point at) if they fall outside the + host pages above; `check` names the exact PA. + Tier-2 shadows (high-VA legs): + * the 16 guest-scratch-PT page DMAs -- module stage-2 chunk list, + built as `guest scratch PT built: pages=16 vaBase=0x200000 + capacity=512`; shadow VA = 0x200000 + (page << 21) + (line << 16); + * the rmcall_gr_shadow chunk DMAs -- engine-write shadow pages for the + high-VA legs (the next wall class after dispatch clears). + Key contract VAs: + qmd-dest 0x204608000 write (QMD destination prewritten by host; + post-mortem reads the same hpa) + release-slot 0x204027f7c write (engine release; relw never fired) + ring 0x100000..0x104000 read/write (GPFIFO ring, push-buffer, + copy src/dst, channel semaphore) + param-alias 0x210000 read (Tier-2 shadow alias; NOT directly + mapped; PTEs come from guest scratch) + program-alias 0x220000 read (same) + indirect-alias 0x260000 read (shadow slot 6; same) + The high-VA param/program guest VAs (0x7e06f0.../0x735aa0...) are NOT + directly mapped: Tier-1 refuses status 0x1f and they resolve through the + shadow aliases above. + +COPY-PASTE SPEC SKELETON (start with empty memory and let `check` drive; fill +each `hex` with the 4096-byte raw dump of the page -- no pretty-printing): +{ + "format": "greeneye-launch-audit-spec/1", + "name": "m7-launch", + "firmware": "570.144", + "gmmu": "ver2", + "page_mode": "4k", + "pdb_pa": "0x12f16e000", + "memory": [], + "contract": [ + {"label": "qmd-dest", "va": "0x204608000", "access": "write", "expect_pa": "0x204608000"}, + {"label": "release-slot", "va": "0x204027f7c", "access": "write"}, + {"label": "gpfifo-ring", "va": "0x100000", "access": "read"}, + {"label": "push-buffer", "va": "0x101000", "access": "write"}, + {"label": "copy-src", "va": "0x102000", "access": "read"}, + {"label": "copy-dst", "va": "0x103000", "access": "write"}, + {"label": "semaphore", "va": "0x104000", "access": "write"}, + {"label": "param-alias", "va": "0x210000", "access": "read"}, + {"label": "program-alias", "va": "0x220000", "access": "read"}, + {"label": "indirect-alias","va": "0x260000", "access": "read"} + ] +} +Then: python3 harness/launch_audit.py check --spec m7-launch.json + +Reading the result: `pa_mismatch` means the engine reads/writes a different PA +than the one you are checking (look at the shadow/copy bookkeeping); +`unmapped` means the walk cannot reach any PA (the engine's access is dropped +or faults); `access_denied` means the leaf is read-only for a write leg. +`ok` means the engine CAN reach every leg -- the silence is downstream of the +address path (front-end/engine/context), not in the mapping. +""" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="launch_audit.py", + description="Audit a submission's engine-visible memory contract (offline).", + ) + sub = parser.add_subparsers(dest="command") + run = sub.add_parser("run", help="run an audit spec") + run.add_argument("--spec", required=True) + run.add_argument("--out", help="write the report JSON to this directory") + check_cmd = sub.add_parser( + "check", help="report the first table pages a partial/empty dump is missing") + check_cmd.add_argument("--spec", required=True) + check_cmd.add_argument("--out", help="write the report JSON to this directory") + demo = sub.add_parser("demo", help="captured contract + QMD/shadow hypotheses") + demo.add_argument("--out", help="output directory") + sub.add_parser("capture-hints", help="print what to capture for a real audit") + return parser + + +def _write_out(report: dict, out: str, name: str) -> str: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, name) + with open(path, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, sort_keys=True) + fh.write("\n") + return os.path.abspath(path) + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "run": + if not os.path.isfile(args.spec): + raise AuditError(f"spec not found: {args.spec}") + with open(args.spec, "r", encoding="utf-8") as fh: + spec = json.load(fh) + report = audit(spec) + for row in report["contract"]: + print(f"{row['verdict']:<14} {row['label']:<28} {row['va']}" + + (f" {row['detail']}" if row.get("detail") else "")) + summary = {"format": report["format"], "ok": report["ok"], + "summary": report["summary"]} + if args.out: + summary["report_json"] = _write_out(report, args.out, "launch-audit.json") + print(json.dumps(summary, indent=2, sort_keys=True)) + return 0 if report["ok"] else 3 + if args.command == "check": + if not os.path.isfile(args.spec): + raise AuditError(f"spec not found: {args.spec}") + with open(args.spec, "r", encoding="utf-8") as fh: + spec = json.load(fh) + report = check(spec) + for row in report["contract"]: + if row["status"] == "missing": + miss = row["first_missing"] + print(f"missing {row['label']:<28} {row['va']}" + f" need {miss['pa']} ({miss['level']}[{miss['index']}])") + elif row["status"] == "complete": + print(f"complete {row['label']:<28} {row['va']}" + f" -> {row['resolved_pa']}") + else: + print(f"fault {row['label']:<28} {row['va']}" + f" {row['fault']}: {row['detail']}") + if report["needed_pages"]: + print("dump next:") + for page in report["needed_pages"]: + print(f" {page['pa']} {'/'.join(page['levels'])}" + f" legs: {', '.join(page['legs'])}") + summary = { + "format": report["format"], + "complete": report["complete"], + "needed_pages": [page["pa"] for page in report["needed_pages"]], + } + if args.out: + summary["report_json"] = _write_out( + report, args.out, "launch-audit-check.json") + print(json.dumps(summary, indent=2, sort_keys=True)) + return 0 if report["complete"] else 3 + if args.command == "demo": + demo = run_demo(args.out) + cap = demo["captured_channel"] + print(f"captured channel contract: ok={cap['ok']} {cap['summary']}") + for scenario in demo["qmd_hypotheses"]: + row = scenario["report"]["contract"][0] + print(f"{row['verdict']:<14} {scenario['name']} {row.get('detail') or ''}") + for scenario in demo["engine_write_shadow"]: + row = scenario["report"]["contract"][0] + print(f"{row['verdict']:<14} {scenario['name']} {row.get('detail') or ''}") + print(json.dumps({"ok": demo["ok"]}, indent=2)) + return 0 if demo["ok"] else 3 + if args.command == "capture-hints": + print(CAPTURE_HINTS) + return 0 + parser.print_help() + return 2 + except (AuditError, WalkError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/log_decode.py b/debug/risc-v-gsp-rm-vm/harness/log_decode.py new file mode 100644 index 0000000..e1d351c --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/log_decode.py @@ -0,0 +1,660 @@ +#!/usr/bin/env python3 +"""Decode the two real GSP-RM log dump forms into one normalized record stream. + +The live GSP-RM writes its "server stderr" into an 8 KiB-word libos log ring. +Two dump forms exist in the campaign evidence and they are the inputs here: + +1. Raw ring dumps (debugfs ``gsp-phys-log-dump`` saved to evidence files). + A ``<LABEL>-raw: <n> bytes (...) VA=... dma=...`` header is followed by + hexdump rows ``00000000: 000002d2 00000000 ...`` (little-endian u64 words). + Each header's rows decode into one ``raw-hex`` record whose ``frames`` are + the u64 words and whose ``raw`` is the decoded byte buffer as lowercase + hex. A later ``<LABEL>-decoded:`` marker switches to the offline + liblogdecode text records ``rec 449 ts=0x.. meta=0x.. args=0x..`` which + become ``record`` entries (``args`` -> ``frames``); both the + ``args=[0x.., 0x..]`` and ``args=0x..,0x..`` spellings are accepted. + +2. Live dmesg / evidence lines, e.g.:: + + vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" faultingEngine="" + vgsp-nocat: rec seq=1 assertCallStack count=1 [0x401000,0x402000] (offsets from load base 0x4000000) + vgsp-nocat: entry[0]: count=2 recType=5 errorCode=0x57f1344 firstSeq=1 lastSeq=2 source="ASSERT" + vgsp: WARNING: ... (NOCAT recType=5 errorCode=0x52a8b04) ... + rec 449 ts=0x18dbb29118ec6200 meta=0x0000000005bb9c48 args=0x1,0x2 + + ``key=value`` pairs fill ``rec_type`` / ``error_code`` / ``source_name`` / + ``seq``; ``assertCallStack [..]`` lines attach their bracketed frame list + to the immediately preceding record with the same ``seq``; anything that + matches none of the shapes is counted in ``unparsed`` (never dropped). + Lines starting with ``#``, ``<LABEL>-raw:``/``<LABEL>-decoded:`` headers and + blank lines are structure, not records. + +Report schema ``greeneye-log-decode/1``:: + + {"schema": "greeneye-log-decode/1", + "source": "<basename of the input>", + "count": <records after filters>, + "unparsed": <lines that matched no shape>, + "records": [{"kind": "raw-hex"|"record", "label"?: str, + "rec_type"?: int|str, "error_code"?: "0x...", + "source_name"?: str, "seq"?: int, + "frames": ["0x...", ...], "raw": str, + "resolved"?: [{"frame": "0x...", "symbol": str|None, + "offset": "0x..."|None}, ...]}], + "histogram": {"rec_type": {..}, "error_code": {..}, "source": {..}}} + +``resolved`` appears only when a symbol table is supplied and is aligned 1:1 +with ``frames``. ``--symbols`` accepts the project symbol-DB shape: a +top-level list, or an object with a ``symbols`` list, of entries +``{"name": str, "addr": int | "0x..."}`` (extra keys ignored). Each frame is +matched to the nearest symbol with ``addr <= frame`` and the offset +(``frame - addr``) is reported; an unreadable or malformed symbols file exits +non-zero with a clear message. + +Filtering (``decode`` subcommand): ``--kind`` selects raw-hex/record, +``--rec-type``/``--error-code`` accept decimal or ``0x`` hex, ``--source`` is +a case-insensitive substring on ``source_name`` and ``--limit`` caps the +records; ``count``/``histogram`` describe the filtered stream. ``--json +out.json`` writes the full report. + +Honesty: offline decoder only (L1 at best; no timing claims). Raw words are +exposed, not structurally ring-decoded; ``assertCallStack`` frames are offsets +from the load base named in the line and are symbolized as given. There is no +hardware, module, QEMU or guest interaction. + +Run:: + + python3 harness/log_decode.py decode <dump> [--symbols symbols.json] + python3 harness/log_decode.py demo +""" +from __future__ import annotations + +import argparse +import bisect +import json +import os +import re +import sys +import tempfile +from typing import Any + +SCHEMA = "greeneye-log-decode/1" +DEMO_FORMAT = "greeneye-log-decode-demo/1" + +#: How many ``resolved`` entries the human-readable output prints per record +#: (the JSON report always carries all of them). +RESOLVED_PRINT_LIMIT = 8 + + +class LogDecodeError(ValueError): + """An input or request the decoder cannot honour.""" + + +# --------------------------------------------------------------------------- +# Symbols: the project symbol-DB shape, nearest-below resolution +# --------------------------------------------------------------------------- + +def _parse_sym_addr(value: Any, where: str) -> int: + if isinstance(value, bool): + raise LogDecodeError(f"{where}: bad symbol addr {value!r}") + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value, 0) + except ValueError as exc: + raise LogDecodeError(f"{where}: bad symbol addr {value!r}") from exc + raise LogDecodeError(f"{where}: bad symbol addr {value!r}") + + +class SymbolTable: + """Nearest-below ``frame -> (name, offset)`` resolver. + + Holds the same entries the project symbol DBs carry; only ``addr`` and + ``name`` are used here. ``resolve`` returns ``None`` when no symbol has an + address at or below the frame. + """ + + def __init__(self, entries: list[tuple[int, str]]) -> None: + self._entries = sorted(entries, key=lambda item: item[0]) + self._addrs = [addr for addr, _ in self._entries] + + def __len__(self) -> int: + return len(self._entries) + + def resolve(self, addr: int) -> tuple[str, int] | None: + pos = bisect.bisect_right(self._addrs, addr) - 1 + if pos < 0: + return None + sym_addr, name = self._entries[pos] + return name, addr - sym_addr + + def annotate_frame(self, frame: str) -> dict: + try: + value = int(frame, 0) + except ValueError: + return {"frame": frame, "symbol": None, "offset": None} + hit = self.resolve(value) + if hit is None: + return {"frame": frame, "symbol": None, "offset": None} + name, offset = hit + return {"frame": frame, "symbol": name, "offset": f"{offset:#x}"} + + +def load_symbols(path: str) -> SymbolTable: + """Load the accepted symbol JSON; raise :class:`LogDecodeError` when bad.""" + if not os.path.isfile(path): + raise LogDecodeError(f"symbols file not found: {path}") + try: + with open(path, "r", encoding="utf-8") as fh: + payload = json.load(fh) + except OSError as exc: + raise LogDecodeError(f"cannot read symbols {path}: {exc}") from exc + except json.JSONDecodeError as exc: + raise LogDecodeError(f"symbols JSON {path} is not valid JSON: {exc}") from exc + if isinstance(payload, dict): + entries = payload.get("symbols") + if entries is None: + raise LogDecodeError( + f"symbols JSON {path}: object must have a 'symbols' list") + elif isinstance(payload, list): + entries = payload + else: + raise LogDecodeError( + f"symbols JSON {path}: expected a list or an object with 'symbols'") + if not isinstance(entries, list): + raise LogDecodeError(f"symbols JSON {path}: 'symbols' must be a list") + table: list[tuple[int, str]] = [] + for index, item in enumerate(entries): + where = f"symbols JSON {path}: entry[{index}]" + if not isinstance(item, dict): + raise LogDecodeError(f"{where} is not an object") + if "addr" not in item: + raise LogDecodeError(f"{where} has no 'addr'") + addr = _parse_sym_addr(item["addr"], where) + name = item.get("name") + if name is None: + name = f"sub_{addr:x}" + if not isinstance(name, str): + raise LogDecodeError(f"{where}: 'name' must be a string") + table.append((addr, name)) + return SymbolTable(table) + + +# --------------------------------------------------------------------------- +# Line shapes observed in the real evidence +# --------------------------------------------------------------------------- + +RAW_SECTION_RE = re.compile(r"^(?P<label>[A-Za-z0-9_.+-]+)-raw:\s+\d+\s+bytes\b") +DECODED_SECTION_RE = re.compile(r"^(?P<label>[A-Za-z0-9_.+-]+)-decoded:") +COMMENT_LABEL_RE = re.compile(r"^#\s*(?P<label>[A-Za-z][A-Za-z0-9_.+-]*)\s*$") +HEX_ROW_RE = re.compile( + r"^(?P<offset>[0-9a-fA-F]{6,16}):\s+" + r"(?P<words>[0-9a-fA-F]{1,16}(?:\s+[0-9a-fA-F]{1,16})*)\s*$") +DMSG_PREFIX_RE = re.compile( + r"^(?:\[[^\]]*\]\s*)?(?P<label>[A-Za-z0-9_.+-]+):\s+(?P<body>.*)$") +KV_RE = re.compile(r"(?P<key>[A-Za-z_][A-Za-z0-9_]*)=" + r"(?P<value>\"[^\"]*\"|[^\s,()]+)") +FRAME_LIST_RE = re.compile(r"\[([^\[\]]*)\]") +CONTINUATION_RE = re.compile(r"\brec\s+seq=(?P<seq>\d+)\s+assertCallStack\b") +DECODED_REC_RE = re.compile( + r"^rec\s+(?P<seq>\d+)\s+ts=(?P<ts>0x[0-9a-fA-F]+)\s+" + r"meta=(?P<meta>\S+)\s+args=(?P<args>.*)$") +REC_SEQ_RE = re.compile(r"\brec\s+seq=(?P<seq>\d+)") + + +def _parse_kv(body: str) -> dict[str, str]: + fields: dict[str, str] = {} + for match in KV_RE.finditer(body): + value = match.group("value") + if value.startswith('"') and value.endswith('"') and len(value) >= 2: + value = value[1:-1] + fields[match.group("key")] = value + return fields + + +def _int_or_text(value: str) -> int | str: + try: + return int(value, 0) + except ValueError: + return value + + +def _norm_hex(value: str) -> str: + try: + return f"{int(value, 0):#x}" + except ValueError: + return value + + +def _parse_frame_tokens(text: str) -> list[str]: + text = text.strip() + if text.startswith("[") and text.endswith("]"): + text = text[1:-1] + frames: list[str] = [] + for token in text.replace(",", " ").split(): + try: + frames.append(f"{int(token, 0):#x}") + except ValueError: + continue + return frames + + +def _frames_from_brackets(text: str) -> list[str]: + match = FRAME_LIST_RE.search(text) + if not match: + return [] + return _parse_frame_tokens(match.group(1)) + + +def _looks_like_record(body: str) -> bool: + if DECODED_REC_RE.match(body): + return True + if "recType=" in body or "errorCode=" in body: + return True + if REC_SEQ_RE.search(body): + return True + return False + + +def _record_from_body(body: str, label: str | None, raw: str | None = None) -> dict: + if raw is None: + raw = body + match = DECODED_REC_RE.match(body) + if match: + return { + "kind": "record", + "label": label, + "seq": int(match.group("seq")), + "frames": _parse_frame_tokens(match.group("args")), + "raw": raw, + } + record: dict[str, Any] = {"kind": "record", "label": label, + "frames": [], "raw": raw} + fields = _parse_kv(body) + if "recType" in fields: + record["rec_type"] = _int_or_text(fields["recType"]) + if "errorCode" in fields: + record["error_code"] = _norm_hex(fields["errorCode"]) + if "source" in fields: + record["source_name"] = fields["source"] + if "seq" in fields: + record["seq"] = int(fields["seq"], 0) + return record + + +def _raw_record(label: str | None, words: list[int]) -> dict: + data = b"".join(word.to_bytes(8, "little") for word in words) + return { + "kind": "raw-hex", + "label": label, + "source_name": label, + "frames": [f"{word:#x}" for word in words], + "raw": data.hex(), + } + + +# --------------------------------------------------------------------------- +# Decode +# --------------------------------------------------------------------------- + +def _histogram(records: list[dict]) -> dict[str, dict[str, int]]: + hist: dict[str, dict[str, int]] = {"rec_type": {}, "error_code": {}, + "source": {}} + for record in records: + if "rec_type" in record: + key = str(record["rec_type"]) + hist["rec_type"][key] = hist["rec_type"].get(key, 0) + 1 + if "error_code" in record: + key = str(record["error_code"]) + hist["error_code"][key] = hist["error_code"].get(key, 0) + 1 + if record.get("source_name"): + key = str(record["source_name"]) + hist["source"][key] = hist["source"].get(key, 0) + 1 + return {name: dict(sorted(values.items())) for name, values in hist.items()} + + +def decode_text(text: str, source: str = "<text>", + symbols: SymbolTable | None = None) -> dict: + """Decode one dump/evidence text into a ``greeneye-log-decode/1`` report.""" + records: list[dict] = [] + unparsed = 0 + raw_state: dict[str, Any] | None = None + current_label: str | None = None + + def flush_raw() -> None: + nonlocal raw_state + if raw_state is None: + return + records.append(_raw_record(raw_state["label"], raw_state["words"])) + raw_state = None + + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + if stripped.startswith("#"): + match = COMMENT_LABEL_RE.match(stripped) + if match: + current_label = match.group("label") + continue + match = RAW_SECTION_RE.match(stripped) + if match: + flush_raw() + raw_state = {"label": match.group("label"), "words": []} + current_label = match.group("label") + continue + match = DECODED_SECTION_RE.match(stripped) + if match: + flush_raw() + current_label = match.group("label") + continue + if raw_state is not None: + match = HEX_ROW_RE.match(stripped) + if match: + raw_state["words"].extend( + int(token, 16) for token in match.group("words").split()) + continue + flush_raw() + else: + match = HEX_ROW_RE.match(stripped) + if match: + raw_state = {"label": None, "words": []} + raw_state["words"].extend( + int(token, 16) for token in match.group("words").split()) + continue + + label = current_label + body = stripped + prefix = DMSG_PREFIX_RE.match(stripped) + if prefix: + label = prefix.group("label") + body = prefix.group("body").strip() + + continuation = CONTINUATION_RE.search(body) + if continuation: + seq = int(continuation.group("seq")) + frames = _frames_from_brackets(body) + last = records[-1] if records else None + if (last is not None and last.get("kind") == "record" + and last.get("seq") == seq and not last.get("frames")): + last["frames"] = frames + else: + records.append({"kind": "record", "label": label, "seq": seq, + "frames": frames, "raw": stripped}) + continue + + if _looks_like_record(body): + records.append(_record_from_body(body, label, stripped)) + continue + unparsed += 1 + + flush_raw() + if symbols is not None: + for record in records: + if record.get("frames"): + record["resolved"] = [symbols.annotate_frame(frame) + for frame in record["frames"]] + return { + "schema": SCHEMA, + "source": source, + "count": len(records), + "unparsed": unparsed, + "records": records, + "histogram": _histogram(records), + } + + +def decode_path(path: str, symbols_path: str | None = None) -> dict: + """Decode the file at ``path``; ``source`` in the report is its basename.""" + if not os.path.isfile(path): + raise LogDecodeError(f"input not found: {path}") + symbols = load_symbols(symbols_path) if symbols_path else None + try: + with open(path, "r", encoding="utf-8", errors="replace") as fh: + text = fh.read() + except OSError as exc: + raise LogDecodeError(f"cannot read input {path}: {exc}") from exc + return decode_text(text, source=os.path.basename(path), symbols=symbols) + + +# --------------------------------------------------------------------------- +# Filters +# --------------------------------------------------------------------------- + +def _want_int(value: Any, what: str) -> int: + if isinstance(value, bool): + raise LogDecodeError(f"bad {what} {value!r}") + if isinstance(value, int): + return value + try: + return int(str(value), 0) + except ValueError as exc: + raise LogDecodeError(f"bad {what} {value!r}") from exc + + +def _match_int(record_value: Any, wanted: int) -> bool: + if record_value is None: + return False + if isinstance(record_value, int): + return record_value == wanted + try: + return int(str(record_value), 0) == wanted + except ValueError: + return False + + +def apply_filters(report: dict, *, rec_type: Any = None, source: str | None = None, + error_code: Any = None, kind: str | None = None, + limit: int | None = None) -> dict: + """Return a copy of ``report`` with the requested record filters applied.""" + records = list(report["records"]) + if kind is not None: + if kind not in ("raw-hex", "record"): + raise LogDecodeError(f"bad kind {kind!r} (want raw-hex or record)") + records = [r for r in records if r["kind"] == kind] + if rec_type is not None: + wanted = _want_int(rec_type, "rec-type") + records = [r for r in records if _match_int(r.get("rec_type"), wanted)] + if error_code is not None: + wanted = _want_int(error_code, "error-code") + records = [r for r in records if _match_int(r.get("error_code"), wanted)] + if source is not None: + needle = source.lower() + records = [r for r in records + if needle in str(r.get("source_name") or "").lower()] + if limit is not None: + if limit < 0: + raise LogDecodeError(f"bad limit {limit!r} (must be >= 0)") + records = records[:limit] + filtered = dict(report) + filtered["records"] = records + filtered["count"] = len(records) + filtered["histogram"] = _histogram(records) + return filtered + + +# --------------------------------------------------------------------------- +# Demo: tiny crafted inputs for both real formats (tempdir only) +# --------------------------------------------------------------------------- + +_DEMO_RAW = """\ +LOGINIT-raw: 64 bytes (allocated 64) VA=0xffff0000 dma=0x1000 +00000000: 00000005 00000000 0000000a deadbeef +00000010: 01020304 05060708 090a0b0c 0d0e0f10 +""" + +_DEMO_RECORDS = """\ +[Tue Oct 6 12:40:52 2026] vgsp-nocat: rec seq=1 recType=5 errorCode=0x57f1344 subsystem=0x0 bugcheck=0x0 payload=1212 source="ASSERT" faultingEngine="" +[Tue Oct 6 12:40:52 2026] vgsp-nocat: rec seq=1 assertCallStack count=1 [0x401000,0x402000,0x0] (offsets from load base 0x4000000) +[Tue Oct 6 12:40:53 2026] vgsp-nocat: entry[0]: count=1 recType=5 errorCode=0x57f1344 firstSeq=1 lastSeq=1 source="ASSERT" +an unmatched filler line +rec 7 ts=0x18dbb29118ec6200 meta=0x0000000005bb9c48 args=0x401000,0x402000 +""" + +_DEMO_SYMBOLS = { + "symbols": [ + {"name": "gr_init", "addr": "0x401000"}, + {"name": "gr_done", "addr": int("0x403000", 0)}, + ] +} + + +def run_demo() -> dict: + """Decode crafted inputs for both formats; deterministic, tempdir only.""" + with tempfile.TemporaryDirectory(prefix="log-decode-demo-") as tmp: + raw_path = os.path.join(tmp, "raw.txt") + records_path = os.path.join(tmp, "records.txt") + symbols_path = os.path.join(tmp, "symbols.json") + with open(raw_path, "w", encoding="utf-8") as fh: + fh.write(_DEMO_RAW) + with open(records_path, "w", encoding="utf-8") as fh: + fh.write(_DEMO_RECORDS) + with open(symbols_path, "w", encoding="utf-8") as fh: + json.dump(_DEMO_SYMBOLS, fh, indent=2, sort_keys=True) + fh.write("\n") + raw_report = decode_path(raw_path) + records_report = decode_path(records_path, symbols_path) + + raw_records = raw_report["records"] + record_records = records_report["records"] + first = record_records[0] if record_records else {} + resolved = first.get("resolved") or [] + ok = ( + raw_report["count"] == 1 + and raw_report["unparsed"] == 0 + and len(raw_records) == 1 + and len(raw_records[0]["frames"]) == 8 + and raw_records[0]["frames"][3] == "0xdeadbeef" + and len(raw_records[0]["raw"]) == 2 * 8 * 8 + and records_report["count"] == 3 + and records_report["unparsed"] == 1 + and first.get("rec_type") == 5 + and first.get("error_code") == "0x57f1344" + and first.get("source_name") == "ASSERT" + and first.get("frames") == ["0x401000", "0x402000", "0x0"] + and len(resolved) == 3 + and resolved[0].get("symbol") == "gr_init" + and resolved[1].get("symbol") == "gr_init" + and resolved[1].get("offset") == "0x1000" + and resolved[2].get("symbol") is None + ) + return { + "format": DEMO_FORMAT, + "ok": ok, + "raw": raw_report, + "records": records_report, + } + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def _format_record(record: dict) -> list[str]: + bits = [f"{record['kind']:<8}"] + for key in ("label", "seq", "rec_type", "error_code", "source_name"): + if record.get(key) is not None: + bits.append(f"{key}={record[key]}") + bits.append(f"frames={len(record.get('frames') or [])}") + lines = [" ".join(bits)] + resolved = record.get("resolved") or [] + for item in resolved[:RESOLVED_PRINT_LIMIT]: + if item.get("symbol") is None: + lines.append(f" {item['frame']} <unknown>") + else: + lines.append( + f" {item['frame']} {item['symbol']}+{item['offset']}") + if len(resolved) > RESOLVED_PRINT_LIMIT: + lines.append(f" ... {len(resolved) - RESOLVED_PRINT_LIMIT} more") + return lines + + +def _write_json(report: dict, path: str) -> str: + target = os.path.abspath(path) + parent = os.path.dirname(target) + if parent: + os.makedirs(parent, exist_ok=True) + with open(target, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, sort_keys=True) + fh.write("\n") + return target + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="log_decode.py", + description="Decode GSP-RM log-ring and NOCAT record dumps (offline).", + ) + sub = parser.add_subparsers(dest="command") + decode = sub.add_parser( + "decode", help="decode a raw ring dump or dmesg/evidence log") + decode.add_argument("input", help="dump/log text file to decode") + decode.add_argument("--symbols", help="symbol DB JSON (list or {symbols:[..]})") + decode.add_argument("--rec-type", help="keep records with this recType") + decode.add_argument("--source", help="case-insensitive source substring") + decode.add_argument("--error-code", help="keep records with this errorCode") + decode.add_argument("--kind", choices=("raw-hex", "record"), + help="keep only this record kind") + decode.add_argument("--limit", type=int, help="max records to emit") + decode.add_argument("--json", help="write the full report to this file") + sub.add_parser("demo", help="decode crafted samples for both real formats") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "decode": + report = decode_path(args.input, args.symbols) + report = apply_filters( + report, + rec_type=args.rec_type, + source=args.source, + error_code=args.error_code, + kind=args.kind, + limit=args.limit, + ) + for record in report["records"]: + for line in _format_record(record): + print(line) + summary = { + "schema": report["schema"], + "source": report["source"], + "count": report["count"], + "unparsed": report["unparsed"], + "histogram": report["histogram"], + } + if args.json: + summary["report_json"] = _write_json(report, args.json) + print(json.dumps(summary, indent=2, sort_keys=True)) + return 0 + if args.command == "demo": + demo = run_demo() + raw = demo["raw"] + records = demo["records"] + print(f"{raw['source']}: records={raw['count']} " + f"unparsed={raw['unparsed']} " + f"frames={len(raw['records'][0]['frames'])} " + f"raw_bytes={len(raw['records'][0]['raw']) // 2}") + print(f"{records['source']}: records={records['count']} " + f"unparsed={records['unparsed']}") + print(f"histogram: {json.dumps(records['histogram'], sort_keys=True)}") + for item in (records["records"][0].get("resolved") or []): + if item.get("symbol") is None: + print(f" {item['frame']} <unknown>") + else: + print(f" {item['frame']} {item['symbol']}+{item['offset']}") + print(json.dumps({"format": demo["format"], "ok": demo["ok"]}, + sort_keys=True)) + return 0 if demo["ok"] else 3 + parser.print_help() + return 2 + except LogDecodeError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/rehost.py b/debug/risc-v-gsp-rm-vm/harness/rehost.py new file mode 100644 index 0000000..fb554db --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/rehost.py @@ -0,0 +1,428 @@ +#!/usr/bin/env python3 +"""Surgical rehost: run a bounded function of a firmware image on the native emulator. + +The standalone simulator boots the real 570.144 image to the message-loop port +wait (B3). Code *after* that wall (the RM handlers, e.g. the GR promote/commit +chain) is only reachable when host traffic exists, which is the B4/B5 rung. +This harness provides the offline alternative the workflow review ranks first +(docs/39 U2): start the machine at a chosen function entry with a crafted +register/memory state, run forward with a bounded budget, and stop on return, +fault, unhandled ecall or budget. Every run records a session and a report; +snapshots before/after a run show what the function changed. + +Honesty rules: + +* This tool executes firmware code; it never touches a GPU, kernel module, + mdev, guest or network. It is development-only. +* A rehosted run is **not** a boot: the function runs with only the state the + caller supplied. The report says so, and any fault (first missing state) is + recorded as evidence, never papered over. +* No timing/cycle claim; fidelity is L0/L1 (docs/06). +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sys +import time +from typing import Any + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from greeneye.elf import load_elf # noqa: E402 +from greeneye.errors import GreenEyeError # noqa: E402 +from greeneye.sim_backend import launch_native # noqa: E402 +from sim.devices import load_profile, registry_from_profile # noqa: E402 +from sim.gsp_boot import _apply_boot_shim # noqa: E402 +from sim.libos import LibosSyscalls # noqa: E402 +from sim.machine import R_BP, R_ECALL, R_FAULT, R_WATCH # noqa: E402 + +#: Schema tag of the rehost report. +REHOST_FORMAT = "greeneye-rehost/1" +#: Schema tag of the demo result. +DEMO_FORMAT = "greeneye-rehost-demo/1" + +DEFAULT_IMAGE = os.path.join( + TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin" +) +DEFAULT_PROFILE = os.path.join(TOOL_ROOT, "sim", "profiles", "tu104-570.144.json") +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") + +#: Wall-clock guard: a rehost that spins in a fault loop must not hang the tool. +DEFAULT_TIMEOUT = 60.0 +#: Instruction budget per default run. +DEFAULT_BUDGET = 5_000_000 +#: Chunk size between stop checks (keeps the loop responsive to timeouts). +RUN_CHUNK = 250_000 +#: Sentinel instruction (``ebreak``) written at the return trampoline. +EBREAK = 0x00100073 + + +class RehostError(GreenEyeError): + """A rehost input, image or argument is invalid.""" + + +def _parse_u64(value: Any) -> int: + if isinstance(value, bool): + return int(value) + if isinstance(value, int): + return value & 0xFFFF_FFFF_FFFF_FFFF + if isinstance(value, str): + text = value.strip() + if not text: + raise RehostError("empty integer") + return int(text, 0) & 0xFFFF_FFFF_FFFF_FFFF + raise RehostError(f"cannot parse integer from {value!r}") + + +def parse_hex_bytes(spec: str) -> bytes: + """Parse ``aabbcc`` (whitespace tolerated) into bytes.""" + text = "".join(spec.split()) + if len(text) % 2: + raise RehostError(f"hex payload must have an even number of digits: {spec!r}") + try: + return bytes.fromhex(text) + except ValueError as exc: + raise RehostError(f"bad hex payload {spec!r}: {exc}") from exc + + +def parse_mem_spec(spec: str) -> tuple[int, bytes]: + """Parse ``<addr>=<hex>`` into ``(address, bytes)``.""" + if "=" not in spec: + raise RehostError(f"memory spec must be <addr>=<hex>: {spec!r}") + addr_text, _, data_text = spec.partition("=") + return _parse_u64(addr_text), parse_hex_bytes(data_text) + + +def parse_snap_spec(spec: str) -> tuple[int, int]: + """Parse ``<addr>:<size>`` into ``(address, size)``.""" + addr_text, sep, size_text = spec.partition(":") + if not sep: + raise RehostError(f"snapshot spec must be <addr>:<size>: {spec!r}") + size = _parse_u64(size_text) + if size <= 0: + raise RehostError(f"snapshot size must be positive: {spec!r}") + return _parse_u64(addr_text), size + + +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() + + +def _region_diff(before: bytes, after: bytes) -> dict: + changed = 0 + first = None + for offset, (a, b) in enumerate(zip(before, after)): + if a != b: + changed += 1 + if first is None: + first = offset + return { + "changed_bytes": changed, + "first_diff_offset": first, + "before_sha256": hashlib.sha256(before).hexdigest(), + "after_sha256": hashlib.sha256(after).hexdigest(), + } + + +def _registers(machine) -> dict: + regs = {f"x{i}": machine.x[i] for i in range(32) if i and machine.x[i]} + regs["pc"] = machine.pc + return regs + + +def rehost( + *, + image: str, + entry: int | str, + profile_path: str | None = None, + args: tuple = (), + mem: tuple = (), + snap: tuple = (), + budget: int = DEFAULT_BUDGET, + timeout: float = DEFAULT_TIMEOUT, + out: str | None = None, + record=None, + name: str = "rehost", + apply_shim: bool = True, + unknown_mmio: str = "lenient", +) -> dict: + """Run one bounded function call on the native emulator; return a report.""" + if not os.path.isfile(image): + raise RehostError(f"image not found: {image}") + if budget <= 0: + raise RehostError(f"budget must be positive, got {budget}") + image_obj = load_elf(image) + if not image_obj.is_riscv: + raise RehostError(f"{image}: not a RISC-V ELF image") + + profile = load_profile(profile_path) if profile_path else None + load = (profile or {}).get("load", {}) + ram_base = _parse_u64(load["ram_base"]) if "ram_base" in load else 0x80000000 + memory_mb = int(load.get("memory_mb", 256)) + + session = launch_native( + kernel=image, + memory_mb=memory_mb, + ram_base=ram_base, + record=record if record is not None else bool(out), + name=name, + device_map=None, + unknown_mmio=unknown_mmio, + bias=0, + ) + machine = session.target.machine + try: + if profile is not None: + machine.devices = registry_from_profile(profile, machine=machine) + machine.translator = getattr(machine.devices, "dma", None) + libos = LibosSyscalls() + machine.syscall_handler = libos.handle + if apply_shim: + shim_meta = {"enabled": True, "assumptions": []} + shim = dict(profile.get("boot_shim", {})) + shim["_image_path"] = image + _apply_boot_shim(machine, shim, image_obj, shim_meta) + + # Resolve a symbol name through the ELF symbol table when possible. + if isinstance(entry, str): + try: + entry_addr = session.symbols.lookup(entry) if session.symbols else None + except GreenEyeError: + entry_addr = None + if entry_addr is None: + raise RehostError(f"symbol {entry!r} not found in {os.path.basename(image)}") + else: + entry_addr = _parse_u64(entry) + + for addr, data in mem: + machine.write_mem(addr, data) + + snapshots = [] + for addr, size in snap: + before = machine.read_mem(addr, size) + snapshots.append({"addr": addr, "size": size, "before": before}) + + # Return trampoline well above the image, inside the RAM arena. + sentinel = (machine.ram_base + machine.ram_size - 0x100) & ~0xF + stack = (sentinel - 0x2000) & ~0xF + trampoline = EBREAK.to_bytes(4, "little") + machine.write_mem(sentinel, trampoline) + machine.write_mem(sentinel + 4, trampoline) + for index, value in enumerate(tuple(args)[:8]): + machine.x[10 + index] = _parse_u64(value) + machine.x[1] = sentinel # ra + machine.x[2] = stack # sp + machine.pc = entry_addr + + entered = _registers(machine) + started = time.monotonic() + status = "budget" + fault: dict | None = None + stop_pc = machine.pc + remaining = budget + while remaining > 0: + if time.monotonic() - started > timeout: + status = "timeout" + break + chunk = min(RUN_CHUNK, remaining) + reason, stop_pc, a, b = machine.run(chunk, breakpoints={sentinel}) + remaining -= chunk + if reason == R_BP: + status = "return" + break + if reason == R_FAULT: + status = "fault" + fault = {"cause": a, "tval": f"{b:#x}"} + break + if reason == R_ECALL: + handled, result = machine.syscall_handler(machine, stop_pc) if machine.syscall_handler else (False, None) + if not handled: + status = "ecall" + break + if result is not None: + machine.x[10] = result & 0xFFFF_FFFF_FFFF_FFFF + continue + if reason == R_WATCH: + continue # watchpoints recorded by the caller's own tooling + # R_BUDGET: keep going until the total budget is spent. + if status == "budget" and remaining <= 0: + status = "budget" + + report: dict[str, Any] = { + "format": REHOST_FORMAT, + "image": os.path.abspath(image), + "image_sha256": _sha256_file(image), + "profile": os.path.basename(profile_path) if profile_path else None, + "entry": f"{entry_addr:#x}", + "args": [_parse_u64(v) for v in tuple(args)[:8]], + "status": status, + "stop_pc": f"{stop_pc:#x}", + "fault": fault, + "retval": machine.x[10], + "instructions": 0 if machine.insns is None else machine.insns, + "wall_seconds": round(time.monotonic() - started, 4), + "registers_entered": {k: f"{v:#x}" if isinstance(v, int) else v for k, v in entered.items()}, + "registers_exited": {k: f"{v:#x}" if isinstance(v, int) else v for k, v in _registers(machine).items()}, + "mmio_unknown": machine.mmio_unknown, + "snapshots": [], + "limits": [ + "Not a boot: the function ran with only the supplied state.", + "No timing/cycle claim (L0/L1 fidelity, docs/06).", + "A fault is the first missing state observed, not proof of a product defect.", + ], + } + for item in snapshots: + after = machine.read_mem(item["addr"], item["size"]) + report["snapshots"].append( + { + "addr": f"{item['addr']:#x}", + "size": item["size"], + **_region_diff(item["before"], after), + } + ) + if getattr(session.recorder, "dir", None) is not None: + session.note("rehost", entry=f"{entry_addr:#x}", status=status, stop_pc=f"{stop_pc:#x}") + report["session"] = session.recorder.dir + if out: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, "rehost-report.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, sort_keys=True) + fh.write("\n") + report["report_json"] = os.path.abspath(path) + return report + finally: + session.close() + + +def run_demo(out: str | None = None) -> dict: + """Rehost a known demo function to prove the loop (synthetic, deterministic).""" + if not os.path.isfile(DEMO_ELF): + raise RehostError( + f"demo firmware not built: {DEMO_ELF} (run `make demo` in the tool dir)" + ) + buffer_addr, payload = 0x80012000, bytes([1, 2, 3, 4]) + expected = sum(payload) + first = rehost( + image=DEMO_ELF, + entry="rpc_sum", + args=(buffer_addr, len(payload)), + mem=((buffer_addr, payload),), + snap=(), + budget=100_000, + timeout=20.0, + record=False, + name="rehost-demo", + ) + second = rehost( + image=DEMO_ELF, + entry="rpc_sum", + args=(buffer_addr, len(payload)), + mem=((buffer_addr, payload),), + snap=(), + budget=100_000, + timeout=20.0, + record=False, + name="rehost-demo", + ) + ok = ( + first["status"] == "return" + and first["retval"] == expected + and first["status"] == second["status"] + and first["retval"] == second["retval"] + ) + payload_result = { + "format": DEMO_FORMAT, + "ok": ok, + "expected": expected, + "first": { + "status": first["status"], + "retval": first["retval"], + "wall_seconds": first["wall_seconds"], + }, + "second": { + "status": second["status"], + "retval": second["retval"], + }, + } + if out: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, "rehost-demo.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload_result, fh, indent=2, sort_keys=True) + fh.write("\n") + payload_result["artifacts"] = {"demo_json": os.path.abspath(path)} + return payload_result + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="rehost.py", + description=( + "Run one bounded firmware function on the native emulator " + "(surgical rehost). Offline only; never touches hardware." + ), + ) + sub = parser.add_subparsers(dest="command") + run_cmd = sub.add_parser("run", help="rehost a function from an image") + run_cmd.add_argument("--image", default=DEFAULT_IMAGE, help="firmware ELF") + run_cmd.add_argument("--profile", default=DEFAULT_PROFILE, help="device profile JSON") + run_cmd.add_argument("--entry", required=True, help="function entry (hex address or symbol)") + run_cmd.add_argument("--arg", action="append", default=[], help="a0..a7 value (repeatable)") + run_cmd.add_argument("--mem", action="append", default=[], help="<addr>=<hex> pre-write (repeatable)") + run_cmd.add_argument("--snap", action="append", default=[], help="<addr>:<size> snapshot (repeatable)") + run_cmd.add_argument("--budget", type=int, default=DEFAULT_BUDGET) + run_cmd.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT) + run_cmd.add_argument("--no-shim", action="store_true", help="do not apply the profile boot shim") + run_cmd.add_argument("--out", help="output directory (report + session)") + demo_cmd = sub.add_parser("demo", help="synthetic self-test on the demo firmware") + demo_cmd.add_argument("--out", help="output directory") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "run": + report = rehost( + image=args.image, + profile_path=args.profile or None, + entry=args.entry, + args=tuple(args.arg), + mem=tuple(parse_mem_spec(spec) for spec in args.mem), + snap=tuple(parse_snap_spec(spec) for spec in args.snap), + budget=args.budget, + timeout=args.timeout, + out=args.out, + apply_shim=not args.no_shim, + ) + elif args.command == "demo": + report = run_demo(args.out) + else: + parser.print_help() + return 2 + except RehostError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + summary_keys = ( + "format", "ok", "entry", "status", "stop_pc", "retval", "fault", + "instructions", "wall_seconds", "report_json", "session", "artifacts", + ) + print(json.dumps({k: report[k] for k in summary_keys if k in report}, indent=2, sort_keys=True)) + if args.command == "demo" and not report.get("ok"): + return 3 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/rm_capture.py b/debug/risc-v-gsp-rm-vm/harness/rm_capture.py new file mode 100644 index 0000000..4aa459b --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/rm_capture.py @@ -0,0 +1,1251 @@ +#!/usr/bin/env python3 +"""Real-client GSP RPC capture ingestion, comparison and replay prep. + +Ingests the text output of the supervised host-client capture (the GSP RPC +send-path instrumentation described by the CAPTURE-RUNBOOK) and turns it into +structured, replayable artifacts: + +* **decode** — a raw capture log (dmesg text with the documented ``GE-SEND``, + ``GE-CTRL``, ``GE-ALLOC``, ``GE-GOLDEN-*`` and ``GE-CTRL-DONE`` tags) into a + ``greeneye-rm-rpc-capture/1`` document: one ordered record per RPC/alloc, + with the wire function id, the named control/class id, handles, caller + return address, and the captured parameter bytes. +* **compare** — two capture documents (for example, the real host client's + sequence vs. the open-iov host ladder's replay, or two firmware/driver + versions) aligned by record identity, with per-record parameter agreement, + coverage and explicit unmatched lists. +* **stimulus** — a ``greeneye-rm-rpc-stimulus/1`` ordered message list for the + simulator's B4/B5 CSB/message-queue seam (not wired yet). + +Honesty rules: + +* Truncated payloads are flagged, never silently padded. The runbook's hex + helper caps a dump at 192 bytes; a capture produced by that helper will mark + every larger payload ``truncated`` so a reader can see what is missing. +* Handles, physical addresses and caller addresses are capture-specific; the + compare report labels them, and the stimulus output says so explicitly. +* This tool only reads files and writes reports. It never touches a GPU, + kernel module, mdev, guest or network. The capture it ingests is a + supervised, human-run physical action (RUNBOOK P1/P7); this tool never runs + or instructs it beyond printing the tag grammar. +* No timing/cycle claim is made anywhere; fidelity is L0/L1 (docs/06). +""" +from __future__ import annotations + +import argparse +import difflib +import hashlib +import json +import os +import re +import sys +import time +from dataclasses import dataclass, field +from typing import Any + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from greeneye.errors import GreenEyeError # noqa: E402 + +#: Schema tag of the decoded capture document. +CAPTURE_FORMAT = "greeneye-rm-rpc-capture/1" +#: Schema tag of the comparison report. +COMPARE_FORMAT = "greeneye-rm-rpc-compare/1" +#: Schema tag of the replay stimulus document. +STIMULUS_FORMAT = "greeneye-rm-rpc-stimulus/1" +#: Schema tag of the demo result line. +DEMO_FORMAT = "greeneye-rm-rpc-capture-demo/1" + +#: Size of ``rpc_message_header_v``: the wire payload starts after it. +RPC_HEADER_BYTES = 32 + +# Wire function ids (``rpc_global_enums.h``): used to derive the embedded +# control/class id from a ``GE-SEND`` payload when no named ``GE-CTRL`` line +# was captured for it. +GSP_RM_CONTROL = 76 +GSP_RM_ALLOC = 103 +# Payload offsets inside the flat parameter blob (runbook §4). +CONTROL_CMD_OFFSET = 8 +ALLOC_CLASS_OFFSET = 12 + +KINDS = ("send", "control", "alloc", "golden", "ctrl-done") +HEX_KIND_TAG = {"send": "SEND", "control": "CTRL", "alloc": "ALLOC"} +TAG_KIND = {v: k for k, v in HEX_KIND_TAG.items()} + +# FROZEN BEFORE THE FIRST MEASUREMENT (2026-10-06). Do not move silently: +# a threshold change requires a new recorded run and a note in docs/45. +#: Conservative ESTIMATE (not measured on a human) of decoding one golden +#: sequence by eye from a capture log: grep the tags, count records, walk each +#: parameter blob against the RPC structs, transcribe into a replay table. +MANUAL_DECODE_BASELINE_SECONDS = 1800 +MANUAL_BASELINE_LABEL = "conservative estimate (not measured on a human)" +#: The tool path must stay at or below this fraction of the estimate. +WORKFLOW_THRESHOLD_FRACTION = 0.05 + +_HEX_RE = re.compile( + r"GE-(SEND|CTRL|ALLOC)\s+([0-9a-fA-F]{1,4})\s*:\s*((?:[0-9a-fA-F]{2}\s*)+)" +) +_KV_RE = re.compile(r"([A-Za-z_][A-Za-z0-9_]*)=(\S+)") +_TS_RE = re.compile(r"^\[([^\]]+)\]\s*") +_GOLDEN_TRIGGER_RE = re.compile( + r"GE-GOLDEN-TRIGGER\s+gspclient=(\d+)(?:\s+mig=(\d+))?" +) +_GOLDEN_END_RE = re.compile(r"GE-GOLDEN-END\s+status=(\S+)") +_GOLDEN_BEGIN_RE = re.compile(r"GE-GOLDEN-BEGIN") +_CTRL_DONE_RE = re.compile(r"GE-CTRL-DONE\b") + + +class RmCaptureError(GreenEyeError): + """A capture input is absent, unreadable or not in a supported schema.""" + + +def _parse_int(value: Any, default: int | None = None) -> int | None: + """Parse ints, decimal strings and ``0x`` strings; ``default`` on failure.""" + if value is None: + return default + if isinstance(value, bool): + return int(value) + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) + if isinstance(value, str): + text = value.strip() + if not text: + return default + try: + return int(text, 0) + except ValueError: + try: + return int(text, 16) + except ValueError: + return default + return default + + +def _sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +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() + + +def canonical_json(payload: Any) -> str: + """Deterministic JSON text (sorted keys, fixed indent, trailing newline).""" + return json.dumps(payload, indent=2, sort_keys=True) + "\n" + + +@dataclass +class CaptureRecord: + """One decoded capture record (an RPC, an alloc, or a golden marker).""" + + index: int + kind: str + line: int + ts: str = "" + fn: int | None = None + length: int | None = None + size: int | None = None + cmd: int | None = None + hclass: int | None = None + hclient: int | None = None + hobject: int | None = None + hparent: int | None = None + caller: int | None = None + caller_raw: str = "" + marker: str = "" + gspclient: int | None = None + mig: int | None = None + status: int | None = None + rpc_status: int | None = None + params_status: int | None = None + chunks: dict[int, bytes] = field(default_factory=dict) + anomalies: list[str] = field(default_factory=list) + + # -- parameter assembly ------------------------------------------------- + @property + def params(self) -> bytes: + """Captured parameter bytes, assembled from the hex chunks in order. + + A gap between chunks is zero-filled and recorded as an anomaly; the + raw chunk map stays available on the record. + """ + if not self.chunks: + return b"" + end = max(offset + len(data) for offset, data in self.chunks.items()) + out = bytearray(end) + for offset, data in sorted(self.chunks.items()): + out[offset : offset + len(data)] = data + return bytes(out) + + @property + def expected_params_size(self) -> int | None: + """Declared parameter size, or ``None`` when the tag does not carry one.""" + if self.kind == "send": + if self.length is None or self.length < RPC_HEADER_BYTES: + return None + return self.length - RPC_HEADER_BYTES + if self.kind in ("control", "alloc"): + return self.size + return None + + @property + def truncated(self) -> bool: + expected = self.expected_params_size + return expected is not None and len(self.params) < expected + + @property + def overrun(self) -> bool: + expected = self.expected_params_size + return expected is not None and len(self.params) > expected + + @property + def wire_id(self) -> int | None: + """Named id: ``cmd``/``hclass`` when captured, else derived from payload.""" + if self.kind == "control": + return self.cmd + if self.kind == "alloc": + return self.hclass + if self.kind == "send": + payload = self.params + if self.fn == GSP_RM_CONTROL and len(payload) >= CONTROL_CMD_OFFSET + 4: + return int.from_bytes( + payload[CONTROL_CMD_OFFSET : CONTROL_CMD_OFFSET + 4], "little" + ) + if self.fn == GSP_RM_ALLOC and len(payload) >= ALLOC_CLASS_OFFSET + 4: + return int.from_bytes( + payload[ALLOC_CLASS_OFFSET : ALLOC_CLASS_OFFSET + 4], "little" + ) + return None + + @property + def signature(self) -> str: + """Identity used for compare alignment (kind + id + declared size).""" + if self.kind == "golden": + return f"golden:{self.marker}" + if self.kind == "ctrl-done": + return f"ctrl-done:{self.cmd:#x}" if self.cmd is not None else "ctrl-done:?" + identifier = self.wire_id + id_text = f"{identifier:#x}" if identifier is not None else "?" + fn_text = f"{self.fn:#x}" if self.fn is not None else "?" + size = self.expected_params_size + size_text = f"{size:#x}" if size is not None else "?" + return f"{self.kind}:{id_text}:{fn_text}:{size_text}" + + @property + def label(self) -> str: + """Short human label for reports.""" + if self.kind == "golden": + return f"GOLDEN {self.marker}" + ( + f" status={self.status:#x}" if self.status is not None else "" + ) + identifier = self.wire_id + id_text = f"{identifier:#x}" if identifier is not None else "?" + if self.kind == "send": + return f"SEND fn={self.fn:#x} id={id_text}" if self.fn is not None else "SEND" + if self.kind == "control": + return f"CTRL cmd={id_text}" + if self.kind == "alloc": + return f"ALLOC cls={id_text}" + return f"CTRL-DONE cmd={id_text}" + + +@dataclass +class RmCapture: + """A decoded capture document.""" + + path: str + sha256: str + records: list[CaptureRecord] = field(default_factory=list) + source: str = "" + format: str = CAPTURE_FORMAT + warnings: list[str] = field(default_factory=list) + + @property + def truncated_records(self) -> list[CaptureRecord]: + return [r for r in self.records if r.truncated] + + def kind_counts(self) -> dict[str, int]: + counts: dict[str, int] = {} + for record in self.records: + counts[record.kind] = counts.get(record.kind, 0) + 1 + return counts + + def fn_counts(self) -> dict[str, int]: + counts: dict[str, int] = {} + for record in self.records: + if record.kind == "send" and record.fn is not None: + key = f"0x{record.fn:x}" + counts[key] = counts.get(key, 0) + 1 + return counts + + def golden_status(self) -> int | None: + for record in self.records: + if record.kind == "golden" and record.marker == "end": + return record.status + return None + + +# --------------------------------------------------------------------------- +# Decode +# --------------------------------------------------------------------------- + + +def _kv_fields(line: str) -> dict[str, str]: + return {key: value for key, value in _KV_RE.findall(line)} + + +def _new_record(index: int, kind: str, line_no: int, ts: str) -> CaptureRecord: + return CaptureRecord(index=index, kind=kind, line=line_no, ts=ts) + + +def decode_text(text: str, source: str = "<text>") -> RmCapture: + """Decode capture-log text into a capture document. + + Lines are matched anywhere in the line (dmesg/NVRM prefixes are fine). + Hex lines attach to the most recent record of their own tag; a hex line + with no open record of that tag is recorded as an ``orphan_hex`` warning + instead of being guessed onto some other record. + """ + records: list[CaptureRecord] = [] + warnings: list[str] = [] + open_records: dict[str, CaptureRecord] = {} + + for line_no, raw_line in enumerate(text.splitlines(), start=1): + ts_match = _TS_RE.search(raw_line) + ts = ts_match.group(1) if ts_match else "" + + hex_match = _HEX_RE.search(raw_line) + if hex_match: + tag = hex_match.group(1) + kind = TAG_KIND[tag] + offset = int(hex_match.group(2), 16) + data = bytes.fromhex(re.sub(r"\s+", "", hex_match.group(3))) + record = open_records.get(kind) + if record is None: + warnings.append(f"line {line_no}: orphan {tag} hex block (no header)") + continue + if offset in record.chunks: + record.anomalies.append(f"duplicate hex offset {offset:#x}") + record.chunks[offset] = data + continue + + if _GOLDEN_TRIGGER_RE.search(raw_line): + match = _GOLDEN_TRIGGER_RE.search(raw_line) + record = _new_record(len(records), "golden", line_no, ts) + record.marker = "trigger" + record.gspclient = int(match.group(1)) if match else None + record.mig = int(match.group(2)) if match and match.group(2) else None + records.append(record) + continue + + if _GOLDEN_BEGIN_RE.search(raw_line): + record = _new_record(len(records), "golden", line_no, ts) + record.marker = "begin" + records.append(record) + continue + + if _GOLDEN_END_RE.search(raw_line): + match = _GOLDEN_END_RE.search(raw_line) + record = _new_record(len(records), "golden", line_no, ts) + record.marker = "end" + record.status = _parse_int(match.group(1) if match else None) + records.append(record) + continue + + if _CTRL_DONE_RE.search(raw_line): + fields = _kv_fields(raw_line) + record = _new_record(len(records), "ctrl-done", line_no, ts) + record.cmd = _parse_int(fields.get("cmd")) + record.rpc_status = _parse_int(fields.get("rpc_status")) + record.params_status = _parse_int(fields.get("params_status")) + records.append(record) + continue + + header_kind = None + for tag, kind in TAG_KIND.items(): + if re.search(rf"GE-{tag}\b", raw_line): + header_kind = kind + break + if header_kind is None: + continue + + fields = _kv_fields(raw_line) + record = _new_record(len(records), header_kind, line_no, ts) + if header_kind == "send": + record.fn = _parse_int(fields.get("fn")) + record.length = _parse_int(fields.get("len")) + elif header_kind == "control": + record.cmd = _parse_int(fields.get("cmd")) + record.hclient = _parse_int(fields.get("hClient")) + record.hobject = _parse_int(fields.get("hObject")) + record.size = _parse_int(fields.get("size")) + elif header_kind == "alloc": + record.hclass = _parse_int(fields.get("cls")) + record.hclient = _parse_int(fields.get("hClient")) + record.hparent = _parse_int(fields.get("hParent")) + record.hobject = _parse_int(fields.get("hObject")) + record.size = _parse_int(fields.get("size")) + caller_raw = fields.get("ret", "") + record.caller_raw = caller_raw + record.caller = _parse_int(caller_raw) + records.append(record) + open_records[header_kind] = record + + # Post-pass: chunk-gap and size anomalies. + for record in records: + expected = record.expected_params_size + if record.chunks and 0 not in record.chunks: + record.anomalies.append("hex chunks do not start at offset 0") + if record.overrun: + record.anomalies.append( + f"captured {len(record.params)} bytes > declared {expected} bytes" + ) + + return RmCapture( + path=os.path.abspath(source), + sha256=_sha256_bytes(text.encode("utf-8", "replace")), + records=records, + source=source, + warnings=warnings, + ) + + +def decode_file(path: str) -> RmCapture: + if not os.path.isfile(path): + raise RmCaptureError(f"capture log not found: {path}") + try: + with open(path, "r", encoding="utf-8", errors="replace") as fh: + text = fh.read() + except OSError as exc: + raise RmCaptureError(f"cannot read capture log {path}: {exc}") from exc + return decode_text(text, source=path) + + +def load_capture(path: str) -> RmCapture: + """Load a ``greeneye-rm-rpc-capture/1`` document or decode a raw log.""" + if not os.path.isfile(path): + raise RmCaptureError(f"capture not found: {path}") + with open(path, "r", encoding="utf-8", errors="replace") as fh: + head = fh.read(4096) + if not head.lstrip().startswith("{"): + return decode_file(path) + try: + with open(path, "r", encoding="utf-8", errors="replace") as fh: + payload = json.load(fh) + except (OSError, json.JSONDecodeError) as exc: + raise RmCaptureError(f"cannot read capture {path}: {exc}") from exc + if not isinstance(payload, dict): + raise RmCaptureError(f"{path}: capture must be an object") + fmt = str(payload.get("format") or payload.get("schema") or "") + if fmt != CAPTURE_FORMAT: + raise RmCaptureError(f"{path}: format {fmt!r} is not {CAPTURE_FORMAT!r}") + raw_records = payload.get("records") + if not isinstance(raw_records, list): + raise RmCaptureError(f"{path}: missing 'records' list") + records: list[CaptureRecord] = [] + for index, raw in enumerate(raw_records): + if not isinstance(raw, dict): + continue + record = _new_record( + index, str(raw.get("kind") or "unknown"), int(raw.get("line") or 0), + str(raw.get("ts") or ""), + ) + record.fn = _parse_int(raw.get("fn")) + record.length = _parse_int(raw.get("length")) + record.size = _parse_int(raw.get("size")) + record.cmd = _parse_int(raw.get("cmd")) + record.hclass = _parse_int(raw.get("hclass")) + record.hclient = _parse_int(raw.get("hclient")) + record.hobject = _parse_int(raw.get("hobject")) + record.hparent = _parse_int(raw.get("hparent")) + record.caller = _parse_int(raw.get("caller")) + record.caller_raw = str(raw.get("caller_raw") or "") + record.marker = str(raw.get("marker") or "") + record.gspclient = _parse_int(raw.get("gspclient")) + record.mig = _parse_int(raw.get("mig")) + record.status = _parse_int(raw.get("status")) + record.rpc_status = _parse_int(raw.get("rpc_status")) + record.params_status = _parse_int(raw.get("params_status")) + params_hex = raw.get("params_hex") + if isinstance(params_hex, str) and params_hex: + try: + record.chunks[0] = bytes.fromhex(re.sub(r"\s+", "", params_hex)) + except ValueError as exc: + raise RmCaptureError(f"{path}: record {index}: bad params hex: {exc}") from exc + records.append(record) + capture = RmCapture( + path=os.path.abspath(path), + sha256=_sha256_file(path), + records=records, + source=str(payload.get("source") or path), + ) + warnings = payload.get("warnings") + if isinstance(warnings, list): + capture.warnings = [str(w) for w in warnings] + return capture + + +# --------------------------------------------------------------------------- +# Compare +# --------------------------------------------------------------------------- + + +def _param_agreement(a: CaptureRecord, b: CaptureRecord) -> dict: + pa, pb = a.params, b.params + matched = sum(1 for x, y in zip(pa, pb) if x == y) + first_diff = None + for offset, (x, y) in enumerate(zip(pa, pb)): + if x != y: + first_diff = offset + break + if first_diff is None and len(pa) != len(pb): + first_diff = min(len(pa), len(pb)) + denominator = max(len(pa), len(pb)) + return { + "a_bytes": len(pa), + "b_bytes": len(pb), + "matched_bytes": matched, + "byte_agreement_pct": ( + round(100.0 * matched / denominator, 2) if denominator else None + ), + "first_diff_offset": first_diff, + "a_truncated": a.truncated, + "b_truncated": b.truncated, + } + + +def _pair_status(a: CaptureRecord, b: CaptureRecord) -> tuple[str, dict]: + if a.kind == "golden": + if ( + a.status == b.status + and a.gspclient == b.gspclient + and a.mig == b.mig + ): + return "agree", {} + return "mismatch", { + "a_status": a.status, + "b_status": b.status, + "a_gspclient": a.gspclient, + "b_gspclient": b.gspclient, + "a_mig": a.mig, + "b_mig": b.mig, + } + if a.kind == "ctrl-done": + if a.rpc_status == b.rpc_status and a.params_status == b.params_status: + return "agree", {} + return "mismatch", { + "a_rpc_status": a.rpc_status, + "b_rpc_status": b.rpc_status, + "a_params_status": a.params_status, + "b_params_status": b.params_status, + } + agreement = _param_agreement(a, b) + if a.truncated or b.truncated: + return "partial", agreement + if agreement["a_bytes"] == agreement["b_bytes"] and agreement["first_diff_offset"] is None: + return "agree", agreement + return "shape-agree", agreement + + +def compare(a: RmCapture, b: RmCapture) -> dict: + """Ordered, per-record comparison of two captures; no timing claims.""" + sig_a = [record.signature for record in a.records] + sig_b = [record.signature for record in b.records] + matcher = difflib.SequenceMatcher(a=sig_a, b=sig_b, autojunk=False) + rows: list[dict] = [] + a_only: list[dict] = [] + b_only: list[dict] = [] + counts = { + "agree": 0, + "shape-agree": 0, + "partial": 0, + "mismatch": 0, + "a-only": 0, + "b-only": 0, + } + for opcode, i1, i2, j1, j2 in matcher.get_opcodes(): + if opcode == "equal": + for offset in range(i2 - i1): + record_a = a.records[i1 + offset] + record_b = b.records[j1 + offset] + status, detail = _pair_status(record_a, record_b) + counts[status] += 1 + rows.append( + { + "a_index": record_a.index, + "b_index": record_b.index, + "a_label": record_a.label, + "b_label": record_b.label, + "signature": record_a.signature, + "status": status, + "detail": detail, + "a_truncated": record_a.truncated, + "b_truncated": record_b.truncated, + "a_caller": record_a.caller, + "b_caller": record_b.caller, + } + ) + else: + for index in range(i1, i2): + counts["a-only"] += 1 + record = a.records[index] + a_only.append( + { + "index": record.index, + "label": record.label, + "signature": record.signature, + "params_bytes": len(record.params), + "line": record.line, + } + ) + for index in range(j1, j2): + counts["b-only"] += 1 + record = b.records[index] + b_only.append( + { + "index": record.index, + "label": record.label, + "signature": record.signature, + "params_bytes": len(record.params), + "line": record.line, + } + ) + matched = counts["agree"] + counts["shape-agree"] + counts["partial"] + counts["mismatch"] + byte_rows = [ + row["detail"].get("byte_agreement_pct") + for row in rows + if isinstance(row.get("detail"), dict) + and row["detail"].get("byte_agreement_pct") is not None + ] + return { + "format": COMPARE_FORMAT, + "inputs": { + "a": { + "path": a.path, + "sha256": a.sha256, + "source": a.source, + "records": len(a.records), + "kinds": a.kind_counts(), + "fn_counts": a.fn_counts(), + "golden_status": a.golden_status(), + "truncated_records": len(a.truncated_records), + }, + "b": { + "path": b.path, + "sha256": b.sha256, + "source": b.source, + "records": len(b.records), + "kinds": b.kind_counts(), + "fn_counts": b.fn_counts(), + "golden_status": b.golden_status(), + "truncated_records": len(b.truncated_records), + }, + }, + "summary": { + "a_records": len(a.records), + "b_records": len(b.records), + "matched": matched, + "agree": counts["agree"], + "shape_agree": counts["shape-agree"], + "partial": counts["partial"], + "mismatch": counts["mismatch"], + "a_only": counts["a-only"], + "b_only": counts["b-only"], + "agreement_pct": round(100.0 * counts["agree"] / matched, 2) if matched else None, + "coverage_pct": round(100.0 * matched / len(a.records), 2) if a.records else None, + "mean_byte_agreement_pct": ( + round(sum(byte_rows) / len(byte_rows), 2) if byte_rows else None + ), + "claim": ( + "per-record identity and parameter agreement on the aligned intersection; " + "not a timing, silicon or full-coverage claim" + ), + }, + "rows": rows, + "a_only": a_only, + "b_only": b_only, + "fidelity": { + "a": f"file-ingested capture ({a.source or 'unlabelled'})", + "b": f"file-ingested capture ({b.source or 'unlabelled'})", + "levels": "L0/L1 only; L2 timing and L3 device environment not claimed (docs/06)", + }, + "limits": [ + "This tool never ran the capture: a supervised human produced both inputs.", + "Handles, physical addresses and caller return addresses are capture-specific and are labelled, not normalised.", + "A truncated payload is reported as partial/truncated, never padded to look complete.", + "A-only/b-only records are ordering or presence differences under this alignment; they are not proof of a wrong value.", + ], + "warnings": (a.warnings if a.warnings else []) + (b.warnings if b.warnings else []), + } + + +def render_compare_markdown(report: dict) -> str: + summary = report["summary"] + lines = [ + "# GSP RPC capture comparison", + "", + f"> Format `{report['format']}`. Fidelity: {report['fidelity']['levels']}", + "> No cycle/timing claim. Underclaim every number (docs/06, docs/45).", + "", + "## Inputs", + "", + f"* A: `{report['inputs']['a']['path']}` " + f"({report['inputs']['a']['records']} records, " + f"sha256 `{report['inputs']['a']['sha256'][:16]}…`" + + ( + f", golden status `{report['inputs']['a']['golden_status']:#x}`" + if report["inputs"]["a"]["golden_status"] is not None + else "" + ) + + ")", + f"* B: `{report['inputs']['b']['path']}` " + f"({report['inputs']['b']['records']} records, " + f"sha256 `{report['inputs']['b']['sha256'][:16]}…`" + + ( + f", golden status `{report['inputs']['b']['golden_status']:#x}`" + if report["inputs"]["b"]["golden_status"] is not None + else "" + ) + + ")", + "", + "## Summary", + "", + "| matched | agree | shape-agree | partial | mismatch | A-only | B-only | agreement % | coverage % | mean byte agreement % |", + "|---|---|---|---|---|---|---|---|---|---|", + f"| {summary['matched']} | {summary['agree']} | {summary['shape_agree']} | " + f"{summary['partial']} | {summary['mismatch']} | {summary['a_only']} | {summary['b_only']} | " + f"{summary['agreement_pct'] if summary['agreement_pct'] is not None else 'n/a'} | " + f"{summary['coverage_pct'] if summary['coverage_pct'] is not None else 'n/a'} | " + f"{summary['mean_byte_agreement_pct'] if summary['mean_byte_agreement_pct'] is not None else 'n/a'} |", + "", + f"* {summary['claim']}", + "", + "## Per-record rows", + "", + "| A | B | status | bytes A/B | byte agreement % | first diff | truncated A/B |", + "|---|---|---|---|---|---|---|", + ] + for row in report["rows"]: + detail = row.get("detail") if isinstance(row.get("detail"), dict) else {} + a_bytes = detail.get("a_bytes", "") + b_bytes = detail.get("b_bytes", "") + byte_pct = detail.get("byte_agreement_pct") + first_diff = detail.get("first_diff_offset") + lines.append( + f"| `{row['a_label']}` | `{row['b_label']}` | {row['status']} | " + f"{a_bytes}/{b_bytes} | {byte_pct if byte_pct is not None else ''} | " + f"{first_diff if first_diff is not None else ''} | " + f"{'Y' if row.get('a_truncated') else '-'}/{'Y' if row.get('b_truncated') else '-'} |" + ) + lines += ["", "## A-only records", ""] + if report["a_only"]: + for item in report["a_only"]: + lines.append( + f"* #{item['index']} `{item['label']}` ({item['params_bytes']} B, line {item['line']})" + ) + else: + lines.append("* none") + lines += ["", "## B-only records", ""] + if report["b_only"]: + for item in report["b_only"]: + lines.append( + f"* #{item['index']} `{item['label']}` ({item['params_bytes']} B, line {item['line']})" + ) + else: + lines.append("* none") + lines += ["", "## Limits", ""] + for limit in report["limits"]: + lines.append(f"* {limit}") + if report.get("warnings"): + lines += ["", "## Decode warnings", ""] + for warning in report["warnings"]: + lines.append(f"* {warning}") + lines.append("") + return "\n".join(lines) + + +def run_decode(log_path: str, out_dir: str | None) -> dict: + capture = decode_file(log_path) + payload = capture_to_dict(capture) + if out_dir: + os.makedirs(out_dir, exist_ok=True) + json_path = os.path.join(out_dir, "rm-capture.json") + with open(json_path, "w", encoding="utf-8") as fh: + fh.write(canonical_json(payload)) + md_path = os.path.join(out_dir, "rm-capture.md") + with open(md_path, "w", encoding="utf-8") as fh: + fh.write(render_capture_markdown(capture)) + payload["artifacts"] = { + "capture_json": os.path.abspath(json_path), + "capture_md": os.path.abspath(md_path), + } + return payload + + +def run_compare(a_path: str, b_path: str, out_dir: str) -> dict: + capture_a = load_capture(a_path) + capture_b = load_capture(b_path) + report = compare(capture_a, capture_b) + os.makedirs(out_dir, exist_ok=True) + json_path = os.path.join(out_dir, "rm-capture-compare.json") + md_path = os.path.join(out_dir, "rm-capture-compare.md") + with open(json_path, "w", encoding="utf-8") as fh: + fh.write(canonical_json(report)) + with open(md_path, "w", encoding="utf-8") as fh: + fh.write(render_compare_markdown(report)) + report["artifacts"] = { + "report_json": os.path.abspath(json_path), + "report_md": os.path.abspath(md_path), + } + return report + + +def run_stimulus(capture_path: str, out_path: str | None) -> dict: + capture = load_capture(capture_path) + payload = stimulus_from_capture(capture) + if out_path: + os.makedirs(os.path.dirname(os.path.abspath(out_path)), exist_ok=True) + with open(out_path, "w", encoding="utf-8") as fh: + fh.write(canonical_json(payload)) + payload["artifacts"] = {"stimulus_json": os.path.abspath(out_path)} + return payload + + +# --------------------------------------------------------------------------- +# Serialisation / stimulus +# --------------------------------------------------------------------------- + + +def record_to_dict(record: CaptureRecord) -> dict: + out: dict[str, Any] = { + "index": record.index, + "kind": record.kind, + "line": record.line, + } + if record.ts: + out["ts"] = record.ts + for key in ( + "fn", + "length", + "size", + "cmd", + "hclass", + "hclient", + "hobject", + "hparent", + "caller", + ): + value = getattr(record, key) + if value is not None: + out[key] = value + if record.caller_raw: + out["caller_raw"] = record.caller_raw + if record.marker: + out["marker"] = record.marker + if record.gspclient is not None: + out["gspclient"] = record.gspclient + if record.mig is not None: + out["mig"] = record.mig + for key in ("status", "rpc_status", "params_status"): + value = getattr(record, key) + if value is not None: + out[key] = value + params = record.params + if params: + out["params_hex"] = params.hex() + out["params_bytes"] = len(params) + out["params_size_expected"] = record.expected_params_size + out["truncated"] = record.truncated + if record.anomalies: + out["anomalies"] = list(record.anomalies) + return out + + +def capture_to_dict(capture: RmCapture) -> dict: + return { + "format": CAPTURE_FORMAT, + "source": capture.source, + "sha256": capture.sha256, + "records": [record_to_dict(record) for record in capture.records], + "summary": { + "records": len(capture.records), + "kinds": capture.kind_counts(), + "fn_counts": capture.fn_counts(), + "truncated_records": len(capture.truncated_records), + "golden_status": capture.golden_status(), + }, + "warnings": list(capture.warnings), + "limits": [ + "Decoded from the documented capture tags only (GE-SEND/GE-CTRL/GE-ALLOC/GE-GOLDEN-*/GE-CTRL-DONE).", + "Handles and addresses are capture-specific; do not reuse them across boots.", + "Truncated payloads are flagged (the runbook hex helper caps at 192 bytes unless extended).", + "This tool never ran the capture; the log is a supervised human artifact.", + ], + } + + +def render_capture_markdown(capture: RmCapture) -> str: + summary = capture_to_dict(capture)["summary"] + lines = [ + "# GSP RPC capture", + "", + f"> Format `{CAPTURE_FORMAT}`; source `{capture.source}`; " + f"sha256 `{capture.sha256[:16]}…`. Fidelity L0/L1; no timing claim.", + "", + "## Summary", + "", + f"* records: {summary['records']} {summary['kinds']}", + f"* wire fn counts: {summary['fn_counts']}", + f"* truncated records: {summary['truncated_records']}", + f"* golden end status: {summary['golden_status']}", + "", + "## Records", + "", + "| # | kind | ts | id | size | captured B | expected B | truncated | caller |", + "|---|---|---|---|---|---|---|---|---|", + ] + for record in capture.records: + identifier = record.wire_id + id_text = f"{identifier:#x}" if identifier is not None else "" + expected = record.expected_params_size + lines.append( + f"| {record.index} | {record.kind} | {record.ts} | {id_text} | " + f"{record.size if record.size is not None else ''} | {len(record.params)} | " + f"{expected if expected is not None else ''} | " + f"{'Y' if record.truncated else '-'} | " + f"{f'{record.caller:#x}' if record.caller is not None else record.caller_raw} |" + ) + if capture.warnings: + lines += ["", "## Decode warnings", ""] + for warning in capture.warnings: + lines.append(f"* {warning}") + lines += ["", "## Limits", ""] + for limit in capture_to_dict(capture)["limits"]: + lines.append(f"* {limit}") + lines.append("") + return "\n".join(lines) + + +def stimulus_from_capture(capture: RmCapture) -> dict: + messages: list[dict] = [] + markers: list[dict] = [] + for record in capture.records: + if record.kind == "golden": + markers.append( + { + "line": record.line, + "marker": record.marker, + "status": record.status, + "gspclient": record.gspclient, + } + ) + continue + if record.kind == "ctrl-done": + continue + message: dict[str, Any] = { + "seq": len(messages), + "kind": record.kind, + "params_hex": record.params.hex(), + "params_size": len(record.params), + "params_size_expected": record.expected_params_size, + "truncated": record.truncated, + } + if record.fn is not None: + message["fn"] = record.fn + if record.cmd is not None: + message["cmd"] = record.cmd + if record.hclass is not None: + message["hclass"] = record.hclass + if record.kind == "send" and record.wire_id is not None: + message["wire_id"] = record.wire_id + messages.append(message) + return { + "format": STIMULUS_FORMAT, + "source": { + "path": capture.path, + "sha256": capture.sha256, + "records": len(capture.records), + }, + "markers": markers, + "messages": messages, + "limits": [ + "Handles, physical addresses and caller addresses are capture-specific and deliberately omitted; a replay must re-map or allocate them.", + "Truncated messages (params_size < params_size_expected) are not replayable as-is; extend the capture dump first.", + "Consumption by the simulator B4/B5 CSB/message-queue seam is not wired yet (docs/45).", + "No ordering claim beyond the captured log order; server replies were not captured by the send-side probes.", + ], + } + + +# --------------------------------------------------------------------------- +# Demo (synthetic; measured value contract) +# --------------------------------------------------------------------------- + + +def _hex_lines(tag: str, payload: bytes, chunk: int = 192) -> list[str]: + lines: list[str] = [] + for start in range(0, len(payload), chunk): + piece = payload[start : start + chunk] + lines.append(f"GE-{tag} {start:03x}:" + "".join(f" {b:02x}" for b in piece)) + return lines + + +def _synthetic_log() -> str: + allocs = [ + ("0x90f1", 0x30), + ("0x003e", 0x100), + ("0x50a0", 0x40), + ("0x0040", 0x1000), + ("0xc46f", 0x180), + ("0xc597", 0x0), + ] + controls = [ + ("0x2080012b", 560), + ("0x20801201", 64), + ("0x20800a2a", 32), + ("0x20809001", 16), + ] + lines = [ + "[ 100.000000] NVRM: GE-GOLDEN-TRIGGER gspclient=1 mig=0", + "[ 100.000010] NVRM: GE-GOLDEN-BEGIN", + ] + step = 1 + for hclass, size in allocs: + hclass_value = int(hclass, 16) + params = bytes((i * 7 + hclass_value) & 0xFF for i in range(size)) + lines.append( + f"[ 100.{step:06d}] NVRM: GE-ALLOC cls={hclass} hClient=0xbaba0001 " + f"hParent=0x0 hObject=0xbaba10{step:02x} size={size:#x} " + f"paramp=0xffffffffc0a00000 ret=0xffffffffc0b00000" + ) + lines.extend(f"[ 100.{step:06d}] NVRM: {line}" for line in _hex_lines("ALLOC", params)) + step += 1 + for cmd, size in controls: + cmd_value = int(cmd, 16) + params = bytes((i * 3 + cmd_value) & 0xFF for i in range(size)) + lines.append( + f"[ 100.{step:06d}] NVRM: GE-CTRL cmd={cmd} hClient=0xc2000006 " + f"hObject=0xabcd0080 size={size:#x} paramp=0xffffffffc0a10000 " + f"ret=0xffffffffc0b00000" + ) + lines.extend(f"[ 100.{step:06d}] NVRM: {line}" for line in _hex_lines("CTRL", params)) + step += 1 + payload = ( + (0xC2000006).to_bytes(4, "little") + + (0xABCD0080).to_bytes(4, "little") + + (0x2080012B).to_bytes(4, "little") + + bytes(range(60)) + ) + lines.append( + f"[ 100.{step:06d}] NVRM: GE-SEND fn=0x0000004c len={len(payload) + RPC_HEADER_BYTES:#x} " + f"ret=0xffffffffc0b00000" + ) + lines.extend(f"[ 100.{step:06d}] NVRM: {line}" for line in _hex_lines("SEND", payload)) + step += 1 + lines.append( + f"[ 100.{step:06d}] NVRM: GE-CTRL-DONE cmd=0x2080012b rpc_status=0x0 params_status=0x0" + ) + lines.append("[ 100.000999] NVRM: GE-GOLDEN-END status=0x0") + return "\n".join(lines) + "\n" + + +def _mutate_log(text: str) -> str: + """Replay-side variant: one parameter byte differs, one alloc is dropped.""" + kept: list[str] = [] + dropping = False + dropped = False + for line in text.splitlines(): + if not dropped and "GE-ALLOC cls=0x003e" in line: + dropped = True + dropping = True + continue + if dropping: + if re.search(r"GE-ALLOC\s+[0-9a-fA-F]{1,4}:", line): + continue + dropping = False + kept.append(line) + text_b = "\n".join(kept) + "\n" + + def flip(match: re.Match) -> str: + value = int(match.group(2), 16) ^ 0x01 + return f"{match.group(1)}{value:02x}" + + text_b, _ = re.subn(r"(GE-CTRL 000: )([0-9a-fA-F]{2})", flip, text_b, count=1) + return text_b + + +def run_demo(out_dir: str | None = None) -> dict: + started = time.perf_counter() + capture_a = decode_text(_synthetic_log(), source="demo-capture-a") + capture_b = decode_text(_mutate_log(_synthetic_log()), source="demo-replay-b") + report = compare(capture_a, capture_b) + stimulus = stimulus_from_capture(capture_a) + wall = time.perf_counter() - started + threshold_seconds = MANUAL_DECODE_BASELINE_SECONDS * WORKFLOW_THRESHOLD_FRACTION + ok = ( + wall <= threshold_seconds + and report["summary"]["agree"] + report["summary"]["shape_agree"] >= 1 + and report["summary"]["a_only"] + report["summary"]["b_only"] >= 1 + and not capture_a.truncated_records + ) + payload = { + "format": DEMO_FORMAT, + "ok": ok, + "wall_seconds": round(wall, 4), + "baseline_seconds": MANUAL_DECODE_BASELINE_SECONDS, + "baseline_label": MANUAL_BASELINE_LABEL, + "threshold_fraction": WORKFLOW_THRESHOLD_FRACTION, + "threshold_seconds": threshold_seconds, + "records_a": len(capture_a.records), + "records_b": len(capture_b.records), + "messages": len(stimulus["messages"]), + "summary": report["summary"], + } + if out_dir: + os.makedirs(out_dir, exist_ok=True) + capture_path = os.path.join(out_dir, "demo-capture-a.json") + replay_path = os.path.join(out_dir, "demo-replay-b.json") + with open(capture_path, "w", encoding="utf-8") as fh: + fh.write(canonical_json(capture_to_dict(capture_a))) + with open(replay_path, "w", encoding="utf-8") as fh: + fh.write(canonical_json(capture_to_dict(capture_b))) + compare_path = os.path.join(out_dir, "demo-compare.json") + with open(compare_path, "w", encoding="utf-8") as fh: + fh.write(canonical_json(report)) + stimulus_path = os.path.join(out_dir, "demo-stimulus.json") + with open(stimulus_path, "w", encoding="utf-8") as fh: + fh.write(canonical_json(stimulus)) + payload["artifacts"] = { + "capture_a": os.path.abspath(capture_path), + "replay_b": os.path.abspath(replay_path), + "compare": os.path.abspath(compare_path), + "stimulus": os.path.abspath(stimulus_path), + } + return payload + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +CAPTURE_HINTS = """\ +================================================================================ +GSP RPC CAPTURE TAG GRAMMAR — SUPERVISED, HUMAN-RUN PHYSICAL ACTION +================================================================================ +This command prints the tag contract the decoder understands; it executes +NOTHING and never runs the capture. + +The capture itself is supervised-only (RUNBOOK P1/P7, SIMULATOR-WORK- +BOUNDARY.txt): driver swap on the physical GPU, one attempt per boot, stop on +Zid/WPR2/pstore; the authored procedure lives in the capture runbook +(authored, not executed). This tool only ingests the resulting log. + +Tags (anywhere in the line; dmesg/NVRM prefixes are fine): + GE-GOLDEN-TRIGGER gspclient=<0|1> mig=<0|1> + GE-GOLDEN-BEGIN + GE-GOLDEN-END status=0x... + GE-SEND fn=0x... len=0x... ret=0x... + GE-CTRL cmd=0x... hClient=0x... hObject=0x... size=0x... paramp=0x... ret=0x... + GE-ALLOC cls=0x... hClient=0x... hParent=0x... hObject=0x... size=0x... paramp=0x... ret=0x... + GE-CTRL-DONE cmd=0x... rpc_status=0x... params_status=0x... + Hex lines attach to the most recent record of their own tag: + GE-<TAG> <offset-hex>: <hex bytes...> + +NOTE — the runbook's hex helper currently caps a dump at 192 bytes; payloads +larger than that (e.g. a 560-byte PROMOTE_CTX) decode as `truncated` and are +not replayable until the helper is extended to chunk. The decoder flags this +instead of padding. + +Offline use (safe): + python3 harness/rm_capture.py decode --log capture.log --out dumps/rm-capture + python3 harness/rm_capture.py compare --a real.json --b replay.json --out dumps/rm-compare + python3 harness/rm_capture.py stimulus --capture real.json --out dumps/rm-stimulus.json + python3 harness/rm_capture.py demo --out /tmp/rm-capture-demo +================================================================================ +""" + + +def print_capture_hints(out=None) -> None: + """Print the supervised capture tag grammar; execute nothing.""" + print(CAPTURE_HINTS, file=out or sys.stdout) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="rm_capture.py", + description=( + "Decode, compare and replay-prep a supervised GSP RPC capture. " + "Offline only; never executes physical commands." + ), + ) + parser.add_argument( + "--print-capture-hints", + action="store_true", + help="print the capture tag grammar (executes nothing)", + ) + sub = parser.add_subparsers(dest="command") + decode_cmd = sub.add_parser("decode", help="decode a raw capture log") + decode_cmd.add_argument("--log", required=True, help="capture log (text)") + decode_cmd.add_argument("--out", help="output directory (optional)") + compare_cmd = sub.add_parser("compare", help="compare two captures") + compare_cmd.add_argument("--a", required=True, help="capture JSON or raw log") + compare_cmd.add_argument("--b", required=True, help="capture JSON or raw log") + compare_cmd.add_argument("--out", required=True, help="output directory") + stimulus_cmd = sub.add_parser("stimulus", help="emit a replay stimulus document") + stimulus_cmd.add_argument("--capture", required=True, help="capture JSON or raw log") + stimulus_cmd.add_argument("--out", help="output JSON path (optional)") + demo_cmd = sub.add_parser("demo", help="synthetic measured end-to-end demo") + demo_cmd.add_argument("--out", help="output directory (optional)") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if args.print_capture_hints: + print_capture_hints() + return 0 + try: + if args.command == "decode": + payload = run_decode(args.log, args.out) + elif args.command == "compare": + payload = run_compare(args.a, args.b, args.out) + elif args.command == "stimulus": + payload = run_stimulus(args.capture, args.out) + elif args.command == "demo": + payload = run_demo(args.out) + else: + parser.print_help() + return 2 + except RmCaptureError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + summary = payload.get("summary", {}) + print( + json.dumps( + { + "format": payload.get("format"), + "ok": payload.get("ok"), + "wall_seconds": payload.get("wall_seconds"), + "summary": summary, + "artifacts": payload.get("artifacts"), + }, + indent=2, + sort_keys=True, + ) + ) + if args.command == "demo" and not payload.get("ok"): + return 3 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/run_contracts.py b/debug/risc-v-gsp-rm-vm/harness/run_contracts.py new file mode 100644 index 0000000..1ea8474 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/run_contracts.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 +"""Run the wall-contract registry (``harness/contracts.py``, docs/50). + +Every solved wall is pinned as a contract over an existing tool entry point; +this CLI runs them and prints pass/fail per contract plus per-wall-class +coverage. Fast contracts are the suite default; slow ones run with ``--all`` +(or by name). + +Offline only: never touches a GPU, kernel module, mdev, guest or network. +Exit codes: 0 = every selected contract passed (or skipped for a missing +prerequisite), 1 = at least one failure, 2 = usage error. +""" +from __future__ import annotations + +import argparse +import json +import os +import sys + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness.contracts import ( # noqa: E402 + Contract, + all_contracts, + fast_contracts, + run_contracts, + summarize, +) + + +def _select(args: argparse.Namespace) -> tuple[list[Contract], list[str]]: + if args.name: + wanted = set(args.name) + chosen = [c for c in all_contracts() if c.name in wanted] + unknown = sorted(wanted - {c.name for c in chosen}) + return chosen, unknown + return (all_contracts() if args.all else fast_contracts()), [] + + +def _table(rows: list[tuple[str, ...]], headers: tuple[str, ...]) -> list[str]: + widths = [ + max(len(headers[index]), *(len(row[index]) for row in rows)) + for index in range(len(headers)) + ] + lines = [" ".join(head.ljust(widths[i]) for i, head in enumerate(headers))] + lines.append(" ".join("-" * width for width in widths)) + for row in rows: + lines.append(" ".join(cell.ljust(widths[i]) for i, cell in enumerate(row))) + return lines + + +def _render(results, summary: dict) -> list[str]: + rows = [ + ( + result.name, + result.wall_class, + result.speed, + result.status.upper(), + f"{result.seconds:.2f}s", + result.detail, + ) + for result in results + ] + lines = _table(rows, ("contract", "class", "speed", "status", "time", "detail")) + lines.append("") + lines.append("wall-class coverage (selected):") + for domain in summary["domains"]: + lines.append( + f" {domain['wall_class']} {domain['domain']:<28} " + f"{domain['total']} contracts: {domain['pass']} pass, " + f"{domain['fail']} fail, {domain['skip']} skip" + ) + counts = summary["counts"] + lines.append("") + lines.append( + f"total: {counts['pass']} pass, {counts['fail']} fail, {counts['skip']} skip " + f"-> {'OK' if summary['ok'] else 'FAIL'}" + ) + return lines + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="run_contracts.py", + description=( + "Run the wall-contract registry: fast contracts by default, all " + "with --all (offline; never touches hardware)." + ), + ) + group = parser.add_mutually_exclusive_group() + group.add_argument( + "--fast", action="store_true", help="run fast contracts only (default)" + ) + group.add_argument("--all", action="store_true", help="run fast and slow contracts") + parser.add_argument( + "--name", + action="append", + default=[], + metavar="NAME", + help="run one named contract (repeatable); overrides --fast/--all selection", + ) + parser.add_argument("--json", metavar="PATH", help="write the run summary as JSON here") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + selected, unknown = _select(args) + if unknown: + parser.error("unknown contract name(s): " + ", ".join(unknown)) + results = run_contracts(selected) + summary = summarize(results) + for line in _render(results, summary): + print(line) + if args.json: + out_path = os.path.abspath(args.json) + parent = os.path.dirname(out_path) + if parent: + os.makedirs(parent, exist_ok=True) + with open(out_path, "w", encoding="utf-8") as fh: + json.dump(summary, fh, indent=2, sort_keys=True) + fh.write("\n") + print(f"json: {out_path}") + return 0 if summary["ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py b/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py new file mode 100644 index 0000000..1099fb0 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py @@ -0,0 +1,698 @@ +#!/usr/bin/env python3 +"""Collect a canonical server-state snapshot from cycle evidence directories. + +The differential loop (live run vs emulated run, run N vs run N+1) needs one +machine-readable view of the server state a cycle produced. Cycles already +write that state as text: labelled field variables (``LABEL[offset]=0xVALUE``), +labelled hexdump blocks (an ``INSTANCE[0000]: w w w w``-style row, or a header +line naming the label followed by ``OFFSET: w w w w`` rows -- e.g. +``LOGINIT-raw`` / ``LOGRM-raw`` raw task-log dumps), and ``vgsp-nocat`` record +lines carrying ``recType=`` / ``errorCode=``. This harness discovers those +files under evidence/session directories, normalizes every dump to +little-endian u32 words, and emits one JSON snapshot that two runs can be +compared with ``diff``. + +Schemas +------- + +``greeneye-server-snapshot/1``:: + + {"schema": "greeneye-server-snapshot/1", + "source": "<string>", + "created": "<ISO-8601>", + "chunks": [{"label": "INSTANCE", "base": "0x0", + "words": ["0x00000000", ...]}], + "records": [{"kind": "record", "raw": "...", + "rec_type": 5, "error_code": 0x57f1344, + "source_name": "ASSERT"}]} + +``greeneye-server-snapshot-diff/1``:: + + {"schema": "greeneye-server-snapshot-diff/1", + "a": {"source": ..., "created": ...}, "b": {...}, + "chunks": [{"label": "INSTANCE", "status": "changed|added|removed|unchanged", + "base": {"a": "0x0", "b": "0x0"}, + "words": {"a": 1024, "b": 1024}, + "added": [{"index": 8, "value": "0x00000000"}], + "removed": [{"index": 8, "value": "0x00000001"}], + "changed": [{"index": 8, "a": "0x...", "b": "0x..."}]}], + "records": {"added": [...], "removed": [...]}, + "summary": {"identical": false, "chunks_added": 0, ...}} + +Normalization rules (the consumer depends on them): + +* words are 8-hex-digit lowercase strings of little-endian u32 values; +* a ``LABEL[offset]=0xVALUE`` value wider than 32 bits becomes two words + (low word first); its offset orders the entries and selects ``base``, and + the values are appended in offset order (the offset is a table index there, + not a byte address); +* hexdump rows are placed by byte offset (``(offset - base) // 4``) when + 4-byte aligned, so a row missing in one run still aligns the rest; +* ``base`` is the smallest offset seen for the chunk (``0x0`` when none); +* chunks are sorted by label; a repeated label is merged only when its base + and word list are byte-identical, otherwise it is kept as ``LABEL#2``, + ``LABEL#3`` ... and the collision is reported in the run stats; +* non-empty records are deduplicated by raw line and sorted by raw line; +* ``created`` is the only variable field: pin it with ``--created`` or + ``SOURCE_DATE_EPOCH`` for deterministic output. + +Honesty: this is an evidence normalizer, not a state oracle. It preserves +what the cycle wrote -- offsets are honoured, identical duplicates are merged, +differing duplicates are renamed, and dump-looking lines that fail to parse +are counted (``unparsed``) instead of being silently dropped. It never +executes anything, touches no GPU and reads only files given on the command +line. +""" +from __future__ import annotations + +import argparse +import fnmatch +import json +import os +import re +import sys +import tempfile +import time +from typing import Any, Iterable + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +SNAPSHOT_FORMAT = "greeneye-server-snapshot/1" +DIFF_FORMAT = "greeneye-server-snapshot-diff/1" +DEMO_FORMAT = "greeneye-server-snapshot-demo/1" + +#: File-name globs discovered under each ``--dir`` unless extended with +#: ``--include``. fnmatch's ``*`` matches leading dots, so the rolling host +#: dmesg captures (``.dmesg-*.rolling``) are covered by ``*.rolling``. +DEFAULT_INCLUDES = ("*.txt", "*.log", "*.rolling") + +#: ``INSTANCE[0000]: w w w w`` -- one labelled dump row inside an arbitrary +#: prefix (dmesg timestamp, subsystem, case id). Offsets are hexadecimal. +_INLINE_ROW_RE = re.compile( + r"(?P<label>[A-Za-z_][A-Za-z0-9_.]*)\[(?P<offset>(?:0x)?[0-9A-Fa-f]+)\]:" + r"[ \t]*(?P<tail>[0-9A-Fa-f]{1,8}(?:[ \t]+[0-9A-Fa-f]{1,8})*)[ \t]*$" +) + +#: ``LABEL[offset]=0xVALUE`` (optionally ``LABEL[0].suffix=``), several per +#: line allowed (``root[0]=... pd2[0]=... pd0[0].small=...``). +_LABELED_VALUE_RE = re.compile( + r"(?P<label>[A-Za-z_][A-Za-z0-9_.]*)\[(?P<offset>(?:0x)?[0-9A-Fa-f]+)\]" + r"(?P<suffix>\.[A-Za-z_][A-Za-z0-9_]*)?[ \t]*=[ \t]*" + r"(?P<value>0[xX][0-9A-Fa-f]+)\b" +) + +#: A bare hexdump row (``00000010: w w w w``) inside a labelled block. +_BLOCK_ROW_RE = re.compile( + r"^(?P<offset>[0-9A-Fa-f]{4,16}):[ \t]+" + r"(?P<tail>[0-9A-Fa-f]{1,8}(?:[ \t]+[0-9A-Fa-f]{1,8})*)[ \t]*$" +) + +#: A block header naming the label for the rows that follow: ``LOGINIT-raw:``, +#: ``LOGRM-raw:``, ``RAMFC:`` ... Used only as one-line lookahead. +_BLOCK_HEADER_RE = re.compile(r"^(?P<label>[A-Za-z_][A-Za-z0-9_.-]*)[ \t]*:") + +#: A GSP-RM error record line (``rec seq=1 recType=5 errorCode=0x...``). +_RECORD_RE = re.compile( + r"\brecType=(?P<rec_type>\d+)\b.*?\berrorCode=(?P<error_code>0[xX][0-9A-Fa-f]+)" +) + +_SOURCE_RE = re.compile(r'\bsource="(?P<source_name>[^"]*)"') + +#: Loose shape check for a line that looks like a dump but failed to parse; +#: these are counted so silent data loss is visible in the run stats. +_UNPARSED_HINT_RE = re.compile( + r"\[[0-9A-Fa-fx]+\][ \t]*[=:]|^[0-9A-Fa-f]{4,}:|\brecType=" +) + + +class SnapshotError(ValueError): + """An input file or snapshot the tool cannot honour.""" + + +def _new_stats() -> dict[str, Any]: + return { + "files": 0, + "lines": 0, + "chunks": 0, + "words": 0, + "records": 0, + "unparsed": 0, + "merged": 0, + "collisions": 0, + "conflicts": 0, + "misaligned": 0, + "collision_list": [], + } + + +class _Builder: + """Words captured for one label occurrence before normalization. + + Entries are ``(offset, [u32 word, ...])`` in encounter order. Hexdump + builders are positional: entries are placed at ``(offset - base) // 4`` + when the byte offset is 4-byte aligned; otherwise they are appended after + the highest placed word. Value-line builders (``LABEL[offset]=0xVALUE``) + are sequential: the offset only orders the entries and selects ``base`` + (its unit is a table entry index there, not a byte), so each entry's words + are appended in offset order. A later value that would overwrite a + different value at the same slot is dropped and counted as a + ``conflict``. + """ + + __slots__ = ("label", "positional", "entries", "conflicts", "misaligned") + + def __init__(self, label: str, positional: bool = True): + self.label = label + self.positional = positional + self.entries: list[tuple[int, list[str]]] = [] + self.conflicts = 0 + self.misaligned = 0 + + def add(self, offset: int, words: list[str]) -> None: + if not words: + return + for existing_offset, existing_words in self.entries: + if existing_offset == offset and existing_words == words: + return + self.entries.append((offset, words)) + + def build(self) -> tuple[int, list[str]]: + base = min(offset for offset, _ in self.entries) + placed: dict[int, str] = {} + append_at = 0 + conflicts = 0 + misaligned = 0 + for offset, words in sorted(self.entries, key=lambda entry: entry[0]): + if not self.positional: + index = append_at + elif (offset - base) % 4 == 0: + index = (offset - base) // 4 + else: + index = append_at + misaligned += 1 + for step, word in enumerate(words): + slot = index + step + if slot in placed and placed[slot] != word: + conflicts += 1 + elif slot not in placed: + placed[slot] = word + append_at = max(append_at, index + len(words)) + self.conflicts = conflicts + self.misaligned = misaligned + return base, [placed[index] for index in sorted(placed)] + + +def _word(value: int) -> str: + return f"0x{value & 0xFFFFFFFF:08x}" + + +def _value_words(value: int) -> list[str]: + """Split a value into little-endian u32 words (low word first).""" + words = [] + while True: + words.append(_word(value)) + value >>= 32 + if value == 0: + return words + + +def _split_words(tail: str) -> list[str]: + return [f"0x{int(token, 16):08x}" for token in tail.split()] + + +def _row_offset(text: str) -> int: + return int(text, 16) + + +def _labeled_offset(text: str) -> int: + if text.lower().startswith("0x"): + return int(text, 16) + try: + return int(text, 10) + except ValueError: + return int(text, 16) + + +def _match_record(line: str) -> dict[str, Any] | None: + match = _RECORD_RE.search(line) + if not match: + return None + record: dict[str, Any] = { + "kind": "record", + "raw": line, + "rec_type": int(match.group("rec_type")), + "error_code": int(match.group("error_code"), 16), + } + source = _SOURCE_RE.search(line) + if source: + record["source_name"] = source.group("source_name") + return record + + +def _parse_text(text: str, stats: dict[str, Any]) -> tuple[list[_Builder], list[dict]]: + occurrences: list[_Builder] = [] + records: list[dict] = [] + by_value_label: dict[str, _Builder] = {} + current_block: _Builder | None = None + inline_block: _Builder | None = None + pending_header: str | None = None + + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + current_block = pending_header = inline_block = None + continue + + record = _match_record(stripped) + if record is not None: + records.append(record) + current_block = pending_header = inline_block = None + continue + + match = _INLINE_ROW_RE.search(stripped) + if match: + label = match.group("label") + words = _split_words(match.group("tail")) + if inline_block is not None and inline_block.label == label: + inline_block.add(_row_offset(match.group("offset")), words) + else: + inline_block = _Builder(label) + inline_block.add(_row_offset(match.group("offset")), words) + occurrences.append(inline_block) + current_block = pending_header = None + continue + + match = _BLOCK_ROW_RE.match(stripped) + if match: + if current_block is None: + if pending_header is None: + stats["unparsed"] += 1 + continue + current_block = _Builder(pending_header) + occurrences.append(current_block) + current_block.add(int(match.group("offset"), 16), + _split_words(match.group("tail"))) + pending_header = inline_block = None + continue + + value_matches = list(_LABELED_VALUE_RE.finditer(stripped)) + if value_matches: + for value_match in value_matches: + label = value_match.group("label") + if value_match.group("suffix"): + label += value_match.group("suffix") + builder = by_value_label.get(label) + if builder is None: + builder = _Builder(label, positional=False) + by_value_label[label] = builder + occurrences.append(builder) + builder.add(_labeled_offset(value_match.group("offset")), + _value_words(int(value_match.group("value"), 16))) + current_block = pending_header = inline_block = None + continue + + if _UNPARSED_HINT_RE.search(stripped): + stats["unparsed"] += 1 + + current_block = inline_block = None + header = _BLOCK_HEADER_RE.match(stripped) + pending_header = header.group("label") if header else None + + return occurrences, records + + +def _chunk_sort_key(label: str) -> tuple[str, int, str]: + stem, separator, tail = label.rpartition("#") + if separator and tail.isdigit(): + return stem, int(tail), label + return label, 0, label + + +def _finalize(occurrences: Iterable[_Builder], + stats: dict[str, Any]) -> list[dict[str, Any]]: + chunks: dict[str, dict[str, Any]] = {} + for builder in occurrences: + base, words = builder.build() + stats["conflicts"] += builder.conflicts + stats["misaligned"] += builder.misaligned + if not words: + continue + key = builder.label + suffix = 1 + merged = False + while key in chunks: + existing = chunks[key] + if existing["base"] == f"{base:#x}" and existing["words"] == words: + stats["merged"] += 1 + merged = True + break + suffix += 1 + key = f"{builder.label}#{suffix}" + if merged: + continue + chunks[key] = {"label": key, "base": f"{base:#x}", "words": words} + if suffix > 1: + stats["collisions"] += 1 + stats["collision_list"].append( + {"label": builder.label, "renamed": key, + "base": f"{base:#x}", "words": len(words)}) + return [chunks[key] for key in sorted(chunks, key=_chunk_sort_key)] + + +def discover(dirs: Iterable[str] | None, files: Iterable[str] | None, + includes: Iterable[str] | None = None) -> list[str]: + """Return the sorted absolute paths matched by the dirs/files/includes.""" + patterns = tuple(DEFAULT_INCLUDES) + tuple(includes or ()) + found: set[str] = set() + for directory in dirs or (): + if not os.path.isdir(directory): + raise SnapshotError(f"not a directory: {directory}") + for root, _subdirs, names in os.walk(directory): + for name in names: + if any(fnmatch.fnmatch(name, pattern) for pattern in patterns): + found.add(os.path.abspath(os.path.join(root, name))) + for path in files or (): + if not os.path.isfile(path): + raise SnapshotError(f"not a file: {path}") + found.add(os.path.abspath(path)) + return sorted(found) + + +def _default_source(dirs: Iterable[str] | None, + files: Iterable[str] | None) -> str: + roots = sorted(os.path.abspath(path) + for path in list(dirs or ()) + list(files or ())) + return ",".join(roots) if roots else "empty" + + +def _created_iso() -> str: + """Build timestamp; stable when ``SOURCE_DATE_EPOCH`` is set.""" + raw = os.environ.get("SOURCE_DATE_EPOCH") + try: + epoch = int(raw) if raw else int(time.time()) + except ValueError: + epoch = 0 + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch)) + + +def collect(dirs: Iterable[str] | None = None, + files: Iterable[str] | None = None, + includes: Iterable[str] | None = None, + source: str | None = None, + created: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]: + """Collect every dump in ``dirs``/``files`` into one snapshot. + + Returns ``(snapshot, stats)``; ``stats`` carries the parse accounting + (files, lines, chunks, words, records, unparsed candidates, merged + duplicates, renamed collisions, conflicts, misaligned offsets) and is + printed by the CLI but is deliberately not part of the consumed schema. + """ + paths = discover(dirs, files, includes) + if not paths: + raise SnapshotError( + "no input files matched the requested --dir/--file/--include set") + stats = _new_stats() + occurrences: list[_Builder] = [] + records: dict[str, dict] = {} + for path in paths: + try: + with open(path, "r", encoding="utf-8", errors="replace") as handle: + text = handle.read() + except OSError as exc: + raise SnapshotError(f"cannot read {path}: {exc}") from exc + stats["files"] += 1 + stats["lines"] += len(text.splitlines()) + file_occurrences, file_records = _parse_text(text, stats) + occurrences.extend(file_occurrences) + for record in file_records: + records.setdefault(record["raw"], record) + chunks = _finalize(occurrences, stats) + stats["chunks"] = len(chunks) + stats["words"] = sum(len(chunk["words"]) for chunk in chunks) + stats["records"] = len(records) + snapshot = { + "schema": SNAPSHOT_FORMAT, + "source": source if source is not None else _default_source(dirs, files), + "created": created if created is not None else _created_iso(), + "chunks": chunks, + "records": sorted(records.values(), key=lambda record: record["raw"]), + } + return snapshot, stats + + +def diff_snapshots(a: dict[str, Any], b: dict[str, Any]) -> dict[str, Any]: + """Compare two ``greeneye-server-snapshot/1`` snapshots word by word. + + Emits a ``greeneye-server-snapshot-diff/1`` report: per shared chunk the + added/removed/changed word lists (with word indices), whole-chunk + added/removed rows, per-chunk and total counts, and record deltas. Output + is sorted by label and by word index, so two diffs of the same inputs are + byte-identical. + """ + for name, snapshot in (("a", a), ("b", b)): + if not isinstance(snapshot, dict) or snapshot.get("schema") != SNAPSHOT_FORMAT: + raise SnapshotError(f"snapshot {name} is not {SNAPSHOT_FORMAT}") + a_chunks = {chunk["label"]: chunk for chunk in a.get("chunks", [])} + b_chunks = {chunk["label"]: chunk for chunk in b.get("chunks", [])} + rows: list[dict[str, Any]] = [] + counts = { + "chunks_added": 0, "chunks_removed": 0, "chunks_changed": 0, + "chunks_unchanged": 0, "words_added": 0, "words_removed": 0, + "words_changed": 0, + } + for label in sorted(set(a_chunks) | set(b_chunks), key=_chunk_sort_key): + left = a_chunks.get(label) + right = b_chunks.get(label) + if left is None: + added = [{"index": index, "value": word} + for index, word in enumerate(right["words"])] + row = {"label": label, "status": "added", + "base": {"a": None, "b": right["base"]}, + "words": {"a": 0, "b": len(right["words"])}, + "added": added, "removed": [], "changed": []} + counts["chunks_added"] += 1 + counts["words_added"] += len(added) + elif right is None: + removed = [{"index": index, "value": word} + for index, word in enumerate(left["words"])] + row = {"label": label, "status": "removed", + "base": {"a": left["base"], "b": None}, + "words": {"a": len(left["words"]), "b": 0}, + "added": [], "removed": removed, "changed": []} + counts["chunks_removed"] += 1 + counts["words_removed"] += len(removed) + else: + a_words, b_words = left["words"], right["words"] + added, removed, changed = [], [], [] + for index in range(max(len(a_words), len(b_words))): + if index >= len(a_words): + added.append({"index": index, "value": b_words[index]}) + elif index >= len(b_words): + removed.append({"index": index, "value": a_words[index]}) + elif a_words[index] != b_words[index]: + changed.append({"index": index, "a": a_words[index], + "b": b_words[index]}) + unchanged = (not added and not removed and not changed + and left["base"] == right["base"]) + row = {"label": label, + "status": "unchanged" if unchanged else "changed", + "base": {"a": left["base"], "b": right["base"]}, + "words": {"a": len(a_words), "b": len(b_words)}, + "added": added, "removed": removed, "changed": changed} + if unchanged: + counts["chunks_unchanged"] += 1 + else: + counts["chunks_changed"] += 1 + counts["words_added"] += len(added) + counts["words_removed"] += len(removed) + counts["words_changed"] += len(changed) + rows.append(row) + + a_records = {record.get("raw", "") for record in a.get("records", [])} + b_records = {record.get("raw", "") for record in b.get("records", [])} + records_added = sorted(b_records - a_records) + records_removed = sorted(a_records - b_records) + counts["records_added"] = len(records_added) + counts["records_removed"] = len(records_removed) + identical = (counts["chunks_added"] == 0 and counts["chunks_removed"] == 0 + and counts["chunks_changed"] == 0 + and not records_added and not records_removed) + return { + "schema": DIFF_FORMAT, + "a": {"source": a.get("source"), "created": a.get("created")}, + "b": {"source": b.get("source"), "created": b.get("created")}, + "chunks": rows, + "records": {"added": records_added, "removed": records_removed}, + "summary": {"identical": identical, **counts}, + } + + +# --------------------------------------------------------------------------- +# Demo: two tiny snapshots in a tempdir, collected and diffed +# --------------------------------------------------------------------------- + +DEMO_A = """\ +[demo] MOCK[0000]: 00000001 00000002 00000003 00000004 +alpha[0]=0x00000011 +alpha[4]=0x00000022 +""" + +DEMO_B = """\ +[demo] MOCK[0000]: 00000001 000000ff 00000003 00000004 00000005 +alpha[0]=0x00000011 +alpha[4]=0x00000033 +""" + + +def _write_text(path: str, text: str) -> str: + with open(path, "w", encoding="utf-8") as handle: + handle.write(text) + return os.path.abspath(path) + + +def _write_json(path: str, payload: dict[str, Any]) -> str: + with open(path, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2, sort_keys=True) + handle.write("\n") + return os.path.abspath(path) + + +def run_demo(workdir: str) -> dict[str, Any]: + """Craft two tiny evidence files in ``workdir``, collect both and diff.""" + a_path = _write_text(os.path.join(workdir, "run-a.txt"), DEMO_A) + b_path = _write_text(os.path.join(workdir, "run-b.txt"), DEMO_B) + snapshot_a, _stats_a = collect(files=[a_path], source="demo-a", + created="2026-01-01T00:00:00Z") + snapshot_b, _stats_b = collect(files=[b_path], source="demo-b", + created="2026-01-01T00:00:00Z") + return { + "schema": DEMO_FORMAT, + "workdir": os.path.abspath(workdir), + "snapshot_a": snapshot_a, + "snapshot_b": snapshot_b, + "diff": diff_snapshots(snapshot_a, snapshot_b), + } + + +def _load_json(path: str) -> dict[str, Any]: + try: + with open(path, "r", encoding="utf-8") as handle: + return json.load(handle) + except (OSError, json.JSONDecodeError) as exc: + raise SnapshotError(f"cannot read {path}: {exc}") from exc + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="server_snapshot.py", + description="Collect and diff canonical server-state snapshots " + "(offline evidence normalizer).", + ) + sub = parser.add_subparsers(dest="command") + + collect_cmd = sub.add_parser( + "collect", help="normalize evidence files into one snapshot") + collect_cmd.add_argument("--dir", action="append", default=[], + help="evidence/session directory (repeatable)") + collect_cmd.add_argument("--file", action="append", default=[], + help="explicit evidence file (repeatable)") + collect_cmd.add_argument( + "--include", action="append", default=[], + help="additional file-name glob under --dir (repeatable); " + f"defaults: {', '.join(DEFAULT_INCLUDES)}") + collect_cmd.add_argument("--source", help="override the snapshot source") + collect_cmd.add_argument( + "--created", help="override the ISO-8601 timestamp (SOURCE_DATE_EPOCH " + "is honoured otherwise)") + collect_cmd.add_argument("--json", help="write the snapshot here instead " + "of stdout") + + diff_cmd = sub.add_parser("diff", help="compare two snapshot JSON files") + diff_cmd.add_argument("--a", required=True, help="first snapshot JSON") + diff_cmd.add_argument("--b", required=True, help="second snapshot JSON") + diff_cmd.add_argument("--json", help="write the diff here instead of stdout") + + demo_cmd = sub.add_parser( + "demo", help="craft two tiny snapshots in a tempdir and diff them") + demo_cmd.add_argument("--out", help="keep the crafted files in this directory") + return parser + + +def _collect_command(args) -> int: + if not args.dir and not args.file: + raise SnapshotError("collect needs at least one --dir or --file") + snapshot, stats = collect(dirs=args.dir, files=args.file, + includes=args.include, source=args.source, + created=args.created) + if args.json: + path = _write_json(args.json, snapshot) + summary = { + "schema": snapshot["schema"], + "source": snapshot["source"], + "created": snapshot["created"], + "chunks": len(snapshot["chunks"]), + "words": stats["words"], + "records": len(snapshot["records"]), + "unparsed": stats["unparsed"], + "collisions": stats["collisions"], + "snapshot_json": path, + } + print(json.dumps(summary, indent=2, sort_keys=True)) + else: + print(json.dumps(snapshot, indent=2, sort_keys=True)) + print(f"server_snapshot: {stats['files']} file(s), {stats['lines']} line(s), " + f"{stats['chunks']} chunk(s), {stats['words']} word(s), " + f"{stats['records']} record(s), {stats['unparsed']} unparsed, " + f"{stats['merged']} merged, {stats['collisions']} collision(s), " + f"{stats['conflicts']} conflict(s)", file=sys.stderr) + return 0 + + +def _diff_command(args) -> int: + report = diff_snapshots(_load_json(args.a), _load_json(args.b)) + if args.json: + path = _write_json(args.json, report) + print(json.dumps({"schema": report["schema"], **report["summary"], + "diff_json": path}, indent=2, sort_keys=True)) + else: + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if report["summary"]["identical"] else 3 + + +def _demo_command(args) -> int: + if args.out: + os.makedirs(args.out, exist_ok=True) + demo = run_demo(args.out) + else: + with tempfile.TemporaryDirectory(prefix="server-snapshot-demo-") as workdir: + demo = run_demo(workdir) + for row in demo["diff"]["chunks"]: + if row["status"] != "unchanged": + print(f"{row['status']:<9} {row['label']:<16} " + f"+{len(row['added'])} -{len(row['removed'])} " + f"~{len(row['changed'])}") + print(json.dumps(demo["diff"]["summary"], indent=2, sort_keys=True)) + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "collect": + return _collect_command(args) + if args.command == "diff": + return _diff_command(args) + if args.command == "demo": + return _demo_command(args) + parser.print_help() + return 2 + except SnapshotError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/state_decode.py b/debug/risc-v-gsp-rm-vm/harness/state_decode.py new file mode 100644 index 0000000..351515a --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/state_decode.py @@ -0,0 +1,1082 @@ +#!/usr/bin/env python3 +"""Decode engine-visible channel state words into named fields, and diff captures. + +During GR bring-up the answer often hides in a handful of 32-bit words inside +the INSTANCE block, the RAMFC chunk and context-buffer headers: one bit turns +"nothing happens" into an engine that runs or refuses. Raw hex hides that; +this tool turns a captured dump into a named field table and makes two +captures differ field-by-field. + +Schemas: + +* ``greeneye-state-decode/1`` -- the decoded report produced by ``decode``. +* ``greeneye-state-diff/1`` -- the word/field diff produced by ``diff``. +* ``greeneye-server-snapshot/1`` -- an input snapshot emitted elsewhere:: + + {"schema": "greeneye-server-snapshot/1", "source": "...", + "chunks": [{"label": "INSTANCE", "base": "0x0", + "words": ["0x...", "..."]}]} + + (one hex string or integer per 32-bit word; accepted exactly). +* ``greeneye-state-map/1`` -- an optional field map (``--map``), see below. + +Input formats (auto-detected per ``--input``): + +(a) text lines ``LABEL[OFFSET]=VALUE`` -- label case-insensitive, offset and + value hex (``0x..``, letters, or leading-zero padded) or decimal; every + other line is ignored; +(b) a dump row with a bracketed chunk offset and words, e.g. + ``... INSTANCE[0040]: 00100000 00030000 ...`` -- this is the shape live + ladder traces print (8 words per row; any count accepted); +(c) plain hexdump rows ``00000000: 000002d2 00000000 ...`` named with + ``--label``/``--base``; +(d) raw little-endian u32 words (``--bin``, or auto-detected when a text parse + finds nothing and the bytes are control-heavy and 4-byte aligned); +(e) the snapshot JSON above. + +Offsets inside a chunk are byte offsets relative to the chunk start (all +inputs above are u32 word dumps, so offsets are multiples of 4). ``base`` is +the engine-visible address of chunk-relative offset 0: it is recorded as +provenance and for the diff, but it does not shift map lookup. + +Bundled map (deliberately small; extend or override it from a map file, never +by guessing here). Provenance is live bring-up observation (2026-10 GSP +ladder runs): + +* ``INSTANCE`` 0x010: magic ``0xface``. +* ``INSTANCE`` 0x048: GPFIFO VA, observed as ``0x100000``. +* ``RAMFC`` 0x048: ``RAMFC_TARGET`` -- bits 16/17 ``ENG_CTX_VALID`` / + ``CE_CTX_VALID``; the observed live success word ``0x00030000`` sets both. + The bit assignment is observation-driven and fully configurable via the map. + +Map file:: + + {"format": "greeneye-state-map/1", "mode": "extend" | "replace", + "entries": [ + {"chunk": "RAMFC", "label": "RAMFC_TARGET", "offset": "0x48", + "width": 4, + "bits": [{"name": "ENG_CTX_VALID", "lo": 16, "hi": 16, + "values": {"0": "clear", "1": "set"}}], + "note": "..."} + ]} + +A user entry with the same ``chunk`` and ``offset`` replaces the bundled +entry; new chunk/offset pairs extend the table. ``mode: replace`` drops the +bundled table first. An entry without ``bits`` decodes as a named value. + +Honesty: this tool names observed words; it does not prove their meaning. An +unmapped word is silence, not absence. Field offsets without evidence stay +unnamed until a map supplies them. Offline only: no hardware, no module, no +guest, no network. +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import sys +import tempfile +from typing import Any + +DECODE_FORMAT = "greeneye-state-decode/1" +DIFF_FORMAT = "greeneye-state-diff/1" +SNAPSHOT_FORMAT = "greeneye-server-snapshot/1" +MAP_FORMAT = "greeneye-state-map/1" +DEMO_FORMAT = "greeneye-state-decode-demo/1" + +WORD_BYTES = 4 +WORD_MASK = 0xFFFFFFFF +WORD_WIDTHS = (4, 8) + + +class StateDecodeError(ValueError): + """An input, map or request the tool cannot honour.""" + + +# --------------------------------------------------------------------------- +# Bundled field table +# --------------------------------------------------------------------------- +# Every entry is an observation from live GSP bring-up traces (2026-10). The +# table is intentionally tiny: unnamed offsets stay unnamed until evidence +# earns them a name, and the map file carries everything else. +BUILTIN_MAP: dict[str, Any] = { + "format": MAP_FORMAT, + "mode": "extend", + "entries": [ + { + "chunk": "INSTANCE", + "label": "MAGIC_0FACE", + "offset": "0x010", + "width": 4, + "note": "Instance magic observed as 0xface at chunk byte 0x010.", + }, + { + "chunk": "INSTANCE", + "label": "GPFIFO_VA", + "offset": "0x048", + "width": 4, + "note": "GPFIFO VA observed as 0x100000 at chunk byte 0x048.", + }, + { + "chunk": "RAMFC", + "label": "RAMFC_TARGET", + "offset": "0x048", + "width": 4, + "bits": [ + {"name": "ENG_CTX_VALID", "lo": 16, "hi": 16}, + {"name": "CE_CTX_VALID", "lo": 17, "hi": 17}, + ], + "note": ("Engine/CE context-valid bits; the observed live success " + "word 0x00030000 sets both. Bits are configurable via " + "--map."), + }, + ], +} + +# --------------------------------------------------------------------------- +# Number and token parsing +# --------------------------------------------------------------------------- + +_EQ_RE = re.compile( + r"^\s*(?P<label>[A-Za-z_][A-Za-z0-9_.-]*)\s*" + r"\[\s*(?P<off>[0-9a-fA-FxX]+)\s*\]\s*=\s*(?P<val>[0-9a-fA-FxX]+)\s*$") + +_ROW_RE = re.compile( + r"(?P<label>[A-Za-z_][A-Za-z0-9_.-]*)\s*" + r"\[\s*(?P<off>[0-9a-fA-FxX]+)\s*\]\s*:\s*" + r"(?P<words>[0-9a-fA-FxX]+(?:\s+[0-9a-fA-FxX]+)*)\s*$") + +_HEXDUMP_RE = re.compile( + r"^\s*(?P<off>[0-9a-fA-FxX]{2,16})\s*:\s*" + r"(?P<words>[0-9a-fA-FxX]+(?:\s+[0-9a-fA-FxX]+)*)\s*$") + +_HEX_RE = re.compile(r"[0-9a-fA-F]+") + +_TEXT_BYTES = frozenset(b"\t\n\r") | frozenset(range(0x20, 0x7F)) + + +def parse_number(value: Any, *, why: str = "number") -> int: + """Parse an int, a ``0x`` string, or a bare token. + + Bare tokens with a hex letter (``face``) or a leading zero and more + digits (``0040``) are read as hex, matching dump padding; other bare + digit strings are decimal. Write ``0x..`` when in doubt. + """ + if isinstance(value, bool): + raise StateDecodeError(f"bad {why} {value!r}") + if isinstance(value, int): + return value + if not isinstance(value, str): + raise StateDecodeError(f"bad {why} {value!r}") + text = value.strip() + if not text: + raise StateDecodeError(f"bad {why} {value!r}") + sign = 1 + if text[0] in "+-": + if text[0] == "-": + sign = -1 + text = text[1:] + low = text.lower() + if low.startswith("0x"): + base, digits = 16, text[2:] + elif re.fullmatch(r"[0-9]+", text): + base = 16 if text.startswith("0") and len(text) > 1 else 10 + digits = text + elif _HEX_RE.fullmatch(text): + base, digits = 16, text + else: + raise StateDecodeError(f"bad {why} {value!r}") + try: + return sign * int(digits, base) + except ValueError as exc: + raise StateDecodeError(f"bad {why} {value!r}") from exc + + +def _parse_word(value: Any, *, where: str) -> int: + number = parse_number(value, why=where) + if not 0 <= number <= WORD_MASK: + raise StateDecodeError(f"{where} {value!r} does not fit in 32 bits") + return number + + +def _hex(value: int, width: int = 4) -> str: + return f"{value:#0{2 + width * 2}x}" + + +def _parse_base(value: Any) -> int: + if value is None: + return 0 + return parse_number(value, why="base") + + +# --------------------------------------------------------------------------- +# Map loading (bundled <- override/extension) +# --------------------------------------------------------------------------- + +def _normalize_entry(raw: Any, *, where: str) -> dict[str, Any]: + if not isinstance(raw, dict): + raise StateDecodeError(f"{where} must be an object") + chunk = raw.get("chunk") + label = raw.get("label") + if not isinstance(chunk, str) or not chunk: + raise StateDecodeError(f"{where} needs a non-empty chunk") + if not isinstance(label, str) or not label: + raise StateDecodeError(f"{where} needs a non-empty label") + if raw.get("offset") is None: + raise StateDecodeError(f"{where} needs an offset") + offset = parse_number(raw["offset"], why=f"{where} offset") + width = raw.get("width", 4) + width = parse_number(width, why=f"{where} width") + if width not in WORD_WIDTHS: + raise StateDecodeError(f"{where} width {width!r} is not 4 or 8") + if offset < 0 or offset % WORD_BYTES: + raise StateDecodeError(f"{where} offset {offset:#x} is not word-aligned") + bits_raw = raw.get("bits") or [] + if not isinstance(bits_raw, list): + raise StateDecodeError(f"{where} bits must be a list") + nbits = width * 8 + bits = [] + for index, bit in enumerate(bits_raw): + if not isinstance(bit, dict): + raise StateDecodeError(f"{where} bits[{index}] must be an object") + name = bit.get("name") + if not isinstance(name, str) or not name: + raise StateDecodeError(f"{where} bits[{index}] needs a name") + lo = parse_number(bit.get("lo"), why=f"{where} bits[{index}].lo") + hi = parse_number(bit.get("hi"), why=f"{where} bits[{index}].hi") + if not 0 <= lo <= hi < nbits: + raise StateDecodeError( + f"{where} bits[{index}] range {lo}..{hi} outside width {width}") + values = bit.get("values") + norm_values: dict[int, str] = {} + if values is not None: + if not isinstance(values, dict): + raise StateDecodeError( + f"{where} bits[{index}].values must be an object") + for key, text in values.items(): + norm_values[parse_number(key, why=f"{where} values key")] = str(text) + bits.append({"name": name, "lo": lo, "hi": hi, "values": norm_values}) + note = raw.get("note") + return { + "chunk": chunk, + "label": label, + "offset": offset, + "width": width, + "bits": bits, + "note": note if isinstance(note, str) else None, + } + + +def load_map(path: str | None = None) -> tuple[dict[tuple[str, int], dict], str]: + """Build the field table; returns ``(table, source)``. + + ``table`` is keyed by ``(chunk.lower(), offset)``. Without ``path`` the + bundled table is used. With ``path``, entries merge over the bundled + table (same chunk+offset replaces, new pairs extend) unless the file sets + ``"mode": "replace"``. + """ + entries: list[Any] = list(BUILTIN_MAP["entries"]) + source = "builtin" + if path: + if not os.path.isfile(path): + raise StateDecodeError(f"map not found: {path}") + with open(path, "r", encoding="utf-8") as handle: + doc = json.load(handle) + source = os.path.abspath(path) + if isinstance(doc, list): + raw_entries, mode = doc, "extend" + elif isinstance(doc, dict): + fmt = doc.get("format", MAP_FORMAT) + if fmt != MAP_FORMAT: + raise StateDecodeError( + f"unsupported map format {fmt!r} (want {MAP_FORMAT})") + raw_entries = doc.get("entries") + mode = doc.get("mode", "extend") + else: + raise StateDecodeError(f"map {path} must be an object or a list") + if mode not in ("extend", "replace"): + raise StateDecodeError(f"map mode {mode!r} is not extend/replace") + if not isinstance(raw_entries, list): + raise StateDecodeError(f"map {path} needs an entries list") + if mode == "replace": + entries = [] + entries.extend(raw_entries) + table: dict[tuple[str, int], dict] = {} + for index, raw in enumerate(entries): + entry = _normalize_entry(raw, where=f"map entry {index}") + table[(entry["chunk"].lower(), entry["offset"])] = entry + return table, source + + +# --------------------------------------------------------------------------- +# Input loading (auto-detected per input) +# --------------------------------------------------------------------------- + +def _read_bytes(path: str) -> bytes: + if not os.path.isfile(path): + raise StateDecodeError(f"input not found: {path}") + with open(path, "rb") as handle: + return handle.read() + + +def _try_json(data: bytes) -> Any: + """Return parsed JSON when the bytes look like JSON, else ``None``.""" + try: + text = data.decode("utf-8") + except UnicodeDecodeError: + return None + stripped = text.lstrip() + if not stripped.startswith("{"): + return None + try: + return json.loads(text) + except json.JSONDecodeError as exc: + raise StateDecodeError(f"invalid JSON input: {exc}") from exc + + +def _try_text(data: bytes) -> str | None: + try: + return data.decode("utf-8") + except UnicodeDecodeError: + return None + + +def _looks_binary(data: bytes) -> bool: + if not data or len(data) % WORD_BYTES: + return False + return any(byte not in _TEXT_BYTES for byte in data) + + +def _chunk_from_binary(data: bytes, label: str, base: int, source: str) -> dict: + if not data: + raise StateDecodeError(f"binary input {source} is empty") + if len(data) % WORD_BYTES: + raise StateDecodeError( + f"binary input {source} size {len(data)} is not a multiple of 4") + words = {offset: int.from_bytes(data[offset:offset + WORD_BYTES], "little") + for offset in range(0, len(data), WORD_BYTES)} + return {"label": label, "base": base, "words": words, + "format": "binary", "source": source} + + +def _document(path: str, fmt: str, label: str | None, base: int, + chunks: list[dict]) -> dict: + return { + "path": os.path.abspath(path), + "format": fmt, + "label": label, + "base": base, + "chunks": chunks, + } + + +def parse_text(text: str, *, default_label: str | None = None, + default_base: int = 0, source: str | None = None) -> list[dict]: + """Parse the text/hexdump/row formats into chunks. + + Chunk names are case-insensitive: ``INSTANCE`` and ``inst`` land in one + chunk, displayed with the first spelling seen. Every line that matches + none of the three patterns is ignored. + """ + chunks: dict[str, dict] = {} + + def chunk_for(label: str) -> dict: + key = label.lower() + chunk = chunks.get(key) + if chunk is None: + chunk = {"label": label, "base": default_base, "words": {}, + "format": "text", "source": source} + chunks[key] = chunk + return chunk + + def put(label: str, offset: int, value: int, where: str) -> None: + if offset < 0 or offset % WORD_BYTES: + raise StateDecodeError(f"{where}: offset {offset:#x} is not word-aligned") + if not 0 <= value <= WORD_MASK: + raise StateDecodeError(f"{where}: value {value:#x} does not fit in 32 bits") + chunk_for(label)["words"][offset] = value + + for lineno, raw_line in enumerate(text.splitlines(), 1): + line = raw_line.strip() + if not line: + continue + where = f"{source or '<text>'}:{lineno}" + match = _EQ_RE.match(line) + if match: + offset = parse_number(match.group("off"), why=f"{where} offset") + value = parse_number(match.group("val"), why=f"{where} value") + put(match.group("label"), offset, value, where) + continue + match = _ROW_RE.search(line) + if match: + offset = parse_number(match.group("off"), why=f"{where} offset") + for index, token in enumerate(match.group("words").split()): + put(match.group("label"), offset + index * WORD_BYTES, + int(token, 16), where) + continue + match = _HEXDUMP_RE.match(line) + if match: + label = default_label or "RAW" + offset = int(match.group("off"), 16) + for index, token in enumerate(match.group("words").split()): + put(label, offset + index * WORD_BYTES, int(token, 16), where) + continue + return [chunks[key] for key in chunks] + + +def load_input(spec: dict) -> dict: + """Load one input spec dict (path/label/base/binary) into a document.""" + path = spec.get("path") + if not path: + raise StateDecodeError("input spec needs a path") + data = _read_bytes(path) + label = spec.get("label") + base = _parse_base(spec.get("base")) + if spec.get("binary"): + chunk = _chunk_from_binary(data, label or "RAW", base, path) + return _document(path, "binary", label, base, [chunk]) + doc = _try_json(data) + if doc is not None: + return _document_from_json(path, doc, label, base) + text = _try_text(data) + if text is not None: + chunks = parse_text(text, default_label=label, default_base=base, + source=path) + if chunks: + return _document(path, "text", label, base, chunks) + if _looks_binary(data): + chunk = _chunk_from_binary(data, label or "RAW", base, path) + return _document(path, "binary", label, base, [chunk]) + raise StateDecodeError( + f"cannot decode {path}: not text fields, not {SNAPSHOT_FORMAT} JSON, " + f"not raw u32 words") + + +def _document_from_json(path: str, doc: Any, label: str | None, + base: int) -> dict: + if isinstance(doc, dict) and doc.get("format") == DECODE_FORMAT: + raise StateDecodeError(f"{path} is a decoded report; use `diff` on it") + if not isinstance(doc, dict) or doc.get("schema") != SNAPSHOT_FORMAT: + got = doc.get("schema") if isinstance(doc, dict) else type(doc).__name__ + raise StateDecodeError( + f"unsupported JSON input {path}: schema={got!r} " + f"(want {SNAPSHOT_FORMAT})") + raw_chunks = doc.get("chunks") + if not isinstance(raw_chunks, list) or not raw_chunks: + raise StateDecodeError(f"snapshot {path} has no chunks") + chunks = [] + for index, raw in enumerate(raw_chunks): + if not isinstance(raw, dict): + raise StateDecodeError(f"{path}:chunks[{index}] must be an object") + chunk_label = raw.get("label") or label or "RAW" + chunk_base = (_parse_base(raw["base"]) if raw.get("base") is not None + else base) + words_raw = raw.get("words") + if not isinstance(words_raw, list): + raise StateDecodeError( + f"{path}:chunks[{index}] needs a words list") + words = {} + for position, word in enumerate(words_raw): + words[position * WORD_BYTES] = _parse_word( + word, where=f"{path}:chunks[{index}].words[{position}]") + chunks.append({"label": str(chunk_label), "base": chunk_base, + "words": words, "format": "snapshot-json", + "source": path}) + return _document(path, "snapshot-json", label, base, chunks) + + +# --------------------------------------------------------------------------- +# Decode +# --------------------------------------------------------------------------- + +def _read_field_value(words: dict[int, int], offset: int, + width: int) -> int | None: + """Little-endian read of a 4/8-byte field; ``None`` when words are absent.""" + data = b"" + for step in range(0, width, WORD_BYTES): + word = words.get(offset + step) + if word is None: + return None + data += word.to_bytes(WORD_BYTES, "little") + return int.from_bytes(data, "little") + + +def _bit_state(bit: dict, value: int) -> str: + values = bit.get("values") or {} + if value in values: + return values[value] + mask = (1 << (bit["hi"] - bit["lo"] + 1)) - 1 + if value == 0: + return "clear" + if value == mask: + return "set" + return f"{value:#x}" + + +def _decode_field(entry: dict, value: int) -> dict: + width = entry["width"] + field: dict[str, Any] = { + "name": entry["label"], + "offset": f"{entry['offset']:#x}", + "width": width, + "kind": "bits" if entry["bits"] else "value", + "raw": _hex(value, width), + "value": value, + } + if entry.get("note"): + field["note"] = entry["note"] + if entry["bits"]: + bitfields = [] + for bit in entry["bits"]: + raw_bit = (value >> bit["lo"]) & ((1 << (bit["hi"] - bit["lo"] + 1)) - 1) + bitfields.append({ + "name": bit["name"], + "lo": bit["lo"], + "hi": bit["hi"], + "value": raw_bit, + "state": _bit_state(bit, raw_bit), + }) + field["bitfields"] = bitfields + return field + + +def decode_chunk(chunk: dict, entries: dict[tuple[str, int], dict]) -> dict: + """Decode one internal chunk into the ``greeneye-state-decode/1`` shape.""" + label = chunk["label"] + words: dict[int, int] = chunk["words"] + matched = sorted( + (entry for (chunk_key, _off), entry in entries.items() + if chunk_key == label.lower()), + key=lambda entry: (entry["offset"], entry["label"])) + fields = [] + covered: set[int] = set() + for entry in matched: + value = _read_field_value(words, entry["offset"], entry["width"]) + if value is None: + continue + fields.append(_decode_field(entry, value)) + covered.update(range(entry["offset"], + entry["offset"] + entry["width"], WORD_BYTES)) + unmapped = [{"offset": f"{offset:#x}", "raw": _hex(words[offset])} + for offset in sorted(words) if offset not in covered] + return { + "label": label, + "base": f"{chunk['base']:#x}", + "source": chunk.get("source"), + "input_format": chunk.get("format"), + "word_width": WORD_BYTES, + "word_count": len(words), + "words": {f"{offset:#x}": _hex(value) + for offset, value in sorted(words.items())}, + "fields": fields, + "unmapped": unmapped, + "unmapped_count": len(unmapped), + } + + +def decode_specs(specs: list[dict], entries: dict[tuple[str, int], dict], + map_source: str = "builtin") -> dict: + """Load and decode input specs into a ``greeneye-state-decode/1`` report.""" + sources = [] + chunks = [] + for spec in specs: + document = load_input(spec) + sources.append({ + "path": document["path"], + "format": document["format"], + "label": document["label"], + "base": f"{document['base']:#x}", + "chunk_labels": [chunk["label"] for chunk in document["chunks"]], + }) + for chunk in document["chunks"]: + chunks.append(decode_chunk(chunk, entries)) + chunks.sort(key=lambda chunk: (chunk["label"].lower(), chunk["label"], + int(chunk["base"], 16))) + return { + "format": DECODE_FORMAT, + "map_source": map_source, + "sources": sources, + "chunks": chunks, + "limits": [ + "Field names come from the bundled/extended map; an unmapped word " + "is silence, not absence.", + "Offsets are chunk-relative bytes; base is recorded provenance and " + "does not shift field lookup.", + "No timing (L2) or device-environment (L3) claim: this is a field " + "decoder over captured words.", + ], + } + + +# --------------------------------------------------------------------------- +# Diff +# --------------------------------------------------------------------------- + +def _validate_report(doc: Any, side: str) -> None: + if not isinstance(doc, dict) or doc.get("format") != DECODE_FORMAT: + raise StateDecodeError(f"{side} is not a {DECODE_FORMAT} report") + if not isinstance(doc.get("chunks"), list): + raise StateDecodeError(f"{side} report has no chunks list") + + +def _words_to_int(chunk: dict) -> dict[int, int]: + return {int(offset, 16): int(raw, 16) + for offset, raw in chunk["words"].items()} + + +def _field_index(chunk: dict | None) -> dict[tuple[int, str], dict]: + if chunk is None: + return {} + return {(int(field["offset"], 16), field["name"]): field + for field in chunk["fields"]} + + +def _field_summary(field: dict | None) -> str | None: + if field is None: + return None + if field.get("kind") == "bits": + parts = [f"{bit['name']}={bit['state']}" + for bit in field.get("bitfields", [])] + return ", ".join(parts) if parts else field["raw"] + return f"value={field['raw']}" + + +def _changed_fields(left: dict | None, right: dict | None) -> list[dict]: + left_fields = _field_index(left) + right_fields = _field_index(right) + rows = [] + for offset, name in sorted(set(left_fields) | set(right_fields)): + left_field = left_fields.get((offset, name)) + right_field = right_fields.get((offset, name)) + if (left_field is not None and right_field is not None + and left_field["raw"] == right_field["raw"]): + continue + row: dict[str, Any] = { + "offset": f"{offset:#x}", + "name": name, + "old": left_field["raw"] if left_field else None, + "new": right_field["raw"] if right_field else None, + "old_summary": _field_summary(left_field), + "new_summary": _field_summary(right_field), + } + if (left_field is not None and right_field is not None + and left_field.get("kind") == right_field.get("kind") == "bits"): + left_bits = {bit["name"]: bit for bit in left_field["bitfields"]} + right_bits = {bit["name"]: bit for bit in right_field["bitfields"]} + bits = [] + for bit_name in sorted(set(left_bits) | set(right_bits)): + bits.append({ + "name": bit_name, + "old": (left_bits[bit_name]["state"] + if bit_name in left_bits else None), + "new": (right_bits[bit_name]["state"] + if bit_name in right_bits else None), + }) + row["bits"] = bits + rows.append(row) + return rows + + +def _word_rows(words: dict[int, int]) -> list[dict]: + return [{"offset": f"{offset:#x}", "raw": _hex(raw)} + for offset, raw in sorted(words.items())] + + +def _diff_chunk(left: dict | None, right: dict | None) -> dict: + if left is None or right is None: + present = right if left is None else left + words = _words_to_int(present) + rows = _word_rows(words) + row = { + "label": present["label"], + "base": present["base"], + "left_base": left["base"] if left else None, + "right_base": right["base"] if right else None, + "status": "right_only" if left is None else "left_only", + "added": rows if left is None else [], + "removed": rows if right is None else [], + "changed": [], + "changed_fields": _changed_fields(left, right), + } + return row + left_words = _words_to_int(left) + right_words = _words_to_int(right) + added = _word_rows({offset: right_words[offset] + for offset in right_words if offset not in left_words}) + removed = _word_rows({offset: left_words[offset] + for offset in left_words if offset not in right_words}) + changed = [{"offset": f"{offset:#x}", "old": _hex(left_words[offset]), + "new": _hex(right_words[offset])} + for offset in sorted(set(left_words) & set(right_words)) + if left_words[offset] != right_words[offset]] + changed_fields = _changed_fields(left, right) + status = "changed" if (added or removed or changed or changed_fields) \ + else "identical" + return { + "label": left["label"], + "base": left["base"], + "left_base": left["base"], + "right_base": right["base"], + "status": status, + "added": added, + "removed": removed, + "changed": changed, + "changed_fields": changed_fields, + } + + +def diff_reports(left: dict, right: dict) -> dict: + """Deterministic word/field diff of two decoded reports.""" + _validate_report(left, "left") + _validate_report(right, "right") + left_groups: dict[str, list[dict]] = {} + right_groups: dict[str, list[dict]] = {} + for chunk in left["chunks"]: + left_groups.setdefault(chunk["label"].lower(), []).append(chunk) + for chunk in right["chunks"]: + right_groups.setdefault(chunk["label"].lower(), []).append(chunk) + labels = sorted(set(left_groups) | set(right_groups)) + chunks = [] + for label in labels: + left_list = left_groups.get(label, []) + right_list = right_groups.get(label, []) + for index in range(max(len(left_list), len(right_list))): + left_chunk = left_list[index] if index < len(left_list) else None + right_chunk = right_list[index] if index < len(right_list) else None + chunks.append(_diff_chunk(left_chunk, right_chunk)) + summary = { + "chunks": len(chunks), + "changed_chunks": sum(1 for chunk in chunks + if chunk["status"] != "identical"), + "added": sum(len(chunk["added"]) for chunk in chunks), + "removed": sum(len(chunk["removed"]) for chunk in chunks), + "changed_words": sum(len(chunk["changed"]) for chunk in chunks), + "changed_fields": sum(len(chunk["changed_fields"]) for chunk in chunks), + } + return { + "format": DIFF_FORMAT, + "left": {"format": left["format"], "map_source": left.get("map_source"), + "chunks": len(left["chunks"])}, + "right": {"format": right["format"], "map_source": right.get("map_source"), + "chunks": len(right["chunks"])}, + "chunks": chunks, + "summary": summary, + "identical": all(chunk["status"] == "identical" for chunk in chunks), + "limits": [ + "Word/field diff only; an unchanged word does not prove untested " + "behaviour.", + "Chunks pair by label and occurrence order; a differing base is " + "recorded but does not re-pair.", + ], + } + + +# --------------------------------------------------------------------------- +# Rendering +# --------------------------------------------------------------------------- + +def render_decode(report: dict) -> str: + """Readable field table for humans; the JSON report stays canonical.""" + lines = [] + for chunk in report["chunks"]: + lines.append( + f"chunk {chunk['label']} base={chunk['base']} " + f"words={chunk['word_count']} mapped={len(chunk['fields'])} " + f"unmapped={chunk['unmapped_count']}") + for field in chunk["fields"]: + if field["kind"] == "bits": + detail = " ".join(f"{bit['name']}={bit['state']}" + for bit in field["bitfields"]) + else: + detail = f"value={field['raw']}" + lines.append(f" {field['offset']:>6} {field['raw']:<12} " + f"{field['name']:<20} {detail}") + nonzero = [word for word in chunk["unmapped"] + if int(word["raw"], 16)] + if nonzero: + shown = " ".join(f"{word['offset']}={word['raw']}" + for word in nonzero[:12]) + more = "" if len(nonzero) <= 12 else f" (+{len(nonzero) - 12} more)" + lines.append(f" unmapped nonzero: {shown}{more}") + return "\n".join(lines) + + +def render_diff(report: dict) -> str: + """Readable diff for humans; the JSON report stays canonical.""" + lines = [] + for chunk in report["chunks"]: + lines.append(f"chunk {chunk['label']} base={chunk['base']}: " + f"{chunk['status']}") + for word in chunk["added"]: + lines.append(f" added {word['offset']} {word['raw']}") + for word in chunk["removed"]: + lines.append(f" removed {word['offset']} {word['raw']}") + for word in chunk["changed"]: + lines.append(f" changed {word['offset']} " + f"{word['old']} -> {word['new']}") + for field in chunk["changed_fields"]: + lines.append(f" field {field['name']} @ {field['offset']}: " + f"{field['old']} -> {field['new']} " + f"({field['old_summary']} -> {field['new_summary']})") + for bit in field.get("bits", []): + lines.append(f" bit {bit['name']}: " + f"{bit['old']} -> {bit['new']}") + lines.append(f"summary: {json.dumps(report['summary'], sort_keys=True)} " + f"identical={report['identical']}") + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Demo +# --------------------------------------------------------------------------- + +def _demo_words(word48: int) -> list[str]: + words = ["0x0"] * 20 # chunk words 0x00..0x4c + words[0x10 // WORD_BYTES] = "0xface" + words[0x48 // WORD_BYTES] = _hex(word48) + return words + + +def _demo_snapshot(name: str, *, gpfifo_va: int, ramfc_target: int) -> dict: + return { + "schema": SNAPSHOT_FORMAT, + "source": name, + "chunks": [ + {"label": "INSTANCE", "base": "0x0", + "words": _demo_words(gpfifo_va)}, + {"label": "RAMFC", "base": "0x0", + "words": _demo_words(ramfc_target)}, + ], + } + + +def run_demo(out: str | None = None) -> dict: + """Craft two snapshots in a tempdir, decode+diff them, and report. + + Expected: INSTANCE word 0x048 moves 0x100000 -> 0x200000 and RAMFC word + 0x048 moves 0x00030000 -> 0x00010000 (CE_CTX_VALID clears), i.e. two + changed words and two changed named fields. + """ + demo_dir = tempfile.mkdtemp(prefix="greeneye-state-decode-demo-") + try: + left_path = os.path.join(demo_dir, "left.json") + right_path = os.path.join(demo_dir, "right.json") + left_snapshot = _demo_snapshot("demo-left", gpfifo_va=0x100000, + ramfc_target=0x00030000) + right_snapshot = _demo_snapshot("demo-right", gpfifo_va=0x200000, + ramfc_target=0x00010000) + _write_json(left_path, left_snapshot) + _write_json(right_path, right_snapshot) + entries, map_source = load_map(None) + left = decode_specs( + [{"path": left_path, "label": None, "base": None, "binary": False}], + entries, map_source) + right = decode_specs( + [{"path": right_path, "label": None, "base": None, "binary": False}], + entries, map_source) + diff = diff_reports(left, right) + ok = (not diff["identical"] + and diff["summary"]["changed_words"] == 2 + and diff["summary"]["changed_fields"] == 2 + and diff["summary"]["added"] == 0 + and diff["summary"]["removed"] == 0) + demo = { + "format": DEMO_FORMAT, + "ok": ok, + "left": left, + "right": right, + "diff": diff, + } + if out: + os.makedirs(out, exist_ok=True) + artifacts = {} + for name, doc in (("state-decode-left.json", left), + ("state-decode-right.json", right), + ("state-diff.json", diff)): + artifacts[name] = _write_json(os.path.join(out, name), doc) + for name, path in (("state-snapshot-left.json", left_path), + ("state-snapshot-right.json", right_path)): + target = os.path.join(out, name) + shutil.copyfile(path, target) + artifacts[name] = os.path.abspath(target) + demo["artifacts"] = artifacts + return demo + finally: + shutil.rmtree(demo_dir, ignore_errors=True) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +class _InputAction(argparse.Action): + """Collect repeatable ``--input``/``--bin`` specs in command order.""" + + def __init__(self, option_strings, dest, binary=False, **kwargs): + self.binary = binary + super().__init__(option_strings, dest, **kwargs) + + def __call__(self, parser, namespace, values, option_string=None): + specs = getattr(namespace, self.dest, None) + if specs is None: + specs = [] + setattr(namespace, self.dest, specs) + spec = {"path": values, "label": None, "base": None, + "binary": self.binary} + specs.append(spec) + setattr(namespace, "_current_spec", spec) + + +class _InputMetaAction(argparse.Action): + """Attach ``--label``/``--base`` to the preceding input, or set a default.""" + + def __call__(self, parser, namespace, values, option_string=None): + current = getattr(namespace, "_current_spec", None) + if current is not None: + current[self.dest] = values + setattr(namespace, self.dest, values) + + +def _apply_defaults(specs: list[dict], label: str | None, + base: str | None) -> list[dict]: + resolved = [] + for spec in specs: + item = dict(spec) + if item.get("label") is None: + item["label"] = label + if item.get("base") is None: + item["base"] = base + resolved.append(item) + return resolved + + +def _peek_report(path: str) -> dict | None: + data = _read_bytes(path) + doc = _try_json(data) + if isinstance(doc, dict) and doc.get("format") == DECODE_FORMAT: + _validate_report(doc, path) + return doc + return None + + +def _load_side(spec: dict, entries: dict[tuple[str, int], dict], + map_source: str) -> dict: + report = _peek_report(spec["path"]) + if report is not None: + return report + return decode_specs([spec], entries, map_source) + + +def _write_json(path: str, doc: Any) -> str: + absolute = os.path.abspath(path) + parent = os.path.dirname(absolute) + if parent: + os.makedirs(parent, exist_ok=True) + with open(absolute, "w", encoding="utf-8") as handle: + json.dump(doc, handle, indent=2, sort_keys=True) + handle.write("\n") + return absolute + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="state_decode.py", + description="Decode engine-visible channel state words into named " + "fields, and diff two captures (offline).") + sub = parser.add_subparsers(dest="command") + + decode = sub.add_parser( + "decode", help="decode one or more raw inputs into a field table") + decode.add_argument("--input", action=_InputAction, dest="inputs", + default=None, metavar="FILE", + help="input file; repeatable, auto-detected format") + decode.add_argument("--bin", action=_InputAction, dest="inputs", + default=None, metavar="FILE", + help="raw little-endian u32 input; repeatable") + decode.add_argument("--label", action=_InputMetaAction, default=None, + help="chunk label for the preceding --input/--bin " + "(default for the rest)") + decode.add_argument("--base", action=_InputMetaAction, default="0x0", + help="engine address of chunk offset 0 for the " + "preceding input (default 0x0)") + decode.add_argument("--map", dest="map_path", default=None, + help=f"{MAP_FORMAT} override/extension file") + decode.add_argument("--json", dest="json_path", default=None, + help="write the decoded report JSON here") + + diff = sub.add_parser( + "diff", help="diff two decoded reports or two raw inputs") + diff.add_argument("left", help=f"{DECODE_FORMAT} report or raw input") + diff.add_argument("right", help=f"{DECODE_FORMAT} report or raw input") + diff.add_argument("--left-label", dest="left_label", default=None, + help="chunk label for a raw left input") + diff.add_argument("--left-base", dest="left_base", default="0x0", + help="base for a raw left input") + diff.add_argument("--left-bin", dest="left_bin", action="store_true", + help="force the left input to raw u32 words") + diff.add_argument("--right-label", dest="right_label", default=None, + help="chunk label for a raw right input") + diff.add_argument("--right-base", dest="right_base", default="0x0", + help="base for a raw right input") + diff.add_argument("--right-bin", dest="right_bin", action="store_true", + help="force the right input to raw u32 words") + diff.add_argument("--map", dest="map_path", default=None, + help=f"{MAP_FORMAT} override/extension file") + diff.add_argument("--json", dest="json_path", default=None, + help="write the diff report JSON here") + + demo = sub.add_parser( + "demo", help="craft two snapshots, decode+diff them, print the result") + demo.add_argument("--out", default=None, + help="write demo snapshots and reports to this directory") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "decode": + specs = _apply_defaults(list(args.inputs or []), args.label, + args.base) + if not specs: + raise StateDecodeError("decode needs at least one --input/--bin") + entries, map_source = load_map(args.map_path) + report = decode_specs(specs, entries, map_source) + print(render_decode(report)) + if args.json_path: + _write_json(args.json_path, report) + return 0 + if args.command == "diff": + entries, map_source = load_map(args.map_path) + left = _load_side({"path": args.left, "label": args.left_label, + "base": args.left_base, + "binary": args.left_bin}, entries, map_source) + right = _load_side({"path": args.right, "label": args.right_label, + "base": args.right_base, + "binary": args.right_bin}, entries, map_source) + report = diff_reports(left, right) + print(render_diff(report)) + if args.json_path: + _write_json(args.json_path, report) + return 0 if report["identical"] else 3 + if args.command == "demo": + demo = run_demo(args.out) + print(f"demo: {demo['format']} ok={demo['ok']}") + print(render_decode(demo["left"])) + print(render_diff(demo["diff"])) + print(json.dumps({"ok": demo["ok"], + "summary": demo["diff"]["summary"]}, + indent=2, sort_keys=True)) + return 0 if demo["ok"] else 3 + parser.print_help() + return 2 + except StateDecodeError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/state_seed.py b/debug/risc-v-gsp-rm-vm/harness/state_seed.py new file mode 100644 index 0000000..e4d9a42 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/state_seed.py @@ -0,0 +1,990 @@ +#!/usr/bin/env python3 +"""RM-object state seeding: declarative state for rehost/valprobe runs. + +Why this exists +--------------- +``harness/rehost.py`` runs one firmware function with caller-supplied bytes and +``harness/valprobe.py`` runs one validation branch with crafted registers and +memory. Both hit the same wall on RM handlers: the function early-outs (or +faults on a NULL callback) before the branch of interest because the RM object +state it walks -- clients, channels, graphics contexts, memdesc/context-buffer +objects -- does not exist. This harness adds the missing layer: a declarative +**state seed** (``greeneye-state-seed/1``) that places symbolic objects at +chosen addresses, fills their fields and pointers, sets registers/entry, and +runs the function under a bounded budget while recording which seeded state the +firmware actually reads and where it first stops. + +Every run reports the ordered first-touch state reads and a ``gate``: the +address and the state the returning/faulting path checked. A clean early-out +with the gate identified is the success criterion; a fault is the first missing +state observed, never a product defect. + +Honesty rules: + +* Not a boot and not a model: only the seeded state exists. The report lists + the layout, every field it wrote, and every field marked ``unknown``. +* The object field map below is partial and cites its evidence per field; a + field without evidence is marked ``unknown`` and must not be invented. +* No timing/cycle claim (L0/L1 fidelity, docs/06). + +Offline only: never touches a GPU, kernel module, mdev, guest or network. +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +from typing import Any + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from greeneye.elf import load_elf # noqa: E402 +from greeneye.errors import GreenEyeError # noqa: E402 +from greeneye.sim_backend import launch_native # noqa: E402 +from sim.devices import load_profile, registry_from_profile # noqa: E402 +from sim.gsp_boot import _apply_boot_shim # noqa: E402 +from sim.libos import LibosSyscalls # noqa: E402 +from sim.machine import Fault, R_BP, R_BUDGET, R_ECALL, R_FAULT, R_WATCH # noqa: E402 + +from harness.rehost import ( # noqa: E402 + DEFAULT_IMAGE, + DEFAULT_PROFILE, + EBREAK, + RUN_CHUNK, + _parse_u64, + _registers, + _sha256_file, +) +from harness.valprobe import ( # noqa: E402 + BUFFER_ID, + INSTANCE_GETTER, + OUTCOMES, + PROMOTE_VALIDATION, + decode_memory_spec, + encode_return_stub, + reg_index, +) + +#: Schema tag of a state seed. +SEED_FORMAT = "greeneye-state-seed/1" +#: Schema tag of a single seeded run. +REPORT_FORMAT = "greeneye-state-seed-report/1" +#: Schema tag of the bundled demo. +DEMO_FORMAT = "greeneye-state-seed-demo/1" + +DEFAULT_BUDGET = 50_000 +DEFAULT_TIMEOUT = 20.0 +DEFAULT_ALLOC_BASE = 0x10000000 +DEFAULT_ALLOC_ALIGN = 16 + +#: Object types that carry a documented field map (evidence per field below). +_CHANNEL_FIELDS = { + 0x0D28: {"name": "instance_index", "width": 4, "status": "known", + "role": "graphics-instance index (u32); selects client+0x37c+idx", + "evidence": "promote-validation predicate (docs/47)"}, + 0x0D01: {"name": "flag_d01", "width": 1, "status": "unknown", + "role": "u8 flag on the failed-lookup path (0 takes the panic path)"}, + 0x0CD1: {"name": "flag_cd1", "width": 1, "status": "unknown", + "role": "u8 flag before the promote calls"}, + 0x0648: {"name": "callback_648", "width": 8, "status": "unknown", + "role": "indirect callback on the invalid-state path"}, + 0x0898: {"name": "callback_898", "width": 8, "status": "unknown", + "role": "indirect callback first reached after the +0x600 gate"}, + 0x03B8: {"name": "marker_3b8", "width": 8, "status": "unknown", + "role": "identity marker compared across objects"}, + 0x00EE: {"name": "obj_ee", "width": 8, "status": "unknown", + "role": "sub-object; +0xe8 read as the gfid/index"}, +} + +_GRAPHICS_INSTANCE_FIELDS = { + 0x0190: {"name": "memdesc_table", "width": 8, "status": "known", + "role": "pointer to the per-buffer memDesc slot table", + "evidence": "promote-validation predicate (docs/47)"}, + 0x0161: {"name": "flag_161", "width": 1, "status": "unknown", + "role": "u8 flag gating the 0x529bd44 call"}, + 0x03B8: {"name": "marker_3b8", "width": 8, "status": "unknown", + "role": "identity marker compared across objects"}, +} + +FIELD_MAP: dict[str, dict[int, dict]] = { + "client": { + 0x1BE0: {"name": "instance_table", "width": 8, "status": "known", + "role": "graphics-instance pointer array base (index 0x37c + idx)", + "evidence": "promote-validation predicate (docs/47)"}, + 0x1AA8: {"name": "obj_1aa8", "width": 8, "status": "unknown", + "role": "sub-object consumed by the construct path (+0x24c callback)"}, + 0x1CC8: {"name": "obj_1cc8", "width": 8, "status": "unknown", + "role": "sub-object consumed by the construct path (+0x150, +0x1d0)"}, + 0x1DD8: {"name": "engine_ctx", "width": 8, "status": "known", + "role": "engine-context pointer; 0 selects the no-engine path", + "evidence": "construct-path disassembly (docs/52)"}, + 0x1DE8: {"name": "subdevice", "width": 8, "status": "known", + "role": "sub-device object whose +0x600 callback is called", + "evidence": "construct-path disassembly (docs/52)"}, + 0x1E48: {"name": "obj_1e48", "width": 8, "status": "unknown", + "role": "sub-object consumed on the no-engine path (+0x110 callback)"}, + }, + "channel": _CHANNEL_FIELDS, + "graphics_context": _CHANNEL_FIELDS, + "graphics_instance": _GRAPHICS_INSTANCE_FIELDS, + "memdesc": { + 0x0000: {"name": "unknown_0", "width": 8, "status": "unknown", + "role": "slot value semantics outside the promote predicate"}, + 0x0008: {"name": "unknown_8", "width": 8, "status": "unknown", + "role": "unknown"}, + }, + "ctx_buffer": { + 0x0000: {"name": "unknown_0", "width": 8, "status": "unknown", + "role": "context-buffer (memdesc) object; field semantics unknown"}, + }, + "rm_object": {}, +} + +#: Object types accepted in a seed. ``rm_object`` carries no field map. +KNOWN_TYPES = tuple(FIELD_MAP) +MISSING_MARKER = "unknown" + + +class StateSeedError(GreenEyeError): + """A state seed, image or scenario is invalid.""" + + +def _require_u64(value: Any, what: str = "value") -> int: + try: + return _parse_u64(value) + except GreenEyeError as exc: + raise StateSeedError(f"{what}: {exc}") from exc + + +def _field_width(obj_type: str, offset: int) -> int: + spec = FIELD_MAP.get(obj_type, {}).get(offset) + return int(spec["width"]) if spec else 8 + + +def _field_spec(obj_type: str, offset: int) -> dict | None: + return FIELD_MAP.get(obj_type, {}).get(offset) + + +def _parse_offset(value: Any) -> int: + text = str(value).strip() + if text.startswith("+"): + text = text[1:] + try: + return int(text, 0) + except ValueError as exc: + raise StateSeedError(f"bad field offset {value!r}") from exc + + +def _encode_int(value: int, width: int) -> bytes: + if width not in (1, 2, 4, 8): + raise StateSeedError(f"unsupported field width {width}") + return (value & ((1 << (8 * width)) - 1)).to_bytes(width, "little") + + +def _is_symbol(value: Any) -> bool: + return isinstance(value, str) and value.strip().startswith("@") + + +def _symbol_expr(value: str) -> tuple[str, int]: + text = value.strip()[1:] + name, sep, offset = text.partition("+") + if not name: + raise StateSeedError(f"bad pointer expression {value!r}") + if sep: + try: + delta = int(offset, 0) + except ValueError as exc: + raise StateSeedError(f"bad pointer expression {value!r}") from exc + else: + delta = 0 + return name, delta + + +class _Symbols: + """Name -> address table for symbolic object/stub pointers.""" + + def __init__(self) -> None: + self._by_name: dict[str, int] = {} + + def add(self, name: str, addr: int, owner: str) -> None: + if name in self._by_name: + raise StateSeedError(f"duplicate symbol {name!r} ({owner})") + self._by_name[name] = addr + + def resolve(self, value: str, *, owner: str) -> int: + name, delta = _symbol_expr(value) + if name not in self._by_name: + raise StateSeedError(f"{owner}: unknown symbol {name!r}") + return (self._by_name[name] + delta) & 0xFFFF_FFFF_FFFF_FFFF + + +def load_seed(seed: dict) -> dict: + """Validate a ``greeneye-state-seed/1`` seed and compile an application plan.""" + if not isinstance(seed, dict): + raise StateSeedError(f"seed must be an object, got {type(seed).__name__}") + fmt = seed.get("format", SEED_FORMAT) + if fmt != SEED_FORMAT: + raise StateSeedError(f"unsupported seed format {fmt!r} (want {SEED_FORMAT})") + name = str(seed.get("name") or "state-seed") + seeded_registers = seed.get("registers") or {} + if not isinstance(seeded_registers, dict): + raise StateSeedError("registers must be an object") + entry_value = seed.get("entry", seeded_registers.get("pc")) + if entry_value is None: + raise StateSeedError("seed needs an entry address (or registers.pc)") + entry = _require_u64(entry_value) + + allocator = seed.get("allocator") or {} + cursor = _require_u64(allocator.get("base", DEFAULT_ALLOC_BASE)) + default_align = _require_u64(allocator.get("align", DEFAULT_ALLOC_ALIGN)) + if default_align <= 0 or default_align & (default_align - 1): + raise StateSeedError("allocator align must be a positive power of two") + + objects = seed.get("objects") or [] + if not isinstance(objects, list): + raise StateSeedError("objects must be a list") + layout: list[dict] = [] + symbols = _Symbols() + for obj in objects: + if not isinstance(obj, dict): + raise StateSeedError(f"object must be an object, got {obj!r}") + obj_type = str(obj.get("type") or "") + if obj_type not in KNOWN_TYPES: + raise StateSeedError(f"unknown object type {obj_type!r} (known: {', '.join(KNOWN_TYPES)})") + obj_name = str(obj.get("name") or "") + if not obj_name: + raise StateSeedError(f"{obj_type} object needs a name") + align = _require_u64(obj.get("align", default_align)) + if align <= 0 or align & (align - 1): + raise StateSeedError(f"{obj_name}: align must be a positive power of two") + fields = obj.get("fields") or {} + if not isinstance(fields, dict): + raise StateSeedError(f"{obj_name}: fields must be an object") + extent = 0 + for key in fields: + offset = _parse_offset(key) + extent = max(extent, offset + _field_width(obj_type, offset)) + size = _require_u64(obj.get("size", max(extent, default_align))) + if size <= 0: + raise StateSeedError(f"{obj_name}: size must be positive") + if "pa" in obj and "va" in obj: + raise StateSeedError(f"{obj_name}: use pa or va, not both") + explicit = obj.get("pa", obj.get("va")) + if explicit is not None: + pa = _require_u64(explicit) + else: + cursor = (cursor + align - 1) & ~(align - 1) + pa = cursor + cursor += (size + align - 1) & ~(align - 1) + record = {"name": obj_name, "type": obj_type, "pa": pa, "size": size, + "fields": {}, "unknown": []} + layout.append(record) + symbols.add(obj_name, pa, obj_name) + for key, value in fields.items(): + record["fields"][_parse_offset(key)] = value + + for i, first in enumerate(layout): + for second in layout[i + 1:]: + if first["pa"] < second["pa"] + second["size"] and second["pa"] < first["pa"] + first["size"]: + raise StateSeedError( + f"objects {first['name']!r} and {second['name']!r} overlap" + ) + + writes: list[dict] = [] + unknown_fields: list[dict] = [] + + stubs = seed.get("stubs") or [] + if not isinstance(stubs, list): + raise StateSeedError("stubs must be a list") + stub_records: list[dict] = [] + for stub in stubs: + if not isinstance(stub, dict) or "return" not in stub: + raise StateSeedError(f"stub needs 'return': {stub!r}") + ret = _require_u64(stub["return"]) + if "addr" in stub: + addr = _require_u64(stub["addr"]) + else: + cursor = (cursor + default_align - 1) & ~(default_align - 1) + addr = cursor + cursor += default_align + stub_name = stub.get("name") + if stub_name: + symbols.add(str(stub_name), addr, f"stub {stub_name}") + code = encode_return_stub(ret) + stub_records.append({"name": stub_name, "addr": addr, "return": ret, + "code_hex": code.hex()}) + writes.append({"pa": addr, "hex": code.hex(), "label": f"stub:{stub_name or hex(addr)}", + "kind": "stub"}) + + def resolve_field(owner: str, obj_type: str, offset: int, value: Any) -> tuple[int | None, int, str]: + width = _field_width(obj_type, offset) + if value == MISSING_MARKER or (isinstance(value, dict) and value.get("unknown")): + return None, width, MISSING_MARKER + if isinstance(value, dict): + if "unknown" in value and value["unknown"]: + return None, width, MISSING_MARKER + if "ptr" in value: + value = value["ptr"] + elif "value" in value: + raw = value["value"] + if raw == MISSING_MARKER: + return None, _require_u64(value.get("size", width)), MISSING_MARKER + width = _require_u64(value.get("size", width)) + return _require_u64(raw), width, "value" + else: + raise StateSeedError(f"{owner}+{offset:#x}: field dict needs value/ptr/unknown") + if _is_symbol(value): + return symbols.resolve(str(value), owner=f"{owner}+{offset:#x}"), width, "ptr" + return _parse_u64(value), width, "value" + + for record in layout: + pa = record["pa"] + for offset in sorted(record["fields"]): + value = record["fields"][offset] + resolved, width, kind = resolve_field(record["name"], record["type"], offset, value) + spec = _field_spec(record["type"], offset) + label = f"{record['name']}+{offset:#x}" + if resolved is None: + unknown_fields.append({ + "object": record["name"], "type": record["type"], "off": f"{offset:#x}", + "name": spec["name"] if spec else None, + "role": spec["role"] if spec else "not documented", + "reason": "seed marks this field unknown; left at zero", + }) + continue + writes.append({"pa": pa + offset, "hex": _encode_int(resolved, width).hex(), + "label": label, "kind": kind}) + + memory = seed.get("memory") or [] + if not isinstance(memory, list): + raise StateSeedError("memory must be a list") + for item in memory: + if not isinstance(item, dict): + raise StateSeedError(f"memory item must be an object: {item!r}") + normalized = dict(item) + if "pa" in normalized and "addr" in normalized: + raise StateSeedError(f"memory item: use pa or addr, not both: {item!r}") + if "pa" in normalized: + normalized["addr"] = normalized.pop("pa") + try: + addr, data = decode_memory_spec(normalized) + except GreenEyeError as exc: + raise StateSeedError(str(exc)) from exc + writes.append({"pa": addr, "hex": data.hex(), + "label": item.get("label") or f"mem+{addr:#x}", "kind": "memory"}) + + registers: dict[str, int] = {} + for reg_name, value in seeded_registers.items(): + if str(reg_name).strip().lower() == "pc": + continue + try: + reg_index(str(reg_name)) + except GreenEyeError as exc: + raise StateSeedError(str(exc)) from exc + if _is_symbol(value): + registers[str(reg_name)] = symbols.resolve(str(value), owner=f"register {reg_name}") + else: + registers[str(reg_name)] = _require_u64(value) + + stack = seed.get("stack") + outcomes = {_require_u64(a): str(label) + for a, label in (seed.get("outcomes") or {}).items()} + reads = [] + for item in seed.get("reads") or (): + if "pa" in item and "addr" not in item: + item = dict(item) + item["addr"] = item.pop("pa") + if "addr" not in item: + raise StateSeedError(f"read item needs addr/pa: {item!r}") + reads.append(item) + watch: list[dict] = [] + seen_labels: dict[str, int] = {} + + def _add_watch(addr: int, size: int, label: str) -> None: + count = seen_labels.get(label, 0) + 1 + seen_labels[label] = count + if count > 1: + label = f"{label}~{count}" + watch.append({"addr": addr, "size": size, "label": label}) + + for item in seed.get("watch") or (): + if "pa" in item and "addr" not in item: + item = dict(item) + item["addr"] = item.pop("pa") + if "addr" not in item: + raise StateSeedError(f"watch item needs addr/pa: {item!r}") + _add_watch(_require_u64(item["addr"]), _require_u64(item.get("size", 8)), + str(item.get("label") or f"{_require_u64(item['addr']):#x}")) + for record in layout: + for offset in record["fields"]: + _add_watch(record["pa"] + offset, _field_width(record["type"], offset), + f"{record['name']}+{offset:#x}") + + return { + "format": SEED_FORMAT, + "name": name, + "entry": entry, + "stack": _require_u64(stack) if stack is not None else None, + "registers": registers, + "writes": writes, + "layout": layout, + "unknown_fields": unknown_fields, + "stubs": stub_records, + "outcomes": outcomes, + "reads": reads, + "watch": watch, + "provenance": seed.get("provenance") or {}, + "notes": seed.get("notes") or [], + } + + +def apply_seed(machine, plan: dict) -> dict: + """Apply a compiled plan to a machine; return the same plan (chainable).""" + for item in plan["writes"]: + machine.write_mem(item["pa"], bytes.fromhex(item["hex"])) + for reg_name, value in plan["registers"].items(): + machine.x[reg_index(reg_name)] = value + return plan + + +def _read_after(machine, reads) -> list[dict]: + out = [] + for item in reads or (): + addr = _require_u64(item["addr"]) + size = _require_u64(item.get("size", 8)) + data = bytes(machine.read_mem(addr, size)) + record = {"addr": f"{addr:#x}", "size": size, "hex": data.hex()} + if item.get("label"): + record["label"] = item["label"] + record["u64"] = int.from_bytes(data.ljust(8, b"\x00"), "little") + out.append(record) + return out + + +def _gate_from_run(status: str, stop_pc: int, fault: dict | None, + state_reads: list[dict]) -> dict | None: + if not state_reads: + return None + last = state_reads[-1] + if status == "fault" and stop_pc == 0 and last.get("value") == 0: + return { + "kind": "null_indirect_call", + "pc": last["pc"], + "addr": last["addr"], + "label": last["label"], + "value": 0, + "detail": ( + f"indirect call target {last['label']} is NULL; execution faulted at " + f"pc 0x0 (fault={fault})" + ), + } + if status == "fault": + return { + "kind": "fault", + "pc": f"{stop_pc:#x}", + "addr": last["addr"], + "label": last["label"], + "value": last.get("value"), + "detail": f"fault before the next seeded-state read (fault={fault})", + } + if status == "outcome": + return None + return { + "kind": f"{status}_before_gate", + "pc": f"{stop_pc:#x}", + "addr": last["addr"], + "label": last["label"], + "value": last.get("value"), + "detail": f"run stopped as {status}; last seeded state read shown", + } + + +def run_seed( + seed: dict, + *, + image: str = DEFAULT_IMAGE, + profile_path: str | None = DEFAULT_PROFILE, + entry: int | str | None = None, + out: str | None = None, + record: bool | None = None, + budget: int = DEFAULT_BUDGET, + timeout: float = DEFAULT_TIMEOUT, + name: str = "state-seed", +) -> dict: + """Load a seed into the native machine, run it bounded, return a report.""" + if not os.path.isfile(image): + raise StateSeedError(f"image not found: {image}") + if budget <= 0: + raise StateSeedError(f"budget must be positive, got {budget}") + image_obj = load_elf(image) + if not image_obj.is_riscv: + raise StateSeedError(f"{image}: not a RISC-V ELF image") + plan = load_seed(seed) + entry_addr = _require_u64(entry) if entry is not None else plan["entry"] + + profile = load_profile(profile_path) if profile_path else None + load = (profile or {}).get("load", {}) + ram_base = _require_u64(load.get("ram_base", 0)) + memory_mb = int(load.get("memory_mb", 256)) + ram_top = ram_base + memory_mb * 1024 * 1024 + + if record is None: + record = bool(out) + session = launch_native( + kernel=image, + memory_mb=memory_mb, + ram_base=ram_base, + record=record, + name=f"{name}-{plan['name']}"[:80], + device_map=None, + unknown_mmio="lenient", + bias=0, + ) + machine = session.target.machine + try: + if profile is not None: + machine.devices = registry_from_profile(profile, machine=machine) + machine.translator = getattr(machine.devices, "dma", None) + machine.syscall_handler = LibosSyscalls().handle + shim = dict(profile.get("boot_shim", {})) + shim["_image_path"] = image + _apply_boot_shim(machine, shim, image_obj, {"enabled": True, "assumptions": []}) + + sentinel = (ram_top - 0x100) & ~0xF + default_sp = (ram_top - 0x8000) & ~0xF + machine.write_mem(sentinel, EBREAK.to_bytes(4, "little")) + + apply_seed(machine, plan) + machine.x[1] = sentinel + if machine.x[2] == 0: + machine.x[2] = plan["stack"] if plan["stack"] is not None else default_sp + machine.pc = entry_addr + + breakpoints = set(plan["outcomes"]) | {sentinel} + watchers = [] + for item in plan["watch"]: + addr = _require_u64(item["addr"]) + size = _require_u64(item.get("size", 8)) + label = str(item.get("label") or f"{addr:#x}") + watchers.append((addr, addr + size, label, 3)) + + entered = {f"x{i}": machine.x[i] for i in range(32) if i and machine.x[i]} + entered["pc"] = entry_addr + state_reads: list[dict] = [] + started = time.monotonic() + status = "budget" + stop_pc = entry_addr + fault: dict | None = None + outcome: str | None = None + remaining = budget + while remaining > 0: + if time.monotonic() - started > timeout: + status = "timeout" + break + chunk = min(RUN_CHUNK, remaining) + reason, stop_pc, a, b = machine.run( + chunk, breakpoints=breakpoints, watchers=tuple(watchers) + ) + if reason == R_WATCH: + width = 8 + for watcher in watchers: + if watcher[2] == b: + width = watcher[1] - watcher[0] + break + try: + value = int.from_bytes(bytes(machine.read_mem(a, width)), "little") + except (GreenEyeError, Fault): + value = None + state_reads.append({"pc": f"{stop_pc:#x}", "addr": f"{a:#x}", + "label": b, "value": value, "width": width}) + watchers = [watcher for watcher in watchers if watcher[2] != b] + continue + remaining -= chunk + if reason == R_BP: + if stop_pc == sentinel: + status = "return" + break + outcome = plan["outcomes"].get(stop_pc, "unlabelled") + status = "outcome" + break + if reason == R_FAULT: + status = "fault" + fault = {"cause": a, "tval": f"{b:#x}"} + break + if reason == R_ECALL: + handled, result = (machine.syscall_handler(machine, stop_pc) + if machine.syscall_handler else (False, None)) + if not handled: + status = "ecall" + break + if result is not None: + machine.x[10] = result & 0xFFFF_FFFF_FFFF_FFFF + continue + if reason == R_BUDGET: + status = "budget" + + gate = _gate_from_run(status, stop_pc, fault, state_reads) + report: dict[str, Any] = { + "format": REPORT_FORMAT, + "seed": plan["name"], + "seed_format": plan["format"], + "seed_provenance": plan["provenance"], + "image": os.path.abspath(image), + "image_sha256": _sha256_file(image), + "profile": os.path.basename(profile_path) if profile_path else None, + "entry": f"{entry_addr:#x}", + "status": status, + "outcome": outcome, + "stop_pc": f"{stop_pc:#x}", + "fault": fault, + "retval": machine.x[10], + "instructions": machine.insns, + "wall_seconds": round(time.monotonic() - started, 4), + "objects": [ + {"name": item["name"], "type": item["type"], "pa": f"{item['pa']:#x}", + "size": item["size"]} + for item in plan["layout"] + ], + "unknown_fields": plan["unknown_fields"], + "stubs": plan["stubs"], + "registers_entered": {k: f"{v:#x}" if isinstance(v, int) else v + for k, v in entered.items()}, + "registers_exited": {k: f"{v:#x}" if isinstance(v, int) else v + for k, v in _registers(machine).items()}, + "state_reads": state_reads, + "gate": gate, + "reads": _read_after(machine, plan["reads"]), + "limits": [ + "Seeded run, not a boot: only the supplied objects/memory exist.", + "A fault is the first missing state observed, not a product defect.", + "No timing/cycle claim (L0/L1 fidelity, docs/06).", + "A field marked unknown was left at zero and reported.", + ], + } + if getattr(session.recorder, "dir", None) is not None: + session.note("state-seed", seed=plan["name"], status=status, + outcome=outcome, stop_pc=f"{stop_pc:#x}", + gate=(gate or {}).get("kind")) + report["session"] = session.recorder.dir + if out: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, "state-seed-report.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, sort_keys=True) + fh.write("\n") + report["report_json"] = os.path.abspath(path) + return report + finally: + session.close() + + +# --------------------------------------------------------------------------- +# Bundled seeds. +# +# 1. fecs_seed: the promote-validation predicate seeded through this loader +# (the mechanism proof -- it works without RM objects). It reproduces the +# valprobe variant memory/registers/stub exactly, so a slot non-NULL seed +# must refuse 0x1f and a slot NULL seed must take the commit path. +# 2. target_seed: the construct target 0x52a70ec with a minimal crafted +# client/channel/graphics-context/memdesc set. The first gate is reported +# (see docs/52); with the +0x600 callback stubbed it advances to the next. +# --------------------------------------------------------------------------- + + +def _fecs_seed(*, slot: int, label: str) -> dict: + scratch = 0x10000000 + layout = { + "scratch": scratch, + "sp": scratch + 0x4000, + "s0": scratch + 0x5000, + "pgpu": scratch + 0x1000, + "getter_out": scratch + 0x1800, + "instance": scratch + 0x3800, + "entry": scratch + 0x6000, + } + slot_addr = layout["instance"] + 0x300 + 6 * 8 + memory = [ + {"pa": f"{layout['entry'] + 0x30:#x}", "u64": 0x1F4150000}, + {"pa": f"{layout['entry'] + 0x38:#x}", "u64": 0x204608000}, + {"pa": f"{layout['entry'] + 0x40:#x}", "u64": 0x10000}, + {"pa": f"{layout['entry'] + 0x48:#x}", "u32": 4}, + {"pa": f"{layout['entry'] + 0x4c:#x}", "u16": BUFFER_ID["FECS_EVENT"]}, + {"pa": f"{layout['entry'] + 0x4e:#x}", "u8": 1}, + {"pa": f"{layout['entry'] + 0x4f:#x}", "u8": 0}, + {"pa": f"{layout['pgpu'] + 0x1be0:#x}", "u64": layout["instance"]}, + {"pa": f"{layout['instance'] + 0x190:#x}", "u64": layout["instance"] + 0x300}, + {"pa": f"{slot_addr:#x}", "u64": slot, "label": "fecs-slot"}, + ] + return { + "format": SEED_FORMAT, + "name": label, + "entry": f"{PROMOTE_VALIDATION:#x}", + "outcomes": dict(OUTCOMES), + "registers": { + "s0": f"{layout['s0']:#x}", + "s2": f"{layout['pgpu']:#x}", + "s5": f"{layout['entry']:#x}", + "sp": f"{layout['sp']:#x}", + "a3": BUFFER_ID["FECS_EVENT"], + "t1": 0xC, + }, + "stubs": [{"addr": f"{INSTANCE_GETTER:#x}", "return": f"{layout['getter_out']:#x}", + "name": "instance_getter"}], + "memory": memory, + "watch": [{"pa": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot"}], + "reads": [{"pa": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot-value"}], + "provenance": { + "source": "promote-validation predicate state (docs/47), rebuilt as a seed", + "slot": slot, + }, + "notes": [ + "Mechanism proof: a path that needs no RM objects, seeded end to end.", + ], + } + + +def fecs_seed(*, slot: int = 1) -> dict: + """Build the FECS promote-validation seed (slot 1 refuses, slot 0 accepts).""" + label = "fecs-promote-slot-nonnull" if slot else "fecs-promote-slot-null" + return _fecs_seed(slot=slot, label=label) + + +def target_seed(*, advance: bool = False) -> dict: + """Build the minimal client/channel/context seed for the 0x52a70ec target. + + ``advance=False`` leaves the +0x600 callback unknown (the first gate). + ``advance=True`` points it at a return-0 stub so the next gate is reached. + """ + callback = "@engine_stub" if advance else MISSING_MARKER + name = "construct-gate-advanced" if advance else "construct-gate-minimal" + return { + "format": SEED_FORMAT, + "name": name, + "entry": "0x52a70ec", + "allocator": {"base": "0x10010000", "align": "0x1000"}, + "stubs": [{"name": "engine_stub", "return": 0}] if advance else [], + "registers": { + "a0": "@client", + "a1": "@channel", + "a2": "0x1003f000", + }, + "stack": "0x1003e000", + "objects": [ + {"name": "client", "type": "client", "size": "0x2000", "fields": { + "0x1be0": "@graphics_instance", + "0x1aa8": "@obj_1aa8", + "0x1cc8": "@obj_1cc8", + "0x1dd8": {"value": 0, "size": 8}, + "0x1de8": "@engine_node", + "0x1e48": "@obj_1e48", + }}, + {"name": "channel", "type": "channel", "size": "0x1000", "fields": { + "0xd28": {"value": 0, "size": 4}, + "0xd01": {"value": 1, "size": 1}, + "0xcd1": {"value": 0, "size": 1}, + "0x898": MISSING_MARKER, + }}, + {"name": "graphics_instance", "type": "graphics_instance", "size": "0x1000", + "fields": { + "0x190": "@memdesc_table", + "0x161": {"value": 0, "size": 1}, + }}, + {"name": "engine_node", "type": "rm_object", "size": "0x1000", "fields": { + "0x600": callback, + }}, + {"name": "obj_1cc8", "type": "rm_object", "size": "0x400", "fields": { + "0x150": MISSING_MARKER, + "0x1d0": {"value": 0x20, "size": 4}, + }}, + {"name": "obj_1aa8", "type": "rm_object", "size": "0x400", "fields": { + "0x24c": MISSING_MARKER, + }}, + {"name": "obj_1e48", "type": "rm_object", "size": "0x400", "fields": { + "0x110": MISSING_MARKER, + }}, + {"name": "memdesc_table", "type": "rm_object", "size": "0x100", "fields": {}}, + {"name": "fecs_memdesc", "type": "memdesc", "size": "0x100", "fields": { + "0x0": MISSING_MARKER, + "0x8": MISSING_MARKER, + }}, + ], + "reads": [{"pa": "0x1003f000", "size": 8, "label": "a2-out-scratch"}], + "provenance": { + "source": "construct-path disassembly (docs/52, this wave)", + "callback": callback, + }, + "notes": [ + "Only the fields the known firmware path reads are seeded; a field marked", + "unknown is left zero and reported so the gate is not mistaken for a seed bug.", + ], + } + + +def run_demo(out: str | None = None) -> dict: + """Prove the seeder on the FECS predicate, then attempt the construct target.""" + if not os.path.isfile(DEFAULT_IMAGE): + raise StateSeedError(f"firmware image absent: {DEFAULT_IMAGE}") + profile = DEFAULT_PROFILE + + fecs_nonnull = run_seed(fecs_seed(slot=1), profile_path=profile, record=False, + name="state-seed-fecs") + fecs_nonnull_again = run_seed(fecs_seed(slot=1), profile_path=profile, record=False, + name="state-seed-fecs") + fecs_null = run_seed(fecs_seed(slot=0), profile_path=profile, record=False, + name="state-seed-fecs") + target = run_seed(target_seed(advance=False), profile_path=profile, record=False, + name="state-seed-target") + advanced = run_seed(target_seed(advance=True), profile_path=profile, record=False, + name="state-seed-target") + + verdicts = [ + {"label": "FECS init slot non-NULL", "outcome": fecs_nonnull["outcome"], + "expect": "refuse-fecs-slot-exists-0x1f", + "pass": fecs_nonnull["outcome"] == "refuse-fecs-slot-exists-0x1f"}, + {"label": "FECS init slot NULL", "outcome": fecs_null["outcome"], + "expect": "accept-commit-path", + "pass": fecs_null["outcome"] == "accept-commit-path"}, + ] + deterministic = ( + fecs_nonnull["outcome"] == fecs_nonnull_again["outcome"] + and fecs_nonnull["gate"] == fecs_nonnull_again["gate"] + ) + result = { + "format": DEMO_FORMAT, + "ok": all(row["pass"] for row in verdicts) and deterministic, + "deterministic": deterministic, + "mechanism": { + "seed": "fecs-promote", + "verdicts": verdicts, + "wall_seconds": round(fecs_nonnull["wall_seconds"] + fecs_null["wall_seconds"], 4), + }, + "target": { + "entry": "0x52a70ec", + "status": target["status"], + "stop_pc": target["stop_pc"], + "gate": target["gate"], + "state_reads": target["state_reads"], + "unknown_fields": target["unknown_fields"], + }, + "target_advanced": { + "entry": "0x52a70ec", + "status": advanced["status"], + "stop_pc": advanced["stop_pc"], + "gate": advanced["gate"], + "state_reads": advanced["state_reads"], + }, + } + if out: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, "state-seed-demo.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(result, fh, indent=2, sort_keys=True) + fh.write("\n") + result["artifacts"] = {"demo_json": os.path.abspath(path)} + return result + + +def _load_seed_file(path: str) -> dict: + if not os.path.isfile(path): + raise StateSeedError(f"seed not found: {path}") + with open(path, "r", encoding="utf-8") as fh: + return json.load(fh) + + +def _emit_seed(kind: str) -> dict: + if kind == "fecs": + return fecs_seed() + if kind == "target-minimal": + return target_seed(advance=False) + if kind == "target-advanced": + return target_seed(advance=True) + raise StateSeedError(f"unknown seed kind {kind!r}") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="state_seed.py", + description=( + "Seed declarative RM object state into a bounded firmware run " + "(offline; never touches hardware)." + ), + ) + sub = parser.add_subparsers(dest="command") + run_cmd = sub.add_parser("run", help="run one seed file") + run_cmd.add_argument("--seed", required=True, help="greeneye-state-seed/1 JSON") + run_cmd.add_argument("--entry", default=None, help="override the seed entry (hex)") + run_cmd.add_argument("--image", default=DEFAULT_IMAGE, help="firmware ELF") + run_cmd.add_argument("--profile", default=DEFAULT_PROFILE, help="device profile JSON") + run_cmd.add_argument("--budget", type=int, default=DEFAULT_BUDGET) + run_cmd.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT) + run_cmd.add_argument("--out", help="output directory (session + report JSON)") + run_cmd.add_argument("--no-record", action="store_true", help="skip session recording") + validate = sub.add_parser("validate", help="validate a seed and print its layout") + validate.add_argument("--seed", required=True) + emit = sub.add_parser("emit", help="write a bundled seed") + emit.add_argument("--kind", required=True, + choices=["fecs", "target-minimal", "target-advanced"]) + emit.add_argument("--out", required=True) + demo = sub.add_parser("demo", help="FECS mechanism proof + construct-gate attempt") + demo.add_argument("--out", help="output directory") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "run": + seed = _load_seed_file(args.seed) + report = run_seed(seed, image=args.image, profile_path=args.profile or None, + entry=args.entry, budget=args.budget, timeout=args.timeout, + out=args.out, record=False if args.no_record else None) + summary = {k: report[k] for k in + ("format", "seed", "entry", "status", "outcome", "stop_pc", + "fault", "gate", "instructions", "wall_seconds", + "report_json", "session") if k in report} + print(json.dumps(summary, indent=2, sort_keys=True)) + return 0 + if args.command == "validate": + seed = _load_seed_file(args.seed) + plan = load_seed(seed) + print(json.dumps({ + "format": plan["format"], + "name": plan["name"], + "entry": f"{plan['entry']:#x}", + "objects": [{"name": item["name"], "type": item["type"], + "pa": f"{item['pa']:#x}", "size": item["size"]} + for item in plan["layout"]], + "unknown_fields": plan["unknown_fields"], + "writes": len(plan["writes"]), + "stubs": plan["stubs"], + }, indent=2, sort_keys=True)) + return 0 + if args.command == "emit": + seed = _emit_seed(args.kind) + os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True) + with open(args.out, "w", encoding="utf-8") as fh: + json.dump(seed, fh, indent=2, sort_keys=True) + fh.write("\n") + print(os.path.abspath(args.out)) + return 0 + if args.command == "demo": + result = run_demo(args.out) + print(json.dumps({k: result[k] for k in + ("format", "ok", "deterministic", "mechanism", + "target", "target_advanced", "artifacts") + if k in result}, indent=2, sort_keys=True)) + return 0 if result["ok"] else 3 + parser.print_help() + return 2 + except StateSeedError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/harness/valprobe.py b/debug/risc-v-gsp-rm-vm/harness/valprobe.py new file mode 100644 index 0000000..55952b9 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/valprobe.py @@ -0,0 +1,651 @@ +#!/usr/bin/env python3 +"""Validation-predicate probe: run a firmware predicate branch with crafted state. + +Why this exists +--------------- +The live GR campaign keeps burning supervised hardware cycles to ask the same +question class: *which control/entry shapes does the 570.144 GSP-RM firmware +accept or refuse, and why?* The answer lives in firmware branches that are +buried behind host traffic (B4/B5) and RM object state, so a plain rehost of +the full handler early-outs before the predicate. This harness instead runs +one **predicate branch** of the real image with the minimum crafted state it +needs: registers, memory, and (where a branch calls out to a lock/table walk +that is not the predicate) one recorded inline stub. Each run stops on a +labelled outcome address, so the verdict is the branch the firmware took, not +an inference. + +It is deliberately not a boot and not a model: the report carries the exact +entry, seeded state, stubs and outcome map, and every limitation in `limits`. +Fidelity is L0/L1 (docs/06); no timing claim. + +Bundled target: ``sweep --target fecs-promote`` reproduces the live V6 FECS +matrix (`PM/PATCH/PRIV/UNR accepted, FECS init refused 0x1f, FECS VA-only +accepted`) and evaluates the untested shapes offline, including the +sysmem-aperture FECS-init variant the handoff proposed paying a hardware cycle +for. + +Offline only: never touches a GPU, kernel module, mdev, guest or network. +""" +from __future__ import annotations + +import argparse +import json +import os +import struct +import sys +import time +from typing import Any + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from greeneye.elf import load_elf # noqa: E402 +from greeneye.errors import GreenEyeError # noqa: E402 +from greeneye.sim_backend import launch_native # noqa: E402 +from sim.devices import load_profile, registry_from_profile # noqa: E402 +from sim.gsp_boot import _apply_boot_shim, _run_with_milestones, _rung_from_run # noqa: E402 +from sim.libos import LibosSyscalls # noqa: E402 +from sim.machine import R_BP, R_BUDGET, R_ECALL, R_FAULT # noqa: E402 + +from harness.rehost import ( # noqa: E402 + DEFAULT_IMAGE, + DEFAULT_PROFILE, + RUN_CHUNK, + _parse_u64, + _sha256_file, + parse_hex_bytes, +) + +#: Schema tag of a single probe run. +PROBE_FORMAT = "greeneye-valprobe/1" +#: Schema tag of a probe spec. +SPEC_FORMAT = "greeneye-valprobe-spec/1" +#: Schema tag of a matrix run (a spec with variants). +MATRIX_FORMAT = "greeneye-valprobe-matrix/1" +#: Schema tag of the bundled demo. +DEMO_FORMAT = "greeneye-valprobe-demo/1" + +DEFAULT_BUDGET = 50_000 +DEFAULT_TIMEOUT = 20.0 +BOOT_INSTRUCTIONS = 2_000_000 +EBREAK = 0x00100073 + +REGISTERS = { + "zero": 0, "ra": 1, "sp": 2, "gp": 3, "tp": 4, + "t0": 5, "t1": 6, "t2": 7, "s0": 8, "fp": 8, "s1": 9, + "a0": 10, "a1": 11, "a2": 12, "a3": 13, "a4": 14, "a5": 15, + "a6": 16, "a7": 17, "s2": 18, "s3": 19, "s4": 20, "s5": 21, + "s6": 22, "s7": 23, "s8": 24, "s9": 25, "s10": 26, "s11": 27, + "t3": 28, "t4": 29, "t5": 30, "t6": 31, +} +for _i in range(32): + REGISTERS.setdefault(f"x{_i}", _i) + +RUNG_RANK = {"B0": 0, "B1": 1, "B2": 2, "B3": 3} + + +class ValprobeError(GreenEyeError): + """A probe spec, image or scenario is invalid.""" + + +def reg_index(name: str) -> int: + key = str(name).strip().lower() + if key not in REGISTERS: + raise ValprobeError(f"unknown register {name!r}") + return REGISTERS[key] + + +def encode_return_stub(value: int) -> bytes: + """Encode ``lui a0,hi; addi a0,a0,lo; jalr zero,0(ra)`` (RV64IMAC).""" + value &= 0xFFFF_FFFF_FFFF_FFFF + hi = (value + 0x800) >> 12 + lo = value - (hi << 12) + lui = ((hi & 0xF_FFFF) << 12) | (10 << 7) | 0x37 + addi = ((lo & 0xFFF) << 20) | (10 << 15) | (10 << 7) | 0x13 + ret = 0x0000_8067 + return struct.pack("<III", lui, addi, ret) + + +def decode_memory_spec(item: dict) -> tuple[int, bytes]: + """Parse one memory item: ``{addr, hex|u64|u32|u8}``.""" + if "addr" not in item: + raise ValprobeError(f"memory item needs addr: {item!r}") + addr = _parse_u64(item["addr"]) + for key, fmt, size in (("hex", None, None), ("u64", "<Q", 8), ("u32", "<I", 4), + ("u16", "<H", 2), ("u8", "<B", 1)): + if key in item: + if key == "hex": + return addr, parse_hex_bytes(str(item["hex"])) + return addr, struct.pack(fmt, _parse_u64(item[key]) & ((1 << (8 * size)) - 1)) + raise ValprobeError(f"memory item needs one of hex/u64/u32/u16/u8: {item!r}") + + +def _load_spec(path: str) -> dict: + if not os.path.isfile(path): + raise ValprobeError(f"spec not found: {path}") + with open(path, "r", encoding="utf-8") as fh: + spec = json.load(fh) + fmt = spec.get("format", SPEC_FORMAT) + if fmt != SPEC_FORMAT: + raise ValprobeError(f"unsupported spec format {fmt!r} (want {SPEC_FORMAT})") + if not spec.get("variants"): + raise ValprobeError("spec has no variants") + return spec + + +def _apply_state(machine, *, registers: dict, memory, stubs, default_sp: int) -> list[dict]: + """Write stubs, memory and registers; return the applied stub notes.""" + stub_notes = [] + for stub in stubs or (): + addr = _parse_u64(stub["addr"]) + if "return" not in stub: + raise ValprobeError(f"stub needs return: {stub!r}") + code = encode_return_stub(_parse_u64(stub["return"])) + machine.write_mem(addr, code) + stub_notes.append({"addr": f"{addr:#x}", "return": f"{_parse_u64(stub['return']):#x}", + "code_hex": code.hex()}) + for item in memory or (): + addr, data = decode_memory_spec(item) + machine.write_mem(addr, data) + for name, value in (registers or {}).items(): + machine.x[reg_index(name)] = _parse_u64(value) + if machine.x[2] == 0: + machine.x[2] = default_sp + return stub_notes + + +def _read_outcomes(machine, reads) -> list[dict]: + out = [] + for item in reads or (): + addr = _parse_u64(item["addr"]) + size = int(item.get("size", 8)) + data = bytes(machine.read_mem(addr, size)) + record = {"addr": f"{addr:#x}", "size": size, "hex": data.hex()} + if item.get("label"): + record["label"] = item["label"] + if size == 8: + record["u64"] = int.from_bytes(data, "little") + elif size == 4: + record["u32"] = int.from_bytes(data, "little") + out.append(record) + return out + + +def run_probe( + *, + image: str = DEFAULT_IMAGE, + profile_path: str | None = DEFAULT_PROFILE, + spec: dict, + variant: dict, + out: str | None = None, + record: bool | None = None, + name: str = "valprobe", +) -> dict: + """Run one variant of a probe spec in a fresh machine; return its report.""" + if not os.path.isfile(image): + raise ValprobeError(f"image not found: {image}") + image_obj = load_elf(image) + if not image_obj.is_riscv: + raise ValprobeError(f"{image}: not a RISC-V ELF image") + + profile = load_profile(profile_path) if profile_path else None + load = (profile or {}).get("load", {}) + ram_base = _parse_u64(load.get("ram_base", 0)) + memory_mb = int(load.get("memory_mb", 256)) + ram_top = ram_base + memory_mb * 1024 * 1024 + + entry = _parse_u64(variant.get("entry", spec.get("target"))) + outcomes = {_parse_u64(a): label for a, label in + (variant.get("outcomes") or spec.get("outcomes") or {}).items()} + if not outcomes: + raise ValprobeError("probe needs at least one outcome address") + + budget = int(variant.get("budget", spec.get("budget", DEFAULT_BUDGET))) + timeout = float(variant.get("timeout", spec.get("timeout", DEFAULT_TIMEOUT))) + label = variant.get("label") or f"variant-{entry:#x}" + boot_to = spec.get("boot_to") or variant.get("boot_to") + + if record is None: + record = bool(out) + session = launch_native( + kernel=image, + memory_mb=memory_mb, + ram_base=ram_base, + record=record, + name=f"{name}-{label}"[:80], + device_map=None, + unknown_mmio="lenient", + bias=0, + ) + machine = session.target.machine + shim_meta: dict[str, Any] = {"enabled": False, "assumptions": []} + libos = LibosSyscalls() + boot: dict | None = None + try: + if profile is not None: + machine.devices = registry_from_profile(profile, machine=machine) + machine.translator = getattr(machine.devices, "dma", None) + machine.syscall_handler = libos.handle + shim = dict(profile.get("boot_shim", {})) + shim["_image_path"] = image + _apply_boot_shim(machine, shim, image_obj, shim_meta) + shim_meta["enabled"] = True + + if boot_to: + milestones: list[str] = [] + reason, pc, _a, _b = _run_with_milestones(machine, BOOT_INSTRUCTIONS, milestones) + rung = _rung_from_run(machine, shim_meta, milestones, libos) + boot = { + "requested": boot_to, + "rung": rung, + "instructions": machine.insns, + "stop_pc": f"{pc:#x}", + "milestones": milestones, + "reached": RUNG_RANK.get(rung, -1) >= RUNG_RANK.get(str(boot_to), 99), + } + + sentinel = (ram_top - 0x100) & ~0xF + default_sp = (ram_top - 0x8000) & ~0xF + machine.write_mem(sentinel, EBREAK.to_bytes(4, "little")) + breakpoints = set(outcomes) + breakpoints.add(sentinel) + + registers = dict(spec.get("registers") or {}) + registers.update(variant.get("registers") or {}) + registers.setdefault("ra", sentinel) + memory = list(spec.get("memory") or ()) + list(variant.get("memory") or ()) + stubs = list(spec.get("stubs") or ()) + list(variant.get("stubs") or ()) + stubs_applied = _apply_state( + machine, registers=registers, memory=memory, stubs=stubs, default_sp=default_sp + ) + machine.pc = entry + + entered = {f"x{i}": machine.x[i] for i in range(32) if i and machine.x[i]} + entered["pc"] = entry + started = time.monotonic() + status = "budget" + stop_pc = entry + fault: dict | None = None + outcome: str | None = None + remaining = budget + while remaining > 0: + if time.monotonic() - started > timeout: + status = "timeout" + break + chunk = min(RUN_CHUNK, remaining) + reason, stop_pc, a, b = machine.run(chunk, breakpoints=breakpoints) + remaining -= chunk + if reason == R_BP: + if stop_pc == sentinel: + status = "return" + break + outcome = outcomes.get(stop_pc, "unlabelled") + status = "outcome" + break + if reason == R_FAULT: + status = "fault" + fault = {"cause": a, "tval": f"{b:#x}"} + break + if reason == R_ECALL: + handled, result = (machine.syscall_handler(machine, stop_pc) + if machine.syscall_handler else (False, None)) + if not handled: + status = "ecall" + break + if result is not None: + machine.x[10] = result & 0xFFFF_FFFF_FFFF_FFFF + continue + if reason == R_BUDGET: + status = "budget" + + report: dict[str, Any] = { + "format": PROBE_FORMAT, + "probe": spec.get("name"), + "label": label, + "image": os.path.abspath(image), + "image_sha256": _sha256_file(image), + "profile": os.path.basename(profile_path) if profile_path else None, + "entry": f"{entry:#x}", + "status": status, + "outcome": outcome, + "expect": variant.get("expect"), + "pass": (None if variant.get("expect") is None else outcome == variant.get("expect")), + "stop_pc": f"{stop_pc:#x}", + "fault": fault, + "retval": machine.x[10], + "instructions": machine.insns, + "wall_seconds": round(time.monotonic() - started, 4), + "boot": boot, + "stubs": stubs_applied, + "registers_entered": {k: f"{v:#x}" if isinstance(v, int) else v + for k, v in entered.items()}, + "reads": _read_outcomes(machine, variant.get("reads") or spec.get("reads")), + "limits": [ + "Predicate-branch probe, not a boot: only the seeded state exists.", + "Stubs (if any) replace an out-of-predicate dependency; they are listed.", + "No timing/cycle claim (L0/L1 fidelity, docs/06).", + "A fault is the first missing state, not a product defect.", + ], + } + if getattr(session.recorder, "dir", None) is not None: + session.note("valprobe", label=label, outcome=outcome, status=status) + report["session"] = session.recorder.dir + return report + finally: + session.close() + + +def run_spec( + *, + spec: dict, + image: str | None = None, + profile_path: str | None = None, + out: str | None = None, + record: bool | None = None, + name: str | None = None, +) -> dict: + """Run every variant of a spec; return a matrix report.""" + image = image or spec.get("image") or DEFAULT_IMAGE + profile_path = profile_path or spec.get("profile") or DEFAULT_PROFILE + started = time.monotonic() + rows = [] + for variant in spec["variants"]: + rows.append(run_probe( + image=image, profile_path=profile_path, spec=spec, variant=variant, + out=out, record=record, name=name or (spec.get("name") or "valprobe"), + )) + matrix: dict[str, Any] = { + "format": MATRIX_FORMAT, + "spec": spec.get("name"), + "spec_format": spec.get("format"), + "image": os.path.abspath(image), + "image_sha256": _sha256_file(image), + "target": spec.get("target"), + "boot_to": spec.get("boot_to"), + "variants": rows, + "passed": sum(1 for r in rows if r["pass"] is True), + "failed": sum(1 for r in rows if r["pass"] is False), + "evaluated": sum(1 for r in rows if r["pass"] is not None), + "wall_seconds": round(time.monotonic() - started, 4), + } + if out: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, "valprobe-matrix.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(matrix, fh, indent=2, sort_keys=True) + fh.write("\n") + matrix["report_json"] = os.path.abspath(path) + return matrix + + +# --------------------------------------------------------------------------- +# Bundled target: the GR PROMOTE_CTX FECS predicate (570.144 firmware). +# +# Firmware facts used (all verified by disassembly of the shipped image): +# * per-buffer validation branch point: 0x5290dd4 (in FUN_ram_05290a48) +# * bInitialize=1 and bufferId==9 -> FECS branch at 0x5290e3c +# * FECS branch: instance = pGpu[(idx+0x37c)], idx = *(u32*)(getter+0xD28); +# table = *(u64*)(instance+0x190); slot = table[mapped(bufferId)*8]; +# refuse (0x1f) iff slot != 0 -- regardless of physAttr/size/VA +# * bufferId -> slot index via the firmware's own 0x531b944 jump table +# (FECS => 6) +# * bInitialize=0: bNonmapped!=0 => 0x1f, else accept +# +# The instance-index getter (0x524a5b8, a lock/table walk outside the +# predicate) is stubbed with a recorded constant return so the branch can run +# with crafted state; the predicate itself, the mapping table call and the +# slot read are the firmware's own code. +# --------------------------------------------------------------------------- + +PROMOTE_VALIDATION = 0x5290DD4 +FECS_INIT_BRANCH = 0x5290E3C +INSTANCE_GETTER = 0x524A5B8 +OUTCOMES = { + "0x5290cd0": "accept-commit-path", + "0x5290ce4": "accept-skip-path", + "0x5290df8": "refuse-invalid-argument-0x1f", + "0x5290ebc": "refuse-fecs-slot-exists-0x1f", +} + +#: Buffer ids (NV2080_CTRL_GPU_PROMOTE_CTX_BUFFER_ID_*), 570.144 interface order. +BUFFER_ID = { + "MAIN": 0, "PM": 1, "PATCH": 2, "BUNDLE_CB": 3, "PAGEPOOL": 4, + "ATTRIBUTE_CB": 5, "RTV_CB_GLOBAL": 6, "GFXP_POOL": 7, "GFXP_CTRL_BLK": 8, + "FECS_EVENT": 9, "PRIV_ACCESS_MAP": 10, "UNRESTRICTED_PRIV_ACCESS_MAP": 11, + "GLOBAL_PRIV_ACCESS_MAP": 12, +} + + +#: Scratch base for crafted predicate state. Must be encodable by the +#: ``lui+addi`` return stub (below 2^31), inside RAM, outside the image +#: (max vaddr ~0x5bb9000), outside the 0x0..0x8000000 DMA window and below +#: the 0x100000000 DMEM aperture the firmware uses after relocation. +FECS_SCRATCH_BASE = 0x10000000 + + +def _fecs_layout() -> dict: + """Deterministic scratch layout (see module doc).""" + scratch = FECS_SCRATCH_BASE + return { + "scratch": scratch, + "sp": scratch + 0x4000, + "s0": scratch + 0x5000, + "pgpu": scratch + 0x1000, + "getter_out": scratch + 0x1800, + "instance": scratch + 0x3800, + "entry": scratch + 0x6000, + } + + +def _fecs_variant(layout: dict, *, label: str, buffer_id: int, b_initialize: int, + b_nonmapped: int, phys_attr: int = 4, size: int = 0x10000, + phys: int = 0x1F4150000, virt: int = 0x204608000, + slot: int | None = None, expect: str | None = None) -> dict: + lay = layout + entry = lay["entry"] + # table = *(u64*)(instance+0x190) = instance+0x300; FECS maps to slot 6. + slot_addr = lay["instance"] + 0x300 + 6 * 8 + memory = [ + {"addr": f"{entry + 0x30:#x}", "u64": b_initialize and phys or 0}, + {"addr": f"{entry + 0x38:#x}", "u64": virt}, + {"addr": f"{entry + 0x40:#x}", "u64": size}, + {"addr": f"{entry + 0x48:#x}", "u32": phys_attr}, + {"addr": f"{entry + 0x4c:#x}", "u16": buffer_id}, + {"addr": f"{entry + 0x4e:#x}", "u8": b_initialize}, + {"addr": f"{entry + 0x4f:#x}", "u8": b_nonmapped}, + # pGpu -> instance pointer; getter result -> idx 0; instance -> table. + {"addr": f"{lay['pgpu'] + 0x1be0:#x}", "u64": lay["instance"]}, + {"addr": f"{lay['instance'] + 0x190:#x}", "u64": lay["instance"] + 0x300}, + ] + if slot is not None: + memory.append({"addr": f"{slot_addr:#x}", "u64": slot}) + return { + "label": label, + "registers": { + "s0": f"{lay['s0']:#x}", + "s2": f"{lay['pgpu']:#x}", + "s5": f"{entry:#x}", + "sp": f"{lay['sp']:#x}", + "a3": buffer_id, + "t1": 0xC, # loop bound set by the function before our entry point + }, + "stubs": [{"addr": f"{INSTANCE_GETTER:#x}", "return": f"{lay['getter_out']:#x}"}], + "memory": memory, + "reads": [{"addr": f"{lay['s0'] - 0x1c8:#x}", "size": 4, + "label": "firmware-mapped-slot-index"}, + {"addr": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot-value"}], + "expect": expect, + } + + +def fecs_spec(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFILE) -> dict: + """Build the FECS promote-validation matrix spec (bundled target).""" + layout = _fecs_layout() + variants = [ + # Live m3j: FECS init (PA) refused -- slot already populated. + _fecs_variant(layout, label="FECS init vidmem, slot non-NULL", + buffer_id=9, b_initialize=1, b_nonmapped=0, slot=1, + expect="refuse-fecs-slot-exists-0x1f"), + # Predicate inverse: the same shape would be accepted on an empty slot. + _fecs_variant(layout, label="FECS init vidmem, slot NULL", + buffer_id=9, b_initialize=1, b_nonmapped=0, slot=0, + expect="accept-commit-path"), + # The handoff's untested variant (a): does a sysmem aperture change it? + _fecs_variant(layout, label="FECS init sysmem attr 0x6, slot non-NULL", + buffer_id=9, b_initialize=1, b_nonmapped=0, phys_attr=6, slot=1, + expect="refuse-fecs-slot-exists-0x1f"), + _fecs_variant(layout, label="FECS init sysmem attr 0x6, slot NULL", + buffer_id=9, b_initialize=1, b_nonmapped=0, phys_attr=6, slot=0, + expect="accept-commit-path"), + # Size/PA content must not matter to the predicate. + _fecs_variant(layout, label="FECS init size 0x1000, slot non-NULL", + buffer_id=9, b_initialize=1, b_nonmapped=0, size=0x1000, slot=1, + expect="refuse-fecs-slot-exists-0x1f"), + # Live m7: FECS VA-only accepted (bInitialize=0 path). + _fecs_variant(layout, label="FECS no-init mapped, slot non-NULL", + buffer_id=9, b_initialize=0, b_nonmapped=0, slot=1, + expect="accept-skip-path"), + _fecs_variant(layout, label="FECS no-init nonmapped (bookkeeping refusal)", + buffer_id=9, b_initialize=0, b_nonmapped=1, slot=1, + expect="refuse-invalid-argument-0x1f"), + # Live m3j: PM and PATCH init accepted. + _fecs_variant(layout, label="PM init", + buffer_id=1, b_initialize=1, b_nonmapped=0, + expect="accept-commit-path"), + _fecs_variant(layout, label="PATCH init", + buffer_id=2, b_initialize=1, b_nonmapped=0, + expect="accept-commit-path"), + _fecs_variant(layout, label="GFXP_CTRL_BLK init (id 8)", + buffer_id=8, b_initialize=1, b_nonmapped=0, + expect="accept-skip-path"), + _fecs_variant(layout, label="BUNDLE_CB init (id 3)", + buffer_id=3, b_initialize=1, b_nonmapped=0, + expect="refuse-invalid-argument-0x1f"), + _fecs_variant(layout, label="PRIV_ACCESS_MAP init (id 10)", + buffer_id=10, b_initialize=1, b_nonmapped=0, + expect="accept-skip-path"), + _fecs_variant(layout, label="UNRESTRICTED_PRIV init (id 11)", + buffer_id=11, b_initialize=1, b_nonmapped=0, + expect="accept-skip-path"), + _fecs_variant(layout, label="GLOBAL_PRIV_ACCESS_MAP init (id 12)", + buffer_id=12, b_initialize=1, b_nonmapped=0, + expect="refuse-invalid-argument-0x1f"), + ] + # id > 0xc is refused one block earlier (0x5290cc4, still in the per-entry + # loop); enter there with the loop's a4/t1 state. + unknown_id = _fecs_variant(layout, label="unknown buffer id 13", + buffer_id=13, b_initialize=1, b_nonmapped=0, + expect="refuse-invalid-argument-0x1f") + unknown_id["entry"] = "0x5290cc0" + unknown_id["registers"]["a4"] = 13 + variants.append(unknown_id) + return { + "format": SPEC_FORMAT, + "name": "fecs-promote-validation", + "target": f"{PROMOTE_VALIDATION:#x}", + "outcomes": OUTCOMES, + "budget": 20_000, + "timeout": 20.0, + "image": image, + "profile": profile_path, + "note": ( + "570.144 per-buffer PROMOTE_CTX validation; instance-index getter " + f"({INSTANCE_GETTER:#x}, lock/table walk outside the predicate) stubbed " + "to a recorded constant; the predicate, the id->slot mapping call and " + "the memDesc slot read are the firmware's own code." + ), + "variants": variants, + } + + +def run_demo(out: str | None = None) -> dict: + """Run the bundled FECS matrix on the real image, twice for determinism.""" + spec = fecs_spec() + first = run_spec(spec=spec, out=out, record=False, name="valprobe-fecs") + second = run_spec(spec=spec, record=False, name="valprobe-fecs") + verdicts_a = [(r["label"], r["outcome"]) for r in first["variants"]] + verdicts_b = [(r["label"], r["outcome"]) for r in second["variants"]] + result = { + "format": DEMO_FORMAT, + "ok": first["failed"] == 0 and verdicts_a == verdicts_b, + "evaluated": first["evaluated"], + "failed": first["failed"], + "deterministic": verdicts_a == verdicts_b, + "verdicts": [{"label": label, "outcome": outcome} for label, outcome in verdicts_a], + } + if out: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, "valprobe-demo.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(result, fh, indent=2, sort_keys=True) + fh.write("\n") + result["artifacts"] = {"demo_json": os.path.abspath(path)} + return result + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="valprobe.py", + description=( + "Run one firmware validation-predicate branch with crafted state " + "(offline; never touches hardware)." + ), + ) + sub = parser.add_subparsers(dest="command") + sweep = sub.add_parser("sweep", help="run a bundled target matrix") + sweep.add_argument("--target", default="fecs-promote", choices=["fecs-promote"]) + sweep.add_argument("--image", default=None) + sweep.add_argument("--profile", default=None) + sweep.add_argument("--boot-to", default=None, help="boot the image to a rung first (e.g. B3)") + sweep.add_argument("--out", help="output directory (matrix JSON + sessions)") + sweep.add_argument("--no-record", action="store_true", help="skip session recording") + run = sub.add_parser("run", help="run a probe spec file") + run.add_argument("--spec", required=True) + run.add_argument("--image", default=None) + run.add_argument("--profile", default=None) + run.add_argument("--out", help="output directory (matrix JSON + sessions)") + run.add_argument("--no-record", action="store_true", help="skip session recording") + demo = sub.add_parser("demo", help="bundled FECS matrix, run twice (determinism)") + demo.add_argument("--out", help="output directory") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "sweep": + spec = fecs_spec(image=args.image or DEFAULT_IMAGE, + profile_path=args.profile or DEFAULT_PROFILE) + if args.boot_to: + spec["boot_to"] = args.boot_to + matrix = run_spec(spec=spec, image=args.image, profile_path=args.profile, + out=args.out, record=False if args.no_record else None) + elif args.command == "run": + spec = _load_spec(args.spec) + matrix = run_spec(spec=spec, image=args.image, profile_path=args.profile, + out=args.out, record=False if args.no_record else None) + elif args.command == "demo": + matrix = run_demo(args.out) + else: + parser.print_help() + return 2 + except ValprobeError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + if matrix.get("format") == MATRIX_FORMAT: + for row in matrix["variants"]: + mark = {True: "PASS", False: "FAIL", None: "----"}[row["pass"]] + print(f"{mark} {row['label']:<48} {row['outcome'] or row['status']}") + print(json.dumps({k: matrix[k] for k in + ("format", "spec", "evaluated", "passed", "failed", + "wall_seconds", "report_json") if k in matrix}, + indent=2, sort_keys=True)) + return 3 if matrix["failed"] else 0 + print(json.dumps(matrix, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) |