mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 19:59:26 +00:00
feat(gate): Phase 4 task 8 — lock the detection gate into CI
Make the >=90% recall / <10% FP detection gate a first-class, score-independent CI step with non-regression guards. - openultrasast.gate: canonical corpus map + DetectionMetrics/DetectionGateVerdict, evaluate_corpus/detection_gate (no score input — cannot be unblocked by score), merge_gate (ANDs the hard detection gate with the advisory-first score gate), within_tolerance, and a `python -m openultrasast.gate` entrypoint that exits non-zero on a breach. Wired as an explicit step in both the GitHub and Dagger pipelines. - 8.1: pipeline tests prove a high/passing score cannot unblock a recall-breaching change; the gate fails below 90% recall or at/above 10% FP. - 8.2: tests/golden/detection_baseline.json + a byte-identical baseline test for the default ruleset / empty ledger (44/47 = 93.62% recall, 0% FP). - 8.3: within_tolerance band logic + a capability-gated HarnessX tolerance test. - 8.4: zero-dependency guard — governance/scoring/benchmark planes never import HarnessX (sys.modules check + module-scope import scan). - test_detection_benchmarks now sources the corpus/aggregation from the gate module. 128 passed, 3 skipped; ruff + mypy clean; detection gate PASS. Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
6e87a58465
commit
649bd42fc3
@@ -0,0 +1,205 @@
|
||||
"""Detection-quality gate — the hard feasibility constraint (Phase 4 task 8).
|
||||
|
||||
Aggregate recall and false-positive rate over the benchmark corpus must clear the
|
||||
project goal: recall >= 90% AND false-positive rate < 10%. This gate is computed
|
||||
**independently of the project score** so a score improvement can never unblock a
|
||||
gate-breaching change — the recall/FP constraint is never folded into the score
|
||||
reward. ``merge_gate`` composes this hard constraint with the (advisory-first)
|
||||
score gate by AND, so the score can only *add* a block, never remove one.
|
||||
|
||||
Run as a module to evaluate the corpus and exit non-zero on a breach::
|
||||
|
||||
python -m openultrasast.gate
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from .benchmark import BenchmarkRun, evaluate_benchmark, load_benchmark_manifest, resolve_benchmark_source
|
||||
from .findings import quick_scan_findings
|
||||
from .mapping import analyze_entry_points, attach_reachability_hints
|
||||
from .preprocess import preprocess_repository
|
||||
from .rank import rank_targets
|
||||
|
||||
# The in-scope corpus: language -> benchmark manifest names (smoke + real repo).
|
||||
# Single source of truth shared by the CI gate and the detection-benchmark test.
|
||||
LANGUAGE_MANIFESTS: dict[str, list[str]] = {
|
||||
"c_cpp": ["c-cpp-smoke", "cpp-damn-vulnerable"],
|
||||
"python": ["python-web-smoke", "python-vulnerable"],
|
||||
"javascript": ["javascript-node-web-smoke", "javascript-vulnerable"],
|
||||
"java": ["java-web-smoke", "java-spring-boot-vulnerable"],
|
||||
}
|
||||
|
||||
RECALL_FLOOR = 0.90
|
||||
FALSE_POSITIVE_CEILING = 0.10
|
||||
DEFAULT_MANIFEST_DIR = Path("benchmarks/manifests")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DetectionMetrics:
|
||||
"""Aggregate detection counts and the two rates derived from them."""
|
||||
|
||||
expected: int
|
||||
matched: int
|
||||
actual: int
|
||||
false_positives: int
|
||||
|
||||
@property
|
||||
def recall(self) -> float:
|
||||
return self.matched / self.expected if self.expected else 1.0
|
||||
|
||||
@property
|
||||
def false_positive_rate(self) -> float:
|
||||
return self.false_positives / self.actual if self.actual else 0.0
|
||||
|
||||
def __add__(self, other: DetectionMetrics) -> DetectionMetrics:
|
||||
return DetectionMetrics(
|
||||
expected=self.expected + other.expected,
|
||||
matched=self.matched + other.matched,
|
||||
actual=self.actual + other.actual,
|
||||
false_positives=self.false_positives + other.false_positives,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DetectionGateVerdict:
|
||||
"""Score-independent verdict: pass only if recall and FP both clear the goal."""
|
||||
|
||||
passed: bool
|
||||
overall: DetectionMetrics
|
||||
per_language: dict[str, DetectionMetrics]
|
||||
recall_floor: float
|
||||
fp_ceiling: float
|
||||
reasons: list[str]
|
||||
|
||||
|
||||
def _quick_scan(target: Path): # type: ignore[no-untyped-def]
|
||||
_, targets = preprocess_repository(target)
|
||||
targets = attach_reachability_hints(targets, analyze_entry_points(target, targets))
|
||||
rankings = rank_targets(targets)
|
||||
return quick_scan_findings(target, targets, rankings)
|
||||
|
||||
|
||||
def language_metrics(manifest_names: list[str], manifest_dir: Path = DEFAULT_MANIFEST_DIR) -> DetectionMetrics:
|
||||
"""Aggregate detection metrics across one language's manifests."""
|
||||
total = DetectionMetrics(0, 0, 0, 0)
|
||||
for name in manifest_names:
|
||||
manifest_path = manifest_dir / f"{name}.toml"
|
||||
manifest = load_benchmark_manifest(manifest_path)
|
||||
target = resolve_benchmark_source(manifest_path, manifest)
|
||||
findings = _quick_scan(target)
|
||||
run = BenchmarkRun(benchmark_run_id="gate", root=Path("/tmp/gate"), manifest=manifest)
|
||||
result = evaluate_benchmark(run=run, mode="quick", findings=findings, scan_id=None, scan_run_dir=None)
|
||||
m = result.metrics
|
||||
total = total + DetectionMetrics(
|
||||
expected=m.expected_findings_total,
|
||||
matched=m.matched_findings_total,
|
||||
actual=m.actual_findings_total,
|
||||
false_positives=m.false_positive_findings_total,
|
||||
)
|
||||
return total
|
||||
|
||||
|
||||
def evaluate_corpus(
|
||||
corpus: Mapping[str, list[str]] = LANGUAGE_MANIFESTS, manifest_dir: Path = DEFAULT_MANIFEST_DIR
|
||||
) -> dict[str, DetectionMetrics]:
|
||||
"""Compute per-language detection metrics across the whole corpus."""
|
||||
return {language: language_metrics(names, manifest_dir) for language, names in sorted(corpus.items())}
|
||||
|
||||
|
||||
def detection_gate(
|
||||
per_language: Mapping[str, DetectionMetrics],
|
||||
*,
|
||||
recall_floor: float = RECALL_FLOOR,
|
||||
fp_ceiling: float = FALSE_POSITIVE_CEILING,
|
||||
) -> DetectionGateVerdict:
|
||||
"""Evaluate the hard recall/FP gate. No score input — it cannot be unblocked by score."""
|
||||
overall = DetectionMetrics(0, 0, 0, 0)
|
||||
for metrics in per_language.values():
|
||||
overall = overall + metrics
|
||||
reasons: list[str] = []
|
||||
for language, metrics in sorted(per_language.items()):
|
||||
if metrics.recall < recall_floor:
|
||||
reasons.append(f"{language} recall {metrics.recall:.2%} below {recall_floor:.0%} ({metrics.matched}/{metrics.expected})")
|
||||
if metrics.false_positive_rate >= fp_ceiling:
|
||||
reasons.append(
|
||||
f"{language} FP rate {metrics.false_positive_rate:.2%} at or above {fp_ceiling:.0%} "
|
||||
f"({metrics.false_positives}/{metrics.actual})"
|
||||
)
|
||||
if overall.recall < recall_floor:
|
||||
reasons.append(f"overall recall {overall.recall:.2%} below {recall_floor:.0%} ({overall.matched}/{overall.expected})")
|
||||
if overall.false_positive_rate >= fp_ceiling:
|
||||
reasons.append(
|
||||
f"overall FP rate {overall.false_positive_rate:.2%} at or above {fp_ceiling:.0%} ({overall.false_positives}/{overall.actual})"
|
||||
)
|
||||
return DetectionGateVerdict(
|
||||
passed=not reasons,
|
||||
overall=overall,
|
||||
per_language=dict(sorted(per_language.items())),
|
||||
recall_floor=recall_floor,
|
||||
fp_ceiling=fp_ceiling,
|
||||
reasons=reasons,
|
||||
)
|
||||
|
||||
|
||||
def merge_gate(detection: DetectionGateVerdict, score_gate: Mapping[str, object] | None = None) -> dict[str, object]:
|
||||
"""Compose the hard detection gate with the (advisory-first) score gate by AND.
|
||||
|
||||
The merge blocks if detection fails OR a *blocking* score gate fails. A passing
|
||||
score gate (however high the score) can never flip a failed detection gate to
|
||||
pass — the recall/FP constraint is hard and never folded into the score reward.
|
||||
"""
|
||||
score_passed = True if score_gate is None else bool(score_gate.get("passed", True))
|
||||
return {
|
||||
"passed": detection.passed and score_passed,
|
||||
"detection_passed": detection.passed,
|
||||
"score_passed": score_passed,
|
||||
"reasons": list(detection.reasons),
|
||||
}
|
||||
|
||||
|
||||
def within_tolerance(
|
||||
baseline: DetectionMetrics,
|
||||
candidate: DetectionMetrics,
|
||||
*,
|
||||
recall_tolerance: float = 0.05,
|
||||
fp_tolerance: float = 0.05,
|
||||
) -> bool:
|
||||
"""True if ``candidate`` recall/FP stay within tolerance of ``baseline``.
|
||||
|
||||
Used to confirm a HarnessX-backed stage does not drift detection beyond the
|
||||
configured band of the zero-dependency baseline. Recall may not fall more than
|
||||
``recall_tolerance`` below baseline; FP may not rise more than ``fp_tolerance``
|
||||
above it.
|
||||
"""
|
||||
recall_ok = candidate.recall >= baseline.recall - recall_tolerance
|
||||
fp_ok = candidate.false_positive_rate <= baseline.false_positive_rate + fp_tolerance
|
||||
return recall_ok and fp_ok
|
||||
|
||||
|
||||
def main() -> int:
|
||||
verdict = detection_gate(evaluate_corpus())
|
||||
o = verdict.overall
|
||||
print(
|
||||
f"detection gate: recall {o.recall:.2%} ({o.matched}/{o.expected}), FP {o.false_positive_rate:.2%} ({o.false_positives}/{o.actual})"
|
||||
)
|
||||
for language, m in verdict.per_language.items():
|
||||
print(
|
||||
f" {language:<11} recall {m.recall:>6.2%} ({m.matched}/{m.expected}) "
|
||||
f"FP {m.false_positive_rate:>6.2%} ({m.false_positives}/{m.actual})"
|
||||
)
|
||||
if verdict.passed:
|
||||
print("detection gate: PASS")
|
||||
return 0
|
||||
print("detection gate: FAIL", file=sys.stderr)
|
||||
for reason in verdict.reasons:
|
||||
print(f" - {reason}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user