ghidra-re-loop/progress.py

#!/usr/bin/env python3
"""Session / progress tracker — adapted port of auto-re-agent's core/session.py,
reports/tracker.py, and reports/formatter.py. JSON-backed, append-only, atomic
writes. Mirrors the original's REPORT format so runs are resumable and progress
claims are auditable against recorded tool evidence (never the model's memory).

A "function" record = one analysis target (a gate/feature/function) the loop has
attempted, keyed by normalized address. Record after EACH attempt so a crash or
new session resumes where it left off.

Usage:
  python progress.py record --file re-progress.json --address 0x401abc \
     --name "CLicense::Validate" --component license \
     --success true --rounds 2 --verdict PASS --parity GREEN \
     --objective PASS --note "jz at 0x401ad0 gates the dialog"
  python progress.py summary --file re-progress.json
  python progress.py table   --file re-progress.json [--component license]
  python progress.py attempted --file re-progress.json --address 0x401abc   # exit 0 if attempted
"""
from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path


def norm_addr(a: str) -> str:
    a = a.strip().lower()
    if a.startswith("0x"):
        try:
            return hex(int(a, 16))
        except ValueError:
            return a
    return a


class Session:
    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.data = {"functions": {}, "runs": []}
        if self.path.exists():
            try:
                self.data = json.loads(self.path.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, OSError):
                pass
        self.data.setdefault("functions", {})
        self.data.setdefault("runs", [])

    def save(self):
        self.path.parent.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_suffix(".tmp")
        tmp.write_text(json.dumps(self.data, indent=2), encoding="utf-8")
        tmp.replace(self.path)

    def record(self, entry: dict):
        addr = norm_addr(entry["address"])
        entry["timestamp"] = time.strftime("%Y-%m-%dT%H:%M:%S")
        self.data["functions"][addr] = entry
        self.data["runs"].append(entry)
        self.save()

    def is_attempted(self, address: str) -> bool:
        return norm_addr(address) in self.data["functions"]

    def summary(self) -> dict:
        funcs = self.data["functions"].values()
        total = len(self.data["functions"])
        passed = sum(1 for f in funcs if f.get("success"))
        comps = {f.get("component", "") for f in funcs if f.get("component")}
        return {"total": total, "passed": passed, "failed": total - passed,
                "components": len(comps)}

    def table(self, component: str | None):
        rows = []
        for f in self.data["functions"].values():
            if component and f.get("component") != component:
                continue
            rows.append(f)
        return rows


def _b(v: str | None) -> bool:
    return str(v).lower() in {"1", "true", "yes", "pass", "y"}


def cmd_record(a):
    s = Session(a.file)
    s.record({
        "address": a.address,
        "function_name": a.name or "",
        "component": a.component or "",
        "success": _b(a.success),
        "rounds_used": a.rounds,
        "verdict": a.verdict,
        "objective_verdict": a.objective,
        "parity_status": a.parity,
        "note": a.note or "",
    })
    print(f"recorded {norm_addr(a.address)} ({'PASS' if _b(a.success) else 'FAIL'})")
    return 0


def cmd_summary(a):
    s = Session(a.file)
    su = s.summary()
    print("RE Progress Summary")
    print("=" * 40)
    print(f"Total targets:    {su['total']}")
    print(f"Passed:           {su['passed']}")
    print(f"Failed:           {su['failed']}")
    print(f"Components:       {su['components']}")
    return 0


def cmd_table(a):
    s = Session(a.file)
    rows = s.table(a.component)
    print("| Address | Function | Comp | Status | Rounds | Parity | Obj |")
    print("|---------|----------|------|--------|--------|--------|-----|")
    for f in rows:
        st = "PASS" if f.get("success") else "FAIL"
        print(f"| {f.get('address','')} | {f.get('function_name','')} | "
              f"{f.get('component','')} | {st} | {f.get('rounds_used','')} | "
              f"{f.get('parity_status') or '-'} | {f.get('objective_verdict') or '-'} |")
    return 0


def cmd_attempted(a):
    return 0 if Session(a.file).is_attempted(a.address) else 1


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)

    r = sub.add_parser("record")
    r.add_argument("--file", required=True)
    r.add_argument("--address", required=True)
    r.add_argument("--name", default="")
    r.add_argument("--component", default="")
    r.add_argument("--success", default="false")
    r.add_argument("--rounds", type=int, default=0)
    r.add_argument("--verdict", default=None)
    r.add_argument("--objective", default=None)
    r.add_argument("--parity", default=None)
    r.add_argument("--note", default="")
    r.set_defaults(fn=cmd_record)

    for name, fn in (("summary", cmd_summary), ("table", cmd_table)):
        p = sub.add_parser(name)
        p.add_argument("--file", required=True)
        if name == "table":
            p.add_argument("--component", default=None)
        p.set_defaults(fn=fn)

    at = sub.add_parser("attempted")
    at.add_argument("--file", required=True)
    at.add_argument("--address", required=True)
    at.set_defaults(fn=cmd_attempted)

    a = ap.parse_args()
    return a.fn(a)


if __name__ == "__main__":
    sys.exit(main())