mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 11:49:28 +00:00
feat(learning-harness): per-pair family scoring
Task 2.3. A pair is scored once per family. Detection means a finding whose family tag relates to the labeled family and whose line sits inside the labeled function; the words in a finding are never read, because a text token is exactly the proxy that made the old hunter number meaningless. Silence means no such finding on the fixed side, and findings of other families or outside the labeled function are counted separately and never called leaks. A fabricated family id counts as neither, and is recorded as noise worth naming. Every evaluation runs a pair several 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, which is what the regression budget will be built from. A pair that cannot be scored keeps its reason and is never run, and a labeled function no parsed range names is unscorable rather than a silent miss. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 5
Claude Opus 5
parent
09898529bd
commit
241af31a9a
@@ -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"]
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user