diff --git a/src/openultrasast/learning/scoring.py b/src/openultrasast/learning/scoring.py new file mode 100644 index 0000000..ed2cd46 --- /dev/null +++ b/src/openultrasast/learning/scoring.py @@ -0,0 +1,161 @@ +"""Class-aware, pair-wise scoring (learning-harness, Req 3, 4.1, 4.5). + +A pair is scored once per family. Detection means a finding whose family tag relates to the labeled +family, sitting inside the labeled function; the words in a finding are never read, because a text +token was exactly the proxy that made the old hunter number meaningless. Silence means no such finding +on the fixed side: findings of other families and findings outside the labeled function are counted +separately and are never called leaks. A pair that cannot be scored keeps its reason and is never run. + +Every evaluation runs a pair K times, because a single run misses most real changes. The recorded +outcome is the majority of the runs, and the runs that disagree with it are the pair's own flakiness. +""" + +from __future__ import annotations + +from collections import Counter +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from typing import Literal + +from ..findings import StaticFinding +from .families import FamilyTaxonomy + +FamilyOutcome = Literal["pair_correct", "both_flagged", "both_silent", "reversed", "unscorable"] +Rung = Literal["suspicion", "static_corroboration", "proven"] +_RUNGS: tuple[Rung, ...] = ("suspicion", "static_corroboration", "proven") +FunctionRanges = Mapping[str, Sequence[tuple[str, int, int]]] + + +@dataclass(frozen=True) +class PairFamilyScore: + """What one pair says about one family.""" + + pair: str + family: str + runs: tuple[FamilyOutcome, ...] + outcome: FamilyOutcome + other_findings_vuln: int = 0 # right place, another family: not a detection, never a leak + other_findings_fixed: int = 0 + fabricated_vuln: int = 0 # a family id outside the taxonomy: noise, and worth naming + fabricated_fixed: int = 0 + flips: int = 0 # runs that disagreed with the majority + rung: Rung = "suspicion" + unscorable_reason: str | None = None + + +def score_pair_family( + case: object, + family: str, + vuln_runs: Sequence[Sequence[StaticFinding]], + fixed_runs: Sequence[Sequence[StaticFinding]], + *, + ranges: FunctionRanges, + taxonomy: FamilyTaxonomy, +) -> PairFamilyScore: + """Score one pair for one family over K runs of the detector on each side.""" + name = str(getattr(case, "name", "?")) + reason = _unscorable(case, ranges) + if reason is not None: + return PairFamilyScore(pair=name, family=family, runs=(), outcome="unscorable", unscorable_reason=reason) + spans = _spans(case, ranges) + outcomes: list[FamilyOutcome] = [] + other_vuln = other_fixed = fabricated_vuln = fabricated_fixed = 0 + rung: Rung = "suspicion" + for index in range(max(len(vuln_runs), len(fixed_runs))): + vuln = list(vuln_runs[index]) if index < len(vuln_runs) else [] + fixed = list(fixed_runs[index]) if index < len(fixed_runs) else [] + hits_vuln = _hits(vuln, family, spans, taxonomy) + hits_fixed = _hits(fixed, family, spans, taxonomy) + other_vuln += _others(vuln, family, spans, taxonomy) + other_fixed += _others(fixed, family, spans, taxonomy) + fabricated_vuln += _fabricated(vuln, taxonomy) + fabricated_fixed += _fabricated(fixed, taxonomy) + rung = _best_rung(rung, hits_vuln) + outcomes.append(_outcome(bool(hits_vuln), bool(hits_fixed))) + if not outcomes: + return PairFamilyScore(pair=name, family=family, runs=(), outcome="unscorable", unscorable_reason="no_runs") + majority = Counter(outcomes).most_common(1)[0][0] + return PairFamilyScore( + pair=name, + family=family, + runs=tuple(outcomes), + outcome=majority, + other_findings_vuln=other_vuln, + other_findings_fixed=other_fixed, + fabricated_vuln=fabricated_vuln, + fabricated_fixed=fabricated_fixed, + flips=sum(1 for item in outcomes if item != majority), + rung=rung, + ) + + +def _outcome(detected: bool, leaked: bool) -> FamilyOutcome: + if detected and not leaked: + return "pair_correct" + if detected and leaked: + return "both_flagged" + return "reversed" if leaked else "both_silent" + + +def _unscorable(case: object, ranges: FunctionRanges) -> str | None: + declared = getattr(case, "unscorable", None) + if declared: + return str(declared) + return None if _spans(case, ranges) else "unresolved_label" + + +def _spans(case: object, ranges: FunctionRanges) -> tuple[tuple[str, int, int], ...]: + """The (path, start, end) spans of the labeled functions, from the parsed vulnerable side.""" + wanted = {str(getattr(row, "function", "") or "") for row in getattr(case, "expected", ()) or ()} + wanted.discard("") + found: list[tuple[str, int, int]] = [] + for path, entries in ranges.items(): + for name, start, end in entries: + if name in wanted: + found.append((path, start, end)) + return tuple(found) + + +def _inside(finding: StaticFinding, spans: Sequence[tuple[str, int, int]]) -> bool: + return finding.line is not None and any(path == finding.path and start <= finding.line <= end for path, start, end in spans) + + +def _families_of(finding: StaticFinding) -> tuple[str, ...]: + return tuple(tag.split(":", 1)[1] for tag in finding.tags if tag.startswith("family:")) + + +def _relation(finding: StaticFinding, family: str, taxonomy: FamilyTaxonomy) -> str: + relations = [taxonomy.related(family, candidate) for candidate in _families_of(finding)] + for wanted in ("same", "parent_child", "lateral"): + if wanted in relations: + return wanted + return "fabricated" if relations else "none" + + +def _hits( + findings: Sequence[StaticFinding], family: str, spans: Sequence[tuple[str, int, int]], taxonomy: FamilyTaxonomy +) -> list[StaticFinding]: + return [finding for finding in findings if _inside(finding, spans) and _relation(finding, family, taxonomy) in {"same", "parent_child"}] + + +def _others(findings: Sequence[StaticFinding], family: str, spans: Sequence[tuple[str, int, int]], taxonomy: FamilyTaxonomy) -> int: + return sum(1 for finding in findings if _relation(finding, family, taxonomy) == "lateral" or not _inside(finding, spans)) + + +def _fabricated(findings: Sequence[StaticFinding], taxonomy: FamilyTaxonomy) -> int: + known = {item.id for item in taxonomy.families} + return sum(1 for finding in findings if any(candidate not in known for candidate in _families_of(finding))) + + +def _best_rung(current: Rung, hits: Sequence[StaticFinding]) -> Rung: + best = current + for finding in hits: + for tag in finding.tags: + if tag.startswith("verifier:"): + candidate = tag.split(":", 1)[1] + if candidate in _RUNGS and _RUNGS.index(candidate) > _RUNGS.index(best): # type: ignore[arg-type] + best = candidate # type: ignore[assignment] + return best + + +__all__ = ["FamilyOutcome", "FunctionRanges", "PairFamilyScore", "Rung", "score_pair_family"] diff --git a/tests/test_learning_scoring.py b/tests/test_learning_scoring.py new file mode 100644 index 0000000..8f736dc --- /dev/null +++ b/tests/test_learning_scoring.py @@ -0,0 +1,152 @@ +"""learning-harness task 2.3: per-pair family scoring (offline).""" + +from __future__ import annotations + +from pathlib import Path + +from openultrasast.benchmark import ExpectedFinding +from openultrasast.findings import StaticFinding +from openultrasast.learning.families import load_families +from openultrasast.pairs import PairCase + +RANGES = {"app.py": (("leaky", 10, 13), ("guarded", 4, 7))} + + +def _finding(family: str, line: int, path: str = "app.py") -> StaticFinding: + return StaticFinding( + finding_id=f"detector:{path}:{line}", + path=path, + title="t", + severity="medium", + confidence="low", + evidence_level="suspicion", + rationale="r", + line=line, + function_name=None, + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=[f"family:{family}", "detector:x@1"], + ranking_priority=0.0, + ) + + +def _case(tmp_path: Path, *, family: str = "access_control", unscorable: str | None = None) -> PairCase: + (tmp_path / "v.py").write_text("x = 1\n") + (tmp_path / "f.py").write_text("x = 2\n") + return PairCase( + name="p", + slice="vibe-py", + language="python", + origin="test", + vuln_file=tmp_path / "v.py", + fixed_file=tmp_path / "f.py", + relpath="app.py", + expected=( + ExpectedFinding( + cwe="CWE-639", vulnerability_class="x", path="app.py", evidence="", function="leaky", mechanism="other", family=family + ), + ), + min_recall=1.0, + fix_policy="silent", + unscorable=unscorable, + ) + + +def test_detection_needs_the_labeled_family_inside_the_labeled_function(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + case = _case(tmp_path) + hit = score_pair_family(case, "access_control", [[_finding("access_control", 12)]], [[]], ranges=RANGES, taxonomy=taxonomy) + assert hit.outcome == "pair_correct" and hit.runs == ("pair_correct",) + outside = score_pair_family(case, "access_control", [[_finding("access_control", 5)]], [[]], ranges=RANGES, taxonomy=taxonomy) + assert outside.outcome == "both_silent" # the right family in the wrong function is not a detection + other_file = score_pair_family( + case, "access_control", [[_finding("access_control", 12, "other.py")]], [[]], ranges=RANGES, taxonomy=taxonomy + ) + assert other_file.outcome == "both_silent" + wrong_family = score_pair_family(case, "access_control", [[_finding("injection", 12)]], [[]], ranges=RANGES, taxonomy=taxonomy) + assert wrong_family.outcome == "both_silent" and wrong_family.other_findings_vuln == 1 + + +def test_a_fabricated_family_never_counts_and_is_recorded(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + score = score_pair_family(_case(tmp_path), "access_control", [[_finding("sql_injection", 12)]], [[]], ranges=RANGES, taxonomy=taxonomy) + assert score.outcome == "both_silent" and score.fabricated_vuln == 1 + assert score.other_findings_vuln == 0 # a fabricated id is not another family, it is noise + + +def test_silence_ignores_findings_of_other_families_and_never_calls_them_leaks(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + case = _case(tmp_path) + quiet = score_pair_family( + case, + "access_control", + [[_finding("access_control", 12)]], + [[_finding("injection", 12), _finding("access_control", 5)]], + ranges=RANGES, + taxonomy=taxonomy, + ) + assert quiet.outcome == "pair_correct" and quiet.other_findings_fixed == 2 + leaked = score_pair_family( + case, "access_control", [[_finding("access_control", 12)]], [[_finding("access_control", 13)]], ranges=RANGES, taxonomy=taxonomy + ) + assert leaked.outcome == "both_flagged" + reversed_pair = score_pair_family(case, "access_control", [[]], [[_finding("access_control", 12)]], ranges=RANGES, taxonomy=taxonomy) + assert reversed_pair.outcome == "reversed" + + +def test_the_outcome_of_several_runs_is_their_majority(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + case = _case(tmp_path) + hit = [_finding("access_control", 12)] + score = score_pair_family(case, "access_control", [hit, [], hit], [[], [], []], ranges=RANGES, taxonomy=taxonomy) + assert score.runs == ("pair_correct", "both_silent", "pair_correct") and score.outcome == "pair_correct" + flaky = score_pair_family(case, "access_control", [hit, [], []], [[], [], []], ranges=RANGES, taxonomy=taxonomy) + assert flaky.outcome == "both_silent" and flaky.flips == 1 # a run that disagrees with the majority is a flip + + +def test_an_unscorable_pair_is_never_run_and_keeps_its_reason(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + case = _case(tmp_path, unscorable="identical_twin") + score = score_pair_family(case, "access_control", [[_finding("access_control", 12)]], [[]], ranges=RANGES, taxonomy=taxonomy) + assert score.outcome == "unscorable" and score.unscorable_reason == "identical_twin" and score.runs == () + + +def test_a_pair_with_no_parsed_function_range_is_unscorable_not_a_miss(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + score = score_pair_family(_case(tmp_path), "access_control", [[]], [[]], ranges={}, taxonomy=taxonomy) + assert score.outcome == "unscorable" and score.unscorable_reason == "unresolved_label" + + +def test_scoring_never_reads_the_text_of_a_finding(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + shouting = _finding("injection", 12) + loud = StaticFinding(**{**shouting.__dict__, "title": "CWE-639 broken object level authorization", "rationale": "IDOR in leaky"}) + score = score_pair_family(_case(tmp_path), "access_control", [[loud]], [[]], ranges=RANGES, taxonomy=taxonomy) + assert score.outcome == "both_silent" # the words say access control; the family tag says injection, and only that counts + + +def test_the_rung_is_the_best_verifier_outcome_among_the_detections(tmp_path: Path) -> None: + from openultrasast.learning.scoring import score_pair_family + + taxonomy = load_families() + corroborated = _finding("access_control", 12) + corroborated.tags.append("verifier:static_corroboration") + score = score_pair_family(_case(tmp_path), "access_control", [[corroborated]], [[]], ranges=RANGES, taxonomy=taxonomy) + assert score.outcome == "pair_correct" and score.rung == "static_corroboration" + plain = score_pair_family(_case(tmp_path), "access_control", [[_finding("access_control", 12)]], [[]], ranges=RANGES, taxonomy=taxonomy) + assert plain.rung == "suspicion"