#!/usr/bin/env python3 """Version diff: old vs new firmware targets (roadmap H4.1, planned docs/21). Compares two firmware images (a built ELF or a raw ``.fwimage``) section by section; every section degrades to ``status: skipped`` with a reason instead of aborting the report: 1. image/shape -- entry, PT_LOAD layout, ELF section sizes, hashes, span; 2. symbol inventory -- ELF ``.symtab`` plus Ghidra names/index/exports TSVs (including the 3-column ``ExportForGreenEye`` form) and green-eye JSON DBs; added/removed/moved/resized with the anonymous-shift matching discipline of ``userspace/cuda/tools/diff_corpus.py``; 3. decompiled hints -- a corpus diff over two ``decompiled_index.tsv`` files, an attached corpus report, or an explicit "not supplied"; 4. probe -- bounded QEMU first-touch probe of both targets with ``parse_qemu_faults`` and firmware-vaddr translation of the epc and access addresses; 5. behavior -- the shared demo scenario (or a ``--stimulus`` script) under QEMU, ``analyze.compare_sessions`` plus hit/watch/write deltas and the docs/06 M1-M6 skeleton. Everything is local and stdlib-only: nothing is downloaded. The RM 610 gate (docs/23) is enforced before any target file is opened. Run from the tool root: python3 harness/version_diff.py --old old.elf --new new.elf --out dumps/vdiff """ from __future__ import annotations import argparse import contextlib import os import re import shutil import statistics import struct import sys import tempfile from collections import defaultdict from dataclasses import dataclass from pathlib import Path try: from .common import ( LOAD_ADDR, MAX_FLAT_BYTES, MAX_LOG_BYTES, SymbolDB, counter_delta, ensure_dir, memory_mb_for, qemu_available, read_log_window, sha256_file, write_json, write_text, ) from .portability_matrix import ( CTRL_RPC_COUNT, METRIC_DEFS, NO_REFERENCE, install_scenario, metric_rows, profile_from_events, read_ctrl, ) except ImportError: # direct script execution (python3 harness/version_diff.py) from common import ( # type: ignore[no-redef] LOAD_ADDR, MAX_FLAT_BYTES, MAX_LOG_BYTES, SymbolDB, counter_delta, ensure_dir, memory_mb_for, qemu_available, read_log_window, sha256_file, write_json, write_text, ) from portability_matrix import ( # type: ignore[no-redef] CTRL_RPC_COUNT, METRIC_DEFS, NO_REFERENCE, install_scenario, metric_rows, profile_from_events, read_ctrl, ) from greeneye.analyze import compare_sessions, parse_qemu_faults # noqa: E402 from greeneye.elf import EM_RISCV, FirmwareImage, Segment, load_elf, write_flat_kernel # noqa: E402 from greeneye.errors import GreenEyeError # noqa: E402 from greeneye.record import load_events # noqa: E402 from greeneye.rsp import RspClient # noqa: E402 from greeneye.session import Session # noqa: E402 from greeneye.symbols import Symbol # noqa: E402 from greeneye.target import QemuTarget # noqa: E402 FORMAT = "greeneye-version-diff/1" RM610_DOC = "docs/23-RM610-READINESS.md" RM610_TOKEN = "RM610-APPROVED" RM610_ENV = "GREENEYE_RM610_APPROVED" RM610_RE = re.compile(r"rm[\s._-]*610", re.IGNORECASE) PT_LOAD = 1 class Rm610GateError(GreenEyeError): """A target/release carries an RM 610 label without explicit approval.""" def is_rm610_label(value: str | None) -> bool: return bool(value) and RM610_RE.search(str(value)) is not None def rm610_approved(token: str | None = None, environ: dict | None = None) -> bool: """Explicit approval is ``--approval-token RM610-APPROVED`` or the env flag.""" env = os.environ if environ is None else environ token_ok = (token or "").strip() == RM610_TOKEN env_ok = str(env.get(RM610_ENV, "")).strip().lower() in ("1", "true", "yes", "approved") return token_ok or env_ok def check_rm610_gate( labels: list[str] | tuple[str, ...], *, approval_token: str | None = None, environ: dict | None = None, ) -> None: """Refuse RM 610 labels before anything is read, opened or downloaded.""" blocked = sorted({str(label) for label in labels if is_rm610_label(label)}) if not blocked: return if rm610_approved(approval_token, environ): return raise Rm610GateError( "target/release label(s) " + ", ".join(repr(label) for label in blocked) + " indicate RM 610. " f"Refused without explicit approval: pass --approval-token {RM610_TOKEN} or set " f"{RM610_ENV}=1. See {RM610_DOC}. Nothing was downloaded and no target file was opened." ) # --------------------------------------------------------------------------- # target loading (ELF or raw) # --------------------------------------------------------------------------- @dataclass class LoadedTarget: path: str name: str kind: str # elf | raw | unsupported | missing image: FirmwareImage | None = None sha256: str = "" size: int = 0 entry: int = 0 bias: int = 0 reason: str | None = None @property def ok(self) -> bool: return self.kind in ("elf", "raw") def _read_bytes(path: str) -> bytes: with open(path, "rb") as fh: return fh.read() def load_any_target(path: str) -> LoadedTarget: """Read a target as RISC-V ELF or as a raw relocated image (no download).""" name = Path(path).stem if not os.path.isfile(path): return LoadedTarget(path=path, name=name, kind="missing", reason="file not found") size = os.path.getsize(path) sha = sha256_file(path) with open(path, "rb") as fh: magic = fh.read(4) if magic == b"\x7fELF": try: image = load_elf(path) except Exception as exc: # malformed ELF -> skipped, never fatal return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size, reason=f"{type(exc).__name__}: {exc}") if not image.is_riscv: return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size, reason=f"not a RISC-V ELF (e_machine={image.e_machine})") span = image.max_vaddr - image.min_vaddr if span > MAX_FLAT_BYTES: return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size, reason=f"image span {span:#x} exceeds the {MAX_FLAT_BYTES:#x} harness cap") return LoadedTarget(path=path, name=name, kind="elf", image=image, sha256=sha, size=size, entry=image.entry, bias=LOAD_ADDR - image.min_vaddr) if size == 0: return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=0, reason="empty raw image") if size > MAX_FLAT_BYTES: return LoadedTarget(path=path, name=name, kind="unsupported", sha256=sha, size=size, reason=f"raw image {size:#x} exceeds the {MAX_FLAT_BYTES:#x} harness cap") data = _read_bytes(path) segment = Segment(p_type=PT_LOAD, p_flags=0x7, p_offset=0, p_vaddr=0, p_paddr=0, p_filesz=len(data), p_memsz=len(data), p_align=0x1000) image = FirmwareImage(path=path, is64=True, e_machine=EM_RISCV, entry=0, segments=[segment], symbols=[], _blob=data) return LoadedTarget(path=path, name=name, kind="raw", image=image, sha256=sha, size=size, entry=0, bias=LOAD_ADDR) # --------------------------------------------------------------------------- # 1. image / shape # --------------------------------------------------------------------------- def _cstr(blob: bytes, offset: int) -> str: if offset >= len(blob): return "" end = blob.find(b"\x00", offset) if end == -1: end = len(blob) return blob[offset:end].decode("utf-8", "replace") def elf_sections(data: bytes) -> list[dict]: """Section name/type/addr/size table (minimal reader; [] when unusable).""" if len(data) < 64 or data[:4] != b"\x7fELF" or data[4] not in (1, 2): return [] is64 = data[4] == 2 try: if is64: sh_off = struct.unpack_from(" len(data): return [] shstr = b"" if 0 <= shstrndx < shnum: base = sh_off + shstrndx * shentsize if base + sh_size <= len(data): sh = struct.unpack_from(sh_fmt, data, base) shstr = data[sh[4] : sh[4] + sh[5]] sections: list[dict] = [] for index in range(shnum): base = sh_off + index * shentsize if base + sh_size > len(data): break sh = struct.unpack_from(sh_fmt, data, base) sections.append( { "name": _cstr(shstr, sh[0]) or f"", "type": sh[1], "addr": sh[3], "offset": sh[4], "size": sh[5], } ) return sections def shape_payload(loaded: LoadedTarget) -> dict: payload = { "kind": loaded.kind, "sha256": loaded.sha256, "size": loaded.size, "reason": loaded.reason, } if not loaded.ok or loaded.image is None: return payload image = loaded.image payload.update( { "entry": loaded.entry, "bias": loaded.bias, "min_vaddr": image.min_vaddr, "max_vaddr": image.max_vaddr, "span": image.max_vaddr - image.min_vaddr, "loads": [ { "index": index, "vaddr": seg.p_vaddr, "paddr": seg.p_paddr, "filesz": seg.p_filesz, "memsz": seg.p_memsz, "flags": seg.p_flags, "align": seg.p_align, } for index, seg in enumerate(image.load_segments) ], } ) if loaded.kind == "elf": payload["sections"] = elf_sections(image.read()) else: payload["sections"] = [] payload["note"] = "raw image: one implicit load at vaddr 0; no ELF section table" return payload def _scalar_delta(a, b, name: str) -> dict: return {"a": a, "b": b, "delta": b - a if isinstance(a, int) and isinstance(b, int) else None, "changed": a != b} def _cap(rows: list, max_rows: int) -> list: return rows[:max_rows] def image_diff(old: LoadedTarget, new: LoadedTarget, max_rows: int) -> dict: a, b = shape_payload(old), shape_payload(new) result = {"status": "skipped", "reason": None, "a": a, "b": b, "same": None, "deltas": None} if not old.ok or not new.ok: result["reason"] = "; ".join( f"{side}: {loaded.reason or loaded.kind}" for side, loaded in (("old", old), ("new", new)) if not loaded.ok ) return result loads_a, loads_b = a["loads"], b["loads"] load_added = [payload for index, payload in enumerate(loads_b) if index >= len(loads_a)] load_removed = [payload for index, payload in enumerate(loads_a) if index >= len(loads_b)] load_changed = [] for index, (la, lb) in enumerate(zip(loads_a, loads_b)): fields = {key: _scalar_delta(la[key], lb[key], key) for key in ("vaddr", "filesz", "memsz", "flags")} if any(row["changed"] for row in fields.values()): load_changed.append({"index": index, "fields": fields}) sec_a = {row["name"]: row for row in a.get("sections", [])} sec_b = {row["name"]: row for row in b.get("sections", [])} sec_added = [{key: sec_b[name][key] for key in ("name", "type", "addr", "size")} for name in sorted(set(sec_b) - set(sec_a))] sec_removed = [{key: sec_a[name][key] for key in ("name", "type", "addr", "size")} for name in sorted(set(sec_a) - set(sec_b))] sec_moved, sec_resized = [], [] for name in sorted(set(sec_a) & set(sec_b)): ra, rb = sec_a[name], sec_b[name] if ra["addr"] != rb["addr"]: sec_moved.append({"name": name, "old_addr": ra["addr"], "new_addr": rb["addr"]}) if ra["size"] != rb["size"]: sec_resized.append({"name": name, "old_size": ra["size"], "new_size": rb["size"]}) deltas = { "entry": _scalar_delta(a["entry"], b["entry"], "entry"), "span": _scalar_delta(a["span"], b["span"], "span"), "file_size": _scalar_delta(a["size"], b["size"], "size"), "loads": { "added": _cap(load_added, max_rows), "removed": _cap(load_removed, max_rows), "changed": _cap(load_changed, max_rows), }, "sections": { "added": _cap(sec_added, max_rows), "removed": _cap(sec_removed, max_rows), "moved": _cap(sec_moved, max_rows), "resized": _cap(sec_resized, max_rows), }, } same = ( not deltas["entry"]["changed"] and not deltas["span"]["changed"] and not deltas["file_size"]["changed"] and not any(deltas["loads"][key] for key in ("added", "removed", "changed")) and not any(deltas["sections"][key] for key in ("added", "removed", "moved", "resized")) ) result.update({"status": "ok", "same": same, "deltas": deltas}) return result # --------------------------------------------------------------------------- # 2. symbol inventory (diff_corpus.py matching discipline) # --------------------------------------------------------------------------- def looks_like_three_column_export(path: str) -> bool: """``addrnamesize`` (ExportForGreenEye) vs names/exports TSVs.""" try: with open(path, "r", encoding="utf-8", errors="replace") as fh: for line in fh: line = line.strip() if not line or line.startswith("#"): continue cols = line.split("\t") if len(cols) >= 3 and [c.strip() for c in cols[:3]] == ["addr", "name", "size"]: return True if len(cols) == 3: try: int(cols[0], 16) int(cols[2], 0) return True except ValueError: return False return False except OSError: return False return False def load_three_column_export(db: SymbolDB, path: str, bias: int = 0) -> int: count = 0 with open(path, "r", encoding="utf-8", errors="replace") as fh: for line in fh: line = line.rstrip("\n") if not line or line.startswith("#"): continue cols = line.split("\t") if len(cols) != 3 or cols[0].strip() == "addr": continue try: addr = int(cols[0], 16) size = int(cols[2], 0) except ValueError: continue if not cols[1]: continue db.add(Symbol(addr=addr + bias, name=cols[1], size=size, source="ghidra-export")) count += 1 return count def load_symbol_sources( loaded: LoadedTarget, *, symbols_json: str | None = None, ghidra_names: str | None = None, ghidra_index: str | None = None, ghidra_exports: str | None = None, ghidra_bias: int = 0, ) -> tuple[SymbolDB, list[str], list[str]]: """Merge every supplied symbol source for one side into one SymbolDB.""" db = SymbolDB() sources: list[str] = [] notes: list[str] = [] if loaded.ok and loaded.image is not None and loaded.image.symbols: count = db.load_elf(loaded.image, bias=loaded.bias, source="elf") if count: sources.append(f"elf:{count}") if symbols_json: if os.path.isfile(symbols_json): count = db.load_json(symbols_json) sources.append(f"{os.path.basename(symbols_json)}:{count}") else: notes.append(f"symbol DB not found: {symbols_json}") if ghidra_names: if os.path.isfile(ghidra_names): count = db.load_names_tsv(ghidra_names, bias=ghidra_bias) sources.append(f"{os.path.basename(ghidra_names)}:{count}") else: notes.append(f"Ghidra names TSV not found: {ghidra_names}") if ghidra_index: if os.path.isfile(ghidra_index): count = db.load_index_tsv(ghidra_index, bias=ghidra_bias) sources.append(f"{os.path.basename(ghidra_index)}:{count}") else: notes.append(f"Ghidra index TSV not found: {ghidra_index}") if ghidra_exports: if os.path.isfile(ghidra_exports): if looks_like_three_column_export(ghidra_exports): count = load_three_column_export(db, ghidra_exports, bias=ghidra_bias) sources.append(f"{os.path.basename(ghidra_exports)}:{count} (3-column export)") else: count = db.load_index_tsv(ghidra_exports, bias=ghidra_bias) sources.append(f"{os.path.basename(ghidra_exports)}:{count}") else: notes.append(f"Ghidra exports TSV not found: {ghidra_exports}") return db, sources, notes def inventory_records(db: SymbolDB) -> list[dict]: records = [ { "name": symbol.name, "addr": symbol.addr, "size": symbol.size, "source": symbol.source, } for symbol in db ] records.sort(key=lambda row: (row["addr"], row["name"], row["size"])) return records def _is_named(name: str) -> bool: return bool(name) and not name.startswith("FUN_") def _record_brief(record: dict | None) -> dict | None: if record is None: return None return {"addr": record["addr"], "name": record["name"], "size": record["size"]} def inventory_diff(a_records: list[dict], b_records: list[dict], max_rows: int = 40) -> dict: """Named-by-name + anonymous-at-shifted-address matching (diff_corpus.py).""" a_by_addr = {row["addr"]: row for row in a_records} b_by_addr = {row["addr"]: row for row in b_records} named_a: dict[str, list[int]] = defaultdict(list) named_b: dict[str, list[int]] = defaultdict(list) for row in a_records: if _is_named(row["name"]): named_a[row["name"]].append(row["addr"]) for row in b_records: if _is_named(row["name"]): named_b[row["name"]].append(row["addr"]) matched_named: list[tuple[str, int, int]] = [] for name in sorted(set(named_a) & set(named_b)): for addr_a, addr_b in zip(sorted(named_a[name]), sorted(named_b[name])): matched_named.append((name, addr_a, addr_b)) deltas = [addr_b - addr_a for _, addr_a, addr_b in matched_named] shift = int(statistics.median(deltas)) if deltas else 0 dominant = (sum(1 for value in deltas if abs(value - shift) <= 0x40) / len(deltas)) if deltas else 0.0 used_a = {addr_a for _, addr_a, _ in matched_named} used_b = {addr_b for _, _, addr_b in matched_named} moved = [(name, addr_a, addr_b) for name, addr_a, addr_b in matched_named if addr_a != addr_b] resized_named = [ (name, addr_a, addr_b) for name, addr_a, addr_b in matched_named if a_by_addr[addr_a]["size"] != b_by_addr[addr_b]["size"] ] shifted: list[tuple[int, int]] = [] for row in sorted(a_records, key=lambda item: item["addr"]): addr_a = row["addr"] if addr_a in used_a or _is_named(row["name"]): continue addr_b = addr_a + shift if addr_b in b_by_addr and addr_b not in used_b: shifted.append((addr_a, addr_b)) used_a.add(addr_a) used_b.add(addr_b) shifted_resized = [(fa, fb) for fa, fb in shifted if a_by_addr[fa]["size"] != b_by_addr[fb]["size"]] shifted_ok = [(fa, fb) for fa, fb in shifted if a_by_addr[fa]["size"] == b_by_addr[fb]["size"]] removed = sorted(addr for addr in a_by_addr if addr not in used_a) added = sorted(addr for addr in b_by_addr if addr not in used_b) matched_total = len(matched_named) + len(shifted) named_count_a = sum(len(v) for v in named_a.values()) named_count_b = sum(len(v) for v in named_b.values()) return { "shift": shift, "shift_dominant": round(dominant, 4), "matched_named": len(matched_named), "matched_shifted": len(shifted_ok), "matched_shifted_resized": len(shifted_resized), "moved": _cap( [{"name": name, "old_addr": fa, "new_addr": fb, "delta": fb - fa} for name, fa, fb in moved], max_rows, ), "resized": _cap( [ { "name": name, "old_addr": fa, "new_addr": fb, "old_size": a_by_addr[fa]["size"], "new_size": b_by_addr[fb]["size"], } for name, fa, fb in resized_named ], max_rows, ), "resized_anonymous": _cap( [ { "old_addr": fa, "new_addr": fb, "old_size": a_by_addr[fa]["size"], "new_size": b_by_addr[fb]["size"], } for fa, fb in shifted_resized ], max_rows, ), "added": _cap([_record_brief(b_by_addr[addr]) for addr in added], max_rows), "removed": _cap([_record_brief(a_by_addr[addr]) for addr in removed], max_rows), "totals": { "a": len(a_records), "b": len(b_records), "named_a": named_count_a, "named_b": named_count_b, "anonymous_a": sum(1 for row in a_records if not _is_named(row["name"])), "anonymous_b": sum(1 for row in b_records if not _is_named(row["name"])), "added": len(added), "removed": len(removed), "moved": len(moved), "resized_named": len(resized_named), }, "rates": { "match_a": round(matched_total / len(a_records), 4) if a_records else 0.0, "match_b": round(matched_total / len(b_records), 4) if b_records else 0.0, "named_transfer": round(len(matched_named) / named_count_a, 4) if named_count_a else 0.0, }, } def symbol_section( old: LoadedTarget, new: LoadedTarget, old_db: SymbolDB, old_sources: list[str], old_notes: list[str], new_db: SymbolDB, new_sources: list[str], new_notes: list[str], options: "VersionDiffOptions", out_dir: str, ) -> dict: result = { "status": "skipped", "reason": None, "a": {"count": 0, "sources": old_sources, "notes": old_notes, "db": None}, "b": {"count": 0, "sources": new_sources, "notes": new_notes, "db": None}, } if not old.ok or not new.ok: result["reason"] = "; ".join( f"{side}: {loaded.reason or loaded.kind}" for side, loaded in (("old", old), ("new", new)) if not loaded.ok ) return result a_records = inventory_records(old_db) b_records = inventory_records(new_db) result["a"]["count"] = len(a_records) result["b"]["count"] = len(b_records) if not a_records and not b_records: result["reason"] = ( "no symbols on either side: no ELF .symtab and no Ghidra/JSON symbol sources supplied" ) return result result.update(inventory_diff(a_records, b_records, options.max_rows)) result["status"] = "ok" for key, db, loaded, sources, notes in ( ("a", old_db, old, old_sources, old_notes), ("b", new_db, new, new_sources, new_notes), ): if not len(db): continue rel = os.path.join("symbols", f"{loaded.name}.json") path = os.path.join(out_dir, rel) ensure_dir(os.path.dirname(path)) db.save_json(path, meta={"firmware": os.path.basename(loaded.path), "bias": loaded.bias, "sources": sources, "notes": notes}) result[key]["db"] = rel return result # --------------------------------------------------------------------------- # 3. decompiled-change hints # --------------------------------------------------------------------------- def corpus_records(path: str) -> list[dict]: db = SymbolDB() if looks_like_three_column_export(path): load_three_column_export(db, path, bias=0) else: db.load_index_tsv(path, bias=0) return inventory_records(db) def corpus_diff(path_a: str, path_b: str, max_rows: int = 40) -> dict: diff = inventory_diff(corpus_records(path_a), corpus_records(path_b), max_rows) diff["a"] = os.path.basename(path_a) diff["b"] = os.path.basename(path_b) return diff def corpus_section(options: "VersionDiffOptions", out_dir: str) -> dict: if options.corpus_report: if not os.path.isfile(options.corpus_report): return {"status": "skipped", "reason": f"corpus report not found: {options.corpus_report}", "mode": "corpus-report"} data = _read_bytes(options.corpus_report) preview = data.decode("utf-8", "replace") return { "status": "ok", "reason": None, "mode": "corpus-report", "report": { "path": os.path.basename(options.corpus_report), "sha256": sha256_file(options.corpus_report), "bytes": len(data), "preview": preview[:4000], "truncated": len(preview) > 4000, }, } if options.corpus_diff: path_a, path_b = options.corpus_diff missing = [path for path in (path_a, path_b) if not os.path.isfile(path)] if missing: return {"status": "skipped", "mode": "corpus-diff", "reason": "corpus TSV not found: " + ", ".join(missing)} diff = corpus_diff(path_a, path_b, options.max_rows) rel = os.path.join("corpus", "corpus-diff.json") write_json(os.path.join(out_dir, rel), diff) return {"status": "ok", "reason": None, "mode": "corpus-diff", "diff": diff, "report": rel} return { "status": "skipped", "reason": "not supplied: pass --corpus-diff OLD_TSV NEW_TSV or --corpus-report FILE", "mode": None, } # --------------------------------------------------------------------------- # QEMU sessions for ELF and raw targets # --------------------------------------------------------------------------- @dataclass class OpenTarget: loaded: LoadedTarget session: Session target: QemuTarget workdir: str owns_workdir: bool = False @contextlib.contextmanager def open_target_session( loaded: LoadedTarget, *, record: str | bool | None, name: str, symbols: SymbolDB | None = None, log_faults: bool = True, workdir: str | None = None, ): """Relocate the target to LOAD_ADDR, start QEMU and yield a live session.""" owns_workdir = workdir is None if workdir is None: workdir = tempfile.mkdtemp(prefix="greeneye-vdiff-") flat_kernel = os.path.join(workdir, "flat.bin") assert loaded.image is not None if loaded.kind == "elf": write_flat_kernel(loaded.path, flat_kernel, load_addr=LOAD_ADDR) else: with open(flat_kernel, "wb") as fh: fh.write(loaded.image.read()) target = QemuTarget( kernel=flat_kernel, memory_mb=memory_mb_for(loaded.image), dlog="guest_errors,unimp,int" if log_faults else None, workdir=workdir, ) target.start(paused=True) client = RspClient(target.host, target.port) client.connect() client.sync() db = symbols if symbols is not None else SymbolDB() session = Session( client=client, target=target, symbols=db, record=record, name=name, meta={ "backend": "qemu", "bias": loaded.bias, "firmware": loaded.path, "kind": loaded.kind, "harness": "version_diff", }, ) session.load_image(loaded.image, bias=loaded.bias, load_symbols=symbols is None) try: yield OpenTarget(loaded=loaded, session=session, target=target, workdir=workdir, owns_workdir=owns_workdir) finally: try: session.close() finally: if owns_workdir: shutil.rmtree(workdir, ignore_errors=True) def enrich_fault(fault: dict, bias: int, span: int) -> dict: """Add firmware-vaddr translations for epc and the access address.""" out = { "cause": fault.get("cause"), "desc": fault.get("desc"), "epc": fault.get("epc"), "tval": fault.get("tval"), "access": fault.get("access"), } epc = fault.get("epc") out["epc_firmware"] = epc - bias if isinstance(epc, int) else None access = fault.get("access") or {} access_addr = access.get("access_addr") if isinstance(access_addr, int): if LOAD_ADDR <= access_addr < LOAD_ADDR + span: out["access_firmware"] = access_addr - bias else: out["access_firmware"] = None else: out["access_firmware"] = None return out # --------------------------------------------------------------------------- # 4. probe first-touch diff # --------------------------------------------------------------------------- def _skipped_probe(reason: str | None) -> dict: return { "status": "skipped", "reason": reason, "session": None, "instructions": None, "stopped_pc": None, "stopped_symbol": None, "fault_count": 0, "faults_capped": False, "first_fault": None, } def run_probe( loaded: LoadedTarget, db: SymbolDB, options: "VersionDiffOptions", out_dir: str, qemu_ok: bool, ) -> dict: if not qemu_ok: return _skipped_probe("qemu-system-riscv64 not found") if not loaded.ok: return _skipped_probe(loaded.reason) result = _skipped_probe(None) result["instructions"] = options.probe_instructions session_rel = os.path.join("sessions", f"{loaded.name}-probe") try: with open_target_session( loaded, record=os.path.join(out_dir, session_rel), name=f"{loaded.name}-probe", symbols=db, log_faults=True, ) as ot: session = ot.session if options.probe_instructions > 0: session.step(options.probe_instructions) else: session.run(duration=options.probe_duration) result["stopped_pc"] = session.pc symbol = session.symbols.resolve(session.pc) result["stopped_symbol"] = symbol.name if symbol else None text, capped = read_log_window(ot.target.qemu_log, 0, options.log_bytes) faults = parse_qemu_faults(text) result["fault_count"] = len(faults) result["faults_capped"] = capped span = loaded.image.max_vaddr - loaded.image.min_vaddr if loaded.image else 0 result["first_fault"] = enrich_fault(faults[0], loaded.bias, span) if faults else None result["session"] = session_rel if text: session.recorder.meta["qemu_log_blob"] = session.recorder.blob("qemu.log", text.encode("utf-8")) session.recorder.meta["faults_capped"] = capped result["status"] = "ok" except Exception as exc: # a failed probe must not abort the diff result["status"] = "error" result["reason"] = f"{type(exc).__name__}: {exc}" return result def _probe_signature(probe: dict) -> tuple: first = probe.get("first_fault") or {} return ( probe.get("stopped_pc"), first.get("desc"), first.get("cause"), first.get("epc_firmware"), first.get("access_firmware"), (first.get("access") or {}).get("access_addr"), ) def probe_diff( old: LoadedTarget, new: LoadedTarget, old_db: SymbolDB, new_db: SymbolDB, options: "VersionDiffOptions", out_dir: str, qemu_ok: bool, ) -> dict: a = run_probe(old, old_db, options, out_dir, qemu_ok) b = run_probe(new, new_db, options, out_dir, qemu_ok) statuses = {a["status"], b["status"]} if statuses == {"ok"}: status, reason = "ok", None elif "error" in statuses: status = "error" reason = "; ".join(row["reason"] for row in (a, b) if row.get("reason")) or "probe failed" else: status = "skipped" reason = "; ".join(row["reason"] for row in (a, b) if row.get("reason")) or "probe not run" changed = _probe_signature(a) != _probe_signature(b) if status == "ok": summary = "first fault and stopped PC unchanged" if not changed else "first fault or stopped PC changed" else: summary = None return {"status": status, "reason": reason, "a": a, "b": b, "changed": changed, "summary": summary} # --------------------------------------------------------------------------- # 5. behavior diff # --------------------------------------------------------------------------- def _skipped_behavior(reason: str | None) -> dict: return { "status": "skipped", "reason": reason, "session": None, "hooks": [], "uart_banner": None, "ctrl": None, "event_counts": {}, "bp": {}, "watch": {}, "writes": {}, "fault_count": 0, "faults_capped": False, "first_fault": None, } def run_behavior( loaded: LoadedTarget, db: SymbolDB, options: "VersionDiffOptions", out_dir: str, qemu_ok: bool, ) -> dict: if not qemu_ok: return _skipped_behavior("qemu-system-riscv64 not found") if not loaded.ok: return _skipped_behavior(loaded.reason) result = _skipped_behavior(None) session_rel = os.path.join("sessions", f"{loaded.name}-behavior") try: with open_target_session( loaded, record=os.path.join(out_dir, session_rel), name=f"{loaded.name}-behavior", symbols=db, log_faults=True, ) as ot: session = ot.session result["hooks"] = install_scenario(session) ctrl_symbol = session.symbols.get("gsp_ctrl") ctrl_addr = ctrl_symbol.addr if ctrl_symbol else None if ctrl_addr is not None: session.watch(ctrl_addr + CTRL_RPC_COUNT, size=8, kind="w", listener="version-diff:rpc_count") if options.stimulus: import greeneye as ge script_path = Path(options.stimulus) namespace = { "ge": ge, "greeneye": ge, "session": session, "s": session, "memory": session.memory, "symbols": session.symbols, "recorder": session.recorder, } exec(compile(script_path.read_text(encoding="utf-8"), str(script_path), "exec"), namespace) else: session.run(duration=options.behavior_duration, max_stops=options.max_stops) if ctrl_addr is not None: session.memory.write_u64(ctrl_addr + 32, 1) # PING session.memory.write_u64(ctrl_addr + 40, 0x1234) session.run(duration=options.behavior_duration, max_stops=options.max_stops) result["ctrl"] = read_ctrl(session, ctrl_symbol) uart = ot.target.read_uart() result["uart_banner"] = uart.strip().splitlines()[0] if uart.strip() else None text, capped = read_log_window(ot.target.qemu_log, 0, options.log_bytes) faults = parse_qemu_faults(text) result["fault_count"] = len(faults) result["faults_capped"] = capped span = loaded.image.max_vaddr - loaded.image.min_vaddr if loaded.image else 0 result["first_fault"] = enrich_fault(faults[0], loaded.bias, span) if faults else None result.update(profile_from_events(load_events(os.path.join(out_dir, session_rel)))) result["session"] = session_rel result["status"] = "ok" except Exception as exc: # a failed scenario must not abort the diff result["status"] = "error" result["reason"] = f"{type(exc).__name__}: {exc}" return result def behavior_diff( old: LoadedTarget, new: LoadedTarget, old_db: SymbolDB, new_db: SymbolDB, probe_a: dict, probe_b: dict, options: "VersionDiffOptions", out_dir: str, qemu_ok: bool, ) -> dict: a = run_behavior(old, old_db, options, out_dir, qemu_ok) b = run_behavior(new, new_db, options, out_dir, qemu_ok) entry = { "status": "skipped", "reason": None, "a": a, "b": b, "compare_report": None, "deltas": {"bp": [], "watch": [], "writes": []}, "all_zero": None, "metrics": [], } a["metrics"] = metric_rows(probe_a, a) b["metrics"] = metric_rows(probe_b, b) if a["status"] != "ok" or b["status"] != "ok": reasons = [row.get("reason") for row in (a, b) if row.get("reason")] entry["reason"] = "; ".join(reasons) or "a behavior session is not available on both targets" entry["status"] = "error" if "error" in {a["status"], b["status"]} else "skipped" return entry a_dir = os.path.join(out_dir, a["session"]) b_dir = os.path.join(out_dir, b["session"]) try: report = compare_sessions(a_dir, b_dir, max_rows=options.max_rows) rel = os.path.join("compare", f"{old.name}--{new.name}.md") write_text(os.path.join(out_dir, rel), report) entry["compare_report"] = rel except Exception as exc: entry["reason"] = f"{type(exc).__name__}: {exc}" deltas = { name: counter_delta(a.get(name) or {}, b.get(name) or {}) for name in ("bp", "watch", "writes") } entry["deltas"] = deltas entry["all_zero"] = all(row["delta"] == 0 for rows in deltas.values() for row in rows) a_metrics = {row["id"]: row for row in a["metrics"]} b_metrics = {row["id"]: row for row in b["metrics"]} entry["metrics"] = [ { "id": metric_id, "name": name, "status": "not measured", "reason": NO_REFERENCE, "a_observed": (a_metrics.get(metric_id) or {}).get("observed"), "b_observed": (b_metrics.get(metric_id) or {}).get("observed"), } for metric_id, name in METRIC_DEFS ] entry["status"] = "ok" return entry # --------------------------------------------------------------------------- # report assembly + rendering # --------------------------------------------------------------------------- @dataclass class VersionDiffOptions: old: str new: str out: str old_label: str | None = None new_label: str | None = None release: str | None = None old_symbols: str | None = None new_symbols: str | None = None old_ghidra_names: str | None = None new_ghidra_names: str | None = None old_ghidra_index: str | None = None new_ghidra_index: str | None = None old_ghidra_exports: str | None = None new_ghidra_exports: str | None = None old_ghidra_bias: int = 0 new_ghidra_bias: int = 0 corpus_diff: tuple[str, str] | None = None corpus_report: str | None = None stimulus: str | None = None probe_instructions: int = 2000 probe_duration: float = 0.5 behavior_duration: float = 0.4 max_stops: int = 5000 max_rows: int = 40 log_bytes: int = MAX_LOG_BYTES approval_token: str | None = None def labels(self) -> list[str]: return [ value for value in ( self.release, self.old_label, self.new_label, os.path.basename(self.old), os.path.basename(self.new), ) if value ] def label_for(self, side: str) -> str: explicit = self.old_label if side == "old" else self.new_label path = self.old if side == "old" else self.new return explicit or Path(path).stem def _target_brief(label: str, loaded: LoadedTarget, qemu_ok: bool, options: VersionDiffOptions) -> dict: return { "label": label, "path": os.path.basename(loaded.path), "kind": loaded.kind, "status": "ok" if loaded.ok else "skipped", "reason": loaded.reason, "sha256": loaded.sha256, "size": loaded.size, "entry": loaded.entry if loaded.ok else None, "bias": loaded.bias if loaded.ok else None, "probe_instructions": options.probe_instructions, "qemu_available": qemu_ok, } def run_version_diff(options: VersionDiffOptions) -> dict: """Run every section, write JSON+markdown under ``--out``, return the report.""" check_rm610_gate(options.labels(), approval_token=options.approval_token) ensure_dir(options.out) old_label, new_label = options.label_for("old"), options.label_for("new") qemu_ok = qemu_available() old = load_any_target(options.old) new = load_any_target(options.new) if old.ok: old_db, old_sources, old_notes = load_symbol_sources( old, symbols_json=options.old_symbols, ghidra_names=options.old_ghidra_names, ghidra_index=options.old_ghidra_index, ghidra_exports=options.old_ghidra_exports, ghidra_bias=options.old_ghidra_bias, ) else: old_db, old_sources, old_notes = SymbolDB(), [], [] if new.ok: new_db, new_sources, new_notes = load_symbol_sources( new, symbols_json=options.new_symbols, ghidra_names=options.new_ghidra_names, ghidra_index=options.new_ghidra_index, ghidra_exports=options.new_ghidra_exports, ghidra_bias=options.new_ghidra_bias, ) else: new_db, new_sources, new_notes = SymbolDB(), [], [] image = image_diff(old, new, options.max_rows) symbols = symbol_section(old, new, old_db, old_sources, old_notes, new_db, new_sources, new_notes, options, options.out) corpus = corpus_section(options, options.out) probe = probe_diff(old, new, old_db, new_db, options, options.out, qemu_ok) behavior = behavior_diff(old, new, old_db, new_db, probe["a"], probe["b"], options, options.out, qemu_ok) sections = {"image": image, "symbols": symbols, "corpus": corpus, "probe": probe, "behavior": behavior} report = { "format": FORMAT, "tool": "harness/version_diff.py", "backend": "qemu", "qemu_available": qemu_ok, "old": _target_brief(old_label, old, qemu_ok, options), "new": _target_brief(new_label, new, qemu_ok, options), "options": { "probe_instructions": options.probe_instructions, "probe_duration": options.probe_duration, "behavior_duration": options.behavior_duration, "max_rows": options.max_rows, "stimulus": os.path.basename(options.stimulus) if options.stimulus else None, "corpus_report": os.path.basename(options.corpus_report) if options.corpus_report else None, "corpus_diff": [os.path.basename(p) for p in options.corpus_diff] if options.corpus_diff else None, }, "sections": sections, "summary": { "sections_ok": sorted(name for name, section in sections.items() if section["status"] == "ok"), "sections_skipped": { name: section.get("reason") for name, section in sections.items() if section["status"] == "skipped" }, "sections_error": { name: section.get("reason") for name, section in sections.items() if section["status"] == "error" }, }, } write_json(os.path.join(options.out, "version-diff.json"), report) write_text(os.path.join(options.out, "version-diff.md"), render_markdown(report, options)) return report # --------------------------------------------------------------------------- # markdown rendering # --------------------------------------------------------------------------- def _table(headers: list[str], rows: list[list]) -> list[str]: lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"] for row in rows: lines.append("| " + " | ".join(str(cell) for cell in row) + " |") return lines def _hex(value) -> str: return f"`{value:#x}`" if isinstance(value, int) else "-" def render_markdown(report: dict, options: VersionDiffOptions) -> str: old, new = report["old"], report["new"] lines = [ "# green-eye version diff", "", "> **# TOOL IN DEVELOPMENT - NOT READY YET!** Generated by `harness/version_diff.py`.", "", f"* format: `{report['format']}`", f"* targets: `{old['label']}` -> `{new['label']}`", f"* backend: `{report['backend']}` (qemu available: {report['qemu_available']})", "", "## Targets", "", ] lines += _table( ["side", "label", "kind", "size", "sha256", "entry"], [ [side, f"`{item['label']}`", item["kind"], item["size"], f"`{(item['sha256'] or '')[:16]}`", _hex(item["entry"])] for side, item in (("old", old), ("new", new)) ], ) lines += [""] image = report["sections"]["image"] lines += ["## 1. Image / shape", "", f"* status: {image['status']}"] if image.get("reason"): lines += [f"* reason: {image['reason']}"] if image["status"] == "ok": deltas = image["deltas"] lines += _table( ["field", "old", "new", "delta"], [ ["entry", _hex(deltas["entry"]["a"]), _hex(deltas["entry"]["b"]), _hex(deltas["entry"]["delta"])], ["span", _hex(deltas["span"]["a"]), _hex(deltas["span"]["b"]), _hex(deltas["span"]["delta"])], ["file size", deltas["file_size"]["a"], deltas["file_size"]["b"], deltas["file_size"]["delta"]], ], ) lines += ["", f"* identical shape: {'yes' if image['same'] else 'no'}", ""] lines += _table( ["load delta", "count"], [[key, len(image["deltas"]["loads"][key])] for key in ("added", "removed", "changed")] + [["section " + key, len(image["deltas"]["sections"][key])] for key in ("added", "removed", "moved", "resized")], ) lines += [""] symbols = report["sections"]["symbols"] lines += ["## 2. Symbol inventory", "", f"* status: {symbols['status']}"] if symbols.get("reason"): lines += [f"* reason: {symbols['reason']}"] if symbols["status"] == "ok": lines += _table( ["side", "symbols", "named", "anonymous", "sources"], [ [side, symbols[side]["count"], symbols["totals"][f"named_{side}"], symbols["totals"][f"anonymous_{side}"], ", ".join(symbols[side]["sources"]) or "-"] for side in ("a", "b") ], ) lines += [""] lines += _table( ["address shift", "dominant", "matched named", "matched anon", "moved", "resized", "added", "removed"], [[ _hex(symbols["shift"]), f"{symbols['shift_dominant'] * 100:.1f}%", symbols["matched_named"], symbols["matched_shifted"], symbols["totals"]["moved"], symbols["totals"]["resized_named"], symbols["totals"]["added"], symbols["totals"]["removed"], ]], ) lines += [""] for key, title in (("moved", "Moved named"), ("resized", "Resized named"), ("added", "Added"), ("removed", "Removed")): rows = symbols.get(key) or [] lines += [f"**{title} (first {options.max_rows})**", ""] if not rows: lines += ["_none_", ""] continue if key in ("added", "removed"): lines += _table(["addr", "name", "size"], [[_hex(r["addr"]), f"`{r['name']}`", r["size"]] for r in rows]) elif key == "moved": lines += _table(["name", "old addr", "new addr", "delta"], [[f"`{r['name']}`", _hex(r["old_addr"]), _hex(r["new_addr"]), _hex(r["delta"])] for r in rows]) else: lines += _table(["name", "old addr", "new addr", "old size", "new size"], [[f"`{r['name']}`", _hex(r["old_addr"]), _hex(r["new_addr"]), r["old_size"], r["new_size"]] for r in rows]) lines += [""] else: lines += [""] corpus = report["sections"]["corpus"] lines += ["## 3. Decompiled-change hints", "", f"* status: {corpus['status']}", f"* mode: {corpus.get('mode') or '-'}"] if corpus.get("reason"): lines += [f"* reason: {corpus['reason']}"] if corpus["status"] == "ok" and corpus.get("mode") == "corpus-diff": diff = corpus["diff"] lines += _table( ["functions A", "functions B", "shift", "added", "removed", "moved", "resized"], [[diff["totals"]["a"], diff["totals"]["b"], _hex(diff["shift"]), diff["totals"]["added"], diff["totals"]["removed"], diff["totals"]["moved"], diff["totals"]["resized_named"]]], ) elif corpus["status"] == "ok" and corpus.get("mode") == "corpus-report": info = corpus["report"] lines += [f"* report: `{info['path']}` ({info['bytes']} bytes, sha256 `{info['sha256'][:16]}`)"] lines += [""] probe = report["sections"]["probe"] lines += ["## 4. Probe first-touch", "", f"* status: {probe['status']}"] if probe.get("reason"): lines += [f"* reason: {probe['reason']}"] if probe["status"] == "ok": lines += [f"* {probe['summary']}", ""] lines += _table( ["side", "stopped pc", "stopped symbol", "fault", "epc (fw vaddr)", "access", "access (fw vaddr)"], [ [ side, _hex(probe[side].get("stopped_pc")), probe[side].get("stopped_symbol") or "-", (probe[side].get("first_fault") or {}).get("desc") or "none", _hex((probe[side].get("first_fault") or {}).get("epc_firmware")), _hex(((probe[side].get("first_fault") or {}).get("access") or {}).get("access_addr")), _hex((probe[side].get("first_fault") or {}).get("access_firmware")), ] for side in ("a", "b") ], ) lines += [""] behavior = report["sections"]["behavior"] lines += ["## 5. Behavior", "", f"* status: {behavior['status']}"] if behavior.get("reason"): lines += [f"* reason: {behavior['reason']}"] if behavior["status"] == "ok": lines += _table( ["side", "session", "hooks", "uart banner", "faults", "events"], [ [ side, f"`{behavior[side].get('session')}`", ", ".join(behavior[side].get("hooks") or []), behavior[side].get("uart_banner") or "-", behavior[side].get("fault_count"), sum((behavior[side].get("event_counts") or {}).values()), ] for side in ("a", "b") ], ) lines += [f"* zero deltas: {'yes' if behavior['all_zero'] else 'no'}"] if behavior.get("compare_report"): lines += [f"* compare report: `{behavior['compare_report']}`"] lines += [""] for name in ("bp", "watch", "writes"): rows = [row for row in behavior["deltas"][name] if row["delta"] != 0] rows.sort(key=lambda row: (-abs(row["delta"]), str(row["key"]))) lines += [f"**{name} delta (nonzero, first {options.max_rows})**", ""] if rows: lines += _table(["key", "old", "new", "delta"], [[f"`{row['key']}`", row["a"], row["b"], row["delta"]] for row in rows[: options.max_rows]]) else: lines += ["_no deltas_"] lines += [""] lines += [f"> {NO_REFERENCE}", ""] lines += _table( ["#", "metric", "old observed", "new observed"], [[row["id"], row["name"], _short_json(row["a_observed"]), _short_json(row["b_observed"])] for row in behavior["metrics"]], ) lines += [""] summary = report["summary"] lines += ["## Summary", ""] lines += [f"* ok: {', '.join(summary['sections_ok']) or '-'}"] if summary["sections_skipped"]: for name, reason in sorted(summary["sections_skipped"].items()): lines += [f"* skipped `{name}`: {reason}"] if summary["sections_error"]: for name, reason in sorted(summary["sections_error"].items()): lines += [f"* error `{name}`: {reason}"] return "\n".join(lines) + "\n" def _short_json(value, limit: int = 72) -> str: import json if value is None: return "-" text = json.dumps(value, sort_keys=True) return f"`{text}`" if len(text) <= limit else f"`{text[: limit - 1]}...`" # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def _add_side_symbols(parser: argparse.ArgumentParser, side: str) -> None: parser.add_argument(f"--{side}-symbols", help="green-eye symbol DB JSON for the %s target" % side) parser.add_argument(f"--{side}-ghidra-names", help="Ghidra names.tsv for the %s target" % side) parser.add_argument(f"--{side}-ghidra-index", help="Ghidra decompiled_index.tsv for the %s target" % side) parser.add_argument( f"--{side}-ghidra-exports", help="Ghidra exports.tsv or 3-column ExportForGreenEye TSV for the %s target" % side, ) parser.add_argument(f"--{side}-ghidra-bias", type=lambda value: int(value, 0), default=0, help="bias added to %s Ghidra symbol addresses" % side) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="version_diff", description=__doc__.splitlines()[0]) parser.add_argument("--old", required=True, help="old target (RISC-V ELF or raw .fwimage)") parser.add_argument("--new", required=True, help="new target (RISC-V ELF or raw .fwimage)") parser.add_argument("--out", required=True, help="output directory for the report, sessions and DBs") parser.add_argument("--old-label", help="report label for the old target (default: file stem)") parser.add_argument("--new-label", help="report label for the new target (default: file stem)") parser.add_argument("--release", help="release label; an rm610 release is gated by docs/23") parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})") _add_side_symbols(parser, "old") _add_side_symbols(parser, "new") parser.add_argument("--corpus-diff", nargs=2, metavar=("OLD_TSV", "NEW_TSV"), help="diff two decompiled_index.tsv corpus exports") parser.add_argument("--corpus-report", help="attach a previously produced corpus diff report") parser.add_argument("--stimulus", help="green-eye script replacing the builtin demo scenario") parser.add_argument("--probe-instructions", type=int, default=2000, help="probe step budget (0 = wall-clock --probe-duration)") parser.add_argument("--probe-duration", type=float, default=0.5, help="probe wall budget when --probe-instructions=0") parser.add_argument("--behavior-duration", type=float, default=0.4, help="budget for each behavior phase") parser.add_argument("--max-stops", type=int, default=5000, help="safety bound on serviced stops per phase") parser.add_argument("--max-rows", type=int, default=40, help="table row cap in reports") parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES, help="QEMU log window per fault check") return parser def options_from_args(args: argparse.Namespace) -> VersionDiffOptions: corpus = tuple(args.corpus_diff) if args.corpus_diff else None return VersionDiffOptions( old=args.old, new=args.new, out=os.path.abspath(args.out), old_label=args.old_label, new_label=args.new_label, release=args.release, old_symbols=args.old_symbols, new_symbols=args.new_symbols, old_ghidra_names=args.old_ghidra_names, new_ghidra_names=args.new_ghidra_names, old_ghidra_index=args.old_ghidra_index, new_ghidra_index=args.new_ghidra_index, old_ghidra_exports=args.old_ghidra_exports, new_ghidra_exports=args.new_ghidra_exports, old_ghidra_bias=args.old_ghidra_bias, new_ghidra_bias=args.new_ghidra_bias, corpus_diff=corpus, # type: ignore[arg-type] corpus_report=args.corpus_report, stimulus=args.stimulus, probe_instructions=args.probe_instructions, probe_duration=args.probe_duration, behavior_duration=args.behavior_duration, max_stops=args.max_stops, max_rows=args.max_rows, log_bytes=args.log_bytes, approval_token=args.approval_token, ) def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) options = options_from_args(args) try: report = run_version_diff(options) except Rm610GateError as exc: print(f"[version-diff] RM610 gate refused: {exc}", file=sys.stderr) return 3 except GreenEyeError as exc: print(f"[version-diff] error: {exc}", file=sys.stderr) return 2 print(f"[version-diff] {report['old']['label']} -> {report['new']['label']}") for name in ("image", "symbols", "corpus", "probe", "behavior"): section = report["sections"][name] print(f"[version-diff] {name}: status={section['status']}" + (f" reason={section['reason']}" if section.get("reason") else "")) print(f"[version-diff] wrote {os.path.join(options.out, 'version-diff.json')}") print(f"[version-diff] wrote {os.path.join(options.out, 'version-diff.md')}") return 0 if __name__ == "__main__": # pragma: no cover raise SystemExit(main())