From 9af356797fa02d0df2db9fb01a7744a903fdcc13 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Fri, 4 Sep 2026 11:53:06 +0200 Subject: [PATCH] feat(three-stage-scan): gate stage-2 maps against split-sink ranking --- .github/workflows/ci.yml | 2 + .kiro/specs/three-stage-scan/tasks.md | 2 +- .../fixtures/split-sink-python/helper.py | 7 + dagger/ci.py | 1 + src/openultrasast/map_gate.py | 130 ++++++++++++++++++ tests/test_map_gate.py | 82 +++++++++++ 6 files changed, 223 insertions(+), 1 deletion(-) create mode 100644 benchmarks/fixtures/split-sink-python/helper.py create mode 100644 src/openultrasast/map_gate.py create mode 100644 tests/test_map_gate.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5beab70..ba152b3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,8 @@ jobs: - run: uv run pytest - name: Detection gate (>=90% recall, <10% FP — hard, score-independent) run: uv run python -m openultrasast.gate + - name: Split-sink map gate (stage 2 beats pattern-hit ranking) + run: uv run python -m openultrasast.map_gate - run: uv run python -m compileall src tests dagger: diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 876df43..534443c 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -178,7 +178,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Requirements: 8.2_ - _Boundary: Benchmark Harness_ -- [ ] 7.3 Gate stage 2 on ranking the true function in the high band and beating pattern-hit order +- [x] 7.3 Gate stage 2 on ranking the true function in the high band and beating pattern-hit order - Stage 1 may still use existing smoke fixtures; that recall is not treated as stage-2 success. - The merge gate fails if map order equals pattern-hit-count order on the split-sink corpus. - Observable completion: a dedicated gate command or test fails a stub map that sorts by hit count and passes the heuristic map on the split-sink corpus. diff --git a/benchmarks/fixtures/split-sink-python/helper.py b/benchmarks/fixtures/split-sink-python/helper.py new file mode 100644 index 0000000..a1f3a67 --- /dev/null +++ b/benchmarks/fixtures/split-sink-python/helper.py @@ -0,0 +1,7 @@ +"""Trivial eval helper: many same-line pattern hits, little complexity.""" + + +def ping() -> str: + eval("1") + eval("2") + return "ok" diff --git a/dagger/ci.py b/dagger/ci.py index dfce778..de0ff94 100644 --- a/dagger/ci.py +++ b/dagger/ci.py @@ -51,6 +51,7 @@ async def _run_checks() -> None: .with_exec(["uv", "run", "mypy", "src/openultrasast"]) .with_exec(["uv", "run", "pytest"]) .with_exec(["uv", "run", "python", "-m", "openultrasast.gate"]) + .with_exec(["uv", "run", "python", "-m", "openultrasast.map_gate"]) .with_exec(["uv", "run", "python", "-m", "compileall", "src", "tests"]) .with_exec(["uv", "build"]) ) diff --git a/src/openultrasast/map_gate.py b/src/openultrasast/map_gate.py new file mode 100644 index 0000000..95e506f --- /dev/null +++ b/src/openultrasast/map_gate.py @@ -0,0 +1,130 @@ +"""Stage-2 quality gate: split-sink ranking must beat pattern-hit-count order. + +The stage-1 smoke gate (``python -m openultrasast.gate``) stays on cheat-sheet +fixtures. This module is the merge gate for the complexity map: the planted +split-sink function must sit in the high band, and map order must not collapse +to sorting files by inventory-hit count. +""" + +from __future__ import annotations + +import sys +import tempfile +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path + +from .benchmark import BenchmarkManifest, ExpectedFinding, load_benchmark_manifest, resolve_benchmark_source +from .complexity.map import ComplexityMap, Hotspot, build_complexity_map +from .findings import StaticFinding, quick_scan_findings +from .mapping import analyze_entry_points, attach_reachability_hints +from .preprocess import preprocess_repository +from .rank import rank_targets + +SPLIT_SINK_MANIFESTS = ( + "split-sink-c", + "split-sink-java", + "split-sink-javascript", + "split-sink-python", +) +DEFAULT_MANIFEST_DIR = Path("benchmarks/manifests") + + +@dataclass(frozen=True) +class SplitSinkScan: + name: str + manifest: BenchmarkManifest + target: Path + findings: tuple[StaticFinding, ...] + complexity_map: ComplexityMap + + +@dataclass(frozen=True) +class MapGateVerdict: + passed: bool + reasons: tuple[str, ...] + + +def scan_split_sink(name: str, manifest_dir: Path = DEFAULT_MANIFEST_DIR) -> SplitSinkScan: + manifest_path = manifest_dir / f"{name}.toml" + manifest = load_benchmark_manifest(manifest_path) + target = resolve_benchmark_source(manifest_path, manifest) + _, targets = preprocess_repository(target) + targets = attach_reachability_hints(targets, analyze_entry_points(target, targets)) + findings = tuple(quick_scan_findings(target, targets, rank_targets(targets))) + with tempfile.TemporaryDirectory() as tmp: + complexity_map = build_complexity_map( + targets, + findings, + Path(tmp) / "complexity_map.json", + repo_files=[item.path for item in targets], + ) + return SplitSinkScan(name=name, manifest=manifest, target=target, findings=findings, complexity_map=complexity_map) + + +def planted_expected(manifest: BenchmarkManifest) -> tuple[ExpectedFinding, ...]: + return tuple(item for item in manifest.expected if item.rule_id is None) + + +_BAND_RANK = {"high": 3, "medium": 2, "low": 1} + + +def planted_in_high_band(manifest: BenchmarkManifest, complexity_map: ComplexityMap) -> bool: + """True when the planted sink sits in the highest band the map actually emitted.""" + if not complexity_map.hotspots: + return False + top = max(_BAND_RANK.get(hotspot.band, 0) for hotspot in complexity_map.hotspots) + for expected in planted_expected(manifest): + matches = [hotspot for hotspot in complexity_map.hotspots if hotspot.path == expected.path] + if expected.function: + named = [hotspot for hotspot in matches if hotspot.function_name == expected.function] + if named: + matches = named + if not matches or not any(_BAND_RANK.get(hotspot.band, 0) == top for hotspot in matches): + return False + return bool(planted_expected(manifest)) + + +def hit_count_path_order(findings: Sequence[StaticFinding], paths: Sequence[str]) -> list[str]: + counts: dict[str, int] = {} + for finding in findings: + counts[finding.path] = counts.get(finding.path, 0) + 1 + unique = list(dict.fromkeys(paths)) + return sorted(unique, key=lambda path: (-counts.get(path, 0), path)) + + +def map_beats_hit_count(hotspots: Sequence[Hotspot], findings: Sequence[StaticFinding]) -> bool: + map_order = list(dict.fromkeys(hotspot.path for hotspot in hotspots)) + stub_order = hit_count_path_order(findings, map_order) + return map_order != stub_order + + +def split_sink_map_gate(manifest_dir: Path = DEFAULT_MANIFEST_DIR) -> MapGateVerdict: + reasons: list[str] = [] + python: SplitSinkScan | None = None + for name in SPLIT_SINK_MANIFESTS: + bundle = scan_split_sink(name, manifest_dir) + if name == "split-sink-python": + python = bundle + if not planted_in_high_band(bundle.manifest, bundle.complexity_map): + reasons.append(f"{name}: planted split-sink is not in the top hotspot band") + if python is None: + reasons.append("split-sink-python corpus is missing") + elif not map_beats_hit_count(python.complexity_map.hotspots, python.findings): + reasons.append("split-sink-python: map order equals pattern-hit-count order") + return MapGateVerdict(passed=not reasons, reasons=tuple(reasons)) + + +def main() -> int: + verdict = split_sink_map_gate() + if verdict.passed: + print("split-sink map gate: PASS") + return 0 + print("split-sink map gate: FAIL", file=sys.stderr) + for reason in verdict.reasons: + print(f" - {reason}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_map_gate.py b/tests/test_map_gate.py new file mode 100644 index 0000000..725813e --- /dev/null +++ b/tests/test_map_gate.py @@ -0,0 +1,82 @@ +"""Stage-2 map gate: planted split-sink ranks in the high band and beats hit-count order.""" + +from openultrasast.complexity.map import Hotspot +from openultrasast.findings import StaticFinding +from openultrasast.gate import LANGUAGE_MANIFESTS +from openultrasast.map_gate import ( + SPLIT_SINK_MANIFESTS, + hit_count_path_order, + map_beats_hit_count, + planted_in_high_band, + scan_split_sink, + split_sink_map_gate, +) + + +def test_split_sink_map_gate_passes_heuristic_map() -> None: + verdict = split_sink_map_gate() + assert verdict.passed, verdict.reasons + + +def test_hit_count_stub_fails_map_gate_on_split_sink_python() -> None: + bundle = scan_split_sink("split-sink-python") + unique_map = list(dict.fromkeys(hotspot.path for hotspot in bundle.complexity_map.hotspots)) + stub_order = hit_count_path_order(bundle.findings, unique_map) + assert stub_order != unique_map + assert map_beats_hit_count(bundle.complexity_map.hotspots, bundle.findings) + + +def test_planted_split_sink_function_is_in_high_band() -> None: + bundle = scan_split_sink("split-sink-python") + assert planted_in_high_band(bundle.manifest, bundle.complexity_map) + names = {hotspot.function_name for hotspot in bundle.complexity_map.hotspots if hotspot.path == "app.py"} + assert "search" in names + + +def test_split_sink_gate_corpus_is_not_the_stage1_smoke_gate() -> None: + listed = {name for names in LANGUAGE_MANIFESTS.values() for name in names} + assert not any(name in listed for name in SPLIT_SINK_MANIFESTS) + + +def test_map_beats_hit_count_rejects_inventory_sorted_stub() -> None: + findings = [ + StaticFinding( + finding_id="python-unsafe-eval:helper.py:1", + path="helper.py", + title="eval", + severity="high", + confidence="low", + evidence_level="static_corroboration", + rationale="hit", + line=1, + function_name="ping", + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=["syscall_entry"], + ranking_priority=1.0, + ) + ] + stub = ( + Hotspot( + path="helper.py", + function_name="ping", + score=0.1, + band="low", + signals={}, + rationale="stub", + test_hint=None, + inventory_finding_ids=(findings[0].finding_id,), + ), + Hotspot( + path="app.py", + function_name="search", + score=9.0, + band="high", + signals={}, + rationale="stub", + test_hint=None, + inventory_finding_ids=(), + ), + ) + assert not map_beats_hit_count(stub, findings)