#!/usr/bin/env python3 """Functional GMMU page-table walker (Turing VER2 geometry). The decoder in :mod:`sim.gmmu_pte` decodes one captured entry. This module follows those entries: given a page-directory base and a sparse physical-memory view of the table pages, it walks a virtual address to its leaf mapping and reports every level, the page size, the aperture and the access flags -- or the exact fault (missing table, invalid aperture, invalid leaf). Why it exists: the engine sees memory through the GMMU, not through the host's bookkeeping. When a forwarded submission goes silent (an engine-side write never appears), the first question is *where would the engine read/write this address?* A capture of the channel's page tables plus this walker answers it offline, deterministically, without hardware. Evidence and honesty: * Geometry, field positions and the entry encodings come from the same evidence base as :mod:`sim.gmmu_pte` (see ``docs/43`` and ``docs/48``). * The 4K and 2M paths are **validated against a live captured channel PDB** (see :func:`channel_pdb_fixture`); the 64K dual path is modelled from the format definitions and reported with ``evidence: "modeled"`` until a capture validates it. * Unsupported encodings fail closed (``E_UNSUPPORTED``); a video-memory leaf address is reported with ``pa_complete: false`` because the decoder cannot claim the high address bits there (never guessed). * No timing claim; this is the engine's *address* view (L1), not its behaviour. """ from __future__ import annotations import argparse import json import os import struct 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 sim.gmmu_pte import ( # noqa: E402 GmmuDecodeError, KIND_VER2, KIND_VER2_DUAL, LEVELS, SUPPORTED_KINDS, decode_dual_pde, decode_entry, decode_pde, decode_pte, ) #: Schema tag of a walk report. WALK_FORMAT = "greeneye-gmmu-walk/1" #: Page modes the walker accepts. ``auto`` prefers the 64K dual half for a #: 64K-aligned VA when that half is valid, else the 4K half. PAGE_MODES = ("4k", "64k", "auto") VA_MASK = (1 << 49) - 1 #: Fault codes (never a guess; each names a concrete missing/invalid piece). FAULT_TABLE_MISSING = "table_missing" FAULT_INVALID_APERTURE = "invalid_aperture" FAULT_PTE_INVALID = "pte_invalid" FAULT_UNSUPPORTED = "unsupported" FAULT_OUT_OF_RANGE = "va_out_of_range" class WalkError(ValueError): """A walk request the evidence cannot answer.""" class MemoryView: """Sparse physical memory: ``pa -> bytes`` blobs (table pages, targets).""" def __init__(self, blobs: dict[int, bytes] | None = None): self._blobs = dict(blobs or {}) def add(self, pa: int, data: bytes) -> None: if pa < 0 or not isinstance(data, (bytes, bytearray)): raise WalkError("add() needs a non-negative pa and bytes") self._blobs[pa] = bytes(data) def has(self, pa: int) -> bool: return pa in self._blobs def read(self, pa: int, size: int) -> bytes | None: for base, blob in self._blobs.items(): if base <= pa and pa + size <= base + len(blob): off = pa - base return blob[off:off + size] return None def spans(self) -> list[dict]: return [{"pa": f"{pa:#x}", "size": len(b)} for pa, b in sorted(self._blobs.items())] # --------------------------------------------------------------------------- # Encoders (fixtures/tests; mirror of the decoder's field positions) # --------------------------------------------------------------------------- def encode_pde_pointer(pa: int, *, volatile: bool = True) -> int: """Encode a VER2 directory PDE: bit0=0, aperture 3 (sysmem non-coherent).""" if pa & 0xFFF: raise WalkError(f"table PA must be 4K aligned, got {pa:#x}") return ((pa >> 12) << 8) | (3 << 1) | (8 if volatile else 0) def encode_pte(pa: int, *, page_size: str = "4k", read_only: bool = False, volatile: bool = True) -> int: """Encode a VER2 leaf PTE: bit0=1, aperture 3 (sysmem non-coherent).""" if page_size == "4k": align = 0xFFF elif page_size == "64k": align = 0xFFFF elif page_size == "2m": align = 0x1F_FFFF else: raise WalkError(f"unknown page_size {page_size!r}") if pa & align: raise WalkError(f"PA {pa:#x} is not {page_size}-aligned") return ((pa >> 12) << 8) | (3 << 1) | 1 | (8 if volatile else 0) | (0x40 if read_only else 0) def encode_dual_slot(big_pt_pa: int | None, small_pt_pa: int) -> bytes: """Encode a VER2 dual PDE0 slot; BIG = 64K table, SMALL = 4K table.""" big = 0 if big_pt_pa is not None: if big_pt_pa & 0xFFF: raise WalkError(f"big table PA must be 4K aligned, got {big_pt_pa:#x}") big = (((big_pt_pa >> 8) & ~1) << 4) | (3 << 1) | 8 small = ((small_pt_pa >> 12) << 8) | (3 << 1) | 8 return struct.pack(" dict: report = { "format": WALK_FORMAT, "status": "fault", "fault": code, "fault_message": message, "steps": steps, } report.update(extra) return report def walk( va: int, *, root_pa: int, mem: MemoryView, kind: str = KIND_VER2, page_mode: str = "4k", ) -> dict: """Walk ``va`` through the page tables rooted at ``root_pa``. Returns a ``greeneye-gmmu-walk/1`` report. ``status`` is ``ok`` with ``pa``/``page_size``/``writable`` on success, otherwise ``fault`` with a concrete ``fault`` code. Every level's raw entry and decode is kept in ``steps`` so the result is auditable. """ if kind not in SUPPORTED_KINDS: return _fault(FAULT_UNSUPPORTED, f"kind {kind!r} not in {SUPPORTED_KINDS}", [], va=va, root_pa=root_pa) if page_mode not in PAGE_MODES: raise WalkError(f"page_mode must be one of {PAGE_MODES}, got {page_mode!r}") if not (0 <= va <= VA_MASK): return _fault(FAULT_OUT_OF_RANGE, f"va {va:#x} outside 49-bit space", [], va=va, root_pa=root_pa) steps: list[dict] = [] pa_complete = True page_used = page_mode evidence = "validated" cur = root_pa for depth in range(5): geo = LEVELS[depth] _hi_bit, lo_bit = geo["va_bits"] idx = (va >> lo_bit) & ((1 << geo["index_bits"]) - 1) table_pa = cur if depth == 3: off = idx * 16 raw = mem.read(table_pa + off, 16) if raw is None: return _fault(FAULT_TABLE_MISSING, f"depth-3 dual slot at pa={table_pa + off:#x} not in view", steps, va=va, root_pa=root_pa) value128 = int.from_bytes(raw, "little") low = value128 & 0xFFFF_FFFF_FFFF_FFFF step = {"depth": depth, "level": geo["hw"], "table_pa": f"{table_pa:#x}", "index": idx, "entry_pa": f"{table_pa + off:#x}", "raw": f"{value128:#x}"} if low & 1: # 2 MB leaf in the low half try: pte = decode_entry(low, kind=KIND_VER2, role="pte", level=3) except GmmuDecodeError as exc: return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step], va=va, root_pa=root_pa) step.update({"role": "pte", "page_size": "2m", "aperture_name": pte["aperture_name"], "valid": pte["valid"], "address": pte["address"]}) steps.append(step) if not pte["valid"] or pte["address"] is None: return _fault(FAULT_PTE_INVALID, f"2M leaf at {table_pa + off:#x} invalid/unsupported", steps, va=va, root_pa=root_pa) if not pte["address_complete"]: pa_complete = False pa = pte["address"] + (va & 0x1F_FFFF) return _ok(va, pa, "2m", pte, steps, root_pa, kind, page_used="2m", evidence=evidence, pa_complete=pa_complete) try: dual = decode_dual_pde(value128, pde_kind=KIND_VER2_DUAL, level=3) except GmmuDecodeError as exc: return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step], va=va, root_pa=root_pa) chosen = None if page_mode == "64k": chosen, page_used, evidence = dual["big"], "64k", "modeled" elif page_mode == "4k": chosen, page_used = dual["small"], "4k" else: # auto: prefer BIG for a 64K-aligned VA when valid if va % 0x10000 == 0 and dual["big"]["address"] is not None: chosen, page_used, evidence = dual["big"], "64k", "modeled" else: chosen, page_used = dual["small"], "4k" step.update({"role": "dual_pde", "page_mode_used": page_used, "big": dual["big"], "small": dual["small"]}) steps.append(step) if chosen["address"] is None: return _fault(FAULT_INVALID_APERTURE, f"dual PDE at {table_pa + off:#x} {page_used} half has " f"aperture {chosen['aperture_name']}", steps, va=va, root_pa=root_pa) if not chosen["address_complete"]: pa_complete = False cur = chosen["address"] continue off = idx * geo["entry_size"] if depth == 4: # A 64K-granularity leaf table is indexed by VA bits 24:16. idx = (va >> (16 if page_used == "64k" else 12)) & 0x1FF off = idx * geo["entry_size"] raw = mem.read(table_pa + off, geo["entry_size"]) if raw is None: return _fault(FAULT_TABLE_MISSING, f"level {depth} entry at pa={table_pa + off:#x} not in view", steps, va=va, root_pa=root_pa) value = int.from_bytes(raw, "little") step = {"depth": depth, "level": geo["hw"], "table_pa": f"{table_pa:#x}", "index": idx, "entry_pa": f"{table_pa + off:#x}", "raw": f"{value:#x}"} if depth < 3: try: pde = decode_pde(value, pde_kind=KIND_VER2, level=depth) except GmmuDecodeError as exc: return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step], va=va, root_pa=root_pa) step.update({"role": "pde", "aperture_name": pde["aperture_name"], "address": pde["address"]}) steps.append(step) if pde["address"] is None: return _fault(FAULT_INVALID_APERTURE, f"level {depth} entry at {table_pa + off:#x} has " f"aperture {pde['aperture_name']}", steps, va=va, root_pa=root_pa) if not pde["address_complete"]: pa_complete = False cur = pde["address"] continue # depth 4: leaf page table try: pte = decode_pte(value, pte_kind=KIND_VER2, level=4) except GmmuDecodeError as exc: return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step], va=va, root_pa=root_pa) leaf_mask = 0xFFFF if page_used == "64k" else 0xFFF step.update({"role": "pte", "page_size": page_used, "aperture_name": pte["aperture_name"], "valid": pte["valid"], "address": pte["address"], "writable": pte["writable"]}) steps.append(step) if not pte["valid"] or pte["address"] is None: return _fault(FAULT_PTE_INVALID, f"leaf PTE at {table_pa + off:#x} invalid/unsupported " f"(raw {value:#x})", steps, va=va, root_pa=root_pa) if not pte["address_complete"]: pa_complete = False pa = pte["address"] + (va & leaf_mask) return _ok(va, pa, page_used, pte, steps, root_pa, kind, page_used=page_used, evidence=evidence, pa_complete=pa_complete) return _fault(FAULT_UNSUPPORTED, "walk exceeded 5 levels", steps, va=va, root_pa=root_pa) def _ok(va, pa, page_size, leaf, steps, root_pa, kind, *, page_used, evidence, pa_complete) -> dict: report = { "format": WALK_FORMAT, "status": "ok", "va": f"{va:#x}", "root_pa": f"{root_pa:#x}", "kind": kind, "page_mode_used": page_used, "page_size": page_size, "evidence": evidence, "pa": f"{pa:#x}", "pa_complete": pa_complete, "aperture_name": leaf.get("aperture_name"), "writable": leaf.get("writable", True), "read_only": leaf.get("read_only", False), "kind_name": leaf.get("kind_name"), "steps": steps, "assumptions": [ "4K/2M paths are validated against a captured channel PDB; the 64K dual path is modelled and reported with evidence='modeled'.", "A video-memory leaf reports pa_complete=false (high address bits are not claimed by the decoder).", ], } if page_size == "64k": report["assumptions"].append("64K page-mode selection for auto walks is an assumption (64K-aligned VA prefers the BIG half).") return report def walk_many(vas, *, root_pa: int, mem: MemoryView, kind: str = KIND_VER2, page_mode: str = "4k") -> list[dict]: return [walk(va, root_pa=root_pa, mem=mem, kind=kind, page_mode=page_mode) for va in vas] # --------------------------------------------------------------------------- # Captured golden vector: a live channel PDB (numeric only) # --------------------------------------------------------------------------- #: Captured channel-PDB entries from a live run (five VAs of one channel #: chunk). Used as the walker's regression vector: the walk must reproduce #: the DMA addresses the capture logged for these VAs. _CHANNEL_PDB = { "root_pa": 0x12F16E000, "root": {0: 0x0000_0000_12F16F0E}, "pd2": {0: 0x0000_0000_11065E0E}, "pd1": {0: 0x0000_0000_1106630E}, "pd0_small": {0: 0x0000_0000_15908D0E}, "pt": { 256: 0x0000_0000_2045D90F, 257: 0x0000_0000_121BD90F, 258: 0x0000_0000_1521150F, 259: 0x0000_0000_15A5F50F, 260: 0x0000_0000_10CF270F, }, } _CHANNEL_VAS = { 0x100000: ("gpfifo", 0x2045D9000), 0x101000: ("push-buffer", 0x121BD9000), 0x102000: ("copy-src", 0x152115000), 0x103000: ("copy-dst", 0x15A5F5000), 0x104000: ("semaphore", 0x10CF27000), } def channel_pdb_fixture() -> tuple[int, MemoryView]: """Rebuild the captured channel PDB as a walkable memory view.""" mem = MemoryView() root = bytearray(0x1000) for idx, value in _CHANNEL_PDB["root"].items(): struct.pack_into("> 8) << 12)) mem.add(0x110663000, bytes(pd0)) pt = bytearray(0x1000) for idx, value in _CHANNEL_PDB["pt"].items(): struct.pack_into(" dict: """Walk the captured channel-PDB vector; PASS iff every VA matches.""" root_pa, mem = channel_pdb_fixture() rows = [] ok = True for va, (label, expected) in sorted(_CHANNEL_VAS.items()): report = walk(va, root_pa=root_pa, mem=mem) got = int(report["pa"], 16) if report["status"] == "ok" else None match = got == expected ok = ok and match rows.append({"label": label, "va": f"{va:#x}", "expected_pa": f"{expected:#x}", "walk": report, "match": match}) demo = {"format": "greeneye-gmmu-walk-demo/1", "ok": ok, "rows": rows, "root_pa": f"{root_pa:#x}"} if out: os.makedirs(out, exist_ok=True) path = os.path.join(out, "gmmu-walk-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 _parse_int(text: str) -> int: try: return int(text, 0) except ValueError as exc: raise WalkError(f"bad integer {text!r}") from exc def _parse_mem(spec: str) -> tuple[int, bytes]: if "=" not in spec: raise WalkError(f"--mem needs =: {spec!r}") pa_text, _, hex_text = spec.partition("=") data = "".join(hex_text.split()) if len(data) % 2: raise WalkError(f"odd hex length in {spec!r}") return _parse_int(pa_text), bytes.fromhex(data) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="gmmu_walk.py", description="Walk a GMMU VA through captured/constructed page tables (offline).", ) sub = parser.add_subparsers(dest="command") run = sub.add_parser("run", help="walk VAs through a page-table view") run.add_argument("--pdb-pa", required=True, help="page-directory base PA") run.add_argument("--va", action="append", required=True, help="virtual address (repeatable)") run.add_argument("--mem", action="append", default=[], help="= table page (repeatable)") run.add_argument("--kind", default=KIND_VER2, choices=list(SUPPORTED_KINDS)) run.add_argument("--page-mode", default="4k", choices=list(PAGE_MODES)) run.add_argument("--out", help="write the JSON report here") demo = sub.add_parser("demo", help="walk the captured channel-PDB vector") 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": mem = MemoryView() for spec in args.mem: pa, data = _parse_mem(spec) mem.add(pa, data) reports = [walk(_parse_int(va), root_pa=_parse_int(args.pdb_pa), mem=mem, kind=args.kind, page_mode=args.page_mode) for va in args.va] doc = {"format": "greeneye-gmmu-walk-set/1", "reports": reports} for report in reports: if report["status"] == "ok": print(f"OK {report['va']} -> {report['pa']} " f"{report['page_size']} {report['aperture_name']} " f"writable={report['writable']}") else: print(f"FAULT {report['va']} {report['fault']}: {report['fault_message']}") if args.out: os.makedirs(args.out, exist_ok=True) path = os.path.join(args.out, "gmmu-walk.json") with open(path, "w", encoding="utf-8") as fh: json.dump(doc, fh, indent=2, sort_keys=True) fh.write("\n") print(json.dumps({"report_json": os.path.abspath(path)}, indent=2)) return 0 if all(r["status"] == "ok" for r in reports) else 3 if args.command == "demo": demo = run_demo(args.out) for row in demo["rows"]: report = row["walk"] got = report.get("pa", report.get("fault_message")) print(f"{'PASS' if row['match'] else 'FAIL'} {row['label']:<12} " f"{row['va']} -> {got} (expected {row['expected_pa']})") print(json.dumps({"ok": demo["ok"]}, indent=2)) return 0 if demo["ok"] else 3 parser.print_help() return 2 except WalkError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())