2dcbdbc75d
Preserve source, binary, and runtime evidence boundaries while making precision and performance regressions reproducible. - decompose analysis orchestration into bounded modules - add differential, metamorphic, organic, and Yams change-time gates - fix detector, P-Code, fuzz sanitizer, and report identity precision
307 lines
11 KiB
Python
Executable file
307 lines
11 KiB
Python
Executable file
#!/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())
|