9c0b5f5e49
- checkAdaptive: cap escalation at 4 min (8x base) instead of 1 hour. The 1-hour ceiling let a genuinely-hard query pin a worker for an hour with zero file completion, stalling the whole scan (observed: 5 workers at 100% CPU, log stopped advancing). 4 min still resolves "just barely over budget" queries while bounding pathological ones. - score_fp.py: guard the last bare .read_text(), drop an unused `elapsed`, log the fold --version failure instead of bare pass, and suppress the runtime sys.path import (reportMissingImports/E402).
1159 lines
41 KiB
Python
Executable file
1159 lines
41 KiB
Python
Executable file
#!/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/<stamp>_fold-juliet
|
|
|
|
Usage (multi-corpus aggregation):
|
|
python3 bench/score_fp.py \
|
|
--results-dir bench/results/<stamp>_fold-juliet \
|
|
--results-dir bench/results/<stamp>_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-<category>.json (built by
|
|
# scripts/benchmarks/build_svcomp_gold.py). Corpus name in findings
|
|
# is `svcomp-<category>` 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())
|