ghidra-re-loop/objective_verify.py

#!/usr/bin/env python3
"""Non-LLM structural verifier — adapted port of auto-re-agent's
verification/objective.py (verify_candidate) + utils/text.py.

Purpose: an INDEPENDENT gate that rejects a sub-agent's reconstructed/reversed
function when it structurally diverges from the Ghidra decompile. Catches a
sub-agent that hand-waved, read the wrong function, or dropped branches/calls.
It never relies on an LLM. It only fails on STRONG mismatches; otherwise PASS or
UNKNOWN (insufficient data). Pair it with an LLM checker sub-agent — accept a
candidate only when checker=PASS AND objective!=FAIL (mirrors run_fix_loop).

Usage:
  python objective_verify.py --candidate cand.c --decompile golden.c
  python objective_verify.py --candidate cand.c --decompile golden.c \
         --callees 9 --asm-calls 11 --call-tol 3 --flow-tol 2

  --candidate   file with the sub-agent's reconstructed C/pseudo-C function
  --decompile   the ground-truth Ghidra decompile of the SAME function
                (extract it from all_decomp.c by its "/* ===== name @ VA ===== */" banner)
  --callees     optional: true callee count if you have it (else counted from decompile)
  --asm-calls   optional: call instruction count from disassembly (capstone)
Exit code: 0 PASS, 1 FAIL, 2 UNKNOWN.
"""
from __future__ import annotations

import argparse
import re
import sys

COMMENT_BLOCK_RE = re.compile(r"/\*.*?\*/", re.S)
COMMENT_LINE_RE = re.compile(r"//.*")
TOKEN_CALL_RE = re.compile(r"\b([A-Za-z_][A-Za-z0-9_:]*)\s*\(")
CONTROL_FLOW_RE = re.compile(r"\b(if|for|while|switch|do|goto)\b")

# Tokens that look like calls but are language constructs, not callees.
CPP_KEYWORDS = {
    "if", "for", "while", "switch", "return", "sizeof", "alignof", "decltype",
    "static_cast", "reinterpret_cast", "const_cast", "dynamic_cast", "catch",
    "new", "delete",
}


def strip_comments(text: str) -> str:
    return COMMENT_LINE_RE.sub("", COMMENT_BLOCK_RE.sub("", text))


def count_calls(body_no_comments: str) -> int:
    total = 0
    for m in TOKEN_CALL_RE.finditer(body_no_comments):
        tok = m.group(1)
        if tok in CPP_KEYWORDS:
            continue
        if tok.endswith("::operator") or tok == "operator":
            continue
        total += 1
    return total


def count_control_flow(body_no_comments: str) -> int:
    return len(CONTROL_FLOW_RE.findall(body_no_comments))


def extract_body(text: str) -> str:
    """Return the outermost { ... } block, or the whole text if none found."""
    open_brace = text.find("{")
    close_brace = text.rfind("}")
    if open_brace == -1 or close_brace == -1 or close_brace <= open_brace:
        return text
    return text[open_brace:close_brace + 1]


def verify(candidate: str, decompile: str, callees: int | None,
           asm_calls: int | None, call_tol: int, flow_tol: int):
    """Port of verify_candidate. Returns (verdict, summary, findings)."""
    if not candidate.strip():
        return "FAIL", "No candidate code produced", ["Candidate code is empty"]

    cand_body = strip_comments(extract_body(candidate))
    cand_calls = count_calls(cand_body)
    cand_flow = count_control_flow(cand_body)

    dec_body = strip_comments(extract_body(decompile))
    dec_calls = count_calls(dec_body)
    dec_flow = count_control_flow(dec_body)

    findings: list[str] = []
    checks_run = 0

    # Callee/call-count check: FAIL only when candidate has FEWER calls than the
    # ground truth by more than tolerance (missing logic), never the reverse.
    ref_callees = callees if callees is not None else dec_calls
    if ref_callees:
        checks_run += 1
        if (ref_callees - cand_calls) >= call_tol and cand_calls < ref_callees:
            findings.append(
                f"Call-count mismatch: ground truth ~{ref_callees} calls, "
                f"candidate has {cand_calls}")

    # Control-flow check (only meaningful when the decompile actually branches).
    if dec_flow >= 2:
        checks_run += 1
        if (dec_flow - cand_flow) >= flow_tol and cand_flow < dec_flow:
            findings.append(
                f"Control-flow mismatch: decompile has {dec_flow} branches/loops, "
                f"candidate has {cand_flow}")

    # Optional independent asm cross-check.
    if asm_calls is not None:
        checks_run += 1
        if (asm_calls - cand_calls) >= call_tol and cand_calls < asm_calls:
            findings.append(
                f"ASM call mismatch: disassembly has {asm_calls} calls, "
                f"candidate has {cand_calls}")

    if findings:
        return "FAIL", "Objective verifier found structural mismatches", findings
    if checks_run == 0:
        return "UNKNOWN", "Insufficient structural data to judge", []
    return "PASS", "No structural mismatches found", []


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--candidate", required=True)
    ap.add_argument("--decompile", required=True)
    ap.add_argument("--callees", type=int, default=None)
    ap.add_argument("--asm-calls", type=int, default=None)
    ap.add_argument("--call-tol", type=int, default=3)
    ap.add_argument("--flow-tol", type=int, default=2)
    a = ap.parse_args()

    with open(a.candidate, encoding="utf-8", errors="ignore") as f:
        cand = f.read()
    with open(a.decompile, encoding="utf-8", errors="ignore") as f:
        dec = f.read()

    verdict, summary, findings = verify(
        cand, dec, a.callees, a.asm_calls, a.call_tol, a.flow_tol)
    print(f"OBJECTIVE: {verdict}")
    print(f"SUMMARY: {summary}")
    for fnd in findings:
        print(f"  - {fnd}")
    return {"PASS": 0, "FAIL": 1, "UNKNOWN": 2}[verdict]


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