diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/version_diff.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/version_diff.py | 1509 |
1 files changed, 1509 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/version_diff.py b/debug/risc-v-gsp-rm-vm/harness/version_diff.py new file mode 100644 index 0000000..a95f244 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/version_diff.py @@ -0,0 +1,1509 @@ +#!/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("<Q", data, 40)[0] + shentsize, shnum, shstrndx = struct.unpack_from("<HHH", data, 58) + sh_fmt = "<IIQQQQIIQQ" + else: + sh_off = struct.unpack_from("<I", data, 32)[0] + shentsize, shnum, shstrndx = struct.unpack_from("<HHH", data, 46) + sh_fmt = "<IIIIIIIIII" + except struct.error: + return [] + sh_size = struct.calcsize(sh_fmt) + if not sh_off or sh_off + sh_size > 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"<section-{index}>", + "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: + """``addr<TAB>name<TAB>size`` (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()) |