diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index b4cd5bd..5d3e46f 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse import json import tempfile +from collections.abc import Sequence from dataclasses import dataclass, replace from pathlib import Path from time import perf_counter @@ -45,6 +46,7 @@ from .pair_gate import print_pair_metrics from .pairs import ( DEFAULT_CATALOG, SLICE_NAMES, + PairCase, evaluate_catalog, load_pair_catalog, make_hunter_scan, @@ -60,6 +62,7 @@ from .provenance import fingerprint from .provider.openrouter import OpenRouterEmbeddingClient, OpenRouterError from .rank import rank_targets, write_rankings from .regress import TRIGGERABLE, CandidateVerdict, run_regression, write_verdicts +from .regress.candidate import hotspot_from_finding from .reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report from .ruleset import DEFAULT_RULESET_DIR, load_ruleset from .run import ScanRun, create_scan_run @@ -159,6 +162,10 @@ def main(argv: list[str] | None = None) -> int: pairs.add_argument("--hunter", action="store_true", help="also score the LLM tool hunter on overlay slices (needs a hunter model)") pairs.add_argument("--hunter-model", default=None, help="model id for --hunter (default: [models].hunter from openultrasast.toml)") pairs.add_argument("--pointers", action="store_true", help="allow network for non-vendored pointer pairs this run (nightly; CI never)") + pairs.add_argument( + "--loo", action="store_true", help="leave-one-out: score each pair against a store seeded from the other trusted pairs" + ) + pairs.add_argument("--loo-out", type=Path, default=None, help="where to write loo.json (default: reports/loo.json)") pairs.add_argument("--json", action="store_true", help="print the pair scoreboard as JSON") subparsers.add_parser("mcp", help="run the narrow MCP server over stdio for OpenCode integration") @@ -203,6 +210,8 @@ def main(argv: list[str] | None = None) -> int: hunter=args.hunter, hunter_model=args.hunter_model, pointers=args.pointers, + loo=args.loo, + loo_out=args.loo_out, ) if args.command == "mechanisms": return _mechanisms_export(args.catalog, args.slice, args.store, json_out=args.json) @@ -823,10 +832,14 @@ def _pairs( hunter: bool = False, hunter_model: str | None = None, pointers: bool = False, + loo: bool = False, + loo_out: Path | None = None, ) -> int: if not catalog.exists() or not catalog.is_file(): raise SystemExit(f"pair catalog is not a file: {catalog}") cases = select_split(select_profile(select_slice(load_pair_catalog(catalog), slice_name), profile), split) + if loo: + return _pairs_loo(cases, loo_out or Path("reports") / "loo.json", json_out=json_out) scan = None if hunter: config_path = Path("openultrasast.toml") @@ -877,16 +890,27 @@ def _search_variants( def _hotspot_from_variant(finding: StaticFinding) -> Hotspot: - return Hotspot( - path=finding.path, - function_name=finding.function_name, - score=0.0, - band="low", - signals={}, - rationale="variant of a known mechanism (suspicion; sandbox may raise)", - test_hint=None, - inventory_finding_ids=(finding.finding_id,), - ) + return hotspot_from_finding(finding, rationale="variant of a known mechanism (suspicion; sandbox may raise)") + + +def _pairs_loo(cases: Sequence[PairCase], out: Path, *, json_out: bool) -> int: + """Req 4: the corpus's own detection rate; written as an artifact, never a gate.""" + from .semantic.loo import evaluate_loo + + result = evaluate_loo(select_vendored(cases)) + payload = result.to_dict() + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + if json_out: + print(json.dumps(payload, indent=2, sort_keys=True)) + return 0 + for slice_name, metrics in result.per_slice.items(): + print( + f"loo {slice_name}: pairs={metrics['pairs']} detected={metrics['detected']} silent={metrics['silent']} " + f"youden={metrics['youden']:+.3f} teaching={metrics['teaching']}" + ) + print(f"artifact={out}") + return 0 def _mechanisms_export(catalog: Path, slice_name: str, store_path: Path, *, json_out: bool) -> int: diff --git a/src/openultrasast/semantic/loo.py b/src/openultrasast/semantic/loo.py new file mode 100644 index 0000000..e0237d7 --- /dev/null +++ b/src/openultrasast/semantic/loo.py @@ -0,0 +1,203 @@ +"""Leave-one-out: the pair corpus's own detection rate (corpus-seeded-mechanisms, Req 4). + +For every scorable pair, a temporary mechanism store is seeded from every *other* trusted pair, both excerpts are +materialized and searched for variants, and the pair is scored with the pair rules: detected when a hit lies inside the +labeled function and names the labeled mechanism (or sink), silent when the fixed side yields no hit. Only ``seeded`` and +``reviewed`` pairs teach; ``advisory`` and ``title`` pairs are scored as held-out targets but never seed. Reported, never +gated. +""" + +from __future__ import annotations + +import tempfile +from collections.abc import Sequence +from dataclasses import dataclass, field +from pathlib import Path + +from ..pairs import GATING_TIERS, PairCase, _materialize, _targets +from .facts import FactLoadError, SemanticFacts, load_facts +from .functions import named_function_ranges, spans_named +from .mechanisms import Mechanism, MechanismStore, append_from_pair, corpus_mechanisms +from .seed import Lesson, _parse_language, pair_lessons +from .variant_search import search_tree +from .variants import Shape, trailing_name + + +@dataclass(frozen=True) +class LooOutcome: + pair: str + slice: str + provenance: str + mechanisms: tuple[str, ...] # labeled mechanism ids of the held-out pair + detected: bool + silent: bool + found_by: tuple[tuple[str, tuple[str, ...]], ...] # (mechanism record id, pairs that taught it) for detecting hits + hits_vuln: int + hits_fixed: int + teaches: bool # the pair contributed at least one shape when others were held out + leaked_by: tuple[str, ...] = () # mechanism record ids that hit the fixed side + + def to_dict(self) -> dict[str, object]: + return { + "pair": self.pair, + "slice": self.slice, + "provenance": self.provenance, + "mechanisms": list(self.mechanisms), + "detected": self.detected, + "silent": self.silent, + "found_by": [{"mechanism_id": mechanism_id, "pairs": list(pairs)} for mechanism_id, pairs in self.found_by], + "hits_vuln": self.hits_vuln, + "hits_fixed": self.hits_fixed, + "teaches": self.teaches, + "leaked_by": list(self.leaked_by), + } + + +@dataclass(frozen=True) +class LooResult: + outcomes: tuple[LooOutcome, ...] + per_slice: dict[str, dict[str, float]] + per_profile: dict[str, dict[str, float]] + per_mechanism: dict[str, dict[str, float]] + skipped: tuple[tuple[str, str], ...] = () + teaching_pairs: int = 0 + degradations: tuple[dict[str, object], ...] = field(default=()) + + def to_dict(self) -> dict[str, object]: + return { + "per_slice": self.per_slice, + "per_profile": self.per_profile, + "per_mechanism": self.per_mechanism, + "outcomes": [outcome.to_dict() for outcome in self.outcomes], + "skipped": [{"pair": pair, "reason": reason} for pair, reason in self.skipped], + "teaching_pairs": self.teaching_pairs, + "degradations": list(self.degradations), + } + + +def evaluate_loo(cases: Sequence[PairCase], *, facts: SemanticFacts | None = None) -> LooResult: + loaded = facts if facts is not None else _facts() + targets: list[PairCase] = [] + skipped: list[tuple[str, str]] = [] + for case in cases: + if case.known_limit: + skipped.append((case.name, f"known_limit:{case.known_limit}")) + elif not case.vuln_file.is_file() or not case.fixed_file.is_file(): + skipped.append((case.name, "pointer_pair_not_cached" if not case.vendored else "excerpt_missing")) + else: + targets.append(case) + lessons = {case.name: pair_lessons(case, loaded) for case in targets if case.review_tier in GATING_TIERS} + outcomes = [_hold_out(case, targets, lessons, loaded) for case in targets] + return LooResult( + outcomes=tuple(outcomes), + per_slice=_group(outcomes, lambda item: (item.slice,)), + per_profile=_group(outcomes, lambda item: (item.provenance,)), + per_mechanism=_group(outcomes, lambda item: item.mechanisms), + skipped=tuple(skipped), + teaching_pairs=sum(1 for shapes in lessons.values() if shapes), + ) + + +def _hold_out(case: PairCase, targets: Sequence[PairCase], lessons: dict[str, list[Lesson]], facts: SemanticFacts) -> LooOutcome: + with tempfile.TemporaryDirectory(prefix="ousast-loo-") as scratch: + root = Path(scratch) + store = MechanismStore(root / "mechanisms.jsonl") + for teacher in targets: + if teacher.name == case.name: + continue + for lesson in lessons.get(teacher.name, ()): + append_from_pair( + store, + lesson.shape, + summary=lesson.summary, + cwe=lesson.cwe, + pair=teacher.name, + provenance=teacher.provenance, + tier=teacher.review_tier, + ) + scored = score_pair_with_store(case, store, facts, root=root) + return LooOutcome(**{**scored.__dict__, "teaches": bool(lessons.get(case.name))}) + + +def score_pair_with_store(case: PairCase, store: MechanismStore, facts: SemanticFacts, *, root: Path | None = None) -> LooOutcome: + """Search both sides of one pair with a fixed store and score it with the pair rules (shared by LOO and the lever).""" + labeled_mechanisms = tuple(sorted({row.mechanism for row in case.expected if row.mechanism})) + records = {record.id: record for record in corpus_mechanisms(store.load())} + with tempfile.TemporaryDirectory(prefix="ousast-loo-pair-") as scratch: + base = root if root is not None else Path(scratch) + vuln_root = _materialize(base / "vuln", case.vuln_file, case.relpath) + fix_root = _materialize(base / "fixed", case.fixed_file, case.relpath) + vuln_hits = search_tree(vuln_root, _targets(vuln_root), store, facts, max_mechanisms=len(records) or 1).hits + fix_hits = search_tree(fix_root, _targets(fix_root), store, facts, max_mechanisms=len(records) or 1).hits + text = case.vuln_file.read_text(errors="ignore") + ranges = named_function_ranges(case.relpath, text, _parse_language(case)) or () + found: list[tuple[str, tuple[str, ...]]] = [] + for hit in vuln_hits: + record = records.get(hit.mechanism_id) + if record is None or not _inside_label(case, hit.line, ranges): + continue + if _names_label(case, record, hit.sink_name): + found.append((record.id, tuple(record.pairs))) + return LooOutcome( + pair=case.name, + slice=case.slice, + provenance=case.provenance, + mechanisms=labeled_mechanisms, + detected=bool(found), + silent=not fix_hits, + found_by=tuple(found), + hits_vuln=len(vuln_hits), + hits_fixed=len(fix_hits), + teaches=False, + leaked_by=tuple(sorted({hit.mechanism_id for hit in fix_hits})), + ) + + +def _inside_label(case: PairCase, line: int, ranges: Sequence[tuple[str, int, int]]) -> bool: + functions = {row.function for row in case.expected if row.function} + if not functions: + return True + return any(start <= line <= end for function in functions for start, end in spans_named(ranges, function)) + + +def _names_label(case: PairCase, record: Mechanism, sink_name: str) -> bool: + """The hit names the pair's labeled mechanism, or its labeled sink when the row carries one.""" + shape = Shape.from_dict(record.shape or {}) + for row in case.expected: + if row.mechanism and row.mechanism == shape.mechanism: + return True + if row.sink and trailing_name(row.sink) == sink_name: + return True + return False + + +def _group(outcomes: Sequence[LooOutcome], keys: object) -> dict[str, dict[str, float]]: + buckets: dict[str, list[LooOutcome]] = {} + for outcome in outcomes: + for key in keys(outcome): # type: ignore[operator] + buckets.setdefault(str(key), []).append(outcome) + return {name: _metrics(items) for name, items in sorted(buckets.items())} + + +def _metrics(outcomes: Sequence[LooOutcome]) -> dict[str, float]: + pairs = len(outcomes) + detected = sum(1 for item in outcomes if item.detected) + silent = sum(1 for item in outcomes if item.silent) + recall = detected / pairs if pairs else 0.0 + silence = silent / pairs if pairs else 0.0 + return { + "pairs": pairs, + "detected": detected, + "silent": silent, + "recall": recall, + "silence": silence, + "youden": recall + silence - 1.0 if pairs else 0.0, + "teaching": sum(1 for item in outcomes if item.teaches), + } + + +def _facts() -> SemanticFacts: + try: + return load_facts() + except FactLoadError: + return SemanticFacts(version="", sources=(), sinks=(), sanitizers=()) diff --git a/tests/test_loo.py b/tests/test_loo.py new file mode 100644 index 0000000..a6edaf2 --- /dev/null +++ b/tests/test_loo.py @@ -0,0 +1,107 @@ +"""corpus-seeded-mechanisms task 3.1: leave-one-out is the corpus's own detection rate (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.benchmark import ExpectedFinding +from openultrasast.pairs import PairCase + +EXEC_A = ( + "import sqlite3\n\n\ndef lookup(user):\n c = sqlite3.connect('x').cursor()\n" + + " c.execute(\"SELECT * FROM t WHERE u = '%s'\" % user)\n" +) +EXEC_A_FIX = ( + "import sqlite3\n\n\ndef lookup(user):\n c = sqlite3.connect('x').cursor()\n c.execute('SELECT * FROM t WHERE u = ?', (user,))\n" +) +EXEC_B = ( + "import sqlite3\n\n\ndef delete(name):\n cur = sqlite3.connect('y').cursor()\n cur.execute('DELETE FROM t WHERE n = ' + name)\n" +) +EXEC_B_FIX = ( + "import sqlite3\n\n\ndef delete(name):\n cur = sqlite3.connect('y').cursor()\n" + + " cur.execute('DELETE FROM t WHERE n = ?', (name,))\n" +) +SYS_C = "import os\n\n\ndef ping(host):\n os.system('ping ' + host)\n" +SYS_C_FIX = "import os\nimport subprocess\n\n\ndef ping(host):\n subprocess.run(['ping', host])\n" + + +def _case(tmp_path: Path, name: str, vuln: str, fixed: str, *, function: str, cwe: str, provenance: str = "human") -> PairCase: + (tmp_path / f"{name}-v.py").write_text(vuln) + (tmp_path / f"{name}-f.py").write_text(fixed) + return PairCase( + name=name, + slice="toy", + language="python", + origin="test", + vuln_file=tmp_path / f"{name}-v.py", + fixed_file=tmp_path / f"{name}-f.py", + relpath="app.py", + expected=( + ExpectedFinding( + cwe=cwe, vulnerability_class="x", path="app.py", evidence="", function=function, mechanism="source_reaches_sink" + ), + ), + min_recall=1.0, + fix_policy="silent", + provenance=provenance, + review_tier="seeded", + ) + + +def _toy(tmp_path: Path) -> list[PairCase]: + return [ + _case(tmp_path, "a", EXEC_A, EXEC_A_FIX, function="lookup", cwe="CWE-89"), + _case(tmp_path, "b", EXEC_B, EXEC_B_FIX, function="delete", cwe="CWE-89", provenance="agent"), + _case(tmp_path, "c", SYS_C, SYS_C_FIX, function="ping", cwe="CWE-78"), + ] + + +def test_leave_one_out_reports_which_pair_taught_which(tmp_path: Path) -> None: + from openultrasast.semantic.loo import evaluate_loo + + result = evaluate_loo(_toy(tmp_path)) + by_name = {outcome.pair: outcome for outcome in result.outcomes} + assert set(by_name) == {"a", "b", "c"} + assert by_name["a"].detected and by_name["a"].silent and by_name["a"].found_by # taught by b's execute shape + assert by_name["b"].detected and by_name["b"].silent and any("a" in pairs for _, pairs in by_name["b"].found_by) + assert not by_name["c"].detected and by_name["c"].found_by == () # nobody else teaches os.system: c only tests + slice_metrics = result.per_slice["toy"] + assert slice_metrics["pairs"] == 3 and slice_metrics["detected"] == 2 and slice_metrics["silent"] == 3 + assert abs(slice_metrics["recall"] - 2 / 3) < 1e-9 and slice_metrics["silence"] == 1.0 and abs(slice_metrics["youden"] - 2 / 3) < 1e-9 + assert result.per_profile["agent"]["pairs"] == 1 and result.per_profile["human"]["pairs"] == 2 + assert "source_reaches_sink" in result.per_mechanism + payload = result.to_dict() + assert set(payload) >= {"per_slice", "per_profile", "per_mechanism", "outcomes"} + json.dumps(payload) # artifact-ready + + +def test_leave_one_out_seeds_only_trusted_pairs_and_skips_known_limit(tmp_path: Path) -> None: + from openultrasast.semantic.loo import evaluate_loo + + cases = _toy(tmp_path) + cases[1] = PairCase(**{**cases[1].__dict__, "review_tier": "advisory"}) # b may be tested, never teaches + cases[2] = PairCase(**{**cases[2].__dict__, "known_limit": "engine"}) + result = evaluate_loo(cases) + by_name = {outcome.pair: outcome for outcome in result.outcomes} + assert set(by_name) == {"a", "b"} # known_limit excluded entirely + assert not by_name["a"].detected # b is advisory: it cannot teach a + assert by_name["b"].detected # a (seeded) teaches b, which is scored as a held-out target + assert result.skipped == (("c", "known_limit:engine"),) + + +def test_cli_pairs_loo_writes_the_artifact_offline(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + import io + from contextlib import redirect_stdout + + from openultrasast.cli import main + + monkeypatch.chdir(Path.cwd()) + buf = io.StringIO() + with redirect_stdout(buf): + assert main(["pairs", "--slice", "vibe-py", "--loo", "--json", "--loo-out", str(tmp_path / "loo.json")]) == 0 + payload = json.loads(buf.getvalue()) + assert payload["per_slice"]["vibe-py"]["pairs"] == 35 and "found_by" in payload["outcomes"][0] + assert json.loads((tmp_path / "loo.json").read_text()) == payload