#!/usr/bin/env python3 """Evaluate pinned adapted upstream vulnerable/fixed pairs with Fold source SAST.""" from __future__ import annotations import argparse import importlib.util import json from pathlib import Path from typing import Any, Mapping from fold_eval.metrics import aggregate_classification_metrics from fold_eval.schema import sha256_file, validate_benchmark_case, write_json from fold_eval.source_lane import run_fold_source REPO_ROOT = Path(__file__).resolve().parents[2] RUNNER_COMMON_PATH = REPO_ROOT / "scripts/benchmarks/runner_common.py" DEFAULT_SPEC = REPO_ROOT / "tests/benchmarks/organic/cve_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_and_verify( entry: Mapping[str, Any], field: str, spec_path: Path ) -> Path: source = common.resolve_path(str(entry.get(field, "")), spec_path.parent) if not source.is_file(): raise FileNotFoundError(source) expected = str(entry[f"{field.removesuffix('_source')}_sha256"]) actual = sha256_file(source) if actual != expected: raise ValueError(f"{field} hash mismatch: expected {expected}, got {actual}") return source def evaluate_entry( *, entry: Mapping[str, Any], case_id: str, spec_path: Path, fold: Path, output_dir: Path, ) -> tuple[dict[str, Any], list[str], dict[str, int]]: name = str(entry["name"]) vulnerable = resolve_and_verify(entry, "vulnerable_source", spec_path) fixed = resolve_and_verify(entry, "fixed_source", spec_path) detector = str(entry["detector"]) rule_id = str(entry["rule_id"]) failures: list[str] = [] vulnerable_run, errors = run_fold_source( fold=fold, source=vulnerable, detector=detector, case_id=case_id, label=f"{name}-vulnerable", output_dir=output_dir, repo_root=REPO_ROOT, ) failures.extend(errors) fixed_run, errors = run_fold_source( fold=fold, source=fixed, detector=detector, case_id=case_id, label=f"{name}-fixed", output_dir=output_dir, repo_root=REPO_ROOT, ) failures.extend(errors) vulnerable_detected = any( finding["rule_id"] == rule_id for finding in vulnerable_run["findings"] ) fixed_reported = any( finding["rule_id"] == rule_id for finding in fixed_run["findings"] ) expected = entry.get("expected", {}) for field, actual in ( ("vulnerable_detected", vulnerable_detected), ("fixed_reported", fixed_reported), ): if field in expected and bool(expected[field]) != actual: failures.append( f"{field} changed from reviewed expectation {bool(expected[field])} to {actual}" ) return { "name": name, "cve": entry["cve"], "project": entry["project"], "hazard": entry["hazard"], "provenance": { key: entry[key] for key in ( "repository", "reference", "vulnerable_revision", "fixed_revision", "upstream_path", "license", "adaptation", "vulnerable_sha256", "fixed_sha256", ) }, "classification": { "vulnerable_detected": vulnerable_detected, "fixed_reported": fixed_reported, }, "known_gap": entry.get("known_gap"), "fold": {"vulnerable": vulnerable_run, "fixed": fixed_run}, "failures": failures, }, failures, { "true_positive": int(vulnerable_detected), "false_positive": int(fixed_reported), "false_negative": int(not vulnerable_detected), "true_negative": int(not fixed_reported), } 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("--spec", type=Path, default=DEFAULT_SPEC) parser.add_argument("--output-dir", type=Path, default=Path("/tmp/fold-organic-pairs")) 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"])) corpus = spec.get("corpus", {}) entries = corpus.get("entries", []) if isinstance(corpus, dict) else [] if not isinstance(entries, list) or not entries: raise ValueError("corpus.entries 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) results: list[dict[str, Any]] = [] failures: list[str] = [] slices: dict[str, dict[str, int]] = {} for entry in entries: if not isinstance(entry, dict): failures.append("corpus entry is not an object") continue try: result, entry_failures, counts = evaluate_entry( entry=entry, case_id=str(spec["case_id"]), spec_path=spec_path, fold=fold, output_dir=args.output_dir, ) results.append(result) failures.extend(f"{entry.get('name')}: {item}" for item in entry_failures) slices[f"organic/{entry['cve']}"] = counts except (OSError, ValueError, TypeError, KeyError) as exc: failures.append(f"{entry.get('name', 'entry')}: {exc}") vulnerable_detected = sum( int(result["classification"]["vulnerable_detected"]) for result in results ) fixed_reported = sum( int(result["classification"]["fixed_reported"]) for result in results ) expected = spec.get("expect", {}) if vulnerable_detected != int(expected.get("vulnerable_detected", vulnerable_detected)): failures.append( f"vulnerable detections {vulnerable_detected} != expected " f"{expected.get('vulnerable_detected')}" ) if fixed_reported != int(expected.get("fixed_reported", fixed_reported)): failures.append( f"fixed reports {fixed_reported} != expected {expected.get('fixed_reported')}" ) summary = { "schema_version": 1, "suite": "organic_vulnerable_fixed", "case_id": spec["case_id"], "origin": "organic", "adaptation_policy": spec["corpus"]["integrity"], "provenance_verified": len(results) == len(entries), "status": "failed" if failures else "ok", "counts": { "pairs": len(results), "vulnerable_detected": vulnerable_detected, "fixed_reported": fixed_reported, "known_gaps": sum(bool(result.get("known_gap")) for result in results), }, "classification_metrics": aggregate_classification_metrics(slices), "results": results, "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())