#!/usr/bin/env python3 """Compare Fold source classifications with Clang Static Analyzer pairs.""" from __future__ import annotations import argparse import importlib.util import json import re import subprocess from pathlib import Path from typing import Any, Mapping from fold_eval.metrics import aggregate_classification_metrics from fold_eval.schema import ( artifact_record, build_run_artifact, sha256_file, utc_now, validate_benchmark_case, validate_run_artifact, write_json, ) from fold_eval.source_lane import run_fold_source from fold_eval.toolchain import find_clang REPO_ROOT = Path(__file__).resolve().parents[2] RUNNER_COMMON_PATH = REPO_ROOT / "scripts" / "benchmarks" / "runner_common.py" DEFAULT_SPEC = REPO_ROOT / "tests/benchmarks/differential/clang_sa_pairs.json" _spec = importlib.util.spec_from_file_location("runner_common", RUNNER_COMMON_PATH) if _spec is None or _spec.loader is None: raise ImportError(f"failed to load {RUNNER_COMMON_PATH}") common = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(common) def resolve_source(pair: Mapping[str, Any], field: str, spec_path: Path) -> Path: source = common.resolve_path(str(pair.get(field, "")), spec_path.parent) if not source.is_file(): raise FileNotFoundError(source) return source def analyzer_command(compiler: str, source: Path) -> list[str]: return [ compiler, "--analyze", "-std=c++20", "-Xanalyzer", "-analyzer-output=text", "-Xanalyzer", "-analyzer-checker=core", str(source), ] def probe_analyzer(compiler: str, output_dir: Path) -> str | None: source = output_dir / "clang-sa-probe.cpp" source.write_text("int main() { return 0; }\n", encoding="utf-8") proc = subprocess.run( analyzer_command(compiler, source), cwd=REPO_ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) return None if proc.returncode == 0 else proc.stdout[-4000:] def run_analyzer( *, compiler: str, source: Path, case_id: str, label: str, output_dir: Path, ) -> tuple[dict[str, Any], list[str]]: lane_dir = output_dir / label / "clang-sa" lane_dir.mkdir(parents=True, exist_ok=True) log_path = lane_dir / "analyze.log" run_path = lane_dir / "run.json" command = analyzer_command(compiler, source) started_at = utc_now() proc = common.run_measured(command, REPO_ROOT, timeout_seconds=60.0) ended_at = utc_now() log_path.write_text(proc.output, encoding="utf-8") failures = [] if proc.returncode == 0 else [f"Clang SA exited {proc.returncode}"] run_artifact = build_run_artifact( experiment_id=f"differential:{case_id}:{label}:clang-sa", case_id=case_id, tool={ "name": "clang-static-analyzer", "path": compiler, "sha256": sha256_file(Path(compiler)), }, target={"source": str(source), "input_fingerprint": sha256_file(source)}, command=command, started_at=started_at, ended_at=ended_at, exit_status=proc.returncode, budgets={"wall_seconds": 60.0, "checker": "core"}, artifacts=[artifact_record("analyzer-log", log_path, "text/plain")], measurements=proc.metrics(), observations={"diagnostic_output": proc.output}, ) failures.extend(f"run artifact: {item}" for item in validate_run_artifact(run_artifact)) write_json(run_path, run_artifact) return { "returncode": proc.returncode, "output": proc.output, "measurements": proc.metrics(), "run_artifact": str(run_path), "log": str(log_path), }, failures def evaluate_pair( *, pair: Mapping[str, Any], spec: Mapping[str, Any], spec_path: Path, fold: Path, compiler: str, output_dir: Path, ) -> tuple[dict[str, Any], list[str], dict[str, dict[str, int]]]: name = str(pair["name"]) detector = str(pair["detector"]) rule_id = str(pair["rule_id"]) vulnerable = resolve_source(pair, "vulnerable_source", spec_path) fixed = resolve_source(pair, "fixed_source", spec_path) pattern = re.compile(str(pair["analyzer_regex"]), re.IGNORECASE) failures: list[str] = [] fold_vulnerable, errors = run_fold_source( fold=fold, source=vulnerable, detector=detector, case_id=str(spec["case_id"]), label=f"{name}-vulnerable", output_dir=output_dir, repo_root=REPO_ROOT, ) failures.extend(errors) fold_fixed, errors = run_fold_source( fold=fold, source=fixed, detector=detector, case_id=str(spec["case_id"]), label=f"{name}-fixed", output_dir=output_dir, repo_root=REPO_ROOT, ) failures.extend(errors) analyzer_vulnerable, errors = run_analyzer( compiler=compiler, source=vulnerable, case_id=str(spec["case_id"]), label=f"{name}-vulnerable", output_dir=output_dir, ) failures.extend(errors) analyzer_fixed, errors = run_analyzer( compiler=compiler, source=fixed, case_id=str(spec["case_id"]), label=f"{name}-fixed", output_dir=output_dir, ) failures.extend(errors) fold_class = { "vulnerable": any(item["rule_id"] == rule_id for item in fold_vulnerable["findings"]), "fixed": any(item["rule_id"] == rule_id for item in fold_fixed["findings"]), } analyzer_class = { "vulnerable": pattern.search(analyzer_vulnerable["output"]) is not None, "fixed": pattern.search(analyzer_fixed["output"]) is not None, } for tool, classification in ( ("Fold", fold_class), ("Clang SA", analyzer_class), ): if not classification["vulnerable"]: failures.append(f"{tool} missed the vulnerable revision") if classification["fixed"]: failures.append(f"{tool} reported the fixed revision") if fold_class != analyzer_class: failures.append("Fold and Clang SA classifications disagree") slices = { f"fold/{pair['hazard']}": { "true_positive": int(fold_class["vulnerable"]), "false_positive": int(fold_class["fixed"]), "false_negative": int(not fold_class["vulnerable"]), "true_negative": int(not fold_class["fixed"]), }, f"clang-sa/{pair['hazard']}": { "true_positive": int(analyzer_class["vulnerable"]), "false_positive": int(analyzer_class["fixed"]), "false_negative": int(not analyzer_class["vulnerable"]), "true_negative": int(not analyzer_class["fixed"]), }, } return { "name": name, "hazard": pair["hazard"], "status": "failed" if failures else "ok", "classifications": {"fold": fold_class, "clang_sa": analyzer_class}, "fold": {"vulnerable": fold_vulnerable, "fixed": fold_fixed}, "clang_sa": { "vulnerable": {key: value for key, value in analyzer_vulnerable.items() if key != "output"}, "fixed": {key: value for key, value in analyzer_fixed.items() if key != "output"}, }, "failures": failures, }, failures, slices def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--build", type=Path, default=Path("build/release")) parser.add_argument("--fold", type=Path) parser.add_argument("--compiler") parser.add_argument("--spec", type=Path, default=DEFAULT_SPEC) parser.add_argument("--output-dir", type=Path, default=Path("/tmp/fold-clang-sa-differential")) args = parser.parse_args() build = args.build if args.build.is_absolute() else REPO_ROOT / args.build fold = common.find_fold_binary(REPO_ROOT, build, args.fold) if fold is None: common.print_missing_fold(args.fold or build / "src/cli/fold") return 2 try: spec_path = args.spec.resolve() spec = common.load_json(spec_path) validation = validate_benchmark_case(spec, allow_legacy=False) if not validation["valid"]: raise ValueError("; ".join(validation["errors"])) pairs = spec.get("pairs", []) if not isinstance(pairs, list) or not pairs: raise ValueError("pairs must be a non-empty array") except (OSError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"status": "failed", "reason": str(exc)}, indent=2)) return 2 args.output_dir.mkdir(parents=True, exist_ok=True) compiler = find_clang(args.compiler) if compiler is None: summary = {"status": "skipped", "reason": "no clang++ compiler found"} write_json(args.output_dir / "summary.json", summary) print(json.dumps(summary, indent=2)) return 77 probe_error = probe_analyzer(compiler, args.output_dir) if probe_error is not None: summary = { "status": "skipped", "reason": "compiler does not provide Clang Static Analyzer", "compiler": compiler, "output": probe_error, } write_json(args.output_dir / "summary.json", summary) print(json.dumps(summary, indent=2)) return 77 version = subprocess.run( [compiler, "--version"], text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT ).stdout.splitlines()[0] results: list[dict[str, Any]] = [] failures: list[str] = [] slices: dict[str, dict[str, int]] = {} for pair in pairs: if not isinstance(pair, dict): failures.append("pair is not an object") continue try: result, pair_failures, pair_slices = evaluate_pair( pair=pair, spec=spec, spec_path=spec_path, fold=fold, compiler=compiler, output_dir=args.output_dir, ) results.append(result) failures.extend(f"{pair.get('name')}: {item}" for item in pair_failures) slices.update(pair_slices) except (OSError, ValueError, TypeError, KeyError) as exc: failures.append(f"{pair.get('name', 'pair')}: {exc}") summary = { "schema_version": 1, "suite": "clang_sa_differential", "case_id": spec["case_id"], "compiler": compiler, "compiler_version": version, "status": "failed" if failures else "ok", "results": results, "classification_metrics": aggregate_classification_metrics(slices), "failures": failures, } write_json(args.output_dir / "summary.json", summary) print(json.dumps(summary, indent=2, sort_keys=True)) return 1 if failures else 0 if __name__ == "__main__": raise SystemExit(main())