#!/usr/bin/env python3 """Per-detector FP rate measurement and scoring infrastructure. Reads fold findings from one or more result directories (produced by run_bench.py or run_fold_on_juliet.py), matches each finding kind against the corresponding gold standard annotations, and produces per-detector precision/recall/F1 stats. Output: bench/results/fp-baseline.json Usage (simple — score existing results): python3 bench/score_fp.py --results-dir bench/results/_fold-juliet Usage (multi-corpus aggregation): python3 bench/score_fp.py \ --results-dir bench/results/_fold-juliet \ --results-dir bench/results/_security-bugs \ --out bench/results/fp-baseline.json Usage (run + score): python3 bench/score_fp.py --run --security-bugs --juliet \ --juliet-cwes 476 369 190 121 122 124 126 127 191 561 \ --fold-bin build/release/src/cli/fold \ --out bench/results/fp-baseline.json """ from __future__ import annotations import argparse import datetime as dt import json import os import re import shlex import subprocess import sys import time from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "scripts" / "benchmarks")) # pi-lens-ignore: reportMissingImports, E402 from fold_eval.metrics import classification_metrics GOLD_ROOT = REPO_ROOT / "bench" / "gold" RESULTS_ROOT = REPO_ROOT / "bench" / "results" # Committed (tracked) precision baseline. bench/results/ is gitignored ("too # large"), which is exactly why the old baseline had no history and drift went # unnoticed — the reference must live at a tracked path. BASELINE_PATH = REPO_ROOT / "bench" / "fp-baseline.json" JULIET_FILTERED = REPO_ROOT / "bench" / "corpora" / "juliet" / "filtered" SECURITY_BUGS_CORPUS = REPO_ROOT / "bench" / "corpora" / "security-bugs" SVCOMP_CORPUS = REPO_ROOT / "bench" / "corpora" / "svcomp" / "sv-benchmarks" / "c" DEFAULT_FOLD = REPO_ROOT / "build" / "release" / "src" / "cli" / "fold" LLVM_BIN = Path(os.environ.get("LLVM_BIN", "/opt/homebrew/opt/llvm@22/bin")) # `latent` is a special, *conditional* classification (proved-modulo-an- # undischarged caller precondition), orthogonal to the strength tiers # (concrete-witness > proved > heuristic > syntactic). By default it is excluded # from the precision/recall metric so it neither penalizes precision nor inflates # recall; gold cases caught only at the latent tier are reported as # `latent_recovered`. Note: `heuristic` findings are genuine (over-approximated) # detections and DO count — only `latent` is conditional. # Set from --include-latent in main(). _INCLUDE_LATENT = False def _is_latent(finding: dict) -> bool: return str(finding.get("confidence", "")).lower() == "latent" def _load_json_file(path): """Read and parse a JSON file, exiting with a clear message on failure.""" try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise SystemExit(f"fold: error reading {path}: {exc}") from exc def _parse_json_line(line): try: return json.loads(line) except ValueError: return None def _as_int(value, default=0): try: return int(value or default) except (TypeError, ValueError): return default def _as_float(value, default=0.0): try: return float(value) except (TypeError, ValueError): return default # --------------------------------------------------------------------------- # Detector-kind registry # --------------------------------------------------------------------------- # Maps each fold detector to the finding kind(s) it emits and the Juliet CWEs # (and security-bugs entries) that provide gold-standard ground truth for it. DETECTOR_REGISTRY: dict[str, dict[str, Any]] = { "null-deref": { "kinds": ["null-deref"], "juliet_cwes": ["476"], "security_bugs_gold_kinds": ["null-deref"], }, "div-by-zero": { "kinds": ["div-by-zero"], "juliet_cwes": ["369"], "security_bugs_gold_kinds": ["div-by-zero"], }, "int-overflow": { "kinds": ["int-overflow"], "juliet_cwes": ["190", "191"], "security_bugs_gold_kinds": ["int-overflow"], }, "oob": { "kinds": ["oob-access"], "juliet_cwes": ["121", "122", "124", "126", "127"], "security_bugs_gold_kinds": ["oob-access"], }, "dead": { "kinds": ["dead-branch", "dead-statement", "unused-function", "dead"], "juliet_cwes": ["561"], "security_bugs_gold_kinds": ["dead-branch"], }, "sign-conversion": { "kinds": ["sign-conversion"], "juliet_cwes": ["194", "195", "196"], "security_bugs_gold_kinds": [], }, "int-truncation": { "kinds": ["int-truncation"], "juliet_cwes": ["197"], "security_bugs_gold_kinds": [], }, "use-after-free": { "kinds": ["use-after-free"], "juliet_cwes": [], "security_bugs_gold_kinds": [], }, "memory-leak": { "kinds": ["memory-leak"], "juliet_cwes": [], "security_bugs_gold_kinds": [], }, } # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def now_iso() -> str: return dt.datetime.now(dt.UTC).strftime("%Y-%m-%dT%H-%M-%SZ") def file_basename(path: str) -> str: return Path(path).name def normalize_kind(kind: str) -> str: """Canonicalize finding kinds for matching.""" m = { "dead-branch": "dead-branch", "dead-statement": "dead-statement", "unused-function": "unused-function", "dead": "dead-branch", # generic dead -> dead-branch "null-deref": "null-deref", "div-by-zero": "div-by-zero", "int-overflow": "int-overflow", "oob-access": "oob-access", "oob": "oob-access", "sign-conversion": "sign-conversion", "int-truncation": "int-truncation", "class-dup": "class-dup", "clone": "clone", "equiv": "equiv", } return m.get(kind, kind) def strip_corpus_prefix(file_path: str, corpus_dir: Path) -> str: """Return file path relative to corpus root.""" try: return str(Path(file_path).relative_to(corpus_dir)) except ValueError: return file_path # --------------------------------------------------------------------------- # Gold-standard loading # --------------------------------------------------------------------------- def load_gold(gold_path: Path) -> list[dict]: """Load gold-standard entries from a JSON file.""" if not gold_path.exists(): return [] doc = _load_json_file(gold_path) return doc.get("expected", []) def load_all_juliet_gold() -> dict[str, list[dict]]: """Return {cwe: gold_entries} for all available Juliet gold files.""" result: dict[str, list[dict]] = {} for gold_file in sorted(GOLD_ROOT.glob("juliet-*.json")): cwe = gold_file.stem.replace("juliet-", "") result[cwe] = load_gold(gold_file) return result # --------------------------------------------------------------------------- # Findings parsing # --------------------------------------------------------------------------- def parse_sarif(sarif_path: Path) -> list[dict]: """Parse a fold SARIF file into normalized findings.""" if not sarif_path.exists(): return [] doc = _load_json_file(sarif_path) findings: list[dict] = [] for run in doc.get("runs", []): for result in run.get("results", []): kind = normalize_kind( result.get("ruleId", "fold.unknown").removeprefix("fold.") ) fid = result.get("properties", {}).get("id", "") for loc in result.get("locations", []): physical = loc.get("physicalLocation", {}) artifact = physical.get("artifactLocation", {}) region = physical.get("region", {}) findings.append( { "file": artifact.get("uri", ""), "line": _as_int(region.get("startLine", 0)), "kind": kind, "id": fid, "confidence": result.get("properties", {}).get( "confidence", "" ), } ) return dedup_findings(findings) def parse_fold_json(json_path: Path) -> list[dict]: """Parse a fold.json payload (from run_bench or run_fold_on_juliet).""" if not json_path.exists(): return [] doc = _load_json_file(json_path) raw_findings = doc.get("findings", []) out: list[dict] = [] for f in raw_findings: if isinstance(f, dict) and "file" in f: out.append( { "file": str(f.get("file", "")), "line": _as_int(f.get("line", 0)), "kind": normalize_kind(str(f.get("kind", "unknown"))), "id": str(f.get("id", "")), "confidence": str(f.get("confidence", "")), } ) return dedup_findings(out) def parse_ndjson_findings(ndjson_path: Path) -> list[dict]: """Parse fold native ndjson/grid output.""" if not ndjson_path.exists(): return [] findings: list[dict] = [] try: text = ndjson_path.read_text() except OSError: return [] for line in text.splitlines(): line = line.strip() if not line: continue try: item = _parse_json_line(line) except json.JSONDecodeError: continue if not isinstance(item, dict): continue kind = item.get("kind", "") data = item.get("data", item) if kind == "finding" and isinstance(data, dict): findings.append( { "file": str(data.get("file", "")), "line": _as_int(data.get("line", 0)), "kind": normalize_kind(str(data.get("kind", "unknown"))), "id": str(data.get("id", "")), "confidence": str(data.get("confidence", "")), } ) return dedup_findings(findings) def dedup_findings(findings: list[dict]) -> list[dict]: """Deduplicate findings by (file, kind). Prefer a non-latent finding over a latent one for the same (file, kind), so a conditional (latent) duplicate cannot mask a genuine Proved finding and drop it from the precision/recall metric. Within the same latent-ness, keep the earliest line for stability. """ dedup: dict[tuple[str, str], dict] = {} for f in findings: key = (str(Path(f["file"]).resolve()), f["kind"]) current = dedup.get(key) if current is None: dedup[key] = f continue cur_latent = _is_latent(current) new_latent = _is_latent(f) if cur_latent and not new_latent: dedup[key] = f # promote: real finding beats a latent dup elif cur_latent == new_latent and f.get("line", 0) < current.get("line", 0): dedup[key] = f return sorted(dedup.values(), key=lambda x: (x["file"], x["line"])) def load_findings_from_dir(dir_path: Path) -> list[dict]: """Try to load findings from a result directory.""" # Prefer fold.json (most complete), then fold.sarif, then NDJSON candidates = [ ("fold.json", parse_fold_json), ("fold.sarif", parse_sarif), ("fold.findings.json", parse_fold_json), ("fold.ndjson", parse_ndjson_findings), ] for name, parser in candidates: path = dir_path / name findings = parser(path) if findings: return findings return [] # --------------------------------------------------------------------------- # File-level scoring # --------------------------------------------------------------------------- def score_detector_on_corpus( findings: list[dict], gold_entries: list[dict], target_kinds: list[str], corpus_dir: Path | None = None, scanned_files: set[str] | None = None, ) -> dict[str, Any]: """Compute precision/recall/F1 for one detector on one corpus. Uses file-level matching: a gold entry is "found" if fold emitted at least one finding of the target kind in that file. A false positive is a finding of the target kind in a file that is not in the gold set. When `scanned_files` is provided (basenames), the gold set is restricted to files that were actually scanned. This is required when the corpus is down-sampled (--max-files-per-cwe): otherwise unscanned gold cases would be counted as false negatives and recall would be meaningless. """ kind_set = set(target_kinds) # Gold files: files where fold *should* emit a finding of target kind gold_files: set[str] = set() for entry in gold_entries: entry_kind = entry.get("kind", "") if entry_kind in kind_set or entry_kind == "any": for f in entry.get("files", []): gold_files.add(Path(f).name) if scanned_files is not None: gold_files &= scanned_files # Finding files: files where fold *did* emit a non-conditional finding of # target kind. `latent` findings are conditional (caller precondition) and, # unless --include-latent, neither count toward TP/recall nor penalize # precision; their gold matches are reported as `latent_recovered`. finding_files: set[str] = set() latent_files: set[str] = set() for f in findings: if f.get("kind") not in kind_set: continue name = Path(f["file"]).name if _is_latent(f) and not _INCLUDE_LATENT: latent_files.add(name) else: finding_files.add(name) # If corpus_dir is provided, convert absolute paths to basenames # (needed when findings use absolute paths from the SARIF output) # Actually the dedup already uses resolve(), but findings may have absolute # paths. We always compare basenames to be safe. tp_files = gold_files & finding_files fp_files = finding_files - gold_files fn_files = gold_files - finding_files # Gold cases missed at the scored tier but caught by a lower-tier (latent) # finding — recall information that the headline metric would otherwise hide. latent_recovered = len(fn_files & latent_files) tp = len(tp_files) fp = len(fp_files) fn = len(fn_files) metrics = classification_metrics(tp, fp, fn) return { "tp": tp, "fp": fp, "fn": fn, "gold_total": len(gold_files), "finding_total": len(finding_files), "precision": round(_as_float(metrics.get("precision")), 4), "recall": round(_as_float(metrics.get("recall")), 4), "f1": round(_as_float(metrics.get("f1")), 4), "latent_recovered": latent_recovered, "tp_files": sorted(tp_files), "fp_files": sorted(fp_files), "fn_files": sorted(fn_files), } # --------------------------------------------------------------------------- # Aggregate scoring across all corpora # --------------------------------------------------------------------------- def aggregate_detector_scores( per_corpus: dict[str, dict[str, Any]], ) -> dict[str, Any]: """Sum TP/FP/FN across corpora and recompute precision/recall/F1.""" tp = sum(s["tp"] for s in per_corpus.values()) fp = sum(s["fp"] for s in per_corpus.values()) fn = sum(s["fn"] for s in per_corpus.values()) gold_total = sum(s["gold_total"] for s in per_corpus.values()) finding_total = sum(s["finding_total"] for s in per_corpus.values()) latent_recovered = sum(s.get("latent_recovered", 0) for s in per_corpus.values()) metrics = classification_metrics(tp, fp, fn) return { "tp": tp, "fp": fp, "fn": fn, "gold_total": gold_total, "finding_total": finding_total, "latent_recovered": latent_recovered, "precision": round(_as_float(metrics.get("precision")), 4), "recall": round(_as_float(metrics.get("recall")), 4), "f1": round(_as_float(metrics.get("f1")), 4), "corpora": per_corpus, } # --------------------------------------------------------------------------- # Compile-database generation # --------------------------------------------------------------------------- def write_compile_db(corpus_dir: Path, db_path: Path, max_files: int = 0) -> list[dict]: """Generate compile_commands.json covering .c/.cpp under corpus. When max_files > 0, deterministically down-sample to ~max_files sources via a fixed stride over the sorted (.c then .cpp) list. Stride sampling (vs a prefix) keeps coverage across Juliet flow-variants (01..82) and both C and C++ files, so a large CWE (121/122 are ~8-10k files) yields a tractable, reproducible gate corpus rather than an hours-long full run. """ entries: list[dict] = [] include_dirs: set[str] = set() for path in sorted(corpus_dir.rglob("*")): if path.is_dir() and path.name == "testcasesupport": include_dirs.add(str(path)) if path.is_dir() and any(path.glob("*.h")): include_dirs.add(str(path)) inc_flags = [flag for d in sorted(include_dirs) for flag in ("-I", d)] clang = LLVM_BIN / "clang" clangxx = LLVM_BIN / "clang++" # (path, kind) for every source, sorted for determinism. kind in # {"c", "cxx", "i"} where "i" = preprocessed C (.i, SV-COMP Juliet_Test) # — compiled with -x c so clang treats the preprocessed file as C. sources: list[tuple[Path, str]] = ( [(s, "c") for s in sorted(corpus_dir.rglob("*.c"))] + [(s, "cxx") for s in sorted(corpus_dir.rglob("*.cpp"))] + [(s, "i") for s in sorted(corpus_dir.rglob("*.i"))] ) if max_files > 0 and len(sources) > max_files: stride = len(sources) // max_files sources = sources[::stride][:max_files] for src, kind in sources: if kind == "cxx": cmd = shlex.join( [str(clangxx), "-std=c++17"] + inc_flags + ["-c", str(src)] ) elif kind == "i": cmd = shlex.join( [str(clang), "-x", "c", "-std=c11"] + inc_flags + ["-c", str(src)] ) else: cmd = shlex.join([str(clang), "-std=c11"] + inc_flags + ["-c", str(src)]) entries.append({"directory": str(corpus_dir), "command": cmd, "file": str(src)}) db_path.write_text(json.dumps(entries, indent=2) + "\n") return entries def scanned_basenames(out_dir: Path) -> set[str]: """Basenames of the sources actually scanned, from the run's compile DB. Used to restrict gold to the scanned subset when --max-files-per-cwe caps a corpus, so recall stays meaningful. """ try: entries = _load_json_file(out_dir / "compile_commands.json") return {Path(e["file"]).name for e in entries if e.get("file")} except Exception: return set() # --------------------------------------------------------------------------- # Running fold # --------------------------------------------------------------------------- def run_fold_on_corpus( corpus_dir: Path, out_dir: Path, fold_bin: Path, detectors: list[str], processes: int = 1, exclude_globs: list[str] | None = None, max_files: int = 0, ) -> list[dict]: """Run fold scan on a corpus directory and return parsed findings.""" out_dir.mkdir(parents=True, exist_ok=True) db_path = out_dir / "compile_commands.json" write_compile_db(corpus_dir, db_path, max_files=max_files) sarif_path = out_dir / "fold.sarif" stdout_path = out_dir / "fold.stdout.log" stderr_path = out_dir / "fold.stderr.log" cmd = [ str(fold_bin), "scan", "--compile-db", str(db_path), "--format", "sarif", "--output", str(sarif_path), "--min-body-stmts", "1", "--processes", str(processes), ] for d in detectors: cmd += ["--detectors", d] if exclude_globs: for g in exclude_globs: cmd += ["--exclude-glob", g] env = os.environ.copy() env["PATH"] = f"{LLVM_BIN}:{env.get('PATH', '')}" t0 = time.monotonic() proc = subprocess.run(cmd, capture_output=True, text=True, env=env) _elapsed = time.monotonic() - t0 stdout_path.write_text(proc.stdout) stderr_path.write_text(proc.stderr) findings = parse_sarif(sarif_path) if not findings: # Fallback: try fold.findings.json findings = parse_fold_json(out_dir / "fold.findings.json") return findings # --------------------------------------------------------------------------- # Main orchestration # --------------------------------------------------------------------------- def collect_results_from_dir( result_dirs: list[Path], ) -> dict[str, list[dict]]: """Scan result directories and collect findings per corpus/cwe. Returns a dict mapping corpus-name -> findings list. """ all_findings: dict[str, list[dict]] = {} for rd in result_dirs: if not rd.is_dir(): continue # Check if this is a juliet multi-cwe output (has cwe subdirs) juliet_cwe_dirs = sorted( p for p in rd.iterdir() if p.is_dir() and re.match(r"^\d{3}$", p.name) ) if juliet_cwe_dirs: for cwe_dir in juliet_cwe_dirs: cwe = cwe_dir.name findings = load_findings_from_dir(cwe_dir) if findings: key = f"juliet-{cwe}" all_findings[key] = findings else: # Single corpus run (security-bugs etc.) corpus_name = rd.name.split("_", 1)[-1] if "_" in rd.name else rd.name findings = load_findings_from_dir(rd) if findings: all_findings[corpus_name] = findings return all_findings def build_fp_baseline( findings_by_corpus: dict[str, list[dict]], juliet_gold: dict[str, list[dict]], security_bugs_gold: list[dict], fold_version: str = "unknown", scanned_by_corpus: dict[str, set[str]] | None = None, ) -> dict[str, Any]: """Compute per-detector scores from collected findings and gold standards.""" detectors: dict[str, Any] = {} scanned_by_corpus = scanned_by_corpus or {} for det_name, det_info in DETECTOR_REGISTRY.items(): target_kinds = det_info["kinds"] per_corpus_scores: dict[str, dict] = {} # Score each available corpus for corpus_name, findings in sorted(findings_by_corpus.items()): gold_entries: list[dict] = [] if corpus_name.startswith("juliet-"): cwe = corpus_name.replace("juliet-", "") if cwe in juliet_gold: gold_entries = juliet_gold[cwe] elif "security" in corpus_name.lower(): gold_entries = security_bugs_gold else: # Try to find a matching gold file gold_path = GOLD_ROOT / f"{corpus_name}.json" if gold_path.exists(): gold_entries = load_gold(gold_path) if not gold_entries: continue score = score_detector_on_corpus( findings, gold_entries, target_kinds, scanned_files=scanned_by_corpus.get(corpus_name), ) # Only include corpora that actually have gold entries for this detector if score["gold_total"] > 0: per_corpus_scores[corpus_name] = score if per_corpus_scores: detectors[det_name] = aggregate_detector_scores(per_corpus_scores) # Literature context for interpretation literature_context = { "codeql_fp_rate": "~5%", "semgrep_fp_rate": "~12%", "source": "Bennett et al. 2024 (EASE), Lipp et al. 2022 (ISSTA)", } return { "fold_version": fold_version, "generated": now_iso(), "scoring_model": "file-level per-detector", "corpora_available": sorted(findings_by_corpus.keys()), "literature_context": literature_context, "detectors": detectors, } def render_summary(baseline: dict[str, Any]) -> str: """Render a Markdown summary table.""" lines = [ "# fold FP baseline", "", f"Generated: {baseline['generated']}", f"Fold version: {baseline['fold_version']}", "", "## Per-detector precision / recall / F1", "", "| Detector | TP | FP | FN | Gold | Findings | Precision | Recall | F1 |", "|----------|----|----|----|------|----------|-----------|--------|----|", ] for det_name, det in sorted(baseline["detectors"].items()): lines.append( f"| {det_name} | {det['tp']} | {det['fp']} | {det['fn']} | " f"{det['gold_total']} | {det['finding_total']} | " f"{det['precision']:.1%} | {det['recall']:.1%} | {det['f1']:.1%} |" ) lines.append("") lines.append("## Literature context") lines.append("") ctx = baseline["literature_context"] lines.append(f"- CodeQL FP rate: {ctx['codeql_fp_rate']}") lines.append(f"- Semgrep FP rate: {ctx['semgrep_fp_rate']}") lines.append(f"- Source: {ctx['source']}") lines.append("") lines.append("## Per-detector per-corpus breakdown") lines.append("") for det_name, det in sorted(baseline["detectors"].items()): lines.append(f"### {det_name}") lines.append("") lines.append("| Corpus | TP | FP | FN | Gold | Findings | Prec | Rec | F1 |") lines.append("|--------|----|----|----|------|----------|------|-----|----|") for corpus, score in sorted(det.get("corpora", {}).items()): lines.append( f"| {corpus} | {score['tp']} | {score['fp']} | {score['fn']} | " f"{score['gold_total']} | {score['finding_total']} | " f"{score['precision']:.1%} | {score['recall']:.1%} | {score['f1']:.1%} |" ) lines.append("") return "\n".join(lines) + "\n" def gate_against_baseline( current: dict[str, Any], baseline_path: Path, tolerance: float ) -> tuple[bool, str]: """Compare freshly-computed per-detector scores to a committed baseline. Returns (passed, report). A detector regresses if its aggregate precision, recall, or F1 drops by more than `tolerance` vs the baseline, or if a detector that scored in the baseline produces no comparable score now. Improvements and new detectors never fail the gate. This is the enforcement the quality gate previously lacked (drift was invisible). """ base = load_json_file(baseline_path) if not base or "detectors" not in base: return False, f"baseline not found or malformed: {baseline_path}" base_dets = base.get("detectors", {}) cur_dets = current.get("detectors", {}) rows: list[str] = [] regressions = 0 # Policy (matches loop-summarization-research.md precision contract): a # regression is a drop in RECALL (never lose a true finding) or F1 (net # quality) beyond tolerance. A precision dip that comes WITH a recall gain # (F1 holds/rises) is allowed — that is the expected, contract-sanctioned # over-approximation tradeoff (e.g. counted-loop summarization), and such # findings are tagged lower-confidence. The recall+F1 gate still catches # the regressions we built it for (those were recall/F1 drops). for det in sorted(base_dets): b = base_dets[det] c = cur_dets.get(det) if not c: rows.append(f" {det:14} MISSING in current run <== REGRESSION") regressions += 1 continue worst = "" det_regressed = False for metric in ("recall", "f1"): bm, cm = _as_float(b.get(metric, 0)), _as_float(c.get(metric, 0)) if cm < bm - tolerance: det_regressed = True worst += f" {metric} {bm:.3f}->{cm:.3f}" # Surface a large precision cliff as advisory even when F1/recall hold. bp, cp = _as_float(b.get("precision", 0)), _as_float(c.get("precision", 0)) note = ( " (precision dip; recall/F1 ok)" if (cp < bp - tolerance and not det_regressed) else "" ) tag = ( " <== REGRESSION" if det_regressed else ( " (gain)" if _as_float(c.get("f1", 0)) > _as_float(b.get("f1", 0)) + tolerance else note ) ) if det_regressed: regressions += 1 rows.append( f" {det:14} P {b.get('precision'):.3f}->{c.get('precision'):.3f}" f" R {b.get('recall'):.3f}->{c.get('recall'):.3f}" f" F1 {b.get('f1'):.3f}->{c.get('f1'):.3f}{worst}{tag}" ) header = ( f"=== drift gate (recall+F1, tolerance {tolerance}) vs {baseline_path.name} ===\n" f"baseline generated: {base.get('generated', '?')}\n" ) verdict = ( "PASS — no per-detector regression beyond tolerance" if regressions == 0 else f"FAIL — {regressions} detector(s) regressed" ) return regressions == 0, header + "\n".join(rows) + f"\n{verdict}" def load_json_file(path: Path) -> Any | None: try: return _load_json_file(Path(path)) except Exception: return None # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main() -> int: parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--results-dir", action="append", default=[], help="Path to a result directory containing fold output " "(may be specified multiple times)", ) parser.add_argument( "--findings-file", action="append", default=[], help="Path to a fold.json or fold.sarif file", ) parser.add_argument( "--out", default=str(RESULTS_ROOT / "fp-baseline.json"), help="Output path for fp-baseline.json", ) parser.add_argument( "--summary", action="store_true", help="Also write a summary.md alongside fp-baseline.json", ) # Run mode parser.add_argument( "--run", action="store_true", help="Run fold analysis before scoring", ) parser.add_argument( "--security-bugs", action="store_true", help="Also run fold on the security-bugs corpus", ) parser.add_argument( "--svcomp", action="store_true", help="Also run fold on SV-COMP categories (gold from svcomp-*.json)", ) parser.add_argument( "--svcomp-categories", nargs="*", default=["memsafety", "array-memsafety", "array-memsafety-realloc"], help="SV-COMP category dirs to scan (default: the memsafety family)", ) parser.add_argument( "--juliet", action="store_true", help="Also run fold on Juliet filtered CWEs", ) parser.add_argument( "--juliet-cwes", nargs="*", default=["476", "369", "190", "121", "122", "124", "126", "127", "191", "561"], ) parser.add_argument("--fold-bin", default=str(DEFAULT_FOLD)) parser.add_argument("--processes", type=int, default=1) parser.add_argument("--run-dir", default=None, help="Directory for run outputs") parser.add_argument( "--max-files-per-cwe", type=int, default=0, help="Deterministically down-sample each Juliet CWE to ~N files " "(0 = all). Keeps the gate tractable on huge CWEs (121/122).", ) # Gate mode: compare the freshly-computed scores against a committed # baseline and fail (exit 1) on any per-detector regression. This is the # local precision gate — drift went unnoticed before because nothing # enforced the baseline. parser.add_argument( "--gate", action="store_true", help="Compare results to --baseline and exit non-zero on regression.", ) parser.add_argument( "--baseline", default=str(BASELINE_PATH), help="Committed baseline to compare against in --gate mode.", ) parser.add_argument( "--tolerance", type=float, default=0.02, help="Max allowed per-detector precision/recall/F1 drop in --gate mode.", ) parser.add_argument( "--include-latent", action="store_true", help="Count `latent` (caller-precondition) findings as normal findings " "in precision/recall. Default: latent is excluded from the metric " "and reported separately as latent_recovered.", ) args = parser.parse_args() if args.include_latent: global _INCLUDE_LATENT _INCLUDE_LATENT = True # --- Collect findings --- findings_by_corpus: dict[str, list[dict]] = {} # corpus -> basenames actually scanned (populated by --run; used to align # gold with a down-sampled corpus so recall stays meaningful). scanned_by_corpus: dict[str, set[str]] = {} # From existing results directories for rd_path in args.results_dir: rd = Path(rd_path).resolve() if not rd.is_dir(): print(f"Results dir not found: {rd}", file=sys.stderr) continue collected = collect_results_from_dir([rd]) for name, findings in collected.items(): if name not in findings_by_corpus or len(findings) > len( findings_by_corpus[name] ): findings_by_corpus[name] = findings if not collected: print( f"No findings found in {rd} — trying direct file parse", file=sys.stderr ) # Try individual files in the directory for fname in ["fold.json", "fold.sarif"]: fp = rd / fname if fp.exists(): corpus = rd.name if ".json" in fp.name: findings_by_corpus[corpus] = parse_fold_json(fp) else: findings_by_corpus[corpus] = parse_sarif(fp) break # From individual findings files for ff_path in args.findings_file: ff = Path(ff_path).resolve() if not ff.exists(): print(f"Findings file not found: {ff}", file=sys.stderr) continue corpus = ff.stem if ff.suffix == ".sarif": findings_by_corpus[corpus] = parse_sarif(ff) elif ff.suffix == ".json": findings_by_corpus[corpus] = parse_fold_json(ff) elif ff.suffix == ".ndjson": findings_by_corpus[corpus] = parse_ndjson_findings(ff) # --- Run fold if requested --- fold_bin = Path(args.fold_bin).resolve() if args.run: if not fold_bin.exists(): print(f"fold binary not found: {fold_bin}", file=sys.stderr) return 2 run_stamp = now_iso() run_root = ( Path(args.run_dir) if args.run_dir else RESULTS_ROOT / f"{run_stamp}_fp-baseline" ) run_root.mkdir(parents=True, exist_ok=True) # Determine which detectors to run based on requested corpora detectors_to_run: set[str] = set() if args.security_bugs: detectors_to_run.update( ["null-deref", "div-by-zero", "int-overflow", "oob", "dead"] ) if args.juliet: for cwe in args.juliet_cwes: for det_name, det_info in DETECTOR_REGISTRY.items(): if cwe in det_info["juliet_cwes"]: detectors_to_run.add(det_name) if not detectors_to_run: detectors_to_run = set(DETECTOR_REGISTRY.keys()) detector_list = sorted(detectors_to_run) # Run on security-bugs if args.security_bugs and SECURITY_BUGS_CORPUS.is_dir(): print("Running fold on security-bugs...", file=sys.stderr) sec_out = run_root / "security-bugs" findings = run_fold_on_corpus( SECURITY_BUGS_CORPUS, sec_out, fold_bin, detector_list, args.processes, ) if findings: findings_by_corpus["security-bugs"] = findings scanned_by_corpus["security-bugs"] = scanned_basenames(sec_out) # Run on Juliet if args.juliet: for cwe in args.juliet_cwes: corpus_dir = JULIET_FILTERED / cwe if not corpus_dir.is_dir(): print( f"Juliet CWE-{cwe} not found at {corpus_dir}", file=sys.stderr ) continue cwe_detectors = [ d for d in detector_list if cwe in DETECTOR_REGISTRY[d]["juliet_cwes"] ] if not cwe_detectors: cwe_detectors = detector_list print(f"Running fold on Juliet CWE-{cwe}...", file=sys.stderr) cwe_out = run_root / f"juliet-{cwe}" findings = run_fold_on_corpus( corpus_dir, cwe_out, fold_bin, cwe_detectors, args.processes, max_files=args.max_files_per_cwe, ) if findings: findings_by_corpus[f"juliet-{cwe}"] = findings scanned_by_corpus[f"juliet-{cwe}"] = scanned_basenames(cwe_out) # Run on SV-COMP categories (user-approved benchmark expansion). # Gold comes from bench/gold/svcomp-.json (built by # scripts/benchmarks/build_svcomp_gold.py). Corpus name in findings # is `svcomp-` so the scoring `else` branch loads the gold. if args.svcomp and SVCOMP_CORPUS.is_dir(): sv_detectors = [ "null-deref", "oob", "div-by-zero", "int-overflow", "use-after-free", "memory-leak", ] for category in args.svcomp_categories: corpus_dir = SVCOMP_CORPUS / category if not corpus_dir.is_dir(): print( f"SV-COMP {category} not found at {corpus_dir}", file=sys.stderr ) continue print(f"Running fold on SV-COMP {category}...", file=sys.stderr) sv_out = run_root / f"svcomp-{category.replace('/', '_')}" findings = run_fold_on_corpus( corpus_dir, sv_out, fold_bin, sv_detectors, args.processes, ) if findings: findings_by_corpus[f"svcomp-{category.replace('/', '_')}"] = ( findings ) scanned_by_corpus[f"svcomp-{category.replace('/', '_')}"] = ( scanned_basenames(sv_out) ) if not findings_by_corpus: print( "No findings collected. Use --results-dir, --findings-file, or --run.", file=sys.stderr, ) return 2 # --- Score --- juliet_gold = load_all_juliet_gold() security_bugs_gold = load_gold(GOLD_ROOT / "security-bugs.json") # Get fold version fold_version = "unknown" try: result = subprocess.run( [str(fold_bin), "--version"], capture_output=True, text=True ) fold_version = result.stdout.strip() or "unknown" except Exception as exc: # `fold --version` failed; fall back to "unknown". print(f"fold --version failed: {exc}", file=sys.stderr) baseline = build_fp_baseline( findings_by_corpus, juliet_gold, security_bugs_gold, fold_version, scanned_by_corpus=scanned_by_corpus, ) # Gate mode: compare to the committed baseline and fail on regression. # Do NOT overwrite the committed baseline in gate mode. if args.gate: passed, report = gate_against_baseline( baseline, Path(args.baseline), args.tolerance ) print(report) if args.out and Path(args.out) != Path(args.baseline): Path(args.out).parent.mkdir(parents=True, exist_ok=True) Path(args.out).write_text( json.dumps(baseline, indent=2, sort_keys=True) + "\n" ) return 0 if passed else 1 out_path = Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(json.dumps(baseline, indent=2, sort_keys=True) + "\n") print(f"Wrote {out_path}") if args.summary: summary_path = out_path.parent / "fp-baseline.md" summary_path.write_text(render_summary(baseline)) print(f"Wrote {summary_path}") # Print summary to stdout print(render_summary(baseline)) return 0 if __name__ == "__main__": raise SystemExit(main())