#!/usr/bin/env python3 """Validation-predicate probe: run a firmware predicate branch with crafted state. Why this exists --------------- The live GR campaign keeps burning supervised hardware cycles to ask the same question class: *which control/entry shapes does the 570.144 GSP-RM firmware accept or refuse, and why?* The answer lives in firmware branches that are buried behind host traffic (B4/B5) and RM object state, so a plain rehost of the full handler early-outs before the predicate. This harness instead runs one **predicate branch** of the real image with the minimum crafted state it needs: registers, memory, and (where a branch calls out to a lock/table walk that is not the predicate) one recorded inline stub. Each run stops on a labelled outcome address, so the verdict is the branch the firmware took, not an inference. It is deliberately not a boot and not a model: the report carries the exact entry, seeded state, stubs and outcome map, and every limitation in `limits`. Fidelity is L0/L1 (docs/06); no timing claim. Bundled target: ``sweep --target fecs-promote`` reproduces the live V6 FECS matrix (`PM/PATCH/PRIV/UNR accepted, FECS init refused 0x1f, FECS VA-only accepted`) and evaluates the untested shapes offline, including the sysmem-aperture FECS-init variant the handoff proposed paying a hardware cycle for. Offline only: never touches a GPU, kernel module, mdev, guest or network. """ from __future__ import annotations import argparse import json import os import struct 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, _run_with_milestones, _rung_from_run # noqa: E402 from sim.libos import LibosSyscalls # noqa: E402 from sim.machine import R_BP, R_BUDGET, R_ECALL, R_FAULT # noqa: E402 from harness.rehost import ( # noqa: E402 DEFAULT_IMAGE, DEFAULT_PROFILE, RUN_CHUNK, _parse_u64, _sha256_file, parse_hex_bytes, ) #: Schema tag of a single probe run. PROBE_FORMAT = "greeneye-valprobe/1" #: Schema tag of a probe spec. SPEC_FORMAT = "greeneye-valprobe-spec/1" #: Schema tag of a matrix run (a spec with variants). MATRIX_FORMAT = "greeneye-valprobe-matrix/1" #: Schema tag of the bundled demo. DEMO_FORMAT = "greeneye-valprobe-demo/1" DEFAULT_BUDGET = 50_000 DEFAULT_TIMEOUT = 20.0 BOOT_INSTRUCTIONS = 2_000_000 EBREAK = 0x00100073 REGISTERS = { "zero": 0, "ra": 1, "sp": 2, "gp": 3, "tp": 4, "t0": 5, "t1": 6, "t2": 7, "s0": 8, "fp": 8, "s1": 9, "a0": 10, "a1": 11, "a2": 12, "a3": 13, "a4": 14, "a5": 15, "a6": 16, "a7": 17, "s2": 18, "s3": 19, "s4": 20, "s5": 21, "s6": 22, "s7": 23, "s8": 24, "s9": 25, "s10": 26, "s11": 27, "t3": 28, "t4": 29, "t5": 30, "t6": 31, } for _i in range(32): REGISTERS.setdefault(f"x{_i}", _i) RUNG_RANK = {"B0": 0, "B1": 1, "B2": 2, "B3": 3} class ValprobeError(GreenEyeError): """A probe spec, image or scenario is invalid.""" def reg_index(name: str) -> int: key = str(name).strip().lower() if key not in REGISTERS: raise ValprobeError(f"unknown register {name!r}") return REGISTERS[key] def encode_return_stub(value: int) -> bytes: """Encode ``lui a0,hi; addi a0,a0,lo; jalr zero,0(ra)`` (RV64IMAC).""" value &= 0xFFFF_FFFF_FFFF_FFFF hi = (value + 0x800) >> 12 lo = value - (hi << 12) lui = ((hi & 0xF_FFFF) << 12) | (10 << 7) | 0x37 addi = ((lo & 0xFFF) << 20) | (10 << 15) | (10 << 7) | 0x13 ret = 0x0000_8067 return struct.pack(" tuple[int, bytes]: """Parse one memory item: ``{addr, hex|u64|u32|u8}``.""" if "addr" not in item: raise ValprobeError(f"memory item needs addr: {item!r}") addr = _parse_u64(item["addr"]) for key, fmt, size in (("hex", None, None), ("u64", " dict: if not os.path.isfile(path): raise ValprobeError(f"spec not found: {path}") with open(path, "r", encoding="utf-8") as fh: spec = json.load(fh) fmt = spec.get("format", SPEC_FORMAT) if fmt != SPEC_FORMAT: raise ValprobeError(f"unsupported spec format {fmt!r} (want {SPEC_FORMAT})") if not spec.get("variants"): raise ValprobeError("spec has no variants") return spec def _apply_state(machine, *, registers: dict, memory, stubs, default_sp: int) -> list[dict]: """Write stubs, memory and registers; return the applied stub notes.""" stub_notes = [] for stub in stubs or (): addr = _parse_u64(stub["addr"]) if "return" not in stub: raise ValprobeError(f"stub needs return: {stub!r}") code = encode_return_stub(_parse_u64(stub["return"])) machine.write_mem(addr, code) stub_notes.append({"addr": f"{addr:#x}", "return": f"{_parse_u64(stub['return']):#x}", "code_hex": code.hex()}) for item in memory or (): addr, data = decode_memory_spec(item) machine.write_mem(addr, data) for name, value in (registers or {}).items(): machine.x[reg_index(name)] = _parse_u64(value) if machine.x[2] == 0: machine.x[2] = default_sp return stub_notes def _read_outcomes(machine, reads) -> list[dict]: out = [] for item in reads or (): addr = _parse_u64(item["addr"]) size = int(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"] if size == 8: record["u64"] = int.from_bytes(data, "little") elif size == 4: record["u32"] = int.from_bytes(data, "little") out.append(record) return out def run_probe( *, image: str = DEFAULT_IMAGE, profile_path: str | None = DEFAULT_PROFILE, spec: dict, variant: dict, out: str | None = None, record: bool | None = None, name: str = "valprobe", ) -> dict: """Run one variant of a probe spec in a fresh machine; return its report.""" if not os.path.isfile(image): raise ValprobeError(f"image not found: {image}") image_obj = load_elf(image) if not image_obj.is_riscv: raise ValprobeError(f"{image}: not a RISC-V ELF image") profile = load_profile(profile_path) if profile_path else None load = (profile or {}).get("load", {}) ram_base = _parse_u64(load.get("ram_base", 0)) memory_mb = int(load.get("memory_mb", 256)) ram_top = ram_base + memory_mb * 1024 * 1024 entry = _parse_u64(variant.get("entry", spec.get("target"))) outcomes = {_parse_u64(a): label for a, label in (variant.get("outcomes") or spec.get("outcomes") or {}).items()} if not outcomes: raise ValprobeError("probe needs at least one outcome address") budget = int(variant.get("budget", spec.get("budget", DEFAULT_BUDGET))) timeout = float(variant.get("timeout", spec.get("timeout", DEFAULT_TIMEOUT))) label = variant.get("label") or f"variant-{entry:#x}" boot_to = spec.get("boot_to") or variant.get("boot_to") 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}-{label}"[:80], device_map=None, unknown_mmio="lenient", bias=0, ) machine = session.target.machine shim_meta: dict[str, Any] = {"enabled": False, "assumptions": []} libos = LibosSyscalls() boot: dict | None = None try: if profile is not None: machine.devices = registry_from_profile(profile, machine=machine) machine.translator = getattr(machine.devices, "dma", None) machine.syscall_handler = libos.handle shim = dict(profile.get("boot_shim", {})) shim["_image_path"] = image _apply_boot_shim(machine, shim, image_obj, shim_meta) shim_meta["enabled"] = True if boot_to: milestones: list[str] = [] reason, pc, _a, _b = _run_with_milestones(machine, BOOT_INSTRUCTIONS, milestones) rung = _rung_from_run(machine, shim_meta, milestones, libos) boot = { "requested": boot_to, "rung": rung, "instructions": machine.insns, "stop_pc": f"{pc:#x}", "milestones": milestones, "reached": RUNG_RANK.get(rung, -1) >= RUNG_RANK.get(str(boot_to), 99), } sentinel = (ram_top - 0x100) & ~0xF default_sp = (ram_top - 0x8000) & ~0xF machine.write_mem(sentinel, EBREAK.to_bytes(4, "little")) breakpoints = set(outcomes) breakpoints.add(sentinel) registers = dict(spec.get("registers") or {}) registers.update(variant.get("registers") or {}) registers.setdefault("ra", sentinel) memory = list(spec.get("memory") or ()) + list(variant.get("memory") or ()) stubs = list(spec.get("stubs") or ()) + list(variant.get("stubs") or ()) stubs_applied = _apply_state( machine, registers=registers, memory=memory, stubs=stubs, default_sp=default_sp ) machine.pc = entry entered = {f"x{i}": machine.x[i] for i in range(32) if i and machine.x[i]} entered["pc"] = entry started = time.monotonic() status = "budget" stop_pc = entry 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) remaining -= chunk if reason == R_BP: if stop_pc == sentinel: status = "return" break outcome = 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" report: dict[str, Any] = { "format": PROBE_FORMAT, "probe": spec.get("name"), "label": label, "image": os.path.abspath(image), "image_sha256": _sha256_file(image), "profile": os.path.basename(profile_path) if profile_path else None, "entry": f"{entry:#x}", "status": status, "outcome": outcome, "expect": variant.get("expect"), "pass": (None if variant.get("expect") is None else outcome == variant.get("expect")), "stop_pc": f"{stop_pc:#x}", "fault": fault, "retval": machine.x[10], "instructions": machine.insns, "wall_seconds": round(time.monotonic() - started, 4), "boot": boot, "stubs": stubs_applied, "registers_entered": {k: f"{v:#x}" if isinstance(v, int) else v for k, v in entered.items()}, "reads": _read_outcomes(machine, variant.get("reads") or spec.get("reads")), "limits": [ "Predicate-branch probe, not a boot: only the seeded state exists.", "Stubs (if any) replace an out-of-predicate dependency; they are listed.", "No timing/cycle claim (L0/L1 fidelity, docs/06).", "A fault is the first missing state, not a product defect.", ], } if getattr(session.recorder, "dir", None) is not None: session.note("valprobe", label=label, outcome=outcome, status=status) report["session"] = session.recorder.dir return report finally: session.close() def run_spec( *, spec: dict, image: str | None = None, profile_path: str | None = None, out: str | None = None, record: bool | None = None, name: str | None = None, ) -> dict: """Run every variant of a spec; return a matrix report.""" image = image or spec.get("image") or DEFAULT_IMAGE profile_path = profile_path or spec.get("profile") or DEFAULT_PROFILE started = time.monotonic() rows = [] for variant in spec["variants"]: rows.append(run_probe( image=image, profile_path=profile_path, spec=spec, variant=variant, out=out, record=record, name=name or (spec.get("name") or "valprobe"), )) matrix: dict[str, Any] = { "format": MATRIX_FORMAT, "spec": spec.get("name"), "spec_format": spec.get("format"), "image": os.path.abspath(image), "image_sha256": _sha256_file(image), "target": spec.get("target"), "boot_to": spec.get("boot_to"), "variants": rows, "passed": sum(1 for r in rows if r["pass"] is True), "failed": sum(1 for r in rows if r["pass"] is False), "evaluated": sum(1 for r in rows if r["pass"] is not None), "wall_seconds": round(time.monotonic() - started, 4), } if out: os.makedirs(out, exist_ok=True) path = os.path.join(out, "valprobe-matrix.json") with open(path, "w", encoding="utf-8") as fh: json.dump(matrix, fh, indent=2, sort_keys=True) fh.write("\n") matrix["report_json"] = os.path.abspath(path) return matrix # --------------------------------------------------------------------------- # Bundled target: the GR PROMOTE_CTX FECS predicate (570.144 firmware). # # Firmware facts used (all verified by disassembly of the shipped image): # * per-buffer validation branch point: 0x5290dd4 (in FUN_ram_05290a48) # * bInitialize=1 and bufferId==9 -> FECS branch at 0x5290e3c # * FECS branch: instance = pGpu[(idx+0x37c)], idx = *(u32*)(getter+0xD28); # table = *(u64*)(instance+0x190); slot = table[mapped(bufferId)*8]; # refuse (0x1f) iff slot != 0 -- regardless of physAttr/size/VA # * bufferId -> slot index via the firmware's own 0x531b944 jump table # (FECS => 6) # * bInitialize=0: bNonmapped!=0 => 0x1f, else accept # # The instance-index getter (0x524a5b8, a lock/table walk outside the # predicate) is stubbed with a recorded constant return so the branch can run # with crafted state; the predicate itself, the mapping table call and the # slot read are the firmware's own code. # --------------------------------------------------------------------------- PROMOTE_VALIDATION = 0x5290DD4 FECS_INIT_BRANCH = 0x5290E3C INSTANCE_GETTER = 0x524A5B8 OUTCOMES = { "0x5290cd0": "accept-commit-path", "0x5290ce4": "accept-skip-path", "0x5290df8": "refuse-invalid-argument-0x1f", "0x5290ebc": "refuse-fecs-slot-exists-0x1f", } #: Buffer ids (NV2080_CTRL_GPU_PROMOTE_CTX_BUFFER_ID_*), 570.144 interface order. BUFFER_ID = { "MAIN": 0, "PM": 1, "PATCH": 2, "BUNDLE_CB": 3, "PAGEPOOL": 4, "ATTRIBUTE_CB": 5, "RTV_CB_GLOBAL": 6, "GFXP_POOL": 7, "GFXP_CTRL_BLK": 8, "FECS_EVENT": 9, "PRIV_ACCESS_MAP": 10, "UNRESTRICTED_PRIV_ACCESS_MAP": 11, "GLOBAL_PRIV_ACCESS_MAP": 12, } #: Scratch base for crafted predicate state. Must be encodable by the #: ``lui+addi`` return stub (below 2^31), inside RAM, outside the image #: (max vaddr ~0x5bb9000), outside the 0x0..0x8000000 DMA window and below #: the 0x100000000 DMEM aperture the firmware uses after relocation. FECS_SCRATCH_BASE = 0x10000000 def _fecs_layout() -> dict: """Deterministic scratch layout (see module doc).""" scratch = FECS_SCRATCH_BASE return { "scratch": scratch, "sp": scratch + 0x4000, "s0": scratch + 0x5000, "pgpu": scratch + 0x1000, "getter_out": scratch + 0x1800, "instance": scratch + 0x3800, "entry": scratch + 0x6000, } def _fecs_variant(layout: dict, *, label: str, buffer_id: int, b_initialize: int, b_nonmapped: int, phys_attr: int = 4, size: int = 0x10000, phys: int = 0x1F4150000, virt: int = 0x204608000, slot: int | None = None, expect: str | None = None) -> dict: lay = layout entry = lay["entry"] # table = *(u64*)(instance+0x190) = instance+0x300; FECS maps to slot 6. slot_addr = lay["instance"] + 0x300 + 6 * 8 memory = [ {"addr": f"{entry + 0x30:#x}", "u64": b_initialize and phys or 0}, {"addr": f"{entry + 0x38:#x}", "u64": virt}, {"addr": f"{entry + 0x40:#x}", "u64": size}, {"addr": f"{entry + 0x48:#x}", "u32": phys_attr}, {"addr": f"{entry + 0x4c:#x}", "u16": buffer_id}, {"addr": f"{entry + 0x4e:#x}", "u8": b_initialize}, {"addr": f"{entry + 0x4f:#x}", "u8": b_nonmapped}, # pGpu -> instance pointer; getter result -> idx 0; instance -> table. {"addr": f"{lay['pgpu'] + 0x1be0:#x}", "u64": lay["instance"]}, {"addr": f"{lay['instance'] + 0x190:#x}", "u64": lay["instance"] + 0x300}, ] if slot is not None: memory.append({"addr": f"{slot_addr:#x}", "u64": slot}) return { "label": label, "registers": { "s0": f"{lay['s0']:#x}", "s2": f"{lay['pgpu']:#x}", "s5": f"{entry:#x}", "sp": f"{lay['sp']:#x}", "a3": buffer_id, "t1": 0xC, # loop bound set by the function before our entry point }, "stubs": [{"addr": f"{INSTANCE_GETTER:#x}", "return": f"{lay['getter_out']:#x}"}], "memory": memory, "reads": [{"addr": f"{lay['s0'] - 0x1c8:#x}", "size": 4, "label": "firmware-mapped-slot-index"}, {"addr": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot-value"}], "expect": expect, } def fecs_spec(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFILE) -> dict: """Build the FECS promote-validation matrix spec (bundled target).""" layout = _fecs_layout() variants = [ # Live m3j: FECS init (PA) refused -- slot already populated. _fecs_variant(layout, label="FECS init vidmem, slot non-NULL", buffer_id=9, b_initialize=1, b_nonmapped=0, slot=1, expect="refuse-fecs-slot-exists-0x1f"), # Predicate inverse: the same shape would be accepted on an empty slot. _fecs_variant(layout, label="FECS init vidmem, slot NULL", buffer_id=9, b_initialize=1, b_nonmapped=0, slot=0, expect="accept-commit-path"), # The handoff's untested variant (a): does a sysmem aperture change it? _fecs_variant(layout, label="FECS init sysmem attr 0x6, slot non-NULL", buffer_id=9, b_initialize=1, b_nonmapped=0, phys_attr=6, slot=1, expect="refuse-fecs-slot-exists-0x1f"), _fecs_variant(layout, label="FECS init sysmem attr 0x6, slot NULL", buffer_id=9, b_initialize=1, b_nonmapped=0, phys_attr=6, slot=0, expect="accept-commit-path"), # Size/PA content must not matter to the predicate. _fecs_variant(layout, label="FECS init size 0x1000, slot non-NULL", buffer_id=9, b_initialize=1, b_nonmapped=0, size=0x1000, slot=1, expect="refuse-fecs-slot-exists-0x1f"), # Live m7: FECS VA-only accepted (bInitialize=0 path). _fecs_variant(layout, label="FECS no-init mapped, slot non-NULL", buffer_id=9, b_initialize=0, b_nonmapped=0, slot=1, expect="accept-skip-path"), _fecs_variant(layout, label="FECS no-init nonmapped (bookkeeping refusal)", buffer_id=9, b_initialize=0, b_nonmapped=1, slot=1, expect="refuse-invalid-argument-0x1f"), # Live m3j: PM and PATCH init accepted. _fecs_variant(layout, label="PM init", buffer_id=1, b_initialize=1, b_nonmapped=0, expect="accept-commit-path"), _fecs_variant(layout, label="PATCH init", buffer_id=2, b_initialize=1, b_nonmapped=0, expect="accept-commit-path"), _fecs_variant(layout, label="GFXP_CTRL_BLK init (id 8)", buffer_id=8, b_initialize=1, b_nonmapped=0, expect="accept-skip-path"), _fecs_variant(layout, label="BUNDLE_CB init (id 3)", buffer_id=3, b_initialize=1, b_nonmapped=0, expect="refuse-invalid-argument-0x1f"), _fecs_variant(layout, label="PRIV_ACCESS_MAP init (id 10)", buffer_id=10, b_initialize=1, b_nonmapped=0, expect="accept-skip-path"), _fecs_variant(layout, label="UNRESTRICTED_PRIV init (id 11)", buffer_id=11, b_initialize=1, b_nonmapped=0, expect="accept-skip-path"), _fecs_variant(layout, label="GLOBAL_PRIV_ACCESS_MAP init (id 12)", buffer_id=12, b_initialize=1, b_nonmapped=0, expect="refuse-invalid-argument-0x1f"), ] # id > 0xc is refused one block earlier (0x5290cc4, still in the per-entry # loop); enter there with the loop's a4/t1 state. unknown_id = _fecs_variant(layout, label="unknown buffer id 13", buffer_id=13, b_initialize=1, b_nonmapped=0, expect="refuse-invalid-argument-0x1f") unknown_id["entry"] = "0x5290cc0" unknown_id["registers"]["a4"] = 13 variants.append(unknown_id) return { "format": SPEC_FORMAT, "name": "fecs-promote-validation", "target": f"{PROMOTE_VALIDATION:#x}", "outcomes": OUTCOMES, "budget": 20_000, "timeout": 20.0, "image": image, "profile": profile_path, "note": ( "570.144 per-buffer PROMOTE_CTX validation; instance-index getter " f"({INSTANCE_GETTER:#x}, lock/table walk outside the predicate) stubbed " "to a recorded constant; the predicate, the id->slot mapping call and " "the memDesc slot read are the firmware's own code." ), "variants": variants, } def run_demo(out: str | None = None) -> dict: """Run the bundled FECS matrix on the real image, twice for determinism.""" spec = fecs_spec() first = run_spec(spec=spec, out=out, record=False, name="valprobe-fecs") second = run_spec(spec=spec, record=False, name="valprobe-fecs") verdicts_a = [(r["label"], r["outcome"]) for r in first["variants"]] verdicts_b = [(r["label"], r["outcome"]) for r in second["variants"]] result = { "format": DEMO_FORMAT, "ok": first["failed"] == 0 and verdicts_a == verdicts_b, "evaluated": first["evaluated"], "failed": first["failed"], "deterministic": verdicts_a == verdicts_b, "verdicts": [{"label": label, "outcome": outcome} for label, outcome in verdicts_a], } if out: os.makedirs(out, exist_ok=True) path = os.path.join(out, "valprobe-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 build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="valprobe.py", description=( "Run one firmware validation-predicate branch with crafted state " "(offline; never touches hardware)." ), ) sub = parser.add_subparsers(dest="command") sweep = sub.add_parser("sweep", help="run a bundled target matrix") sweep.add_argument("--target", default="fecs-promote", choices=["fecs-promote"]) sweep.add_argument("--image", default=None) sweep.add_argument("--profile", default=None) sweep.add_argument("--boot-to", default=None, help="boot the image to a rung first (e.g. B3)") sweep.add_argument("--out", help="output directory (matrix JSON + sessions)") sweep.add_argument("--no-record", action="store_true", help="skip session recording") run = sub.add_parser("run", help="run a probe spec file") run.add_argument("--spec", required=True) run.add_argument("--image", default=None) run.add_argument("--profile", default=None) run.add_argument("--out", help="output directory (matrix JSON + sessions)") run.add_argument("--no-record", action="store_true", help="skip session recording") demo = sub.add_parser("demo", help="bundled FECS matrix, run twice (determinism)") 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 == "sweep": spec = fecs_spec(image=args.image or DEFAULT_IMAGE, profile_path=args.profile or DEFAULT_PROFILE) if args.boot_to: spec["boot_to"] = args.boot_to matrix = run_spec(spec=spec, image=args.image, profile_path=args.profile, out=args.out, record=False if args.no_record else None) elif args.command == "run": spec = _load_spec(args.spec) matrix = run_spec(spec=spec, image=args.image, profile_path=args.profile, out=args.out, record=False if args.no_record else None) elif args.command == "demo": matrix = run_demo(args.out) else: parser.print_help() return 2 except ValprobeError as exc: print(f"error: {exc}", file=sys.stderr) return 2 if matrix.get("format") == MATRIX_FORMAT: for row in matrix["variants"]: mark = {True: "PASS", False: "FAIL", None: "----"}[row["pass"]] print(f"{mark} {row['label']:<48} {row['outcome'] or row['status']}") print(json.dumps({k: matrix[k] for k in ("format", "spec", "evaluated", "passed", "failed", "wall_seconds", "report_json") if k in matrix}, indent=2, sort_keys=True)) return 3 if matrix["failed"] else 0 print(json.dumps(matrix, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())