diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/re_demo.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/re_demo.py | 832 |
1 files changed, 832 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/re_demo.py b/debug/risc-v-gsp-rm-vm/harness/re_demo.py new file mode 100644 index 0000000..3f9c317 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/re_demo.py @@ -0,0 +1,832 @@ +#!/usr/bin/env python3 +"""Real reverse-engineering demo: GSP entry -> first fault/milestone (U51). + +Answers one concrete question on the real TU104 570.144 image with the +simulator's own primitives (``launch_native`` session, hooks, memory reads, +snapshots, exec trace, recorded sessions): + + What happens on the real GSP image between entry and the first + fault/milestone, and what does the state tell us about boot? + +Method (both sides measured, thresholds frozen in this file before any run): + +* **baseline** - the QEMU ``probe`` path (bounded 0.5 s, measured) plus a + conservative manual static-reading estimate (``MANUAL_ESTIMATE_S``; the + method note says how the estimate is derived). QEMU reports the first + fault; the "what happens between entry and the fault" part is hand work + in a disassembler. +* **emulator** - three recorded native sessions: (A) lenient MMIO (B0), which + reproduces the QEMU first touch and follows the firmware to its fail-closed + halt; (A') the same run repeated for determinism; (B) the same run with the + chip-ID register served a Turing value from a one-entry device map, which + advances past the QEMU point to the next (relocation) blocker. + +Everything is stdlib-only, bounded (<= 2M instructions, <= 10 s QEMU), and +touches no hardware, GPU, guest, module or network. This is simulator +evidence at L0/L1 (docs/06): no timing or silicon claim. + +CLI:: + + python3 harness/re_demo.py + python3 harness/re_demo.py --json + python3 harness/re_demo.py --out harness/re_demo --report harness/re-demo-report.md +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.dirname(HERE) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from greeneye.device_map import DeviceEntry, DeviceMap, QEMU_REFERENCE # noqa: E402 +from greeneye.sim_backend import launch_native # noqa: E402 +from harness.sim_qemu_line import run_qemu_baseline, session_hashes # noqa: E402 +from sim.decode import ( # noqa: E402 + K_AUIPC, + K_BRANCH, + K_CSR, + K_JAL, + K_JALR, + K_LOAD, + K_LUI, + K_OP, + K_OP32, + K_OPI, + K_OPI32, + decode16, + decode32, +) + +DEFAULT_IMAGE = os.path.normpath( + os.path.join( + TOOL_ROOT, "..", "..", "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin" + ) +) +DEFAULT_OUT = os.path.join(HERE, "re_demo") +DEFAULT_REPORT = os.path.join(HERE, "re-demo-report.md") + +#: GPU physical addresses are relocated into simulator RAM at this bias +#: (same convention as the probe/device-map/gsp-profile paths). +BIAS = 0x80000000 + +ENTRY = 0x4009000 +FIRST_TOUCH_PC = 0x4009014 +HALT_PC = 0x4009070 +TU10X_ENTRY = 0x4007000 +GA100_ENTRY = 0x4005000 +TU10X_ABS_TARGET = 0x40070F8 + +#: The chip-ID field is bits [28:20] of the register read at the first touch; +#: these are the comparison values the image itself uses (session A trace). +TU10X_IDS = (354, 355, 356, 357, 358, 359, 360) +GA100_ID = 368 +#: Exploratory modeled register value: Turing TU10x ID in the field position. +#: This is an *assumption* chosen to exercise the dispatch path, not an +#: observed hardware value (the QEMU baseline faults before reading it). +MODELED_ID = TU10X_IDS[0] << 20 + +#: Frozen thresholds - recorded 2026-10-03, before the first measurement. +#: Never move silently; if a threshold changes, date the change in docs/37. +FROZEN_THRESHOLDS = { + "frozen_on": "2026-10-03", + "emulator_wall_s_max": 120.0, + "manual_static_estimate_s": 600.0, + "answers_required": 5, + "identical_lenient_runs": 2, + "milestone_past_qemu": True, +} + +#: How the manual estimate is derived (a floor, not a ceiling): read the +#: 6-instruction .bootstrap prologue + the two per-architecture text sections +#: (~90 instructions), identify the pointer construction, the chip-ID bit +#: field, the two dispatch targets and the halt path, then correlate against +#: the QEMU fault report. No human fast path is measured; 10 minutes is the +#: conservative floor for one engineer with the artifacts already open. +MANUAL_METHOD_NOTE = ( + "floor estimate: ~90 instructions across .bootstrap + .tu10x_text + " + ".ga100_text read by hand, dispatch table identified and correlated " + "with the QEMU first-fault report; not measured in this run" +) + + +def _relpath(path: str | None) -> str | None: + if path is None: + return None + return os.path.relpath(os.path.abspath(path), TOOL_ROOT) + + +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 _read_json_lines(path: str) -> list[dict]: + rows: list[dict] = [] + try: + with open(path, "r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if line: + rows.append(json.loads(line)) + except (OSError, ValueError): + pass + return rows + + +def _trace_exec_pcs(session_dir: str) -> list[int]: + rows = _read_json_lines(os.path.join(session_dir, "blobs", "trace.jsonl")) + return [int(row["pc"], 16) for row in rows if row.get("t") == "exec"] + + +def _fw(pc: int) -> int: + return pc - BIAS if pc >= BIAS else pc + + +# -- ELF section names (stdlib) ---------------------------------------------- + + +def elf_progbits_sections(path: str) -> dict: + """{name: (addr, size)} for PROGBITS sections, parsed directly from ELF.""" + with open(path, "rb") as fh: + data = fh.read() + e_shoff = int.from_bytes(data[0x28:0x30], "little") + e_shentsize = int.from_bytes(data[0x3A:0x3C], "little") + e_shnum = int.from_bytes(data[0x3C:0x3E], "little") + e_shstrndx = int.from_bytes(data[0x3E:0x40], "little") + + def header(index: int): + off = e_shoff + index * e_shentsize + return ( + int.from_bytes(data[off : off + 4], "little"), # sh_name + int.from_bytes(data[off + 4 : off + 8], "little"), # sh_type + int.from_bytes(data[off + 16 : off + 24], "little"), # sh_addr + int.from_bytes(data[off + 24 : off + 32], "little"), # sh_offset + int.from_bytes(data[off + 32 : off + 40], "little"), # sh_size + ) + + _, _, _, strtab_off, strtab_size = header(e_shstrndx) + strtab = data[strtab_off : strtab_off + strtab_size] + sections: dict = {} + for index in range(e_shnum): + name_off, sh_type, addr, _offset, size = header(index) + end = strtab.find(b"\0", name_off) + name = strtab[name_off:end].decode("utf-8", "replace") if end >= 0 else "" + if sh_type == 1 and size: # SHT_PROGBITS + sections[name] = (addr, size) + return sections + + +def section_for(sections: dict, fw_addr: int) -> str | None: + for name, (base, size) in sections.items(): + if base <= fw_addr < base + size: + return name + return None + + +# -- minimal disassembly (decode tuples -> mnemonics) ------------------------ + +_LOAD_NAMES = ("lb", "lh", "lw", "ld", "lbu", "lhu", "lwu") +_BRANCH_NAMES = ("beq", "bne", "blt", "bge", "bltu", "bgeu") +_OPI_NAMES = {0: "addi", 1: "slli", 2: "slti", 3: "sltiu", 4: "xori", 5: "srli", 6: "ori", 7: "andi", 0x15: "srai"} +_OPI32_NAMES = {0: "addiw", 1: "slliw", 5: "srliw", 0x15: "sraiw"} +_OP_NAMES = { + 0: "add", 1: "sll", 2: "slt", 3: "sltu", 4: "xor", 5: "srl", 6: "or", 7: "and", + 8: "sub", 0x15: "sra", +} +_OP32_NAMES = {0: "addw", 1: "sllw", 5: "srlw", 8: "subw", 0x15: "sraw"} +_CSR_NAMES = {0x300: "mstatus"} +_CSR_OPS = {1: "csrrw", 2: "csrs", 3: "csrc", 5: "csrrwi", 6: "csrsi", 7: "csrci"} + + +def _abi(index: int) -> str: + return ( + "zero", "ra", "sp", "gp", "tp", "t0", "t1", "t2", "s0", "s1", + "a0", "a1", "a2", "a3", "a4", "a5", "a6", "a7", + "s2", "s3", "s4", "s5", "s6", "s7", "s8", "s9", "s10", "s11", + "t3", "t4", "t5", "t6", + )[index] + + +def disasm(pc: int, raw: int, decoded: tuple) -> str: + kind, sub, rd, rs1, rs2, imm, _next = decoded + if kind == K_OPI: + name = _OPI_NAMES.get(sub, f"opi:{sub}") + text = f"{name} {_abi(rd)},{_abi(rs1)},{imm}" + elif kind == K_OP: + name = _OP_NAMES.get(sub, f"op:{sub}") + text = f"{name} {_abi(rd)},{_abi(rs1)},{_abi(rs2)}" + elif kind == K_OPI32: + name = _OPI32_NAMES.get(sub, f"opi32:{sub}") + text = f"{name} {_abi(rd)},{_abi(rs1)},{imm}" + elif kind == K_OP32: + name = _OP32_NAMES.get(sub, f"op32:{sub}") + text = f"{name} {_abi(rd)},{_abi(rs1)},{_abi(rs2)}" + elif kind == K_LOAD: + name = _LOAD_NAMES[sub] if sub < len(_LOAD_NAMES) else f"load:{sub}" + text = f"{name} {_abi(rd)},{imm}({_abi(rs1)})" + elif kind == K_BRANCH: + name = _BRANCH_NAMES[sub] if sub < len(_BRANCH_NAMES) else f"branch:{sub}" + text = f"{name} {_abi(rs1)},{_abi(rs2)},{_next:#x}" + elif kind == K_LUI: + text = f"lui {_abi(rd)},{imm >> 12:#x}" + elif kind == K_AUIPC: + text = f"auipc {_abi(rd)},{imm >> 12:#x}" + elif kind == K_JALR: + text = f"jalr {_abi(rd)},{imm}({_abi(rs1)})" + elif kind == K_JAL: + text = f"jal {_abi(rd)},{_next:#x}" + elif kind == K_CSR: + csr = _CSR_NAMES.get(imm, f"0x{imm:x}") + name = _CSR_OPS.get(sub, f"csr:{imm:#x}") + text = f"{name} {csr},{_abi(rs1)}" + else: + text = f".raw 0x{raw:08x}" + return f"{pc:#x}: {raw:08x} {text}" + + +def read_disasm(session, addr: int, count: int) -> list[str]: + lines: list[str] = [] + for _ in range(count): + raw16 = int.from_bytes(session.memory.read(addr, 2), "little") + if raw16 & 3 == 3: + raw = int.from_bytes(session.memory.read(addr, 4), "little") + decoded = decode32(addr, raw) + step = 4 + else: + raw = raw16 + decoded = decode16(addr, raw16) + step = 2 + lines.append(disasm(addr, raw, decoded)) + addr += step + return lines + + +# -- sessions ---------------------------------------------------------------- + + +def _launch(image: str, out_dir: str, name: str, device_map, trace_spec): + session_dir = os.path.join(out_dir, "sessions", name) + shutil.rmtree(session_dir, ignore_errors=True) + session = launch_native( + kernel=image, + record=session_dir, + name=name, + memory_mb=512, + bias=BIAS, + device_map=device_map, + unknown_mmio="lenient", + learn_mmio=device_map is not None, + trace=trace_spec, + ) + return session, session_dir + + +def run_lenient_session(image: str, out_dir: str, tag: str = "native-lenient") -> dict: + """Session A: lenient MMIO; reproduce the QEMU first touch, follow to halt.""" + dm = DeviceMap(meta={"source": "re-demo-lenient", "bias": BIAS}) + session, session_dir = _launch( + image, + out_dir, + tag, + dm, + {"exec": True, "start": ENTRY + BIAS, "end": HALT_PC + BIAS, "max_events": 256}, + ) + start = time.monotonic() + entry_regs = session.read_regs() + snapshot = session.snapshot(ENTRY + BIAS, 0x78, name="bootstrap") + stops: list[dict] = [] + + def record(label): + def callback(hit): + regs = hit.regs + stops.append( + { + "label": label, + "pc": hit.pc, + "pc_fw": _fw(hit.pc), + "t0": regs.get("t0", 0), + "t1": regs.get("t1", 0), + "t2": regs.get("t2", 0), + } + ) + return callback + + session.hook(FIRST_TOUCH_PC + BIAS, on_enter=record("first_touch_pc")) + session.hook(HALT_PC + BIAS, on_enter=record("halt")) + session.run(duration=0.05, max_stops=2) + wall = time.monotonic() - start + machine = session.target.machine + halt_pc = session.pc + disassembly = read_disasm(session, ENTRY + BIAS, 8) + session.note( + "re_demo_lenient", + first_unknown_mmio=machine.mmio_log[0] if machine.mmio_log else None, + stop_pc_fw=_fw(halt_pc), + instructions=machine.insns, + hook_stops=stops, + ) + session.close() + trace_pcs = _trace_exec_pcs(session_dir) + first_touch = machine.mmio_log[0] if machine.mmio_log else None + return { + "session": _relpath(session_dir), + "session_hashes": session_hashes(session_dir), + "wall_s": round(wall, 4), + "entry_regs_zero": all(value == 0 for value in entry_regs.values()), + "snapshot_bytes": len(snapshot), + "first_unknown_mmio": first_touch, + "first_touch_fw_pc": _fw(first_touch["pc"]) if first_touch else None, + "first_touch_matches_qemu": bool( + first_touch and first_touch["addr"] == QEMU_REFERENCE["access_addr"] + ), + "hook_stops": stops, + "instructions": machine.insns, + "stop_pc": halt_pc, + "stop_pc_fw": _fw(halt_pc), + "trace_exec_count": len(trace_pcs), + "trace_path_fw": [_fw(pc) for pc in trace_pcs], + "disassembly": disassembly, + } + + +def run_modeled_session(image: str, out_dir: str) -> dict: + """Session B: serve the chip-ID register and follow the Turing dispatch.""" + dm = DeviceMap( + meta={ + "source": "re-demo-modeled-chip-id", + "bias": BIAS, + "note": ( + "exploratory assumption: register value carries the chip-ID field " + f"{TU10X_IDS[0]} in bits[28:20]; not observed (QEMU faults before reading)" + ), + } + ) + dm.add( + DeviceEntry( + address=QEMU_REFERENCE["access_addr"], + width=4, + kind="r", + values=(MODELED_ID,), + hits=1, + first_epc=FIRST_TOUCH_PC + BIAS, + first_epc_fw=FIRST_TOUCH_PC, + source="manual", + notes="assumed Turing TU10x chip ID", + ) + ) + session, session_dir = _launch( + image, + out_dir, + "native-modeled-id", + dm, + {"exec": True, "start": 0x84000000, "end": 0x84010000, "max_events": 1024}, + ) + start = time.monotonic() + jumps: list[dict] = [] + + def on_jr(hit): + jumps.append( + { + "pc_fw": _fw(hit.pc), + "t0_absolute_target": hit.regs.get("t0", 0), + "target_fw": _fw(hit.regs.get("t0", 0)), + } + ) + + stops: list[dict] = [] + session.on_event( + "stop", + lambda hit: stops.append( + { + "pc_fw": _fw(hit.pc), + "signal": hit.stop.signal if hit.stop is not None else None, + } + ), + ) + session.hook(TU10X_ABS_TARGET - 4 + BIAS, on_enter=on_jr) + session.run(duration=0.05, max_stops=1) + after_jump_pc = session.pc + session.run(duration=0.05, max_stops=1) + stop_pc = session.pc + fault_stop = stops[-1] if stops else None + wall = time.monotonic() - start + machine = session.target.machine + tail_disasm = read_disasm(session, TU10X_ABS_TARGET - 0x14 + BIAS, 5) + session.note( + "re_demo_modeled", + jumps=jumps, + after_jump_pc_fw=_fw(after_jump_pc), + stop_pc_fw=_fw(stop_pc), + instructions=machine.insns, + fault_stops=stops, + ) + session.close() + trace_pcs = _trace_exec_pcs(session_dir) + return { + "session": _relpath(session_dir), + "session_hashes": session_hashes(session_dir), + "wall_s": round(wall, 4), + "modeled_id": MODELED_ID, + "assumed_chip_id": TU10X_IDS[0], + "jumps": jumps, + "instructions": machine.insns, + "stop_pc": stop_pc, + "stop_pc_fw": _fw(stop_pc), + "stop_outside_ram": stop_pc < BIAS, + "fault_signal": fault_stop.get("signal") if fault_stop else None, + "fault_stops": stops, + "disassembly_tail": tail_disasm, + "trace_exec_count": len(trace_pcs), + "trace_path_fw": [_fw(pc) for pc in trace_pcs], + } + + +# -- answers / evaluation / rendering ---------------------------------------- + + +def build_answers(lenient: dict, modeled: dict) -> dict: + first = lenient.get("first_unknown_mmio") or {} + path = lenient.get("trace_path_fw") or [] + halt_seen = any( + stop.get("label") == "halt" for stop in (lenient.get("hook_stops") or []) + ) + answers = { + "q1_first_touch": { + "answer": ( + f"the first MMIO is a 4-byte read of PA {first.get('addr'):#x} at fw " + f"PC {lenient.get('first_touch_fw_pc'):#x}; it is the QEMU-documented " + "first fault" + ), + "addr": first.get("addr"), + "size": first.get("size"), + "rw": first.get("rw"), + "pc_fw": lenient.get("first_touch_fw_pc"), + "matches_qemu_reference": lenient.get("first_touch_matches_qemu"), + }, + "q2_path": { + "answer": ( + f"entry state is all-zero (no bootloader handoff modelled); the " + f"bootstrap builds the device PA with 5 arithmetic instructions " + f"(li/slli/addi/slli/addi), reads it, then runs the chip-ID " + f"comparison chain: {lenient.get('trace_exec_count')} instructions " + f"recorded from entry until the halt at fw " + f"{lenient.get('stop_pc_fw'):#x}" + ), + "entry_regs_zero": lenient.get("entry_regs_zero"), + "instructions_to_halt": lenient.get("instructions"), + "halt_pc_fw": lenient.get("stop_pc_fw"), + "halt_seen": halt_seen, + "trace_path_fw": path, + }, + "q3_id_field": { + "answer": ( + "the read value's bits [28:20] are extracted (srli 20, andi 0x1ff) " + f"and compared against {TU10X_IDS[0]}..{TU10X_IDS[-1]} (Turing) and " + f"{GA100_ID} (Ampere); the image itself defines this as a chip-ID " + "field" + ), + "shift": 20, + "mask": 0x1FF, + "turing_ids": list(TU10X_IDS), + "ampere_id": GA100_ID, + "disassembly": lenient.get("disassembly"), + }, + "q4_dispatch": { + "answer": ( + f"ID {TU10X_IDS[0]}..{TU10X_IDS[-1]} dispatches to .tu10x_text at " + f"{TU10X_ENTRY:#x}; ID {GA100_ID} dispatches to .ga100_text at " + f"{GA100_ENTRY:#x}; any other value (including the lenient zero) " + f"falls to the fail-closed `j .` loop at {HALT_PC:#x}" + ), + "tu10x_target": TU10X_ENTRY, + "ga100_target": GA100_ENTRY, + "halt_pc": HALT_PC, + "sections": None, # filled by the caller (needs the image path) + }, + "q5_next_blocker": { + "answer": ( + "with a Turing ID served, .tu10x_text programs CSRs 0x7c8-0x7cc " + "(region descriptors) and enables mstatus FP/SIMD state, then " + f"builds the absolute link-time address {TU10X_ABS_TARGET:#x} and " + "jumps to it; because the loader relocates segments but not " + "absolute immediates, that jump lands outside RAM and faults - a " + "relocation artifact, so the next rung needs a chip-ID device " + "model plus load-at-link-address/alias translation" + ), + "modeled_run": { + "jumps": modeled.get("jumps"), + "instructions": modeled.get("instructions"), + "stop_pc_fw": modeled.get("stop_pc_fw"), + "stop_outside_ram": modeled.get("stop_outside_ram"), + "fault_signal": modeled.get("fault_signal"), + }, + "disassembly_tail": modeled.get("disassembly_tail"), + }, + } + return answers + + +def evaluate( + *, + answers: dict, + lenient: dict, + repeat: dict, + emulator_wall_s: float, + baseline_total_s: float, + thresholds: dict, +) -> dict: + keys = ("instructions", "stop_pc_fw", "first_unknown_mmio", "trace_exec_count") + identical = all(lenient.get(key) == repeat.get(key) for key in keys) + checks = { + "answers_complete": all( + isinstance(answer, dict) and bool(answer.get("answer")) + for answer in answers.values() + ) + and len(answers) >= int(thresholds["answers_required"]), + "deterministic_lenient_runs": identical, + "emulator_wall_s": emulator_wall_s <= float(thresholds["emulator_wall_s_max"]), + "milestone_past_qemu": bool( + lenient.get("first_touch_matches_qemu") + and lenient.get("stop_pc_fw") == HALT_PC + and (repeat.get("stop_pc_fw") == HALT_PC) + ), + "baseline_beat": emulator_wall_s < baseline_total_s, + } + checks["pass"] = all(checks.values()) + return checks + + +def render_markdown(report: dict) -> str: + answers = report["answers"] + baseline = report["baseline"] + emulator = report["emulator"] + lenient = emulator["lenient"] + modeled = emulator["modeled"] + checks = report["threshold_checks"] + first = lenient.get("first_unknown_mmio") or {} + lines = [ + "# Real RE demo - what the GSP does between entry and the first fault", + "", + f"* schema: `{report['schema']}`", + f"* script: `{report['script']}` (stdlib-only, no hardware/guest/module)", + f"* firmware: `{report['firmware']['name']}` sha256 " + f"`{report['firmware']['sha256'][:16]}...`", + f"* verdict: **{'PASS' if checks.get('pass') else 'FAIL'}**", + "", + "## Scenario", + "", + f"> {report['scenario']}", + "", + "## Answer", + "", + ] + for index in sorted(answers): + lines.append(f"**{index}: {answers[index]['answer']}**") + lines.append("") + lines += [ + "## Baseline vs emulator (measured)", + "", + "| side | method | wall s | what it answers |", + "|---|---|---:|---|", + f"| baseline | `{baseline['qemu_method']}` | {baseline['qemu_probe_wall_s']} | first fault only |", + f"| baseline | manual static reading (estimate, {MANUAL_METHOD_NOTE}) | " + f"{baseline['manual_estimate_s']} | entry->fault path, by hand |", + f"| baseline | total (conservative) | {baseline['total_s']} | full question |", + f"| emulator | three recorded native sessions | {emulator['wall_s']} | full question + evidence |", + "", + "Per-session `wall_s` in the milestone table is the recorded run only; the " + "emulator total includes session launch, repeat and report writing.", + "", + "## Rung / milestone table", + "", + "| run | first touch | stop/milestone | instructions | wall s | evidence |", + "|---|---|---|---:|---:|---|", + f"| QEMU probe | {QEMU_REFERENCE['access_addr']:#x} (fault) | fault, then mtvec=0 loop | - | " + f"{baseline['qemu_probe_wall_s']} | `{baseline['qemu_session']}` |", + f"| native lenient (B0) | {first.get('addr'):#x} | halt loop fw " + f"{lenient['stop_pc_fw']:#x} | {lenient['instructions']} | {lenient['wall_s']} | " + f"`{lenient['session']}` |", + f"| native modeled chip-ID (B1 candidate) | {first.get('addr'):#x} | instruction " + f"fault at fw {modeled['stop_pc_fw']:#x} (outside RAM) | {modeled['instructions']} | " + f"{modeled['wall_s']} | `{modeled['session']}` |", + "", + "## Frozen thresholds", + "", + "| check | pass |", + "|---|---|", + ] + for name, value in checks.items(): + lines.append(f"| {name} | {value} |") + lines += [ + "", + f"Thresholds frozen {report['thresholds']['frozen_on']} before the run: " + f"emulator wall <= {report['thresholds']['emulator_wall_s_max']} s; " + f"{report['thresholds']['identical_lenient_runs']} identical lenient runs; " + f"all {report['thresholds']['answers_required']} answers present; milestone past " + "the QEMU point. Failure signal: any check false -> the run is a failed " + "experiment and the feature is not claimed.", + "", + "## Sessions (evidence)", + "", + "| session | file | sha256 |", + "|---|---|---|", + ] + for label, session in ( + ("qemu-probe", baseline.get("qemu_session_hashes")), + ("native-lenient", lenient.get("session_hashes")), + ("native-lenient-repeat", report.get("lenient_repeat_hashes")), + ("native-modeled-id", modeled.get("session_hashes")), + ): + if not session: + continue + for name, digest in sorted(session.items()): + lines.append(f"| {label} | {name} | `{digest}` |") + lines += [ + "", + "## Disassembly of the entry prologue (from the lenient session memory read)", + "", + "```", + *lenient.get("disassembly", []), + "```", + "", + "### Turing dispatch tail (from the modeled session memory read)", + "", + "```", + *(modeled.get("disassembly_tail") or []), + "```", + "", + "## Not verified (honest limits)", + "", + ] + for item in report.get("not_verified") or []: + lines.append(f"* {item}") + lines += [ + "", + "## Reproduce", + "", + "```bash", + f"cd {report['rerun']['cwd']}", + report["rerun"]["command"], + "```", + "", + "## Re-run contract", + "", + ] + for item in report.get("rerun_notes") or []: + lines.append(f"* {item}") + lines.append("") + return "\n".join(lines) + + +# -- main -------------------------------------------------------------------- + + +def run_demo(image: str = DEFAULT_IMAGE, out_dir: str = DEFAULT_OUT, + report_path: str = DEFAULT_REPORT, run_qemu: bool = True) -> dict: + os.makedirs(out_dir, exist_ok=True) + if not os.path.isfile(image): + return { + "schema": "greeneye-re-demo/1", + "ran": False, + "reason": f"firmware absent: {image}", + } + + baseline_start = time.monotonic() + qemu = ( + run_qemu_baseline(image=image, out_dir=out_dir, duration=0.5) + if run_qemu + else {"ran": False, "reason": "qemu disabled", "wall_s": 0.0} + ) + baseline_wall = time.monotonic() - baseline_start + baseline_total = ( + (qemu.get("wall_s") or 0.0 if qemu.get("ran") else baseline_wall) + + FROZEN_THRESHOLDS["manual_static_estimate_s"] + ) + + emulator_start = time.monotonic() + lenient = run_lenient_session(image, out_dir, tag="native-lenient") + repeat = run_lenient_session(image, out_dir, tag="native-lenient-repeat") + modeled = run_modeled_session(image, out_dir) + emulator_wall = time.monotonic() - emulator_start + + sections = elf_progbits_sections(image) + answers = build_answers(lenient, modeled) + answers["q4_dispatch"]["sections"] = { + "tu10x_target": section_for(sections, TU10X_ENTRY), + "ga100_target": section_for(sections, GA100_ENTRY), + "bootstrap": section_for(sections, ENTRY), + } + checks = evaluate( + answers=answers, + lenient=lenient, + repeat=repeat, + emulator_wall_s=emulator_wall, + baseline_total_s=baseline_total, + thresholds=FROZEN_THRESHOLDS, + ) + report = { + "schema": "greeneye-re-demo/1", + "ran": True, + "script": "harness/re_demo.py", + "scenario": ( + "What happens on the real GSP image between entry and the first " + "fault/milestone, and what does the state tell us about boot?" + ), + "firmware": {"name": os.path.basename(image), "sha256": _sha256_file(image)}, + "qemu_reference_first_fault": QEMU_REFERENCE, + "thresholds": FROZEN_THRESHOLDS, + "threshold_checks": checks, + "baseline": { + "qemu_method": "python3 -m greeneye.cli probe <image> --duration 1.0 (here run via harness.run_qemu_baseline, 0.5 s bound)", + "qemu_probe_wall_s": qemu.get("wall_s"), + "qemu_session": qemu.get("session"), + "qemu_session_hashes": qemu.get("session_hashes"), + "qemu_first_fault": qemu.get("first_fault"), + "manual_estimate_s": FROZEN_THRESHOLDS["manual_static_estimate_s"], + "manual_method": MANUAL_METHOD_NOTE, + "total_s": round(baseline_total, 4), + }, + "emulator": { + "wall_s": round(emulator_wall, 4), + "lenient": lenient, + "modeled": modeled, + }, + "lenient_repeat": repeat, + "lenient_repeat_hashes": repeat.get("session_hashes"), + "answers": answers, + "not_verified": [ + "the modeled chip-ID value is an assumption (the QEMU baseline faults before reading it); only the field position and the compared IDs are firmware evidence", + "no timing/cycle claim: cycles are modeled (L2 not modelled, docs/06) and none are used in this answer", + "L3 device environment is absent: the register's on-silicon semantics are inferred from firmware comparisons, not from a device model", + "load-at-link-address and absolute-immediate relocation behaviour are not implemented; the B1 candidate fault is a relocation artifact of this harness", + "watchpoints are not exercised on this path: the bootstrap performs no RAM data access before the fault (RAM watchpoints are exercised by harness/sim_qemu_line.py on the demo firmware)", + "no hardware, GPU, mdev, module, guest or network involvement by design", + ], + "rerun": { + "cwd": "open-iov/debug/risc-v-gsp-rm-vm", + "command": "python3 harness/re_demo.py", + }, + "rerun_notes": [ + "thresholds live in harness/re_demo.py (FROZEN_THRESHOLDS) and are not moved after the run", + "the QEMU baseline needs qemu-system-riscv64; without it the emulator columns still reproduce and the baseline is marked unverified", + "sessions are under harness/re_demo/sessions/ (hash-stamped); the report regenerates with the same command", + ], + } + json_path = os.path.join(out_dir, "result.json") + with open(json_path, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2, sort_keys=True) + fh.write("\n") + markdown = render_markdown(report) + os.makedirs(os.path.dirname(report_path) or ".", exist_ok=True) + with open(report_path, "w", encoding="utf-8") as fh: + fh.write(markdown) + report["result_json"] = _relpath(json_path) + report["report_md"] = _relpath(report_path) + return report + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="python3 harness/re_demo.py", + description=( + "U51 real RE demo: entry->first fault on the real GSP image, with " + "frozen thresholds and recorded sessions. No hardware." + ), + ) + parser.add_argument("--image", default=DEFAULT_IMAGE) + parser.add_argument("--out", default=DEFAULT_OUT) + parser.add_argument("--report", default=DEFAULT_REPORT) + parser.add_argument("--no-qemu", action="store_true") + parser.add_argument("--json", action="store_true") + args = parser.parse_args(argv) + + report = run_demo( + image=args.image, out_dir=args.out, report_path=args.report, run_qemu=not args.no_qemu + ) + if not report.get("ran"): + print(f"[re-demo] skipped: {report.get('reason')}", file=sys.stderr) + return 2 + checks = report["threshold_checks"] + if args.json: + print(json.dumps(report, indent=1, sort_keys=True, default=str)) + print(f"[re-demo] answers: {len(report['answers'])}") + print(f"[re-demo] emulator wall: {report['emulator']['wall_s']} s") + print(f"[re-demo] baseline (qemu+manual): {report['baseline']['total_s']} s") + print(f"[re-demo] verdict: {'PASS' if checks.get('pass') else 'FAIL'} ({checks})") + print(f"[re-demo] report: {report.get('report_md')} + {report.get('result_json')}") + return 0 if checks.get("pass") else 1 + + +if __name__ == "__main__": + sys.exit(main()) |