1 files changed, 428 insertions, 0 deletions
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())