diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/portability_matrix.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/portability_matrix.py | 779 |
1 files changed, 779 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/portability_matrix.py b/debug/risc-v-gsp-rm-vm/harness/portability_matrix.py new file mode 100644 index 0000000..82af458 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/portability_matrix.py @@ -0,0 +1,779 @@ +#!/usr/bin/env python3 +"""Multi-version portability orchestration (roadmap R6, H4.1). + +Given one or more firmware ELF targets (a demo or re-hosted build, a raw +``.fwimage``, a future version), run the same deterministic scenario against +each on the QEMU backend: + +1. bounded probe: relocate into virt RAM and run/step a fixed budget, + capturing the first exception via ``parse_qemu_faults``; +2. symbol DB build: ELF symbols plus optional Ghidra exports / an existing + green-eye JSON DB, saved per target; +3. scripted demo run with handler hooks and a control-block watchpoint, + recorded as a normal green-eye session; +4. cross-target comparison with ``greeneye.analyze.compare_sessions`` plus + explicit hit/watch/write profile deltas and the docs/06 M1-M6 skeleton + (metrics without a reference capture are marked "not measured"). + +Outputs ``portability-report.json`` and ``portability-report.md`` under +``--out DIR`` with stable ordering. Unsupported targets are skipped with a +reason instead of aborting the matrix. + +Run from the tool root: + + python3 harness/portability_matrix.py --target firmware/uproc_demo/build/uproc_demo.elf --out dumps/matrix +""" +from __future__ import annotations + +import argparse +import hashlib +import os +import struct +from collections import Counter +from dataclasses import dataclass +from pathlib import Path + +try: + from .common import ( + LOAD_ADDR, + MAX_LOG_BYTES, + SymbolDB, + counter_delta, + ensure_dir, + load_target, + open_flat_session, + qemu_available, + read_log_window, + sha256_file, + write_json, + write_text, + ) +except ImportError: # direct script execution (python3 harness/portability_matrix.py) + from common import ( # type: ignore[no-redef] + LOAD_ADDR, + MAX_LOG_BYTES, + SymbolDB, + counter_delta, + ensure_dir, + load_target, + open_flat_session, + qemu_available, + read_log_window, + sha256_file, + write_json, + write_text, + ) + +from greeneye.analyze import compare_sessions, parse_qemu_faults # noqa: E402 +from greeneye.record import load_events # noqa: E402 + +FORMAT = "greeneye-portability-matrix/1" +RPC_HANDLERS = ("rpc_ping", "rpc_echo", "rpc_fill", "rpc_sum", "rpc_state", "rpc_fault") +CTRL_SIZE = 400 +CTRL_RPC_COUNT = 16 +CTRL_STATE = 8 +CTRL_HANDLER_COUNT = 328 + +METRIC_DEFS = ( + ("M1", "Inventory coverage"), + ("M2", "Reply agreement"), + ("M3", "Ordering agreement"), + ("M4", "State agreement"), + ("M5", "Error-path agreement"), + ("M6", "Memory side-effects"), +) +NO_REFERENCE = "agreement is not measured: needs a reference capture (docs/06 5)" + + +@dataclass +class TargetSpec: + path: str + name: str + + +@dataclass +class MatrixOptions: + out: str + probe_instructions: int = 2000 + probe_duration: float = 0.5 + scenario_duration: float = 0.4 + max_stops: int = 5000 + max_rows: int = 40 + symbols_json: str | None = None + ghidra_names: str | None = None + ghidra_index: str | None = None + ghidra_exports: str | None = None + ghidra_callgraph: str | None = None + ghidra_bias: int = 0 + scenario_script: str | None = None + log_bytes: int = MAX_LOG_BYTES + + +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 _skipped_scenario(reason: str | None) -> dict: + return { + "status": "skipped", + "reason": reason, + "session": None, + "hooks": [], + "uart_banner": None, + "ctrl": None, + "snapshot": None, + "event_counts": {}, + "bp": {}, + "watch": {}, + "writes": {}, + "fault_count": 0, + "faults_capped": False, + "first_fault": None, + } + + +# --------------------------------------------------------------------------- +# per-target steps +# --------------------------------------------------------------------------- + + +def build_symbols(spec: TargetSpec, bias: int, out_dir: str, options: MatrixOptions): + """ELF symbols + optional Ghidra exports / JSON DB, saved per target.""" + db = SymbolDB() + db.load_elf(spec.path, bias=bias, source="elf") + sources = ["elf"] + if options.symbols_json: + if not os.path.isfile(options.symbols_json): + raise FileNotFoundError(f"symbol DB not found: {options.symbols_json}") + db.load_json(options.symbols_json) + sources.append(os.path.basename(options.symbols_json)) + ghidra_inputs = any( + ( + options.ghidra_names, + options.ghidra_index, + options.ghidra_exports, + options.ghidra_callgraph, + ) + ) + if ghidra_inputs: + from greeneye.ghidra import build_symbol_db + + gdb = build_symbol_db( + names_tsv=options.ghidra_names, + index_tsv=options.ghidra_index, + exports_tsv=options.ghidra_exports, + callgraph_tsv=options.ghidra_callgraph, + bias=options.ghidra_bias, + ) + for symbol in gdb: + db.add(symbol) + sources.append("ghidra") + rel = os.path.join("symbols", f"{spec.name}.json") + out_path = os.path.join(out_dir, rel) + ensure_dir(os.path.dirname(out_path)) + db.save_json( + out_path, + meta={"bias": bias, "sources": sources, "firmware": os.path.basename(spec.path)}, + ) + return db, sources, rel + + +def run_probe(spec: TargetSpec, image, db: SymbolDB, out_dir: str, options: MatrixOptions, qemu_ok: bool) -> dict: + if not qemu_ok: + return _skipped_probe("qemu-system-riscv64 not found") + result = _skipped_probe(None) + result["instructions"] = options.probe_instructions + session_rel = os.path.join("sessions", f"{spec.name}-probe") + try: + with open_flat_session( + image, + spec.path, + record=os.path.join(out_dir, session_rel), + name=f"{spec.name}-probe", + symbols=db, + log_faults=True, + ) as fs: + session = fs.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(fs.target.qemu_log, 0, options.log_bytes) + faults = parse_qemu_faults(text) + result["fault_count"] = len(faults) + result["faults_capped"] = capped + result["first_fault"] = faults[0] if faults else None + result["session"] = session_rel + if text: + fs.session.recorder.meta["qemu_log_blob"] = fs.session.recorder.blob("qemu.log", text.encode("utf-8")) + fs.session.recorder.meta["faults_capped"] = capped + result["status"] = "ok" + except Exception as exc: # a failed step must not abort the matrix + result["status"] = "error" + result["reason"] = f"{type(exc).__name__}: {exc}" + return result + + +def install_scenario(session) -> list[str]: + hooks: list[str] = [] + for name in RPC_HANDLERS: + symbol = session.symbols.get(name) + if symbol is None: + continue + session.hook(symbol.addr, on_enter=lambda hit: None, listener=f"matrix:{name}") + hooks.append(name) + return hooks + + +def read_ctrl(session, symbol) -> dict | None: + if symbol is None: + return None + raw = session.memory.read(symbol.addr, CTRL_SIZE) + u32 = lambda offset: struct.unpack_from("<I", raw, offset)[0] # noqa: E731 + u64 = lambda offset: struct.unpack_from("<Q", raw, offset)[0] # noqa: E731 + return { + "addr": f"{symbol.addr:#x}", + "magic": u32(0), + "version": u32(4), + "state": u32(CTRL_STATE), + "rpc_count": u64(CTRL_RPC_COUNT), + "last_rpc": u64(24), + "result": u64(64), + "handler_count": u64(CTRL_HANDLER_COUNT), + } + + +def profile_from_events(events: list[dict]) -> dict: + counts: Counter = Counter() + bp: Counter = Counter() + watch: Counter = Counter() + writes: Counter = Counter() + for event in events: + kind = event.get("type", "?") + counts[kind] += 1 + if kind == "bp_hit": + bp[event.get("symbol") or f"{event.get('pc', 0):#x}"] += 1 + elif kind == "watch_hit": + watch[f"{event.get('addr', 0):#x}:{event.get('kind') or 'w'}"] += 1 + elif kind == "mem_write": + writes[f"{(event.get('addr', 0) >> 12) << 12:#x}"] += 1 + return { + "event_counts": dict(sorted(counts.items())), + "bp": {key: bp[key] for key in sorted(bp)}, + "watch": {key: watch[key] for key in sorted(watch)}, + "writes": {key: writes[key] for key in sorted(writes)}, + } + + +def run_scenario(spec: TargetSpec, image, db: SymbolDB, out_dir: str, options: MatrixOptions, qemu_ok: bool) -> dict: + if not qemu_ok: + return _skipped_scenario("qemu-system-riscv64 not found") + result = _skipped_scenario(None) + session_rel = os.path.join("sessions", f"{spec.name}-scenario") + try: + with open_flat_session( + image, + spec.path, + record=os.path.join(out_dir, session_rel), + name=f"{spec.name}-scenario", + symbols=db, + log_faults=True, + ) as fs: + session = fs.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="matrix:rpc_count") + if options.scenario_script: + import greeneye as ge + + script_path = Path(options.scenario_script) + 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.scenario_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.scenario_duration, max_stops=options.max_stops) + result["ctrl"] = read_ctrl(session, ctrl_symbol) + if ctrl_addr is not None: + data = session.snapshot(ctrl_addr, CTRL_SIZE, name="ctrl") + result["snapshot"] = {"label": "ctrl", "size": len(data), "sha256": hashlib.sha256(data).hexdigest()} + uart = fs.target.read_uart() + result["uart_banner"] = uart.strip().splitlines()[0] if uart.strip() else None + text, capped = read_log_window(fs.target.qemu_log, 0, options.log_bytes) + faults = parse_qemu_faults(text) + result["fault_count"] = len(faults) + result["faults_capped"] = capped + result["first_fault"] = faults[0] 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 step must not abort the matrix + result["status"] = "error" + result["reason"] = f"{type(exc).__name__}: {exc}" + return result + + +# --------------------------------------------------------------------------- +# metrics + comparisons +# --------------------------------------------------------------------------- + + +def metric_rows(probe: dict, scenario: dict) -> list[dict]: + """docs/06 M1-M6 skeleton; observed data is local, agreement stays unmeasured.""" + bp = scenario.get("bp") or {} + ctrl = scenario.get("ctrl") or {} + snapshot = scenario.get("snapshot") or {} + probe_fault = (probe.get("first_fault") or {}).get("desc") + scenario_fault = (scenario.get("first_fault") or {}).get("desc") + rows = [ + { + "id": "M1", + "name": "Inventory coverage", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": { + "rpc_handlers_hit": sorted(k for k in bp if k in RPC_HANDLERS), + "other_hit_sites": sorted(k for k in bp if k not in RPC_HANDLERS), + "declared_handlers": ctrl.get("handler_count"), + }, + }, + { + "id": "M2", + "name": "Reply agreement", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": {"note": "reply fields are not decoded by the builtin scenario", "result": ctrl.get("result")}, + }, + { + "id": "M3", + "name": "Ordering agreement", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": {"events": sum((scenario.get("event_counts") or {}).values())}, + }, + { + "id": "M4", + "name": "State agreement", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": {"final_state": ctrl.get("state"), "rpc_state_hits": bp.get("rpc_state", 0)}, + }, + { + "id": "M5", + "name": "Error-path agreement", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": { + "probe_first_fault": probe_fault, + "scenario_first_fault": scenario_fault, + "scenario_fault_count": scenario.get("fault_count", 0), + }, + }, + { + "id": "M6", + "name": "Memory side-effects", + "status": "not measured", + "reason": NO_REFERENCE, + "observed": { + "written_pages": len(scenario.get("writes") or {}), + "ctrl_snapshot_sha256": snapshot.get("sha256"), + }, + }, + ] + return rows + + +def compare_targets(a: dict, b: dict, out_dir: str, options: MatrixOptions) -> dict: + entry = { + "a": a["name"], + "b": b["name"], + "status": "ok", + "reason": None, + "compare_report": None, + "deltas": {"bp": [], "watch": [], "writes": []}, + "all_zero": True, + "metrics": [], + } + a_scenario = a.get("scenario") or {} + b_scenario = b.get("scenario") or {} + if a_scenario.get("status") != "ok" or b_scenario.get("status") != "ok": + entry["status"] = "skipped" + entry["reason"] = "a scenario session is not available on both targets" + return entry + a_dir = os.path.join(out_dir, a_scenario.get("session") or "") + b_dir = os.path.join(out_dir, b_scenario.get("session") or "") + try: + report = compare_sessions(a_dir, b_dir, max_rows=options.max_rows) + rel = os.path.join("compare", f"{a['name']}--{b['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 = {"bp": [], "watch": [], "writes": []} + for name in ("bp", "watch", "writes"): + deltas[name] = counter_delta(a_scenario.get(name) or {}, b_scenario.get(name) or {}) + 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.get("metrics") or []} + b_metrics = {row["id"]: row for row in b.get("metrics") or []} + 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 + ] + return entry + + +# --------------------------------------------------------------------------- +# report assembly + rendering +# --------------------------------------------------------------------------- + + +def analyze_target(spec: TargetSpec, options: MatrixOptions, qemu_ok: bool) -> dict: + target = { + "name": spec.name, + "path": spec.path, + "sha256": "", + "kind": "elf", + "entry": None, + "status": "skipped", + "skip_reason": None, + "probe": _skipped_probe(None), + "symbols": {"count": 0, "sources": [], "path": None}, + "scenario": _skipped_scenario(None), + "metrics": metric_rows({}, {}), + } + if not os.path.isfile(spec.path): + target["skip_reason"] = "file not found" + return target + target["sha256"] = sha256_file(spec.path) + try: + image = load_target(spec.path) + except Exception as exc: + target["kind"] = "unknown" + target["skip_reason"] = f"{type(exc).__name__}: {exc}" + return target + target["entry"] = f"{image.entry:#x}" + bias = LOAD_ADDR - image.min_vaddr + db = SymbolDB() + try: + db, sources, sym_rel = build_symbols(spec, bias, options.out, options) + target["symbols"] = {"count": len(db), "sources": sources, "path": sym_rel} + except Exception as exc: + target["symbols"] = {"count": 0, "sources": [], "path": None, "reason": f"{type(exc).__name__}: {exc}"} + db.load_elf(spec.path, bias=bias, source="elf") + target["symbols"]["count"] = len(db) + target["symbols"]["sources"] = ["elf"] + target["probe"] = run_probe(spec, image, db, options.out, options, qemu_ok) + target["scenario"] = run_scenario(spec, image, db, options.out, options, qemu_ok) + target["metrics"] = metric_rows(target["probe"], target["scenario"]) + statuses = {target["probe"]["status"], target["scenario"]["status"]} + if statuses == {"ok"}: + target["status"] = "ok" + elif "ok" in statuses: + target["status"] = "partial" + elif statuses == {"skipped"}: + target["status"] = "skipped" + target["skip_reason"] = target["probe"].get("reason") or target["scenario"].get("reason") + else: + target["status"] = "error" + target["skip_reason"] = target["probe"].get("reason") or target["scenario"].get("reason") + return target + + +def run_matrix(specs: list[TargetSpec], options: MatrixOptions) -> dict: + ensure_dir(options.out) + qemu_ok = qemu_available() + targets = [analyze_target(spec, options, qemu_ok) for spec in specs] + comparisons: list[dict] = [] + for i in range(len(targets)): + for j in range(i + 1, len(targets)): + comparisons.append(compare_targets(targets[i], targets[j], options.out, options)) + report = { + "format": FORMAT, + "tool": "harness/portability_matrix.py", + "backend": "qemu", + "qemu_available": qemu_ok, + "options": { + "probe_instructions": options.probe_instructions, + "probe_duration": options.probe_duration, + "scenario_duration": options.scenario_duration, + "scenario_script": options.scenario_script, + "max_rows": options.max_rows, + "symbols_json": options.symbols_json, + "ghidra_bias": options.ghidra_bias, + }, + "targets": targets, + "comparisons": comparisons, + "skipped": [ + {"name": target["name"], "reason": target["skip_reason"]} + for target in targets + if target["status"] == "skipped" + ], + } + return report + + +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 _profile_tables(title: str, profile: dict, max_rows: int) -> list[str]: + lines = [f"**{title}**", ""] + if not profile: + return lines + ["_none observed_", ""] + rows = [[f"`{key}`", count] for key, count in sorted(profile.items(), key=lambda kv: (-int(kv[1]), str(kv[0])))[:max_rows]] + lines += _table(["key", "hits"], rows) + [""] + return lines + + +def render_markdown(report: dict, options: MatrixOptions) -> str: + lines = [ + "# green-eye portability delta report", + "", + "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/portability_matrix.py`.", + "", + f"* format: `{report['format']}` (green-eye {_greeneye_version()})", + f"* backend: `{report['backend']}` (qemu available: {report['qemu_available']})", + f"* targets: {len(report['targets'])}", + "", + "## Targets", + "", + ] + lines += _table( + ["target", "status", "sha256", "probe", "symbols", "scenario events"], + [ + [ + f"`{target['name']}`", + target["status"], + f"`{(target['sha256'] or '')[:16]}`", + f"{target['probe']['status']} ({target['probe']['fault_count']} faults)", + target["symbols"]["count"], + sum((target["scenario"].get("event_counts") or {}).values()), + ] + for target in report["targets"] + ], + ) + lines += [""] + + for target in report["targets"]: + lines += [f"### {target['name']}", ""] + lines += [f"* path: `{target['path']}`", f"* entry: `{target['entry']}`", f"* sha256: `{target['sha256']}`"] + if target["skip_reason"]: + lines += [f"* note: {target['skip_reason']}"] + lines += [""] + + probe = target["probe"] + lines += ["#### Probe", ""] + fault = probe.get("first_fault") + stopped_pc = probe.get("stopped_pc") + probe_rows = [ + ["status", probe["status"]], + ["instructions", probe.get("instructions")], + ["stopped pc", f"`{stopped_pc:#x}`" if stopped_pc is not None else None], + ["stopped symbol", probe.get("stopped_symbol")], + ["fault count", probe.get("fault_count")], + ["faults capped", probe.get("faults_capped")], + ] + if fault: + probe_rows.append( + ["first fault", f"cause {fault.get('cause')} {fault.get('desc')} @ epc `{fault.get('epc', 0):#x}`"] + ) + if probe.get("reason"): + probe_rows.append(["reason", probe["reason"]]) + lines += _table(["field", "value"], probe_rows) + [""] + + lines += ["#### Symbols", ""] + lines += _table( + ["count", "sources", "path"], + [[target["symbols"]["count"], ", ".join(target["symbols"]["sources"]), target["symbols"]["path"]]], + ) + [""] + + scenario = target["scenario"] + lines += ["#### Scenario", ""] + lines += _table( + ["status", "session", "hooks", "uart banner", "faults"], + [[ + scenario["status"], + f"`{scenario.get('session')}`", + ", ".join(scenario.get("hooks") or []), + scenario.get("uart_banner"), + scenario.get("fault_count"), + ]], + ) + [""] + lines += _profile_tables("Breakpoint hits", scenario.get("bp") or {}, options.max_rows) + lines += _profile_tables("Watchpoint hits", scenario.get("watch") or {}, options.max_rows) + lines += _profile_tables("Written pages", scenario.get("writes") or {}, options.max_rows) + + lines += ["#### M1-M6", ""] + lines += [f"> {NO_REFERENCE}", ""] + lines += _table( + ["#", "metric", "status", "observed"], + [ + [row["id"], row["name"], row["status"], _short_json(row["observed"])] + for row in target["metrics"] + ], + ) + [""] + + lines += ["## Cross-target comparisons", ""] + if not report["comparisons"]: + lines += ["_single target: no comparison_", ""] + for comp in report["comparisons"]: + lines += [f"### {comp['a']} vs {comp['b']}", ""] + lines += [f"* status: {comp['status']}", f"* zero deltas: {'yes' if comp['all_zero'] else 'no'}"] + if comp.get("compare_report"): + lines += [f"* compare report: `{comp['compare_report']}`"] + if comp.get("reason"): + lines += [f"* note: {comp['reason']}"] + lines += [""] + for name in ("bp", "watch", "writes"): + rows = [row for row in comp["deltas"][name] if row["delta"] != 0] + rows.sort(key=lambda row: (-abs(row["delta"]), row["key"])) + lines += [f"**{name} delta (nonzero, first {options.max_rows})**", ""] + if rows: + lines += _table( + ["key", comp["a"], comp["b"], "delta"], + [[f"`{row['key']}`", row["a"], row["b"], row["delta"]] for row in rows[: options.max_rows]], + ) + [""] + else: + lines += ["_no deltas_", ""] + lines += ["**M1-M6 (not measured: reference capture required)**", ""] + lines += _table( + ["#", "metric", "A observed", "B observed"], + [ + [row["id"], row["name"], _short_json(row["a_observed"]), _short_json(row["b_observed"])] + for row in comp["metrics"] + ], + ) + [""] + + if report["skipped"]: + lines += ["## Skipped targets", ""] + lines += _table(["target", "reason"], [[f"`{item['name']}`", item["reason"]] for item in report["skipped"]]) + lines += [""] + return "\n".join(lines) + + +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]}...`" + + +def _greeneye_version() -> str: + try: + import greeneye + + return greeneye.__version__ + except Exception: # pragma: no cover + return "unknown" + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="portability_matrix", description=__doc__.splitlines()[0]) + parser.add_argument("targets", nargs="*", help="firmware ELF paths") + parser.add_argument("--target", action="append", default=[], help="firmware ELF path (repeatable)") + parser.add_argument("--out", required=True, help="output directory for the report, sessions and symbols") + parser.add_argument("--symbols", dest="symbols_json", help="existing greeneye symbol DB JSON") + parser.add_argument("--ghidra-names", help="Ghidra names.tsv export") + parser.add_argument("--ghidra-index", help="Ghidra decompiled_index.tsv export") + parser.add_argument("--ghidra-exports", help="Ghidra exports.tsv") + parser.add_argument("--ghidra-callgraph", help="Ghidra callgraph.tsv") + parser.add_argument("--ghidra-bias", type=lambda value: int(value, 0), default=0) + parser.add_argument("--probe-instructions", type=int, default=2000, help="probe step budget (0 = wall-clock run)") + parser.add_argument("--probe-duration", type=float, default=0.5, help="probe wall budget when --probe-instructions=0") + parser.add_argument("--scenario-duration", type=float, default=0.4, help="budget for each scenario phase") + parser.add_argument("--max-stops", type=int, default=5000, help="safety bound on serviced stops per scenario phase") + parser.add_argument("--max-rows", type=int, default=40, help="table row cap in reports") + parser.add_argument("--scenario-script", help="replace the builtin scenario with a green-eye script") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + targets = list(args.targets) + list(args.target) + if not targets: + parser.error("at least one target is required (positional or --target)") + options = MatrixOptions( + out=os.path.abspath(args.out), + probe_instructions=args.probe_instructions, + probe_duration=args.probe_duration, + scenario_duration=args.scenario_duration, + max_stops=args.max_stops, + max_rows=args.max_rows, + symbols_json=args.symbols_json, + ghidra_names=args.ghidra_names, + ghidra_index=args.ghidra_index, + ghidra_exports=args.ghidra_exports, + ghidra_callgraph=args.ghidra_callgraph, + ghidra_bias=args.ghidra_bias, + scenario_script=args.scenario_script, + ) + specs = _name_specs(targets) + report = run_matrix(specs, options) + json_path = write_json(os.path.join(options.out, "portability-report.json"), report) + md_path = write_text(os.path.join(options.out, "portability-report.md"), render_markdown(report, options)) + print(f"[matrix] {len(specs)} target(s), out={options.out}") + for target in report["targets"]: + print(f"[matrix] {target['name']}: status={target['status']} probe={target['probe']['status']} " + f"scenario={target['scenario']['status']}") + print(f"[matrix] wrote {json_path}") + print(f"[matrix] wrote {md_path}") + return 0 + + +def _name_specs(paths: list[str]) -> list[TargetSpec]: + seen: dict[str, int] = {} + specs: list[TargetSpec] = [] + for path in paths: + base = Path(path).stem + seen[base] = seen.get(base, 0) + 1 + name = base if seen[base] == 1 else f"{base}-{seen[base]}" + specs.append(TargetSpec(path=os.path.abspath(path), name=name)) + return specs + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) |