diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/launch_audit.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/launch_audit.py | 691 |
1 files changed, 691 insertions, 0 deletions
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()) |