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
213 lines
7.7 KiB
Python
Executable file
213 lines
7.7 KiB
Python
Executable file
#!/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())
|