summaryrefslogtreecommitdiffstats
path: root/debug/risc-v-gsp-rm-vm/harness/package_sim.py
diff options
context:
space:
mode:
authorArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-02 11:30:46 -0400
committerArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-02 11:30:46 -0400
commit4467eb66a4c8a776e35e9644960eba699dc85e6f (patch)
treee75a53992746a902ba035d7b4ae76a5a401150e9 /debug/risc-v-gsp-rm-vm/harness/package_sim.py
parentfaabc48f6d9b2d531b5b7a7e1320ccd1e030e85c (diff)
downloadopen-iov-4467eb66a4c8a776e35e9644960eba699dc85e6f.tar.gz
open-iov-4467eb66a4c8a776e35e9644960eba699dc85e6f.zip
debug: green-eye porting automation - new-device and driver-version port scripts, simulator packaging, porting docs; control-discovery fix
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.py1139
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())