diff options
| author | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-10-06 19:28:39 -0400 |
|---|---|---|
| committer | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-10-06 19:28:39 -0400 |
| commit | 6c3fd5df61ff2f5cb8833ead6c05eaff9fdc92bf (patch) | |
| tree | 3b3680902cd2223cbe60aaec0a0799cfc7b26be6 /debug/risc-v-gsp-rm-vm/harness/state_seed.py | |
| parent | a5473c91b8869b017385582ea969783b33553ded (diff) | |
| download | open-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.py | 990 |
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()) |
