#!/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\S+)\s+" r"queries=(?P\d+)\s+sat=(?P\d+)\s+unsat=(?P\d+)\s+" r"unknown=(?P\d+)\s+decisions=(?P\d+)\s+" r"conflicts=(?P\d+)\s+restarts=(?P\d+)\s+" r"propagations=(?P\d+)\s+z3_time_ms=(?P\d+)\s+" r"wall_ms=(?P\d+)" ) EMITTED_RE = re.compile(r"emitted\s+(?P\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())