fold/bench/juliet/expansion/run_expanded.py
trvon 56550a93a1
feat: symex-fuzz landscape, SWAN SIL parser, Juliet expansion, FP scoring
- docs/internal/symex-fuzz-landscape.md: 11 tools across 6 categories
- tools/fold-swift/swan_sil_parser: 75 SIL operator kinds, JSON converter
- bench/juliet/expansion/: 118-CWE catalog, gold generator, runner
- bench/score_fp.py: precision/recall/F1 per detector
2026-05-27 18:00:06 -04:00

608 lines
21 KiB
Python

#!/usr/bin/env python3
"""Run fold against expanded Juliet CWE categories and score precision/recall.
Uses the catalog at bench/juliet/expansion/cwe_catalog.json to determine which
CWEs to run and which detectors to enable for each. Generates per-CWE SARIF
output, a fold.json payload, and per-detector aggregate scores.
Usage:
# Run all CWEs mapped to the 6 target detectors
python3 bench/juliet/expansion/run_expanded.py --processes 4
# Run only specific detectors
python3 bench/juliet/expansion/run_expanded.py --detectors null-deref,div-by-zero --processes 1
# Run specific CWEs
python3 bench/juliet/expansion/run_expanded.py --cwes 476,690,369
# Quick smoke test (limit testcases)
python3 bench/juliet/expansion/run_expanded.py --smoke --limit-files 20
# Dry run (print what would be executed)
python3 bench/juliet/expansion/run_expanded.py --dry-run
"""
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
REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent
HERE = Path(__file__).resolve().parent
CATALOG_PATH = HERE / "cwe_catalog.json"
JULIET_ROOT = REPO_ROOT / "bench" / "corpora" / "juliet" / "filtered"
GOLD_ROOT = REPO_ROOT / "bench" / "gold"
RESULTS_ROOT = REPO_ROOT / "bench" / "results"
DEFAULT_FOLD = REPO_ROOT / "build" / "release" / "src" / "cli" / "fold"
LLVM_BIN = Path(os.environ.get("LLVM_BIN", "/opt/homebrew/opt/llvm@22/bin"))
CLANG = LLVM_BIN / "clang"
CLANGXX = LLVM_BIN / "clang++"
# ── CWE → detector mapping (extended beyond the original run_fold_on_juliet.py) ──
CWE_DETECTOR_MAP: dict[str, dict] = {
"476": {"kinds": ["null-deref"], "detectors": ["null-deref"]},
"369": {
"kinds": ["div-by-zero"],
"detectors": ["div-by-zero"],
"exclude_globs": [
"*CWE369_Divide_by_Zero__float_listenSocket_73a.cpp",
"*CWE369_Divide_by_Zero__float_listenSocket_74a.cpp",
"*CWE369_Divide_by_Zero__int_listen_socket_divide_73a.cpp",
"*CWE369_Divide_by_Zero__int_listen_socket_divide_74a.cpp",
"*CWE369_Divide_by_Zero__int_listen_socket_modulo_73a.cpp",
"*CWE369_Divide_by_Zero__int_listen_socket_modulo_74a.cpp",
],
},
"190": {"kinds": ["int-overflow"], "detectors": ["int-overflow"]},
"191": {"kinds": ["int-overflow"], "detectors": ["int-overflow"]},
"194": {"kinds": ["sign-conversion"], "detectors": ["sign-conversion"]},
"195": {"kinds": ["sign-conversion"], "detectors": ["sign-conversion"]},
"196": {"kinds": ["sign-conversion"], "detectors": ["sign-conversion"]},
"197": {"kinds": ["int-truncation"], "detectors": ["int-truncation"]},
"121": {"kinds": ["oob-access"], "detectors": ["oob"]},
"122": {"kinds": ["oob-access"], "detectors": ["oob"]},
"124": {"kinds": ["oob-access"], "detectors": ["oob"]},
"126": {"kinds": ["oob-access"], "detectors": ["oob"]},
"127": {"kinds": ["oob-access"], "detectors": ["oob"]},
"561": {
"kinds": ["dead", "dead-statement", "unused-function"],
"detectors": ["dead", "dead-statement", "unused-function"],
},
# ── Expanded CWEs ──
"690": {"kinds": ["null-deref"], "detectors": ["null-deref"]},
"252": {"kinds": ["null-deref"], "detectors": ["null-deref"]},
"588": {"kinds": ["null-deref"], "detectors": ["null-deref"]},
"562": {"kinds": ["null-deref"], "detectors": ["null-deref"]},
"680": {"kinds": ["int-overflow"], "detectors": ["int-overflow"]},
"789": {"kinds": ["int-overflow"], "detectors": ["int-overflow"]},
"681": {"kinds": ["int-truncation"], "detectors": ["int-truncation"]},
"467": {"kinds": ["sign-conversion"], "detectors": ["sign-conversion"]},
"123": {"kinds": ["oob-access"], "detectors": ["oob"]},
"468": {"kinds": ["oob-access"], "detectors": ["oob"]},
"785": {"kinds": ["oob-access"], "detectors": ["oob"]},
}
# ── Compiled regexes ──
SOLVER_RE = re.compile(
r"^\[solver-profile\]\s+(?P<detector>\S+)\s+"
r"queries=(?P<queries>\d+)\s+sat=(?P<sat>\d+)\s+unsat=(?P<unsat>\d+)\s+"
r"unknown=(?P<unknown>\d+)\s+decisions=(?P<decisions>\d+)\s+"
r"conflicts=(?P<conflicts>\d+)\s+restarts=(?P<restarts>\d+)\s+"
r"propagations=(?P<propagations>\d+)\s+z3_time_ms=(?P<z3_time_ms>\d+)\s+"
r"wall_ms=(?P<wall_ms>\d+)"
)
EMITTED_RE = re.compile(r"emitted\s+(?P<count>\d+)\s+findings total")
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 load_catalog() -> dict:
return json.loads(CATALOG_PATH.read_text())
CATALOG = load_catalog()
def resolve_cwes(
detectors: list[str] | None,
cwe_list: list[str] | None,
all_mapped: bool,
) -> list[str]:
"""Resolve which CWEs to run."""
if cwe_list:
return sorted(set(cwe_list), key=int)
cwe_entries = CATALOG["cwe_entries"]
if all_mapped:
return sorted(
cwe for cwe in CWE_DETECTOR_MAP
)
if detectors:
target_detectors = set(detectors)
if "oob" in target_detectors:
target_detectors.add("oob-access")
return sorted(
cwe
for cwe, cfg in CWE_DETECTOR_MAP.items()
if any(d in target_detectors for d in cfg["detectors"])
)
return sorted(CWE_DETECTOR_MAP.keys())
def load_gold(cwe: str) -> dict | None:
gold_path = GOLD_ROOT / f"juliet-{cwe}.json"
if not gold_path.exists():
return None
return json.loads(gold_path.read_text())
def score_file_level(
findings: list[dict],
gold_entries: list[dict],
kinds: list[str],
) -> dict:
"""Score at file granularity — a file is TP if it has a finding of the expected kind."""
kind_set = set(kinds)
gold_files = {
file_basename(entry["files"][0])
for entry in gold_entries
if entry.get("kind") in kind_set
}
finding_files = {
file_basename(f["file"])
for f in findings
if f.get("kind") in kind_set
}
tp_files = gold_files & finding_files
tp = len(tp_files)
fp = len(finding_files - gold_files)
fn = len(gold_files - finding_files)
precision = tp / len(finding_files) if finding_files else 0.0
recall = tp / len(gold_files) if gold_files else 0.0
f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) else 0.0
return {
"true_positive": tp,
"false_positive": fp,
"false_negative": fn,
"gold_total": len(gold_files),
"finding_groups": len(finding_files),
"precision": precision,
"recall": recall,
"f1": f1,
"matched_files": sorted(tp_files),
"missed_files": sorted(gold_files - finding_files),
}
def normalize_kind(kind: str) -> str:
return {
"dead-branch": "dead",
"duplicate-class": "class-dup",
}.get(kind, kind)
def write_compile_db(
corpus_dir: Path,
db_path: Path,
limit_files: int = 0,
) -> list[dict]:
"""Write compile_commands.json for a CWE directory."""
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)]
src_files = sorted(corpus_dir.rglob("*.c")) + sorted(corpus_dir.rglob("*.cpp"))
if limit_files:
src_files = src_files[:limit_files]
for src in src_files:
compiler = CLANG if src.suffix == ".c" else CLANGXX
std_flag = "-std=c11" if src.suffix == ".c" else "-std=c++17"
entries.append({
"directory": str(corpus_dir),
"command": [str(compiler), std_flag, *inc_flags, "-c", str(src)],
"file": str(src),
})
db_path.parent.mkdir(parents=True, exist_ok=True)
db_path.write_text(json.dumps([
{"directory": e["directory"], "command": shlex.join(e["command"]), "file": e["file"]}
for e in entries
], indent=2) + "\n")
return entries
def normalize_findings(findings: list[dict]) -> list[dict]:
dedup: dict[tuple[str, str], dict] = {}
for finding in findings:
key = (str(Path(finding["file"]).resolve()), finding["kind"])
current = dedup.get(key)
if current is None or finding.get("line", 0) < current.get("line", 0):
dedup[key] = finding
return sorted(dedup.values(), key=lambda item: (item["file"], item["line"]))
def parse_sarif(sarif_path: Path) -> list[dict]:
if not sarif_path.exists():
return []
doc = json.loads(sarif_path.read_text())
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": int(region.get("startLine", 0) or 0),
"kind": kind,
"id": fid,
})
return normalize_findings(findings)
def parse_profiles(log_text: str, expected_detectors: list[str]) -> dict:
totals = {
"queries": 0, "sat": 0, "unsat": 0, "unknown": 0,
"decisions": 0, "conflicts": 0, "restarts": 0,
"propagations": 0, "z3_time_ms": 0, "wall_ms": 0,
}
expected = set(expected_detectors)
for line in log_text.splitlines():
match = SOLVER_RE.match(line.strip())
if not match:
continue
if match.group("detector") not in expected:
continue
for key in totals:
totals[key] += int(match.group(key))
emitted = 0
for match in EMITTED_RE.finditer(log_text):
emitted = int(match.group("count"))
return {"solver": totals, "findings_emitted_total": emitted}
def run_cwe(
cwe: str,
fold_bin: Path,
processes: int,
out_root: Path,
limit_files: int = 0,
dry_run: bool = False,
) -> dict | None:
config = CWE_DETECTOR_MAP[cwe]
kinds = list(config["kinds"])
detectors = list(config["detectors"])
exclude_globs = config.get("exclude_globs", [])
corpus_dir = JULIET_ROOT / cwe
if not corpus_dir.is_dir():
print(f"CWE-{cwe}: skipped — filtered corpus not found at {corpus_dir}",
file=sys.stderr)
return None
gold = load_gold(cwe)
if gold is None:
print(f"CWE-{cwe}: skipped — gold file not found for juliet-{cwe}.json",
file=sys.stderr)
print(f" Run: python3 bench/juliet/expansion/_generate_gold.py",
file=sys.stderr)
return None
cwe_out = out_root / cwe
cwe_out.mkdir(parents=True, exist_ok=True)
db_path = cwe_out / "compile_commands.json"
sarif_path = cwe_out / "fold.sarif"
stdout_path = cwe_out / "fold.stdout.log"
stderr_path = cwe_out / "fold.stderr.log"
command = [
str(fold_bin), "scan",
"--compile-db", str(db_path),
"--format", "sarif",
"--output", str(sarif_path),
"--min-body-stmts", "1",
"--processes", str(processes),
"--profile-solver",
]
for detector in detectors:
command.extend(["--detectors", detector])
for pattern in exclude_globs:
command.extend(["--exclude-glob", pattern])
if dry_run:
entries = write_compile_db(corpus_dir, db_path, limit_files)
return {
"cwe": cwe,
"kind": ",".join(kinds),
"detectors": detectors,
"entries": len(entries),
"command": shlex.join(command),
"dry_run": True,
}
entries = write_compile_db(corpus_dir, db_path, limit_files)
env = os.environ.copy()
env["PATH"] = f"{LLVM_BIN}:{env.get('PATH', '')}"
print(f" [{cwe}] {len(entries)} TUs, detectors={','.join(detectors)}", file=sys.stderr)
start = time.monotonic()
proc = subprocess.run(command, capture_output=True, text=True, env=env)
elapsed = time.monotonic() - start
stdout_path.write_text(proc.stdout)
stderr_path.write_text(proc.stderr)
combined_log = proc.stdout + "\n" + proc.stderr
if "subprocess_fanout: bad NDJSON line:" in combined_log:
raise RuntimeError(
f"parallel worker serialization dropped findings for CWE-{cwe}; "
f"rerun with --processes 1"
)
findings = parse_sarif(sarif_path)
score = score_file_level(findings, gold["expected"], kinds)
profile = parse_profiles(combined_log, detectors + kinds)
version_info = subprocess.run(
[str(fold_bin), "--version"], capture_output=True, text=True
).stdout.strip()
payload = {
"tool": "fold",
"cwe": cwe,
"kind": ",".join(kinds),
"kinds": kinds,
"detectors": detectors,
"version": version_info or "unknown",
"elapsed_s": elapsed,
"exit_code": proc.returncode,
"entries": len(entries),
"excluded_globs": exclude_globs,
"findings": findings,
"findings_source": "sarif",
"score": score,
"profile": profile,
"artifacts": {
"compile_db": str(db_path),
"sarif": str(sarif_path),
"stdout_log": str(stdout_path),
"stderr_log": str(stderr_path),
},
}
(cwe_out / "fold.json").write_text(
json.dumps(payload, indent=2, sort_keys=True) + "\n"
)
return payload
# ── Aggregate scoring ──
def aggregate_scores(results: list[dict]) -> dict[str, dict]:
"""Compute per-detector aggregate precision/recall across all CWEs."""
detector_stats: dict[str, dict] = {}
for result in results:
if result.get("dry_run"):
continue
score = result["score"]
for detector in result["detectors"]:
if detector not in detector_stats:
detector_stats[detector] = {
"tp": 0, "fp": 0, "fn": 0, "gold_total": 0,
"finding_groups": 0, "cwe_count": 0, "elapsed_s": 0.0,
}
stats = detector_stats[detector]
stats["tp"] += score["true_positive"]
stats["fp"] += score["false_positive"]
stats["fn"] += score["false_negative"]
stats["gold_total"] += score["gold_total"]
stats["finding_groups"] += score.get("finding_groups", 0)
stats["cwe_count"] += 1
stats["elapsed_s"] += result.get("elapsed_s", 0)
output = {}
for detector, stats in sorted(detector_stats.items()):
tp, fp, fn = stats["tp"], stats["fp"], stats["fn"]
precision = tp / (tp + fp) if (tp + fp) else 0.0
recall = tp / (tp + fn) if (tp + fn) else 0.0
f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) else 0.0
output[detector] = {
**stats,
"precision": precision,
"recall": recall,
"f1": f1,
}
return output
def render_summary(results: list[dict], per_detector: dict[str, dict]) -> str:
lines = [
"# fold on Juliet — Expanded Coverage",
"",
f"Run: {now_iso()}",
f"CWEs evaluated: {len([r for r in results if not r.get('dry_run')])}",
"",
"## Per-CWE Results",
"",
"| CWE | Detector | TP | FP | FN | Findings | Gold | Precision | Recall | F1 | Time |",
"|-----|----------|----|----|----|----------|------|-----------|--------|----|------|",
]
for result in results:
if result.get("dry_run"):
continue
score = result["score"]
lines.append(
f"| CWE-{result['cwe']} | {result['kind']} | {score['true_positive']} | "
f"{score['false_positive']} | {score['false_negative']} | "
f"{score['finding_groups']} | {score['gold_total']} | "
f"{score['precision']:.1%} | {score['recall']:.1%} | {score['f1']:.1%} | "
f"{result['elapsed_s']:.1f}s |"
)
lines.extend([
"",
"## Per-Detector Aggregate",
"",
"| Detector | CWEs | TP | FP | FN | Gold | Precision | Recall | F1 | Total Time |",
"|----------|------|----|----|----|------|-----------|--------|----|------------|",
])
for detector, stats in sorted(per_detector.items()):
lines.append(
f"| {detector} | {stats['cwe_count']} | {stats['tp']} | {stats['fp']} | "
f"{stats['fn']} | {stats['gold_total']} | "
f"{stats['precision']:.1%} | {stats['recall']:.1%} | {stats['f1']:.1%} | "
f"{stats['elapsed_s']:.1f}s |"
)
lines.append("")
return "\n".join(lines)
def main() -> int:
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
cwe_group = parser.add_mutually_exclusive_group()
cwe_group.add_argument(
"--detectors",
help="Comma-separated detector names (default: all 6 target detectors)",
)
cwe_group.add_argument(
"--cwes",
help="Comma-separated CWE numbers (e.g., 476,690,369)",
)
parser.add_argument(
"--all-mapped",
action="store_true",
help="Run all CWEs that have entries in CWE_DETECTOR_MAP (including non-target detectors)",
)
parser.add_argument("--fold-bin", default=str(DEFAULT_FOLD))
parser.add_argument("--processes", type=int, default=1)
parser.add_argument("--out-root", default=str(RESULTS_ROOT))
parser.add_argument("--limit-files", type=int, default=0,
help="Limit files per CWE (for smoke tests)")
parser.add_argument("--smoke", action="store_true",
help="Smoke test: run 1 CWE with --limit-files 10")
parser.add_argument("--dry-run", action="store_true",
help="Print what would run without executing")
args = parser.parse_args()
fold_bin = Path(args.fold_bin).resolve()
if not fold_bin.exists() and not args.dry_run:
print(f"fold binary not found: {fold_bin}", file=sys.stderr)
return 2
if not CLANG.exists() or not CLANGXX.exists():
print(f"clang/clang++ not found under {LLVM_BIN}", file=sys.stderr)
return 2
detector_list = (
[d.strip() for d in args.detectors.split(",") if d.strip()]
if args.detectors
else None
)
cwe_list = (
[c.strip() for c in args.cwes.split(",") if c.strip()]
if args.cwes
else None
)
cwes = resolve_cwes(detector_list, cwe_list, args.all_mapped)
if not cwes:
print("No CWEs resolved. Check detector names or use --all-mapped.", file=sys.stderr)
return 1
limit_files = args.limit_files
if args.smoke:
cwes = cwes[:1]
limit_files = 10
print(f"Smoke mode: running CWE-{cwes[0]} with {limit_files} files", file=sys.stderr)
stamp = now_iso()
out_root = Path(args.out_root).resolve() / f"{stamp}_fold-juliet-expanded"
out_root.mkdir(parents=True, exist_ok=True)
print(f"Running fold on {len(cwes)} Juliet CWEs...", file=sys.stderr)
print(f"Output directory: {out_root}", file=sys.stderr)
results = []
for cwe in cwes:
print(f"CWE-{cwe}...", file=sys.stderr)
result = run_cwe(
cwe, fold_bin, args.processes, out_root,
limit_files=limit_files, dry_run=args.dry_run,
)
if result:
results.append(result)
if args.dry_run:
for r in results:
print(f"\nCWE-{r['cwe']}: {r['entries']} files, detectors={r['detectors']}")
print(f" {r['command']}")
print(f"\n{len(results)} CWEs would be executed.")
return 0
per_detector = aggregate_scores(results)
summary = render_summary(results, per_detector)
summary_path = out_root / "summary.md"
summary_path.write_text(summary + "\n")
print(summary)
# Write per-detector JSON summary
det_summary_path = out_root / "detector_summary.json"
det_summary_path.write_text(
json.dumps(per_detector, indent=2, sort_keys=True) + "\n"
)
# Write manifest
manifest = {
"tool": "fold",
"timestamp": stamp,
"cwes_run": [r["cwe"] for r in results],
"cwes_total": len(cwes),
"out_root": str(out_root),
"per_detector": {
k: {"precision": v["precision"], "recall": v["recall"], "f1": v["f1"]}
for k, v in per_detector.items()
},
}
(out_root / "manifest.json").write_text(
json.dumps(manifest, indent=2) + "\n"
)
print(f"\nResults written to {out_root}")
print(f" summary.md — human-readable table")
print(f" detector_summary.json — per-detector aggregate scores")
print(f" manifest.json — run metadata")
return 0
if __name__ == "__main__":
raise SystemExit(main())