summaryrefslogtreecommitdiffstats
path: root/debug/risc-v-gsp-rm-vm/harness/state_seed.py
diff options
context:
space:
mode:
authorArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-06 19:28:39 -0400
committerArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-06 19:28:39 -0400
commit6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf (patch)
tree3b3680902cd2223cbe60aaec0a0799cfc7b26be6 /debug/risc-v-gsp-rm-vm/harness/state_seed.py
parenta5473c91b8869b017385582ea969783b33553ded (diff)
downloadopen-iov-6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf.tar.gz
open-iov-6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf.zip
risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with tests
rm_capture (chunked decode), launch_audit, state_seed/state_decode, valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk, msgq_attach, libos_port, log_decode, server_snapshot; doc index updates in the tool overview, operations pages and the repo README.
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/state_seed.py')
-rw-r--r--debug/risc-v-gsp-rm-vm/harness/state_seed.py990
1 files changed, 990 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/state_seed.py b/debug/risc-v-gsp-rm-vm/harness/state_seed.py
new file mode 100644
index 0000000..e4d9a42
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/state_seed.py
@@ -0,0 +1,990 @@
+#!/usr/bin/env python3
+"""RM-object state seeding: declarative state for rehost/valprobe runs.
+
+Why this exists
+---------------
+``harness/rehost.py`` runs one firmware function with caller-supplied bytes and
+``harness/valprobe.py`` runs one validation branch with crafted registers and
+memory. Both hit the same wall on RM handlers: the function early-outs (or
+faults on a NULL callback) before the branch of interest because the RM object
+state it walks -- clients, channels, graphics contexts, memdesc/context-buffer
+objects -- does not exist. This harness adds the missing layer: a declarative
+**state seed** (``greeneye-state-seed/1``) that places symbolic objects at
+chosen addresses, fills their fields and pointers, sets registers/entry, and
+runs the function under a bounded budget while recording which seeded state the
+firmware actually reads and where it first stops.
+
+Every run reports the ordered first-touch state reads and a ``gate``: the
+address and the state the returning/faulting path checked. A clean early-out
+with the gate identified is the success criterion; a fault is the first missing
+state observed, never a product defect.
+
+Honesty rules:
+
+* Not a boot and not a model: only the seeded state exists. The report lists
+ the layout, every field it wrote, and every field marked ``unknown``.
+* The object field map below is partial and cites its evidence per field; a
+ field without evidence is marked ``unknown`` and must not be invented.
+* No timing/cycle claim (L0/L1 fidelity, docs/06).
+
+Offline only: never touches a GPU, kernel module, mdev, guest or network.
+"""
+from __future__ import annotations
+
+import argparse
+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 Fault, R_BP, R_BUDGET, R_ECALL, R_FAULT, R_WATCH # noqa: E402
+
+from harness.rehost import ( # noqa: E402
+ DEFAULT_IMAGE,
+ DEFAULT_PROFILE,
+ EBREAK,
+ RUN_CHUNK,
+ _parse_u64,
+ _registers,
+ _sha256_file,
+)
+from harness.valprobe import ( # noqa: E402
+ BUFFER_ID,
+ INSTANCE_GETTER,
+ OUTCOMES,
+ PROMOTE_VALIDATION,
+ decode_memory_spec,
+ encode_return_stub,
+ reg_index,
+)
+
+#: Schema tag of a state seed.
+SEED_FORMAT = "greeneye-state-seed/1"
+#: Schema tag of a single seeded run.
+REPORT_FORMAT = "greeneye-state-seed-report/1"
+#: Schema tag of the bundled demo.
+DEMO_FORMAT = "greeneye-state-seed-demo/1"
+
+DEFAULT_BUDGET = 50_000
+DEFAULT_TIMEOUT = 20.0
+DEFAULT_ALLOC_BASE = 0x10000000
+DEFAULT_ALLOC_ALIGN = 16
+
+#: Object types that carry a documented field map (evidence per field below).
+_CHANNEL_FIELDS = {
+ 0x0D28: {"name": "instance_index", "width": 4, "status": "known",
+ "role": "graphics-instance index (u32); selects client+0x37c+idx",
+ "evidence": "promote-validation predicate (docs/47)"},
+ 0x0D01: {"name": "flag_d01", "width": 1, "status": "unknown",
+ "role": "u8 flag on the failed-lookup path (0 takes the panic path)"},
+ 0x0CD1: {"name": "flag_cd1", "width": 1, "status": "unknown",
+ "role": "u8 flag before the promote calls"},
+ 0x0648: {"name": "callback_648", "width": 8, "status": "unknown",
+ "role": "indirect callback on the invalid-state path"},
+ 0x0898: {"name": "callback_898", "width": 8, "status": "unknown",
+ "role": "indirect callback first reached after the +0x600 gate"},
+ 0x03B8: {"name": "marker_3b8", "width": 8, "status": "unknown",
+ "role": "identity marker compared across objects"},
+ 0x00EE: {"name": "obj_ee", "width": 8, "status": "unknown",
+ "role": "sub-object; +0xe8 read as the gfid/index"},
+}
+
+_GRAPHICS_INSTANCE_FIELDS = {
+ 0x0190: {"name": "memdesc_table", "width": 8, "status": "known",
+ "role": "pointer to the per-buffer memDesc slot table",
+ "evidence": "promote-validation predicate (docs/47)"},
+ 0x0161: {"name": "flag_161", "width": 1, "status": "unknown",
+ "role": "u8 flag gating the 0x529bd44 call"},
+ 0x03B8: {"name": "marker_3b8", "width": 8, "status": "unknown",
+ "role": "identity marker compared across objects"},
+}
+
+FIELD_MAP: dict[str, dict[int, dict]] = {
+ "client": {
+ 0x1BE0: {"name": "instance_table", "width": 8, "status": "known",
+ "role": "graphics-instance pointer array base (index 0x37c + idx)",
+ "evidence": "promote-validation predicate (docs/47)"},
+ 0x1AA8: {"name": "obj_1aa8", "width": 8, "status": "unknown",
+ "role": "sub-object consumed by the construct path (+0x24c callback)"},
+ 0x1CC8: {"name": "obj_1cc8", "width": 8, "status": "unknown",
+ "role": "sub-object consumed by the construct path (+0x150, +0x1d0)"},
+ 0x1DD8: {"name": "engine_ctx", "width": 8, "status": "known",
+ "role": "engine-context pointer; 0 selects the no-engine path",
+ "evidence": "construct-path disassembly (docs/52)"},
+ 0x1DE8: {"name": "subdevice", "width": 8, "status": "known",
+ "role": "sub-device object whose +0x600 callback is called",
+ "evidence": "construct-path disassembly (docs/52)"},
+ 0x1E48: {"name": "obj_1e48", "width": 8, "status": "unknown",
+ "role": "sub-object consumed on the no-engine path (+0x110 callback)"},
+ },
+ "channel": _CHANNEL_FIELDS,
+ "graphics_context": _CHANNEL_FIELDS,
+ "graphics_instance": _GRAPHICS_INSTANCE_FIELDS,
+ "memdesc": {
+ 0x0000: {"name": "unknown_0", "width": 8, "status": "unknown",
+ "role": "slot value semantics outside the promote predicate"},
+ 0x0008: {"name": "unknown_8", "width": 8, "status": "unknown",
+ "role": "unknown"},
+ },
+ "ctx_buffer": {
+ 0x0000: {"name": "unknown_0", "width": 8, "status": "unknown",
+ "role": "context-buffer (memdesc) object; field semantics unknown"},
+ },
+ "rm_object": {},
+}
+
+#: Object types accepted in a seed. ``rm_object`` carries no field map.
+KNOWN_TYPES = tuple(FIELD_MAP)
+MISSING_MARKER = "unknown"
+
+
+class StateSeedError(GreenEyeError):
+ """A state seed, image or scenario is invalid."""
+
+
+def _require_u64(value: Any, what: str = "value") -> int:
+ try:
+ return _parse_u64(value)
+ except GreenEyeError as exc:
+ raise StateSeedError(f"{what}: {exc}") from exc
+
+
+def _field_width(obj_type: str, offset: int) -> int:
+ spec = FIELD_MAP.get(obj_type, {}).get(offset)
+ return int(spec["width"]) if spec else 8
+
+
+def _field_spec(obj_type: str, offset: int) -> dict | None:
+ return FIELD_MAP.get(obj_type, {}).get(offset)
+
+
+def _parse_offset(value: Any) -> int:
+ text = str(value).strip()
+ if text.startswith("+"):
+ text = text[1:]
+ try:
+ return int(text, 0)
+ except ValueError as exc:
+ raise StateSeedError(f"bad field offset {value!r}") from exc
+
+
+def _encode_int(value: int, width: int) -> bytes:
+ if width not in (1, 2, 4, 8):
+ raise StateSeedError(f"unsupported field width {width}")
+ return (value & ((1 << (8 * width)) - 1)).to_bytes(width, "little")
+
+
+def _is_symbol(value: Any) -> bool:
+ return isinstance(value, str) and value.strip().startswith("@")
+
+
+def _symbol_expr(value: str) -> tuple[str, int]:
+ text = value.strip()[1:]
+ name, sep, offset = text.partition("+")
+ if not name:
+ raise StateSeedError(f"bad pointer expression {value!r}")
+ if sep:
+ try:
+ delta = int(offset, 0)
+ except ValueError as exc:
+ raise StateSeedError(f"bad pointer expression {value!r}") from exc
+ else:
+ delta = 0
+ return name, delta
+
+
+class _Symbols:
+ """Name -> address table for symbolic object/stub pointers."""
+
+ def __init__(self) -> None:
+ self._by_name: dict[str, int] = {}
+
+ def add(self, name: str, addr: int, owner: str) -> None:
+ if name in self._by_name:
+ raise StateSeedError(f"duplicate symbol {name!r} ({owner})")
+ self._by_name[name] = addr
+
+ def resolve(self, value: str, *, owner: str) -> int:
+ name, delta = _symbol_expr(value)
+ if name not in self._by_name:
+ raise StateSeedError(f"{owner}: unknown symbol {name!r}")
+ return (self._by_name[name] + delta) & 0xFFFF_FFFF_FFFF_FFFF
+
+
+def load_seed(seed: dict) -> dict:
+ """Validate a ``greeneye-state-seed/1`` seed and compile an application plan."""
+ if not isinstance(seed, dict):
+ raise StateSeedError(f"seed must be an object, got {type(seed).__name__}")
+ fmt = seed.get("format", SEED_FORMAT)
+ if fmt != SEED_FORMAT:
+ raise StateSeedError(f"unsupported seed format {fmt!r} (want {SEED_FORMAT})")
+ name = str(seed.get("name") or "state-seed")
+ seeded_registers = seed.get("registers") or {}
+ if not isinstance(seeded_registers, dict):
+ raise StateSeedError("registers must be an object")
+ entry_value = seed.get("entry", seeded_registers.get("pc"))
+ if entry_value is None:
+ raise StateSeedError("seed needs an entry address (or registers.pc)")
+ entry = _require_u64(entry_value)
+
+ allocator = seed.get("allocator") or {}
+ cursor = _require_u64(allocator.get("base", DEFAULT_ALLOC_BASE))
+ default_align = _require_u64(allocator.get("align", DEFAULT_ALLOC_ALIGN))
+ if default_align <= 0 or default_align & (default_align - 1):
+ raise StateSeedError("allocator align must be a positive power of two")
+
+ objects = seed.get("objects") or []
+ if not isinstance(objects, list):
+ raise StateSeedError("objects must be a list")
+ layout: list[dict] = []
+ symbols = _Symbols()
+ for obj in objects:
+ if not isinstance(obj, dict):
+ raise StateSeedError(f"object must be an object, got {obj!r}")
+ obj_type = str(obj.get("type") or "")
+ if obj_type not in KNOWN_TYPES:
+ raise StateSeedError(f"unknown object type {obj_type!r} (known: {', '.join(KNOWN_TYPES)})")
+ obj_name = str(obj.get("name") or "")
+ if not obj_name:
+ raise StateSeedError(f"{obj_type} object needs a name")
+ align = _require_u64(obj.get("align", default_align))
+ if align <= 0 or align & (align - 1):
+ raise StateSeedError(f"{obj_name}: align must be a positive power of two")
+ fields = obj.get("fields") or {}
+ if not isinstance(fields, dict):
+ raise StateSeedError(f"{obj_name}: fields must be an object")
+ extent = 0
+ for key in fields:
+ offset = _parse_offset(key)
+ extent = max(extent, offset + _field_width(obj_type, offset))
+ size = _require_u64(obj.get("size", max(extent, default_align)))
+ if size <= 0:
+ raise StateSeedError(f"{obj_name}: size must be positive")
+ if "pa" in obj and "va" in obj:
+ raise StateSeedError(f"{obj_name}: use pa or va, not both")
+ explicit = obj.get("pa", obj.get("va"))
+ if explicit is not None:
+ pa = _require_u64(explicit)
+ else:
+ cursor = (cursor + align - 1) & ~(align - 1)
+ pa = cursor
+ cursor += (size + align - 1) & ~(align - 1)
+ record = {"name": obj_name, "type": obj_type, "pa": pa, "size": size,
+ "fields": {}, "unknown": []}
+ layout.append(record)
+ symbols.add(obj_name, pa, obj_name)
+ for key, value in fields.items():
+ record["fields"][_parse_offset(key)] = value
+
+ for i, first in enumerate(layout):
+ for second in layout[i + 1:]:
+ if first["pa"] < second["pa"] + second["size"] and second["pa"] < first["pa"] + first["size"]:
+ raise StateSeedError(
+ f"objects {first['name']!r} and {second['name']!r} overlap"
+ )
+
+ writes: list[dict] = []
+ unknown_fields: list[dict] = []
+
+ stubs = seed.get("stubs") or []
+ if not isinstance(stubs, list):
+ raise StateSeedError("stubs must be a list")
+ stub_records: list[dict] = []
+ for stub in stubs:
+ if not isinstance(stub, dict) or "return" not in stub:
+ raise StateSeedError(f"stub needs 'return': {stub!r}")
+ ret = _require_u64(stub["return"])
+ if "addr" in stub:
+ addr = _require_u64(stub["addr"])
+ else:
+ cursor = (cursor + default_align - 1) & ~(default_align - 1)
+ addr = cursor
+ cursor += default_align
+ stub_name = stub.get("name")
+ if stub_name:
+ symbols.add(str(stub_name), addr, f"stub {stub_name}")
+ code = encode_return_stub(ret)
+ stub_records.append({"name": stub_name, "addr": addr, "return": ret,
+ "code_hex": code.hex()})
+ writes.append({"pa": addr, "hex": code.hex(), "label": f"stub:{stub_name or hex(addr)}",
+ "kind": "stub"})
+
+ def resolve_field(owner: str, obj_type: str, offset: int, value: Any) -> tuple[int | None, int, str]:
+ width = _field_width(obj_type, offset)
+ if value == MISSING_MARKER or (isinstance(value, dict) and value.get("unknown")):
+ return None, width, MISSING_MARKER
+ if isinstance(value, dict):
+ if "unknown" in value and value["unknown"]:
+ return None, width, MISSING_MARKER
+ if "ptr" in value:
+ value = value["ptr"]
+ elif "value" in value:
+ raw = value["value"]
+ if raw == MISSING_MARKER:
+ return None, _require_u64(value.get("size", width)), MISSING_MARKER
+ width = _require_u64(value.get("size", width))
+ return _require_u64(raw), width, "value"
+ else:
+ raise StateSeedError(f"{owner}+{offset:#x}: field dict needs value/ptr/unknown")
+ if _is_symbol(value):
+ return symbols.resolve(str(value), owner=f"{owner}+{offset:#x}"), width, "ptr"
+ return _parse_u64(value), width, "value"
+
+ for record in layout:
+ pa = record["pa"]
+ for offset in sorted(record["fields"]):
+ value = record["fields"][offset]
+ resolved, width, kind = resolve_field(record["name"], record["type"], offset, value)
+ spec = _field_spec(record["type"], offset)
+ label = f"{record['name']}+{offset:#x}"
+ if resolved is None:
+ unknown_fields.append({
+ "object": record["name"], "type": record["type"], "off": f"{offset:#x}",
+ "name": spec["name"] if spec else None,
+ "role": spec["role"] if spec else "not documented",
+ "reason": "seed marks this field unknown; left at zero",
+ })
+ continue
+ writes.append({"pa": pa + offset, "hex": _encode_int(resolved, width).hex(),
+ "label": label, "kind": kind})
+
+ memory = seed.get("memory") or []
+ if not isinstance(memory, list):
+ raise StateSeedError("memory must be a list")
+ for item in memory:
+ if not isinstance(item, dict):
+ raise StateSeedError(f"memory item must be an object: {item!r}")
+ normalized = dict(item)
+ if "pa" in normalized and "addr" in normalized:
+ raise StateSeedError(f"memory item: use pa or addr, not both: {item!r}")
+ if "pa" in normalized:
+ normalized["addr"] = normalized.pop("pa")
+ try:
+ addr, data = decode_memory_spec(normalized)
+ except GreenEyeError as exc:
+ raise StateSeedError(str(exc)) from exc
+ writes.append({"pa": addr, "hex": data.hex(),
+ "label": item.get("label") or f"mem+{addr:#x}", "kind": "memory"})
+
+ registers: dict[str, int] = {}
+ for reg_name, value in seeded_registers.items():
+ if str(reg_name).strip().lower() == "pc":
+ continue
+ try:
+ reg_index(str(reg_name))
+ except GreenEyeError as exc:
+ raise StateSeedError(str(exc)) from exc
+ if _is_symbol(value):
+ registers[str(reg_name)] = symbols.resolve(str(value), owner=f"register {reg_name}")
+ else:
+ registers[str(reg_name)] = _require_u64(value)
+
+ stack = seed.get("stack")
+ outcomes = {_require_u64(a): str(label)
+ for a, label in (seed.get("outcomes") or {}).items()}
+ reads = []
+ for item in seed.get("reads") or ():
+ if "pa" in item and "addr" not in item:
+ item = dict(item)
+ item["addr"] = item.pop("pa")
+ if "addr" not in item:
+ raise StateSeedError(f"read item needs addr/pa: {item!r}")
+ reads.append(item)
+ watch: list[dict] = []
+ seen_labels: dict[str, int] = {}
+
+ def _add_watch(addr: int, size: int, label: str) -> None:
+ count = seen_labels.get(label, 0) + 1
+ seen_labels[label] = count
+ if count > 1:
+ label = f"{label}~{count}"
+ watch.append({"addr": addr, "size": size, "label": label})
+
+ for item in seed.get("watch") or ():
+ if "pa" in item and "addr" not in item:
+ item = dict(item)
+ item["addr"] = item.pop("pa")
+ if "addr" not in item:
+ raise StateSeedError(f"watch item needs addr/pa: {item!r}")
+ _add_watch(_require_u64(item["addr"]), _require_u64(item.get("size", 8)),
+ str(item.get("label") or f"{_require_u64(item['addr']):#x}"))
+ for record in layout:
+ for offset in record["fields"]:
+ _add_watch(record["pa"] + offset, _field_width(record["type"], offset),
+ f"{record['name']}+{offset:#x}")
+
+ return {
+ "format": SEED_FORMAT,
+ "name": name,
+ "entry": entry,
+ "stack": _require_u64(stack) if stack is not None else None,
+ "registers": registers,
+ "writes": writes,
+ "layout": layout,
+ "unknown_fields": unknown_fields,
+ "stubs": stub_records,
+ "outcomes": outcomes,
+ "reads": reads,
+ "watch": watch,
+ "provenance": seed.get("provenance") or {},
+ "notes": seed.get("notes") or [],
+ }
+
+
+def apply_seed(machine, plan: dict) -> dict:
+ """Apply a compiled plan to a machine; return the same plan (chainable)."""
+ for item in plan["writes"]:
+ machine.write_mem(item["pa"], bytes.fromhex(item["hex"]))
+ for reg_name, value in plan["registers"].items():
+ machine.x[reg_index(reg_name)] = value
+ return plan
+
+
+def _read_after(machine, reads) -> list[dict]:
+ out = []
+ for item in reads or ():
+ addr = _require_u64(item["addr"])
+ size = _require_u64(item.get("size", 8))
+ data = bytes(machine.read_mem(addr, size))
+ record = {"addr": f"{addr:#x}", "size": size, "hex": data.hex()}
+ if item.get("label"):
+ record["label"] = item["label"]
+ record["u64"] = int.from_bytes(data.ljust(8, b"\x00"), "little")
+ out.append(record)
+ return out
+
+
+def _gate_from_run(status: str, stop_pc: int, fault: dict | None,
+ state_reads: list[dict]) -> dict | None:
+ if not state_reads:
+ return None
+ last = state_reads[-1]
+ if status == "fault" and stop_pc == 0 and last.get("value") == 0:
+ return {
+ "kind": "null_indirect_call",
+ "pc": last["pc"],
+ "addr": last["addr"],
+ "label": last["label"],
+ "value": 0,
+ "detail": (
+ f"indirect call target {last['label']} is NULL; execution faulted at "
+ f"pc 0x0 (fault={fault})"
+ ),
+ }
+ if status == "fault":
+ return {
+ "kind": "fault",
+ "pc": f"{stop_pc:#x}",
+ "addr": last["addr"],
+ "label": last["label"],
+ "value": last.get("value"),
+ "detail": f"fault before the next seeded-state read (fault={fault})",
+ }
+ if status == "outcome":
+ return None
+ return {
+ "kind": f"{status}_before_gate",
+ "pc": f"{stop_pc:#x}",
+ "addr": last["addr"],
+ "label": last["label"],
+ "value": last.get("value"),
+ "detail": f"run stopped as {status}; last seeded state read shown",
+ }
+
+
+def run_seed(
+ seed: dict,
+ *,
+ image: str = DEFAULT_IMAGE,
+ profile_path: str | None = DEFAULT_PROFILE,
+ entry: int | str | None = None,
+ out: str | None = None,
+ record: bool | None = None,
+ budget: int = DEFAULT_BUDGET,
+ timeout: float = DEFAULT_TIMEOUT,
+ name: str = "state-seed",
+) -> dict:
+ """Load a seed into the native machine, run it bounded, return a report."""
+ if not os.path.isfile(image):
+ raise StateSeedError(f"image not found: {image}")
+ if budget <= 0:
+ raise StateSeedError(f"budget must be positive, got {budget}")
+ image_obj = load_elf(image)
+ if not image_obj.is_riscv:
+ raise StateSeedError(f"{image}: not a RISC-V ELF image")
+ plan = load_seed(seed)
+ entry_addr = _require_u64(entry) if entry is not None else plan["entry"]
+
+ profile = load_profile(profile_path) if profile_path else None
+ load = (profile or {}).get("load", {})
+ ram_base = _require_u64(load.get("ram_base", 0))
+ memory_mb = int(load.get("memory_mb", 256))
+ ram_top = ram_base + memory_mb * 1024 * 1024
+
+ if record is None:
+ record = bool(out)
+ session = launch_native(
+ kernel=image,
+ memory_mb=memory_mb,
+ ram_base=ram_base,
+ record=record,
+ name=f"{name}-{plan['name']}"[:80],
+ device_map=None,
+ unknown_mmio="lenient",
+ 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)
+ machine.syscall_handler = LibosSyscalls().handle
+ shim = dict(profile.get("boot_shim", {}))
+ shim["_image_path"] = image
+ _apply_boot_shim(machine, shim, image_obj, {"enabled": True, "assumptions": []})
+
+ sentinel = (ram_top - 0x100) & ~0xF
+ default_sp = (ram_top - 0x8000) & ~0xF
+ machine.write_mem(sentinel, EBREAK.to_bytes(4, "little"))
+
+ apply_seed(machine, plan)
+ machine.x[1] = sentinel
+ if machine.x[2] == 0:
+ machine.x[2] = plan["stack"] if plan["stack"] is not None else default_sp
+ machine.pc = entry_addr
+
+ breakpoints = set(plan["outcomes"]) | {sentinel}
+ watchers = []
+ for item in plan["watch"]:
+ addr = _require_u64(item["addr"])
+ size = _require_u64(item.get("size", 8))
+ label = str(item.get("label") or f"{addr:#x}")
+ watchers.append((addr, addr + size, label, 3))
+
+ entered = {f"x{i}": machine.x[i] for i in range(32) if i and machine.x[i]}
+ entered["pc"] = entry_addr
+ state_reads: list[dict] = []
+ started = time.monotonic()
+ status = "budget"
+ stop_pc = entry_addr
+ fault: dict | None = None
+ outcome: str | None = None
+ 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=breakpoints, watchers=tuple(watchers)
+ )
+ if reason == R_WATCH:
+ width = 8
+ for watcher in watchers:
+ if watcher[2] == b:
+ width = watcher[1] - watcher[0]
+ break
+ try:
+ value = int.from_bytes(bytes(machine.read_mem(a, width)), "little")
+ except (GreenEyeError, Fault):
+ value = None
+ state_reads.append({"pc": f"{stop_pc:#x}", "addr": f"{a:#x}",
+ "label": b, "value": value, "width": width})
+ watchers = [watcher for watcher in watchers if watcher[2] != b]
+ continue
+ remaining -= chunk
+ if reason == R_BP:
+ if stop_pc == sentinel:
+ status = "return"
+ break
+ outcome = plan["outcomes"].get(stop_pc, "unlabelled")
+ status = "outcome"
+ 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_BUDGET:
+ status = "budget"
+
+ gate = _gate_from_run(status, stop_pc, fault, state_reads)
+ report: dict[str, Any] = {
+ "format": REPORT_FORMAT,
+ "seed": plan["name"],
+ "seed_format": plan["format"],
+ "seed_provenance": plan["provenance"],
+ "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}",
+ "status": status,
+ "outcome": outcome,
+ "stop_pc": f"{stop_pc:#x}",
+ "fault": fault,
+ "retval": machine.x[10],
+ "instructions": machine.insns,
+ "wall_seconds": round(time.monotonic() - started, 4),
+ "objects": [
+ {"name": item["name"], "type": item["type"], "pa": f"{item['pa']:#x}",
+ "size": item["size"]}
+ for item in plan["layout"]
+ ],
+ "unknown_fields": plan["unknown_fields"],
+ "stubs": plan["stubs"],
+ "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()},
+ "state_reads": state_reads,
+ "gate": gate,
+ "reads": _read_after(machine, plan["reads"]),
+ "limits": [
+ "Seeded run, not a boot: only the supplied objects/memory exist.",
+ "A fault is the first missing state observed, not a product defect.",
+ "No timing/cycle claim (L0/L1 fidelity, docs/06).",
+ "A field marked unknown was left at zero and reported.",
+ ],
+ }
+ if getattr(session.recorder, "dir", None) is not None:
+ session.note("state-seed", seed=plan["name"], status=status,
+ outcome=outcome, stop_pc=f"{stop_pc:#x}",
+ gate=(gate or {}).get("kind"))
+ report["session"] = session.recorder.dir
+ if out:
+ os.makedirs(out, exist_ok=True)
+ path = os.path.join(out, "state-seed-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()
+
+
+# ---------------------------------------------------------------------------
+# Bundled seeds.
+#
+# 1. fecs_seed: the promote-validation predicate seeded through this loader
+# (the mechanism proof -- it works without RM objects). It reproduces the
+# valprobe variant memory/registers/stub exactly, so a slot non-NULL seed
+# must refuse 0x1f and a slot NULL seed must take the commit path.
+# 2. target_seed: the construct target 0x52a70ec with a minimal crafted
+# client/channel/graphics-context/memdesc set. The first gate is reported
+# (see docs/52); with the +0x600 callback stubbed it advances to the next.
+# ---------------------------------------------------------------------------
+
+
+def _fecs_seed(*, slot: int, label: str) -> dict:
+ scratch = 0x10000000
+ layout = {
+ "scratch": scratch,
+ "sp": scratch + 0x4000,
+ "s0": scratch + 0x5000,
+ "pgpu": scratch + 0x1000,
+ "getter_out": scratch + 0x1800,
+ "instance": scratch + 0x3800,
+ "entry": scratch + 0x6000,
+ }
+ slot_addr = layout["instance"] + 0x300 + 6 * 8
+ memory = [
+ {"pa": f"{layout['entry'] + 0x30:#x}", "u64": 0x1F4150000},
+ {"pa": f"{layout['entry'] + 0x38:#x}", "u64": 0x204608000},
+ {"pa": f"{layout['entry'] + 0x40:#x}", "u64": 0x10000},
+ {"pa": f"{layout['entry'] + 0x48:#x}", "u32": 4},
+ {"pa": f"{layout['entry'] + 0x4c:#x}", "u16": BUFFER_ID["FECS_EVENT"]},
+ {"pa": f"{layout['entry'] + 0x4e:#x}", "u8": 1},
+ {"pa": f"{layout['entry'] + 0x4f:#x}", "u8": 0},
+ {"pa": f"{layout['pgpu'] + 0x1be0:#x}", "u64": layout["instance"]},
+ {"pa": f"{layout['instance'] + 0x190:#x}", "u64": layout["instance"] + 0x300},
+ {"pa": f"{slot_addr:#x}", "u64": slot, "label": "fecs-slot"},
+ ]
+ return {
+ "format": SEED_FORMAT,
+ "name": label,
+ "entry": f"{PROMOTE_VALIDATION:#x}",
+ "outcomes": dict(OUTCOMES),
+ "registers": {
+ "s0": f"{layout['s0']:#x}",
+ "s2": f"{layout['pgpu']:#x}",
+ "s5": f"{layout['entry']:#x}",
+ "sp": f"{layout['sp']:#x}",
+ "a3": BUFFER_ID["FECS_EVENT"],
+ "t1": 0xC,
+ },
+ "stubs": [{"addr": f"{INSTANCE_GETTER:#x}", "return": f"{layout['getter_out']:#x}",
+ "name": "instance_getter"}],
+ "memory": memory,
+ "watch": [{"pa": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot"}],
+ "reads": [{"pa": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot-value"}],
+ "provenance": {
+ "source": "promote-validation predicate state (docs/47), rebuilt as a seed",
+ "slot": slot,
+ },
+ "notes": [
+ "Mechanism proof: a path that needs no RM objects, seeded end to end.",
+ ],
+ }
+
+
+def fecs_seed(*, slot: int = 1) -> dict:
+ """Build the FECS promote-validation seed (slot 1 refuses, slot 0 accepts)."""
+ label = "fecs-promote-slot-nonnull" if slot else "fecs-promote-slot-null"
+ return _fecs_seed(slot=slot, label=label)
+
+
+def target_seed(*, advance: bool = False) -> dict:
+ """Build the minimal client/channel/context seed for the 0x52a70ec target.
+
+ ``advance=False`` leaves the +0x600 callback unknown (the first gate).
+ ``advance=True`` points it at a return-0 stub so the next gate is reached.
+ """
+ callback = "@engine_stub" if advance else MISSING_MARKER
+ name = "construct-gate-advanced" if advance else "construct-gate-minimal"
+ return {
+ "format": SEED_FORMAT,
+ "name": name,
+ "entry": "0x52a70ec",
+ "allocator": {"base": "0x10010000", "align": "0x1000"},
+ "stubs": [{"name": "engine_stub", "return": 0}] if advance else [],
+ "registers": {
+ "a0": "@client",
+ "a1": "@channel",
+ "a2": "0x1003f000",
+ },
+ "stack": "0x1003e000",
+ "objects": [
+ {"name": "client", "type": "client", "size": "0x2000", "fields": {
+ "0x1be0": "@graphics_instance",
+ "0x1aa8": "@obj_1aa8",
+ "0x1cc8": "@obj_1cc8",
+ "0x1dd8": {"value": 0, "size": 8},
+ "0x1de8": "@engine_node",
+ "0x1e48": "@obj_1e48",
+ }},
+ {"name": "channel", "type": "channel", "size": "0x1000", "fields": {
+ "0xd28": {"value": 0, "size": 4},
+ "0xd01": {"value": 1, "size": 1},
+ "0xcd1": {"value": 0, "size": 1},
+ "0x898": MISSING_MARKER,
+ }},
+ {"name": "graphics_instance", "type": "graphics_instance", "size": "0x1000",
+ "fields": {
+ "0x190": "@memdesc_table",
+ "0x161": {"value": 0, "size": 1},
+ }},
+ {"name": "engine_node", "type": "rm_object", "size": "0x1000", "fields": {
+ "0x600": callback,
+ }},
+ {"name": "obj_1cc8", "type": "rm_object", "size": "0x400", "fields": {
+ "0x150": MISSING_MARKER,
+ "0x1d0": {"value": 0x20, "size": 4},
+ }},
+ {"name": "obj_1aa8", "type": "rm_object", "size": "0x400", "fields": {
+ "0x24c": MISSING_MARKER,
+ }},
+ {"name": "obj_1e48", "type": "rm_object", "size": "0x400", "fields": {
+ "0x110": MISSING_MARKER,
+ }},
+ {"name": "memdesc_table", "type": "rm_object", "size": "0x100", "fields": {}},
+ {"name": "fecs_memdesc", "type": "memdesc", "size": "0x100", "fields": {
+ "0x0": MISSING_MARKER,
+ "0x8": MISSING_MARKER,
+ }},
+ ],
+ "reads": [{"pa": "0x1003f000", "size": 8, "label": "a2-out-scratch"}],
+ "provenance": {
+ "source": "construct-path disassembly (docs/52, this wave)",
+ "callback": callback,
+ },
+ "notes": [
+ "Only the fields the known firmware path reads are seeded; a field marked",
+ "unknown is left zero and reported so the gate is not mistaken for a seed bug.",
+ ],
+ }
+
+
+def run_demo(out: str | None = None) -> dict:
+ """Prove the seeder on the FECS predicate, then attempt the construct target."""
+ if not os.path.isfile(DEFAULT_IMAGE):
+ raise StateSeedError(f"firmware image absent: {DEFAULT_IMAGE}")
+ profile = DEFAULT_PROFILE
+
+ fecs_nonnull = run_seed(fecs_seed(slot=1), profile_path=profile, record=False,
+ name="state-seed-fecs")
+ fecs_nonnull_again = run_seed(fecs_seed(slot=1), profile_path=profile, record=False,
+ name="state-seed-fecs")
+ fecs_null = run_seed(fecs_seed(slot=0), profile_path=profile, record=False,
+ name="state-seed-fecs")
+ target = run_seed(target_seed(advance=False), profile_path=profile, record=False,
+ name="state-seed-target")
+ advanced = run_seed(target_seed(advance=True), profile_path=profile, record=False,
+ name="state-seed-target")
+
+ verdicts = [
+ {"label": "FECS init slot non-NULL", "outcome": fecs_nonnull["outcome"],
+ "expect": "refuse-fecs-slot-exists-0x1f",
+ "pass": fecs_nonnull["outcome"] == "refuse-fecs-slot-exists-0x1f"},
+ {"label": "FECS init slot NULL", "outcome": fecs_null["outcome"],
+ "expect": "accept-commit-path",
+ "pass": fecs_null["outcome"] == "accept-commit-path"},
+ ]
+ deterministic = (
+ fecs_nonnull["outcome"] == fecs_nonnull_again["outcome"]
+ and fecs_nonnull["gate"] == fecs_nonnull_again["gate"]
+ )
+ result = {
+ "format": DEMO_FORMAT,
+ "ok": all(row["pass"] for row in verdicts) and deterministic,
+ "deterministic": deterministic,
+ "mechanism": {
+ "seed": "fecs-promote",
+ "verdicts": verdicts,
+ "wall_seconds": round(fecs_nonnull["wall_seconds"] + fecs_null["wall_seconds"], 4),
+ },
+ "target": {
+ "entry": "0x52a70ec",
+ "status": target["status"],
+ "stop_pc": target["stop_pc"],
+ "gate": target["gate"],
+ "state_reads": target["state_reads"],
+ "unknown_fields": target["unknown_fields"],
+ },
+ "target_advanced": {
+ "entry": "0x52a70ec",
+ "status": advanced["status"],
+ "stop_pc": advanced["stop_pc"],
+ "gate": advanced["gate"],
+ "state_reads": advanced["state_reads"],
+ },
+ }
+ if out:
+ os.makedirs(out, exist_ok=True)
+ path = os.path.join(out, "state-seed-demo.json")
+ with open(path, "w", encoding="utf-8") as fh:
+ json.dump(result, fh, indent=2, sort_keys=True)
+ fh.write("\n")
+ result["artifacts"] = {"demo_json": os.path.abspath(path)}
+ return result
+
+
+def _load_seed_file(path: str) -> dict:
+ if not os.path.isfile(path):
+ raise StateSeedError(f"seed not found: {path}")
+ with open(path, "r", encoding="utf-8") as fh:
+ return json.load(fh)
+
+
+def _emit_seed(kind: str) -> dict:
+ if kind == "fecs":
+ return fecs_seed()
+ if kind == "target-minimal":
+ return target_seed(advance=False)
+ if kind == "target-advanced":
+ return target_seed(advance=True)
+ raise StateSeedError(f"unknown seed kind {kind!r}")
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ prog="state_seed.py",
+ description=(
+ "Seed declarative RM object state into a bounded firmware run "
+ "(offline; never touches hardware)."
+ ),
+ )
+ sub = parser.add_subparsers(dest="command")
+ run_cmd = sub.add_parser("run", help="run one seed file")
+ run_cmd.add_argument("--seed", required=True, help="greeneye-state-seed/1 JSON")
+ run_cmd.add_argument("--entry", default=None, help="override the seed entry (hex)")
+ 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("--budget", type=int, default=DEFAULT_BUDGET)
+ run_cmd.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
+ run_cmd.add_argument("--out", help="output directory (session + report JSON)")
+ run_cmd.add_argument("--no-record", action="store_true", help="skip session recording")
+ validate = sub.add_parser("validate", help="validate a seed and print its layout")
+ validate.add_argument("--seed", required=True)
+ emit = sub.add_parser("emit", help="write a bundled seed")
+ emit.add_argument("--kind", required=True,
+ choices=["fecs", "target-minimal", "target-advanced"])
+ emit.add_argument("--out", required=True)
+ demo = sub.add_parser("demo", help="FECS mechanism proof + construct-gate attempt")
+ demo.add_argument("--out", help="output directory")
+ return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = build_parser()
+ args = parser.parse_args(argv)
+ try:
+ if args.command == "run":
+ seed = _load_seed_file(args.seed)
+ report = run_seed(seed, image=args.image, profile_path=args.profile or None,
+ entry=args.entry, budget=args.budget, timeout=args.timeout,
+ out=args.out, record=False if args.no_record else None)
+ summary = {k: report[k] for k in
+ ("format", "seed", "entry", "status", "outcome", "stop_pc",
+ "fault", "gate", "instructions", "wall_seconds",
+ "report_json", "session") if k in report}
+ print(json.dumps(summary, indent=2, sort_keys=True))
+ return 0
+ if args.command == "validate":
+ seed = _load_seed_file(args.seed)
+ plan = load_seed(seed)
+ print(json.dumps({
+ "format": plan["format"],
+ "name": plan["name"],
+ "entry": f"{plan['entry']:#x}",
+ "objects": [{"name": item["name"], "type": item["type"],
+ "pa": f"{item['pa']:#x}", "size": item["size"]}
+ for item in plan["layout"]],
+ "unknown_fields": plan["unknown_fields"],
+ "writes": len(plan["writes"]),
+ "stubs": plan["stubs"],
+ }, indent=2, sort_keys=True))
+ return 0
+ if args.command == "emit":
+ seed = _emit_seed(args.kind)
+ os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
+ with open(args.out, "w", encoding="utf-8") as fh:
+ json.dump(seed, fh, indent=2, sort_keys=True)
+ fh.write("\n")
+ print(os.path.abspath(args.out))
+ return 0
+ if args.command == "demo":
+ result = run_demo(args.out)
+ print(json.dumps({k: result[k] for k in
+ ("format", "ok", "deterministic", "mechanism",
+ "target", "target_advanced", "artifacts")
+ if k in result}, indent=2, sort_keys=True))
+ return 0 if result["ok"] else 3
+ parser.print_help()
+ return 2
+ except StateSeedError as exc:
+ print(f"error: {exc}", file=sys.stderr)
+ return 2
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())