diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/package_sim.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/package_sim.py | 1139 |
1 files changed, 1139 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/package_sim.py b/debug/risc-v-gsp-rm-vm/harness/package_sim.py new file mode 100644 index 0000000..28f54c4 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/package_sim.py @@ -0,0 +1,1139 @@ +#!/usr/bin/env python3 +"""Package the native green-eye simulator together with injected code. + +Builds a self-contained trial directory (or deterministic ``.tar.gz``): + + README.md what this is, how to run, what to expect, safety + run.sh single entry: native simulator + scenario, writes results + manifest.json format/version, hashes, component list, run contract + payload/ target image/ELF, service blob, device map, preinject config + scenario.py the pre-scripted scenario as run (builtin or --script) + results/ produced by run.sh (results.json, results.md, session) + +The trial boots the target natively, loads/registers the injected service, +invokes it directly and through the firmware dispatcher, runs one hook/watch +pass, and writes machine-readable and human-readable results. ``--pack-image`` +additionally emits a combined ELF (demo flattened plus the service at a fixed +offset) and a preinject config, so the trial starts with the injected code +already part of the simulated system. + +Honesty: the real GSP ``.fwimage`` is packaged with a lenient device map but +cannot boot on the native simulator yet (H2.2 device model absent); its trial +records ``status=blocked`` and a bounded first-touch probe, never a boot. + +Everything is stdlib-only; builds are deterministic apart from the manifest +``created`` field (set ``SOURCE_DATE_EPOCH`` to make even that stable). +""" +from __future__ import annotations + +import argparse +import gzip +import hashlib +import json +import os +import shutil +import struct +import subprocess +import sys +import tarfile +import time +from dataclasses import dataclass + +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 import __version__ as GREENEYE_VERSION # noqa: E402 +from greeneye.elf import EM_RISCV, load_elf # noqa: E402 +from greeneye.errors import GreenEyeError # noqa: E402 +from greeneye.inject import ENTRY_SYMBOL # noqa: E402 + +FORMAT = "greeneye-trial/1" +RESULTS_SCHEMA = "greeneye-trial-results/1" +PREINJECT_SCHEMA = "greeneye-preinject/1" +DEVICE_MAP_SCHEMA = "greeneye-device-map/1" + +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +DEFAULT_SERVICE = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf") +REPO_ROOT = os.path.dirname(os.path.dirname(TOOL_ROOT)) +REAL_IMAGE = os.path.normpath( + os.path.join(REPO_ROOT, "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin") +) + +RAM_BASE = 0x80000000 +REAL_IMAGE_BIAS = 0x80000000 +#: Service placement mirrors greeneye.inject.load_service's default load address. +SERVICE_OFFSET = 0x01000000 +DEFAULT_SLOT = 6 +PING_XOR = 0xBEEF +IMAGE_BUDGET = 200_000 +DEMO_BOOT_SECONDS = 0.25 +DEMO_SETTLE_SECONDS = 0.25 +DEFAULT_RUN_TIMEOUT = 180.0 + + +# -- small deterministic helpers ------------------------------------------- + + +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 sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _portable(path: str) -> str: + """Manifest-friendly source name: repo-relative when possible, else basename.""" + absolute = os.path.abspath(path) + relative = os.path.relpath(absolute, TOOL_ROOT) + if relative.startswith(".."): + return os.path.basename(absolute) + return relative + + +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 _write_bytes(path: str, data: bytes) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as fh: + fh.write(data) + return path + + +def _write_text(path: str, text: str) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +def _write_json(path: str, payload, indent: int = 2) -> str: + return _write_text(path, json.dumps(payload, indent=indent, sort_keys=True) + "\n") + + +# -- combined demo+service ELF --------------------------------------------- + + +def _load_plan(image, bias: int) -> list[tuple[int, int, bytes]]: + """``(vaddr, p_flags, memsz-padded bytes)`` for every non-empty PT_LOAD.""" + plan = [] + for seg in image.load_segments: + data = bytearray(image.segment_bytes(seg)) + if len(data) < seg.p_memsz: + data += b"\x00" * (seg.p_memsz - len(data)) + if data: + plan.append((seg.p_vaddr + bias, seg.p_flags, bytes(data))) + return plan + + +def write_combined_elf( + demo_path: str, + service_path: str, + out_path: str, + *, + service_offset: int = SERVICE_OFFSET, + slot: int = DEFAULT_SLOT, +) -> dict: + """Merge a target ELF and a relocated service ELF into one runnable ELF64. + + The service is placed exactly as ``greeneye.inject.load_service`` would place + it (segments at ``service_load_addr + p_vaddr``, entry at + ``service_load_addr + e_entry``), so a pre-injected boot is address-for-address + equivalent to the load/register path. Returns the placement config. + """ + demo = load_elf(demo_path) + service = load_elf(service_path) + if not demo.is_riscv or not service.is_riscv: + raise GreenEyeError("combined image requires RISC-V ELF inputs") + target_load = demo.min_vaddr + service_load = target_load + service_offset + demo_plan = _load_plan(demo, 0) + service_plan = _load_plan(service, service_load) + if not demo_plan: + raise GreenEyeError(f"{demo_path}: no loadable segments") + + spans = sorted((vaddr, vaddr + len(data)) for vaddr, _flags, data in demo_plan + service_plan) + for (a1, b1), (a2, b2) in zip(spans, spans[1:]): + if a2 < b1: + raise GreenEyeError( + f"combined image overlap: [{a2:#x},{b2:#x}) collides with [{a1:#x},{b1:#x})" + ) + + segments = demo_plan + service_plan + phnum = len(segments) + ident = b"\x7fELF" + bytes([2, 1, 1, 0]) + b"\x00" * 8 # ELF64, LE, v1 + header = ident + struct.pack( + "<HHIQQQIHHHHHH", 2, EM_RISCV, 1, demo.entry, 64, 0, 0, 64, 56, phnum, 64, 0, 0 + ) + offset = 64 + 56 * phnum + phdrs = bytearray() + body = bytearray() + for vaddr, flags, data in segments: + phdrs += struct.pack( + "<IIQQQQQQ", 1, flags, offset + len(body), vaddr, vaddr, len(data), len(data), 0x1000 + ) + body += data + _write_bytes(out_path, header + bytes(phdrs) + bytes(body)) + + check = load_elf(out_path) + if check.is_riscv is False or check.entry != demo.entry or len(check.load_segments) != phnum: + raise GreenEyeError(f"combined ELF failed its own round-trip: {out_path}") + + info = { + "target_load_addr": target_load, + "target_entry": demo.entry, + "service_load_addr": service_load, + "service_offset": service_offset, + "service_entry": service.entry, + "entry_addr": service_load + service.entry, + "slot": slot, + "size": sum(len(data) for _v, _f, data in service_plan), + "segments": phnum, + } + return info + + +# -- deterministic tar ------------------------------------------------------ + + +def write_deterministic_tar(out_dir: str, tar_path: str) -> str: + """Write ``out_dir`` as a gzip tar with zero timestamps and sorted members. + + Member mtimes, uid/gid, uname/gname and the gzip mtime are fixed, so two + builds of identical content produce byte-identical archives. + """ + names = [] + for root, dirs, files in os.walk(out_dir): + dirs.sort() + for name in sorted(files): + names.append(os.path.relpath(os.path.join(root, name), out_dir)) + names.sort() + with open(tar_path, "wb") as raw: + with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as gz: + with tarfile.open(fileobj=gz, mode="w", format=tarfile.GNU_FORMAT) as tar: + for name in names: + full = os.path.join(out_dir, name) + info = tar.gettarinfo(full, arcname=name) + info.mtime = 0 + info.uid = info.gid = 0 + info.uname = info.gname = "" + info.mode = 0o755 if name == "run.sh" else 0o644 + with open(full, "rb") as fh: + tar.addfile(info, fh) + return tar_path + + +# -- request/build ---------------------------------------------------------- + +RUN_SH = """#!/bin/sh +# green-eye trial runner (generated by harness/package_sim.py; inspect before use). +# Usage: ./run.sh [scenario args] Environment: GREENEYE_HOME overrides the tool root. +set -eu +here=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) +if [ -n "${GREENEYE_HOME:-}" ]; then + root=$GREENEYE_HOME +else + root=$here + while [ "$root" != "/" ] && [ ! -f "$root/greeneye/__init__.py" ]; do + root=$(dirname "$root") + done + if [ ! -f "$root/greeneye/__init__.py" ]; then + echo "run.sh: green-eye tool root not found; set GREENEYE_HOME" >&2 + exit 3 + fi +fi +export PYTHONPATH="$root${PYTHONPATH:+:$PYTHONPATH}" +exec python3 "$here/scenario.py" --package "$here" "$@" +""" + + +@dataclass +class PackageRequest: + out: str + target: str = "demo" # demo | image + image: str | None = None + service: str | None = None + device_map: str | None = None + script: str | None = None + slot: int = DEFAULT_SLOT + pack_image: bool = False + tar: bool = False + + +def _readme_text( + *, + target: str, + target_name: str, + target_sha: str, + service_name: str, + service_sha: str, + slot: int, + device_map_name: str | None, + preinject: dict | None, +) -> str: + lines = [ + "# green-eye trial package", + "", + "Generated by `harness/package_sim.py` (`greeneye-trial/1`). This package", + "bundles a target, an injected service blob and a pre-scripted scenario", + "that runs on the deterministic native RV64 simulator in the green-eye", + "tool tree. It needs no QEMU, no GPU and no guest.", + "", + "## Contents", + "", + f"- target: `payload/{target_name}` (`{target}`), sha256 `{target_sha}`", + f"- service: `payload/{service_name}`, slot {slot}, sha256 `{service_sha}`", + f"- device map: `{('payload/' + device_map_name) if device_map_name else 'none'}`", + f"- pre-injected combined image: `{preinject['file'] if preinject else 'none'}`", + "- scenario: `scenario.py`; results are written to `results/`", + "", + "## How to run", + "", + "```sh", + "./run.sh", + "# image packages may bound the lenient probe explicitly:", + "./run.sh --budget 200000", + "```", + "", + "Requires Python 3.10+ and the green-eye tool tree. `run.sh` finds the", + "tree by walking up from this directory; set `GREENEYE_HOME=/path/to/risc-v-gsp-rm-vm`", + "if the package was moved. Exit code 0 means the assertions passed (or an", + "image package produced its documented blocked status), 1 means an", + "assertion failure, 3 means the tool tree was not found.", + "", + "## What to expect", + "", + ] + if target == "demo": + lines += [ + "The demo uproc boots natively, the service is loaded and registered in", + f"`gsp_ctrl.handlers[{slot}]`, then invoked directly and through mailbox", + "command 16 + slot. `results/results.json` should report `status=ok` with", + "matching call results, at least one `rpc_dispatch` hook hit, at least one", + "watch hit on `gsp_ctrl.result`, and the UART banner/self-test lines.", + ] + if preinject: + lines += [ + "", + "This package uses `--pack-image`: the combined ELF", + f"`{preinject['file']}` already contains the service at", + f"{preinject['service_load_addr']:#x}, so no separate load step runs.", + ] + else: + lines += [ + "**Honest status: the native simulator cannot boot the real GSP", + "`.fwimage` yet** (the H2.2 device model is absent). The scenario runs a", + "bounded lenient first-touch probe with the bundled device map and", + "records `status=blocked`, `blocked_on_device_model=true` in", + "`results/results.json`; it is not a boot and makes no hardware claim.", + "No RM service registration point exists yet (M5), so the service blob", + "is packaged but not loaded or registered.", + ] + if preinject: + lines += [ + "", + f"A combined artifact `{preinject['file']}` is present but marked", + "`runnable: false` for the same reason; it is not executed.", + ] + lines += [ + "", + "## How to read the results", + "", + "- `results/results.json` - machine-readable: status, call results, hook and", + " watch hits, control block, UART excerpt, session path, assertion list.", + "- `results/results.md` - the same rendered for humans.", + "- `results/session/` - the recorded greeneye session (`events.jsonl`,", + " `session.json`, snapshots) for `greeneye analyze`/`compare`.", + "- `manifest.json` - format/version, input hashes and the component list;", + " every component hash covers the file as packaged.", + "", + "## Safety", + "", + "Run only packages from sources you trust. Injected code executes inside", + "the simulator process with the simulator's full privileges. This package", + "contains no hardware access and must never be used to inject into a guest,", + "a shipped environment or real hardware.", + "", + "## Determinism", + "", + "Manifest component order and hashes are stable, and `--tar` archives use", + "zeroed member timestamps. The only variable field is `manifest.created`;", + "set `SOURCE_DATE_EPOCH` to freeze it and reproduce byte-identical builds.", + "", + ] + return "\n".join(lines) + + +def build_package(request: PackageRequest) -> dict: + """Build the trial package; returns the manifest that was written.""" + out = os.path.abspath(request.out) + if os.path.isdir(out) and os.listdir(out): + raise GreenEyeError(f"output directory is not empty: {out}") + if request.target not in ("demo", "image"): + raise GreenEyeError(f"target must be demo or image, got {request.target!r}") + if not 0 <= request.slot <= 7: + raise GreenEyeError(f"slot must be 0..7, got {request.slot}") + + if request.target == "demo": + target_input = os.path.abspath(request.image or DEMO_ELF) + else: + target_input = os.path.abspath(request.image or REAL_IMAGE) + if not os.path.isfile(target_input): + raise GreenEyeError(f"target not found: {target_input}") + service_input = os.path.abspath(request.service or DEFAULT_SERVICE) + if not os.path.isfile(service_input): + raise GreenEyeError(f"service blob not found: {service_input} (run: make sdk)") + + target_image = load_elf(target_input) + if not target_image.is_riscv: + raise GreenEyeError(f"{target_input}: not a RISC-V ELF image") + service_image = load_elf(service_input) + if not service_image.is_riscv: + raise GreenEyeError(f"{service_input}: not a RISC-V ELF image") + + os.makedirs(out, exist_ok=True) + payload_dir = os.path.join(out, "payload") + os.makedirs(payload_dir, exist_ok=True) + components: list[tuple[str, str]] = [] + + target_name = os.path.basename(target_input) + shutil.copyfile(target_input, os.path.join(payload_dir, target_name)) + components.append((f"payload/{target_name}", "target-image")) + + service_name = os.path.basename(service_input) + shutil.copyfile(service_input, os.path.join(payload_dir, service_name)) + components.append((f"payload/{service_name}", "service-blob")) + + device_map_info: dict | None = None + if request.device_map: + dm_input = os.path.abspath(request.device_map) + if not os.path.isfile(dm_input): + raise GreenEyeError(f"device map not found: {dm_input}") + dm_name = os.path.basename(dm_input) + shutil.copyfile(dm_input, os.path.join(payload_dir, dm_name)) + components.append((f"payload/{dm_name}", "device-map")) + device_map_info = { + "file": f"payload/{dm_name}", + "input": _portable(dm_input), + "sha256": sha256_file(dm_input), + "lenient": request.target == "image", + "learn": False, + } + elif request.target == "image": + dm_name = "device_map.json" + _write_json( + os.path.join(payload_dir, dm_name), + {"schema": DEVICE_MAP_SCHEMA, "meta": {"bias": REAL_IMAGE_BIAS, "source": "lenient"}, "entries": []}, + indent=1, + ) + components.append((f"payload/{dm_name}", "device-map")) + device_map_info = { + "file": f"payload/{dm_name}", + "input": None, + "sha256": sha256_file(os.path.join(payload_dir, dm_name)), + "lenient": True, + "learn": True, + } + + preinject: dict | None = None + if request.pack_image: + service_offset = SERVICE_OFFSET + if request.target == "image": + span = target_image.max_vaddr - target_image.min_vaddr + service_offset = max(SERVICE_OFFSET, ((span + 0xFFF) // 0x1000) * 0x1000) + combined_rel = "payload/combined.elf" + placement = write_combined_elf( + target_input, + service_input, + os.path.join(out, combined_rel), + service_offset=service_offset, + slot=request.slot, + ) + components.append((combined_rel, "combined-elf")) + runnable = request.target == "demo" + reason = ( + "" + if runnable + else "the native simulator cannot boot the real GSP image yet (H2.2 device model absent)" + ) + preinject = { + "file": combined_rel, + "schema": PREINJECT_SCHEMA, + "runnable": runnable, + "reason": reason, + **placement, + } + _write_json(os.path.join(payload_dir, "preinject.json"), preinject) + components.append(("payload/preinject.json", "preinject-config")) + + if request.script: + script_input = os.path.abspath(request.script) + if not os.path.isfile(script_input): + raise GreenEyeError(f"scenario script not found: {script_input}") + shutil.copyfile(script_input, os.path.join(out, "scenario.py")) + script_info = { + "kind": "custom", + "name": os.path.basename(script_input), + "input": _portable(script_input), + "sha256": sha256_file(script_input), + } + else: + _write_text(os.path.join(out, "scenario.py"), BUILTIN_SCENARIO) + script_info = { + "kind": "builtin", + "name": "builtin-demo-v1", + "input": None, + "sha256": sha256_bytes(BUILTIN_SCENARIO.encode("utf-8")), + } + components.append(("scenario.py", "scenario")) + + run_sh_path = os.path.join(out, "run.sh") + _write_text(run_sh_path, RUN_SH) + os.chmod(run_sh_path, 0o755) + components.append(("run.sh", "runner")) + + _write_text(os.path.join(out, "results", ".gitkeep"), "") + components.append(("results/.gitkeep", "results-placeholder")) + + if request.target == "image": + if preinject and preinject.get("runnable"): + end = max( + target_image.max_vaddr + REAL_IMAGE_BIAS, + preinject["service_load_addr"] + preinject["size"], + ) + else: + end = target_image.max_vaddr + REAL_IMAGE_BIAS + need = end - RAM_BASE + (1 << 20) + memory_mb = max(256, (need + (1 << 20) - 1) // (1 << 20)) + limitations = [ + "The native simulator cannot boot the real GSP .fwimage yet: the H2.2 device model is absent.", + "The run is a bounded lenient first-touch probe, not a boot; results.json records status=blocked.", + "No RM service registration point exists yet (M5): the blob is packaged but not loaded or registered.", + "No L2 timing/concurrency and no L3 device fidelity is claimed (docs/06-ACCURACY.md).", + ] + else: + memory_mb = 256 + limitations = [ + "The demo uproc is a stand-in for GSP-RM, not the real device; results describe the injection path only.", + "Injected code runs with the simulator's full in-process privileges; only run trusted packages.", + ] + + _write_text( + os.path.join(out, "README.md"), + _readme_text( + target=request.target, + target_name=target_name, + target_sha=sha256_file(target_input), + service_name=service_name, + service_sha=sha256_file(service_input), + slot=request.slot, + device_map_name=device_map_info["file"].split("/", 1)[1] if device_map_info else None, + preinject=preinject, + ), + ) + components.append(("README.md", "documentation")) + + manifest = { + "format": FORMAT, + "greeneye_version": GREENEYE_VERSION, + "created": _created_iso(), + "results_schema": RESULTS_SCHEMA, + "target": { + "kind": request.target, + "file": f"payload/{target_name}", + "input": _portable(target_input), + "sha256": sha256_file(target_input), + "entry": target_image.entry, + "min_vaddr": target_image.min_vaddr, + "max_vaddr": target_image.max_vaddr, + "bias": REAL_IMAGE_BIAS if request.target == "image" else 0, + "runnable": request.target == "demo", + "limitations": limitations, + }, + "service": { + "file": f"payload/{service_name}", + "input": _portable(service_input), + "sha256": sha256_file(service_input), + "entry": service_image.entry, + "entry_symbol": ENTRY_SYMBOL, + "slot": request.slot, + }, + "device_map": device_map_info, + "script": script_info, + "preinject": preinject, + "scenario": { + "kind": script_info["kind"], + "demo_boot_seconds": DEMO_BOOT_SECONDS, + "demo_settle_seconds": DEMO_SETTLE_SECONDS, + "image_budget_instructions": IMAGE_BUDGET, + "memory_mb": memory_mb, + "expected": { + "cmd_ping": 1, + "cmd_sum": 2, + "ping_arg": 0x41, + "sum_args": [0x10, 0x20], + "ping_result": 0x41 ^ PING_XOR, + "sum_result": 0x30, + }, + }, + "run": { + "command": "./run.sh", + "results": ["results/results.json", "results/results.md"], + "exit_nonzero_on": "assertion failure", + }, + "safety": ( + "Local, development-only study tooling: no hardware access, no guest, " + "never shipped. Run only packages from trusted sources." + ), + "determinism": ( + "Component order and hashes are stable; --tar members use mtime 0. " + "manifest.created is the only timestamp (SOURCE_DATE_EPOCH freezes it)." + ), + "components": [], + } + for relpath, role in sorted(components): + full = os.path.join(out, relpath) + manifest["components"].append( + { + "path": relpath, + "role": role, + "bytes": os.path.getsize(full), + "sha256": sha256_file(full), + } + ) + + _write_json(os.path.join(out, "manifest.json"), manifest) + if request.tar: + write_deterministic_tar(out, out + ".tar.gz") + return manifest + + +def run_package(out: str, *, timeout: float = DEFAULT_RUN_TIMEOUT, args: list[str] | None = None) -> int: + """Run ``run.sh`` in a bounded subprocess; returns its exit code.""" + run_sh = os.path.join(os.path.abspath(out), "run.sh") + if not os.path.isfile(run_sh): + raise GreenEyeError(f"no run.sh in {out}; build the package first") + environment = dict(os.environ) + environment.setdefault("GREENEYE_HOME", TOOL_ROOT) + try: + proc = subprocess.run( + ["/bin/sh", run_sh, *(args or [])], + cwd=os.path.abspath(out), + env=environment, + capture_output=True, + text=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as exc: + print(f"[package-sim] run.sh timed out after {timeout}s", file=sys.stderr) + print(exc.stdout or "", end="") + print(exc.stderr or "", end="", file=sys.stderr) + return 124 + print(proc.stdout, end="") + print(proc.stderr, end="", file=sys.stderr) + return proc.returncode + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="python3 harness/package_sim.py", + description="Build a self-contained green-eye trial package (simulator + injected service).", + ) + parser.add_argument("--target", choices=("demo", "image"), default="demo", + help="demo uproc (default) or a real .fwimage") + parser.add_argument("--image", help="target ELF/.fwimage path (image target; overrides the demo ELF)") + parser.add_argument("--service", help=f"service blob ELF (default: {os.path.relpath(DEFAULT_SERVICE, TOOL_ROOT)})") + parser.add_argument("--device-map", help="greeneye-device-map/1 JSON (lenient MMIO evidence)") + parser.add_argument("--script", help="custom scenario script (default: builtin demo scenario)") + parser.add_argument("--out", required=True, help="output directory (must be empty)") + parser.add_argument("--slot", type=int, default=DEFAULT_SLOT, help="handler slot 0..7 (default 6)") + parser.add_argument("--pack-image", action="store_true", + help="also emit payload/combined.elf + preinject config") + parser.add_argument("--tar", action="store_true", help="write a deterministic <out>.tar.gz") + parser.add_argument("--run", action="store_true", help="run the packaged trial after building") + parser.add_argument("--run-timeout", type=float, default=DEFAULT_RUN_TIMEOUT, help="run.sh timeout seconds") + args = parser.parse_args(argv) + + request = PackageRequest( + out=args.out, + target=args.target, + image=args.image, + service=args.service, + device_map=args.device_map, + script=args.script, + slot=args.slot, + pack_image=args.pack_image, + tar=args.tar, + ) + try: + manifest = build_package(request) + except GreenEyeError as exc: + print(f"[package-sim] error: {exc}", file=sys.stderr) + return 1 + print( + f"[package-sim] built {args.out}: {len(manifest['components'])} components, " + f"target={manifest['target']['kind']}" + ) + if args.tar: + print(f"[package-sim] archive {args.out}.tar.gz") + if manifest["target"]["runnable"] is False: + print("[package-sim] note: packaged target cannot boot yet; results will report status=blocked") + if manifest.get("preinject"): + print("[package-sim] note: preinject/unused combined artifact is informational only") + if args.run: + return run_package(args.out, timeout=args.run_timeout) + return 0 + + +# -- generated scenario ----------------------------------------------------- + +BUILTIN_SCENARIO = r'''#!/usr/bin/env python3 +"""Pre-scripted green-eye trial scenario (generated by harness/package_sim.py). + +Boots the packaged target on the native simulator, loads and registers the +injected service, invokes it directly and through the firmware mailbox, runs +one hook/watch pass, and writes results/results.json + results/results.md. +Exit 0 = assertions passed (or the documented blocked image status), 1 = a +verification assertion failed. +""" +from __future__ import annotations + +import argparse +import json +import os +import sys + +CTRL_HANDLERS = 0x150 +CTRL_RESULT = 0x40 +CTRL_RPC_COUNT = 0x10 +PING_ARG = 0x41 +SUM_A0 = 0x10 +SUM_A1 = 0x20 +EXPECTED_PING = PING_ARG ^ 0xBEEF +EXPECTED_SUM = SUM_A0 + SUM_A1 +UART_BANNER = "GREEN-EYE UPROC DEMO v1" +UART_SELFTEST = "SELFTEST DONE" + + +def bootstrap_tool_root(pkg): + """Make `import greeneye` work when scenario.py is run directly.""" + try: + import greeneye # noqa: F401 + return + except ImportError: + pass + candidates = [] + if os.environ.get("GREENEYE_HOME"): + candidates.append(os.path.abspath(os.environ["GREENEYE_HOME"])) + here = pkg + while True: + candidates.append(here) + parent = os.path.dirname(here) + if parent == here: + break + here = parent + for candidate in candidates: + if os.path.isfile(os.path.join(candidate, "greeneye", "__init__.py")): + sys.path.insert(0, candidate) + return + + +def load_json(path): + with open(path, "r", encoding="utf-8") as fh: + return json.load(fh) + + +def write_json(path, payload): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + fh.write("\n") + + +def write_text(path, text): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + + +def component(manifest, role): + for comp in manifest.get("components", ()): + if comp.get("role") == role: + return comp["path"] + return None + + +def uart_excerpt(text, first=4, last=6, cap=14): + lines = [line for line in text.splitlines() if line.strip()] + if len(lines) > cap: + return lines[:first] + ["..."] + lines[-last:] + return lines + + +def hexv(value): + return f"0x{int(value):x}" + + +def render_md(manifest, data): + target = data.get("target", {}) + lines = ["# green-eye trial results", ""] + lines.append(f"- status: **{data.get('status')}**") + lines.append(f"- target: `{target.get('file', '?')}` ({target.get('kind', '?')})") + service = data.get("service") + if service: + lines.append( + f"- service: `{service.get('file', '?')}` slot {service.get('slot', '?')} " + f"pre-injected {'yes' if service.get('preinjected') else 'no'}" + ) + lines.append(f"- backend: {data.get('backend', 'native')}") + if data.get("error"): + lines += ["", "## Error", "", "```", str(data["error"]), "```"] + if data.get("status") == "blocked": + lines += [ + "", + "## Honest status", + "", + str(data.get("blocked_reason", "the native simulator cannot boot this target yet")), + "", + f"- instructions executed: {data.get('instructions_executed')}", + f"- stopped: pc {hexv(data.get('stopped', {}).get('pc', 0))} " + f"(firmware {hexv(data.get('stopped', {}).get('fw_pc', 0))})", + f"- unmodelled MMIO accesses served leniently: {data.get('mmio_unknown')}", + ] + if data.get("calls"): + calls = data["calls"] + expected = calls.get("expected", {}) + lines += [ + "", + "## Injected call results", + "", + "| route | cmd | result | expected | ok |", + "|---|---|---|---|---|", + ] + for route, pair in (("remote (direct)", "ping"), ("remote (direct)", "sum"), + ("via firmware mailbox", "ping"), ("via firmware mailbox", "sum")): + key = "remote" if route.startswith("remote") else "via_firmware" + actual = calls.get(key, {}).get(pair) + want = expected.get(pair) + lines.append(f"| {route} | {pair} | {hexv(actual)} | {hexv(want)} | " + f"{'yes' if actual == want else 'no'} |") + lines.append("") + lines.append(f"routes agree: {'yes' if calls.get('agree') else 'no'}") + if data.get("hooks"): + hooks = data["hooks"]["rpc_dispatch"] + lines += ["", "## Hook / watch pass", "", + f"- `rpc_dispatch` hook hits: {hooks['hits']}"] + for hit in hooks.get("first", [])[:4]: + lines.append(f" - pc {hit['pc']} cmd {hit['cmd']} a0 {hit['a0']} a1 {hit['a1']}") + if data.get("watches"): + watch = data["watches"]["gsp_ctrl.result"] + lines.append(f"- `gsp_ctrl.result` watch hits: {watch['hits']}") + for hit in watch.get("first", [])[:4]: + lines.append(f" - pc {hit['pc']} ({hit.get('symbol')})") + if data.get("control"): + control = data["control"] + lines.append( + f"- control block: result {hexv(control['result'])}, rpc_count {control['rpc_count']}" + ) + lines.append(f"- handlers slot value: {hexv(data.get('handlers_slot', 0))}") + if data.get("uart_excerpt"): + lines += ["", "## UART excerpt", "", "```"] + lines += list(data["uart_excerpt"]) + lines += ["```"] + lines += ["", "## Session", "", f"`{data.get('session', 'results/session')}`"] + if data.get("learned_device_map"): + lines.append(f"`{data['learned_device_map']}`") + assertions = data.get("assertions", {}) + lines += ["", "## Assertions", ""] + for name in sorted(assertions.get("checks", {})): + lines.append(f"- [{'x' if assertions['checks'][name] else ' '}] {name}") + failed = assertions.get("failed", []) + lines.append("") + lines.append("all checks passed" if not failed else f"FAILED: {', '.join(failed)}") + if data.get("limitations"): + lines += ["", "## Limits", ""] + lines += [f"- {item}" for item in data["limitations"]] + lines.append("") + return "\n".join(lines) + + +def run_demo(pkg, manifest): + from greeneye.elf import load_elf + from greeneye.inject import InjectedService, invoke_remote, invoke_via_firmware, load_service + from greeneye.sim_backend import launch_native + + results_dir = os.path.join(pkg, "results") + target_rel = component(manifest, "target-image") + service_rel = component(manifest, "service-blob") + dm_rel = component(manifest, "device-map") + combined_rel = component(manifest, "combined-elf") + preinject_rel = component(manifest, "preinject-config") + target_path = os.path.join(pkg, target_rel) + service_path = os.path.join(pkg, service_rel) + device_map_path = os.path.join(pkg, dm_rel) if dm_rel else None + preinject = load_json(os.path.join(pkg, preinject_rel)) if preinject_rel else None + preinjected = bool(preinject and preinject.get("runnable", True)) + kernel = os.path.join(pkg, combined_rel) if preinjected else target_path + + dispatch_total = [0] + watch_total = [0] + dispatch_hits = [] + watch_hits = [] + data = { + "schema": "greeneye-trial-results/1", + "backend": "native", + "target": {"kind": "demo", "file": target_rel, "sha256": manifest["target"]["sha256"]}, + "session": "results/session", + "service": { + "file": service_rel, + "sha256": manifest["service"]["sha256"], + "slot": manifest["service"]["slot"], + "preinjected": preinjected, + }, + } + checks = {} + session = None + try: + session = launch_native( + kernel=kernel, + record=os.path.join(results_dir, "session"), + name="trial-demo", + memory_mb=int(manifest["scenario"].get("memory_mb", 256)), + device_map=device_map_path, + unknown_mmio="lenient" if device_map_path else "fault", + ) + if preinjected: + session.symbols.load_elf(load_elf(target_path)) + ctrl = session.symbols.lookup("gsp_ctrl") + if preinjected: + session.memory.write_u64( + ctrl + CTRL_HANDLERS + int(preinject["slot"]) * 8, int(preinject["entry_addr"]) + ) + service = InjectedService( + path=service_rel, + name=os.path.basename(service_rel), + load_addr=int(preinject["service_load_addr"]), + entry_addr=int(preinject["entry_addr"]), + dispatch_offset=int(preinject["service_entry"]), + slot=int(preinject["slot"]), + size=int(preinject["size"]), + ) + session.recorder.event( + "service_injected", + path=service.path, + name=service.name, + load_addr=service.load_addr, + entry_addr=service.entry_addr, + slot=service.slot, + size=service.size, + preinjected=True, + ) + else: + service = load_service(session, service_path, slot=int(manifest["service"]["slot"])) + + def on_dispatch(hit): + dispatch_total[0] += 1 + if len(dispatch_hits) < 64: + dispatch_hits.append( + { + "pc": hexv(hit.pc), + "cmd": hexv(hit.arg(0)), + "a0": hexv(hit.arg(1)), + "a1": hexv(hit.arg(2)), + } + ) + + def on_result(hit): + watch_total[0] += 1 + if len(watch_hits) < 64: + watch_hits.append( + {"pc": hexv(hit.pc), "symbol": hit.symbol.name if hit.symbol else None} + ) + + session.hook("rpc_dispatch", on_enter=on_dispatch) + session.watch(ctrl + CTRL_RESULT, 8, "w", callback=on_result) + session.run(duration=float(manifest["scenario"]["demo_boot_seconds"])) + + remote_ping = invoke_remote(session, service, 1, PING_ARG) + remote_sum = invoke_remote(session, service, 2, SUM_A0, SUM_A1) + fw_ping = invoke_via_firmware(session, service, 1, PING_ARG) + fw_sum = invoke_via_firmware(session, service, 2, SUM_A0, SUM_A1) + + handlers = session.memory.read_u64(ctrl + CTRL_HANDLERS + service.slot * 8) + result_value = session.memory.read_u64(ctrl + CTRL_RESULT) + rpc_count = session.memory.read_u64(ctrl + CTRL_RPC_COUNT) + uart = session.target.read_uart() + + data["service"].update( + {"load_addr": hexv(service.load_addr), "entry_addr": hexv(service.entry_addr)} + ) + data["calls"] = { + "remote": {"ping": remote_ping, "sum": remote_sum}, + "via_firmware": {"ping": fw_ping, "sum": fw_sum}, + "expected": {"ping": EXPECTED_PING, "sum": EXPECTED_SUM}, + "agree": remote_ping == fw_ping and remote_sum == fw_sum, + } + data["handlers_slot"] = handlers + data["control"] = {"result": result_value, "rpc_count": rpc_count} + data["hooks"] = {"rpc_dispatch": {"hits": dispatch_total[0], "first": dispatch_hits}} + data["watches"] = {"gsp_ctrl.result": {"hits": watch_total[0], "first": watch_hits}} + data["uart_excerpt"] = uart_excerpt(uart) + checks = { + "handlers_slot_registered": handlers == service.entry_addr, + "remote_ping": remote_ping == EXPECTED_PING, + "remote_sum": remote_sum == EXPECTED_SUM, + "via_firmware_ping": fw_ping == EXPECTED_PING, + "via_firmware_sum": fw_sum == EXPECTED_SUM, + "routes_agree": remote_ping == fw_ping and remote_sum == fw_sum, + "dispatch_hook_hits": dispatch_total[0] >= 1, + "result_watch_hits": watch_total[0] >= 1, + "uart_banner": UART_BANNER in uart, + "uart_selftest": UART_SELFTEST in uart, + } + except Exception as exc: # record the honest failure, never fake a pass + data["error"] = f"{type(exc).__name__}: {exc}" + checks = {"no_exception": False} + finally: + if session is not None: + try: + session.close() + except Exception: + pass + + failures = sorted(name for name, ok in checks.items() if not ok) + data["status"] = "ok" if not failures else "failed" + data["success"] = not failures + data["assertions"] = {"checks": checks, "failed": failures} + data["limitations"] = manifest["target"].get("limitations", []) + write_json(os.path.join(results_dir, "results.json"), data) + write_text(os.path.join(results_dir, "results.md"), render_md(manifest, data)) + print(f"trial: {data['status']} (results/results.json)") + return 0 if not failures else 1 + + +def run_image(pkg, manifest, budget=None): + from greeneye.sim_backend import launch_native + + results_dir = os.path.join(pkg, "results") + target_rel = component(manifest, "target-image") + service_rel = component(manifest, "service-blob") + dm_rel = component(manifest, "device-map") + scenario = manifest["scenario"] + target_path = os.path.join(pkg, target_rel) + dm_path = os.path.join(pkg, dm_rel) if dm_rel else None + budget = int(budget if budget is not None else scenario["image_budget_instructions"]) + bias = int(manifest["target"].get("bias", 0x80000000)) + learn = bool(manifest.get("device_map") and manifest["device_map"].get("learn")) + + data = { + "schema": "greeneye-trial-results/1", + "backend": "native", + "target": { + "kind": "image", + "file": target_rel, + "sha256": manifest["target"]["sha256"], + "entry": manifest["target"]["entry"], + "bias": bias, + "runnable": False, + }, + "service": { + "file": service_rel, + "sha256": manifest["service"]["sha256"], + "loaded": False, + "reason": ( + "no RM service registration point exists yet (M5); loading the blob " + "would overwrite the image's own RAM" + ), + }, + "session": "results/session", + "budget_instructions": budget, + "preinject": manifest.get("preinject"), + "limitations": manifest["target"].get("limitations", []), + "blocked_reason": ( + "the native simulator cannot boot the real GSP image yet: the H2.2 device " + "model is absent. This is a bounded lenient first-touch probe, not a boot." + ), + } + checks = {"bounded_lenient_run": False} + session = None + try: + session = launch_native( + kernel=target_path, + memory_mb=int(scenario.get("memory_mb", 256)), + record=os.path.join(results_dir, "session"), + name="trial-image", + device_map=dm_path, + unknown_mmio="lenient", + learn_mmio=learn, + bias=bias, + ) + session.run(duration=budget / 1_000_000.0) + machine = session.target.machine + pc = machine.pc + data["instructions_executed"] = machine.insns + data["stopped"] = {"pc": pc, "fw_pc": pc - bias if pc >= bias else pc} + data["mmio_unknown"] = machine.mmio_unknown + data["first_unknown_mmio"] = [ + {"pc": hexv(entry["pc"]), "addr": hexv(entry["addr"]), + "size": entry["size"], "rw": entry["rw"]} + for entry in machine.mmio_log[:8] + ] + data["uart_excerpt"] = uart_excerpt(session.target.read_uart()) + if learn and machine.device_map is not None: + machine.device_map.save(os.path.join(results_dir, "learned_device_map.json")) + data["learned_device_map"] = "results/learned_device_map.json" + checks = {"bounded_lenient_run": True, "booted": False} + except Exception as exc: + data["error"] = f"{type(exc).__name__}: {exc}" + finally: + if session is not None: + try: + session.close() + except Exception: + pass + + data["status"] = "blocked" + data["success"] = False + data["blocked_on_device_model"] = True + data["assertions"] = {"checks": checks, "failed": []} + write_json(os.path.join(results_dir, "results.json"), data) + write_text(os.path.join(results_dir, "results.md"), render_md(manifest, data)) + print(f"trial: blocked (results/results.json): {data['blocked_reason']}") + return 0 if "error" not in data else 1 + + +def main(argv=None): + parser = argparse.ArgumentParser(description="Run a packaged green-eye trial.") + parser.add_argument("--package", default=os.path.dirname(os.path.abspath(__file__))) + parser.add_argument("--budget", type=int, default=None, + help="image probe budget in instructions (default from manifest)") + args = parser.parse_args(argv) + pkg = os.path.abspath(args.package) + bootstrap_tool_root(pkg) + manifest = load_json(os.path.join(pkg, "manifest.json")) + if manifest["target"]["kind"] == "image": + return run_image(pkg, manifest, budget=args.budget) + return run_demo(pkg, manifest) + + +if __name__ == "__main__": + raise SystemExit(main()) +''' + + +if __name__ == "__main__": + raise SystemExit(main()) |