diff --git a/src/openultrasast/pairs.py b/src/openultrasast/pairs.py index fedf8f3..4dc00cd 100644 --- a/src/openultrasast/pairs.py +++ b/src/openultrasast/pairs.py @@ -163,6 +163,8 @@ class PairOutcome: unadjudicated_vuln: int = 0 unadjudicated_fixed: int = 0 detection_kinds: tuple[str, ...] = () + family: str = "" # the family this pair was scored under (hunter path only) + unscorable: str | None = None # why it could not be scored; it then leaves every denominator unresolved_labels: int = 0 # expected rows whose function names no range on the vulnerable side review_tier: str = "advisory" @@ -193,6 +195,7 @@ class PairEvalResult: per_profile: dict[str, PairCorpusMetrics] = field(default_factory=dict) per_tier: dict[str, PairCorpusMetrics] = field(default_factory=dict) per_mechanism: dict[str, PairCorpusMetrics] = field(default_factory=dict) + per_family: dict[str, dict[str, object]] = field(default_factory=dict) # learning-harness Req 3.3: family metrics beside the rest achievable: dict[str, PairCorpusMetrics] = field(default_factory=dict) known_limit: tuple[str, ...] = () loss: dict[str, dict[str, int]] = field(default_factory=dict) @@ -425,13 +428,16 @@ def evaluate_catalog( cases = tuple(ready) outcomes = tuple(evaluate_pair(case, ruleset=ruleset) for case in cases) scorers: dict[str, dict[str, PairCorpusMetrics]] = {} + hunter_outcomes: list[PairOutcome] = [] # the family scorer runs on the hunter path only for slice_name in sorted({case.slice for case in cases if case.slice in OVERLAY_SLICES}): slice_cases = tuple(case for case in cases if case.slice == slice_name) scorers[slice_name] = {"overlay": _metrics(tuple(item for item in outcomes if item.slice == slice_name))} if inventory_sidecar: scorers[slice_name]["inventory"] = _metrics(tuple(_inventory_only(case, ruleset) for case in slice_cases)) if hunter is not None: - scorers[slice_name]["hunter"] = _metrics(tuple(evaluate_pair(case, hunter=hunter) for case in slice_cases)) + hunted = tuple(evaluate_pair(case, hunter=hunter) for case in slice_cases) + hunter_outcomes.extend(hunted) + scorers[slice_name]["hunter"] = _metrics(hunted) if hunter is None and scorers: degradations.append({"stage": "pairs", "reason": "hunter_model_unavailable"}) achievable_outcomes = tuple(item for item in outcomes if item.known_limit is None) @@ -457,6 +463,7 @@ def evaluate_catalog( for tier in sorted({item.review_tier for item in outcomes}) }, per_mechanism={mechanism: _metrics(tuple(items)) for mechanism, items in sorted(per_mechanism.items())}, + per_family=_family_metrics(hunter_outcomes), achievable={ slice_name: _metrics(tuple(item for item in achievable_outcomes if item.slice == slice_name)) for slice_name in sorted({item.slice for item in achievable_outcomes}) @@ -467,6 +474,43 @@ def evaluate_catalog( ) +def _family_metrics(outcomes: Sequence[PairOutcome]) -> dict[str, dict[str, object]]: + """Per-family numbers for the pairs the family scorer ran, with unscorable rows out of the denominator.""" + from .learning.families import load_families + from .learning.scoring import PairFamilyScore, aggregate + + scored = [item for item in outcomes if item.family] + if not scored: + return {} + taxonomy = load_families() + scores = [ + PairFamilyScore( + pair=item.name, + family=item.family, + runs=() + if item.unscorable + else ( + ( + ("pair_correct" if item.pair_correct else "both_flagged") + if item.detected_vuln + else ("reversed" if not item.silent_fix else "both_silent") + ), + ), + outcome="unscorable" + if item.unscorable + else ( + ("pair_correct" if item.pair_correct else "both_flagged") + if item.detected_vuln + else ("reversed" if not item.silent_fix else "both_silent") + ), + slice=item.slice, + unscorable_reason=item.unscorable, + ) + for item in scored + ] + return {name: metrics.to_dict() for name, metrics in aggregate(scores, taxonomy=taxonomy).items()} + + def build_pair_signals(outcomes: Sequence[PairOutcome], *, split: str | None = None) -> list[dict[str, object]]: """Same miss/fp vocabulary as ``build_rule_signals``, tagged with the pair name, profile, and mechanisms. @@ -641,6 +685,7 @@ def result_payload(result: PairEvalResult) -> dict[str, object]: "per_profile": {name: asdict(metrics) for name, metrics in result.per_profile.items()}, "per_tier": {name: asdict(metrics) for name, metrics in result.per_tier.items()}, "per_mechanism": {name: asdict(metrics) for name, metrics in result.per_mechanism.items()}, + "per_family": dict(result.per_family), "achievable": {name: asdict(metrics) for name, metrics in result.achievable.items()}, "known_limit": list(result.known_limit), "loss": {name: dict(counts) for name, counts in result.loss.items()}, @@ -1018,43 +1063,71 @@ def _evaluate_overlay_pair(case: PairCase, vuln_root: Path, fix_root: Path, rule ) -def _evaluate_hunter_pair(case: PairCase, vuln_root: Path, fix_root: Path, hunter: HunterScan) -> PairOutcome: - """Score the LLM hunter with the same detection and leak rules (Req 1.7).""" - vuln_findings = hunter(vuln_root) - fix_findings = hunter(fix_root) +def _evaluate_hunter_pair(case: PairCase, vuln_root: Path, fix_root: Path, hunter: HunterScan, runs: int = 1) -> PairOutcome: + """Score the model-driven hunter by family and function (learning-harness Req 3.1-3.3, 3.5). + + Detection is a finding of the labeled family inside the labeled function; a finding of another family + or outside it is counted and never called a leak; the words in a finding are never read. A pair that + cannot be scored is reported with its reason and the detector is never run on it. + """ + from .learning.classify import classify_pair + from .learning.families import load_families + from .learning.scoring import score_pair_family, unscorable_reason + + taxonomy = load_families() vuln_ranges = _overlay_scan(vuln_root).ranges + reason = unscorable_reason(case, parse_ok=bool(vuln_ranges), ranges=vuln_ranges) + family = _family_of(case, taxonomy) + if reason is not None: + return _hunter_outcome(case, family=family, unscorable=reason) fix_ranges = _overlay_scan(fix_root).ranges - expected_total = len(case.expected) - matched = sum(1 for row in case.expected if any(_hunter_matches_expected(row, finding, vuln_ranges) for finding in vuln_findings)) - recall = matched / expected_total if expected_total else 1.0 - detected = recall + 1e-12 >= case.min_recall - if case.fix_policy == "silent": - leaks = tuple(finding.finding_id for finding in fix_findings) - else: - leaks = tuple( - finding.finding_id - for finding in fix_findings - if any(_hunter_matches_expected(row, finding, fix_ranges) for row in case.expected) - ) - misses = tuple( - f"{row.rule_id or '-'}:{row.path}:{row.cwe}" - for row in case.expected - if not any(_hunter_matches_expected(row, finding, vuln_ranges) for finding in vuln_findings) - ) + vuln_runs = [hunter(vuln_root) for _ in range(max(runs, 1))] + fix_runs = [hunter(fix_root) for _ in range(max(runs, 1))] + score = score_pair_family(case, family, vuln_runs, fix_runs, ranges=vuln_ranges, taxonomy=taxonomy) + del classify_pair, fix_ranges + return _hunter_outcome(case, family=family, score=score, vuln=vuln_runs[0], fixed=fix_runs[0]) + + +def _family_of(case: PairCase, taxonomy: object) -> str: + """The family this pair is scored under: its label when it has one, else what the classifier says.""" + from .learning.classify import classify_pair + + for row in case.expected: + label = getattr(row, "family", None) + if label: + return str(label) + answer = classify_pair(case, taxonomy, client=None) # type: ignore[arg-type] + return answer.families[0] if answer.families else "unknown" + + +def _hunter_outcome( + case: PairCase, + *, + family: str, + score: object | None = None, + vuln: Sequence[StaticFinding] = (), + fixed: Sequence[StaticFinding] = (), + unscorable: str | None = None, +) -> PairOutcome: + outcome = getattr(score, "outcome", "unscorable") + detected = outcome in {"pair_correct", "both_flagged"} + silent = outcome in {"pair_correct", "both_silent"} + leaks = tuple(finding.finding_id for finding in fixed) if outcome in {"both_flagged", "reversed"} else () + misses = () if detected else tuple(f"{row.rule_id or '-'}:{row.path}:{row.cwe}" for row in case.expected) return PairOutcome( name=case.name, slice=case.slice, language=case.language, origin=case.origin, detected_vuln=detected, - silent_fix=not leaks, - pair_correct=detected and not leaks, - vuln_matched=matched, - vuln_expected=expected_total, - vuln_findings=len(vuln_findings), - fix_findings=len(fix_findings), + silent_fix=silent, + pair_correct=outcome == "pair_correct", + vuln_matched=1 if detected else 0, + vuln_expected=len(case.expected), + vuln_findings=len(vuln), + fix_findings=len(fixed), fix_leaks=len(leaks), - recall=recall, + recall=1.0 if detected else 0.0, misses=misses, leaks=leaks, commit_url=case.commit_url, @@ -1064,8 +1137,10 @@ def _evaluate_hunter_pair(case: PairCase, vuln_root: Path, fix_root: Path, hunte mechanisms=_mechanisms(case), known_limit=case.known_limit, split=case.split, - weak_labels=sum(1 for row in case.expected if is_weak_label(row)), - detection_kinds=("hunter",) if vuln_findings else (), + weak_labels=0, + detection_kinds=("hunter",) if detected else (), + family=family, + unscorable=unscorable or (str(getattr(score, "unscorable_reason", "") or "") or None), ) @@ -1140,23 +1215,6 @@ def _overlay_matches_expected( return False -def _hunter_matches_expected( - expected: ExpectedFinding, finding: StaticFinding, ranges: dict[str, tuple[tuple[str, int, int], ...]] -) -> bool: - if is_weak_label(expected): - return False - text = f"{finding.title} {finding.rationale} {' '.join(finding.tags)}".lower() - if expected.function: - spans = spans_named(ranges.get(finding.path, ()), expected.function) - if finding.line is None or not any(start <= finding.line <= end for start, end in spans): - return False - if expected.cwe and expected.cwe.lower() in text: - return True - if expected.sink and expected.sink.lower() != "unknown" and expected.sink.lower() in text: - return True - return bool(expected.rule_id and expected.rule_id in finding.finding_id) - - def _obligation_matches_expected(expected: ExpectedFinding, statics: Sequence[StaticFinding], ranges: FunctionRanges | None) -> bool: """The one additive rule (Req 7.2): a row with `obligation` is detected by a finding tagged `obligation:` whose line lies inside the labeled function. Nothing else about the scorer changes; rows without the label never reach here.""" @@ -1214,6 +1272,11 @@ def _mechanisms(case: PairCase) -> tuple[str, ...]: # --- metrics --------------------------------------------------------------- +def _scorable(outcomes: Sequence[PairOutcome]) -> tuple[PairOutcome, ...]: + """Rows that could be scored. An unscorable row leaves every rate rather than counting as a miss (Req 4.1).""" + return tuple(item for item in outcomes if item.unscorable is None) + + def _metrics(outcomes: Sequence[PairOutcome]) -> PairCorpusMetrics: pairs = len(outcomes) detected = sum(1 for item in outcomes if item.detected_vuln) diff --git a/tests/test_learning_pairs_path.py b/tests/test_learning_pairs_path.py new file mode 100644 index 0000000..cab5abe --- /dev/null +++ b/tests/test_learning_pairs_path.py @@ -0,0 +1,110 @@ +"""learning-harness task 3.2: family scoring on the pairs hunter path (offline).""" + +from __future__ import annotations + +from pathlib import Path + +from openultrasast.benchmark import ExpectedFinding +from openultrasast.findings import StaticFinding +from openultrasast.pairs import PairCase, evaluate_pair + +VULN = "from flask import request\nimport os\n\n\ndef run():\n os.system(request.args.get('cmd'))\n" +FIXED = "from flask import request\nimport subprocess\n\n\ndef run():\n subprocess.run(['echo', request.args.get('cmd')], check=False)\n" + + +def _case(tmp_path: Path, *, family: str | None = "injection", unscorable: str | None = None) -> PairCase: + (tmp_path / "v.py").write_text(VULN) + (tmp_path / "f.py").write_text(FIXED) + 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-78", vulnerability_class="x", path="app.py", evidence="", function="run", mechanism="other", family=family + ), + ), + min_recall=1.0, + fix_policy="silent", + unscorable=unscorable, + ) + + +def _finding(family: str, line: int) -> StaticFinding: + return StaticFinding( + finding_id=f"detector:app.py:{line}", + path="app.py", + title="t", + severity="high", + confidence="low", + evidence_level="suspicion", + rationale="r", + line=line, + function_name="run", + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=[f"family:{family}", "detector:injection@1"], + ranking_priority=0.0, + ) + + +def _hunter(vuln: list[StaticFinding], fixed: list[StaticFinding]): # type: ignore[no-untyped-def] + seen: list[Path] = [] + + def scan(root: Path) -> list[StaticFinding]: + seen.append(root) + return vuln if root.name == "vuln" else fixed + + scan.seen = seen # type: ignore[attr-defined] + return scan + + +def test_the_hunter_path_scores_by_family_and_function_not_by_text(tmp_path: Path) -> None: + right = evaluate_pair(_case(tmp_path), hunter=_hunter([_finding("injection", 6)], [])) + assert right.detected_vuln and right.silent_fix and right.pair_correct + assert right.detection_kinds == ("hunter",) and right.family == "injection" + wrong_family = evaluate_pair(_case(tmp_path), hunter=_hunter([_finding("access_control", 6)], [])) + assert not wrong_family.detected_vuln # the right place, the wrong family + loud = StaticFinding(**{**_finding("access_control", 6).__dict__, "title": "CWE-78 command injection in run"}) + assert not evaluate_pair(_case(tmp_path), hunter=_hunter([loud], [])).detected_vuln # the words never count + + +def test_a_finding_of_another_family_on_the_fixed_side_is_not_a_leak(tmp_path: Path) -> None: + outcome = evaluate_pair(_case(tmp_path), hunter=_hunter([_finding("injection", 6)], [_finding("access_control", 6)])) + assert outcome.pair_correct and outcome.silent_fix + assert outcome.fix_findings == 1 and outcome.fix_leaks == 0 # counted, never charged + leaked = evaluate_pair(_case(tmp_path), hunter=_hunter([_finding("injection", 6)], [_finding("injection", 6)])) + assert leaked.detected_vuln and not leaked.silent_fix and leaked.fix_leaks == 1 + + +def test_an_unscorable_pair_is_reported_as_such_and_never_run(tmp_path: Path) -> None: + hunter = _hunter([_finding("injection", 6)], []) + outcome = evaluate_pair(_case(tmp_path, unscorable="identical_twin"), hunter=hunter) + assert outcome.unscorable == "identical_twin" and not outcome.detected_vuln + assert hunter.seen == [] # type: ignore[attr-defined] + + +def test_a_pair_with_no_family_label_falls_back_to_the_classifier(tmp_path: Path) -> None: + outcome = evaluate_pair(_case(tmp_path, family=None), hunter=_hunter([_finding("injection", 6)], [])) + assert outcome.family == "injection" and outcome.pair_correct # CWE-78 classifies as injection + + +def test_the_overlay_path_is_untouched_by_the_family_scorer(tmp_path: Path) -> None: + overlay = evaluate_pair(_case(tmp_path)) + assert overlay.detection_kinds != ("hunter",) and overlay.family == "" + + +def test_the_scoreboard_carries_the_family_block(tmp_path: Path) -> None: + from openultrasast.pairs import evaluate_catalog, result_payload + + result = evaluate_catalog([_case(tmp_path)], hunter=_hunter([_finding("injection", 6)], []), inventory_sidecar=False) + payload = result_payload(result) + assert "per_family" in payload and payload["per_family"]["injection"]["scorable"] == 1 + assert payload["per_family"]["injection"]["recall"] == 1.0 + assert payload["per_family"]["injection"]["hierarchical_credit"] is False + assert payload["per_family"]["injection"]["taxonomy_version"] diff --git a/tests/test_pair_corpus.py b/tests/test_pair_corpus.py index 5b6c965..cf947f5 100644 --- a/tests/test_pair_corpus.py +++ b/tests/test_pair_corpus.py @@ -373,7 +373,8 @@ def test_known_limit_pairs_are_evaluated_but_excluded_from_achievable(tmp_path: assert result.degradations and result.degradations[0]["reason"] == "hunter_model_unavailable" -def test_hunter_scorer_uses_same_rules_with_a_scripted_client(tmp_path: Path) -> None: +def test_hunter_scorer_credits_a_family_tag_and_never_the_words(tmp_path: Path) -> None: + """learning-harness Req 3.1/3.6 replaced this path's rule: the family tag decides, the rationale never does.""" from openultrasast.findings import StaticFinding src = "from flask import request\n\ndef f():\n return eval(request.args.get('x'))\n" @@ -387,34 +388,37 @@ def test_hunter_scorer_uses_same_rules_with_a_scripted_client(tmp_path: Path) -> mechanism="source_reaches_sink", ) - def scripted(root: Path) -> list[StaticFinding]: - text = (root / "app.py").read_text() - if "eval(" not in text: - return [] - return [ - StaticFinding( - finding_id="tool-hunter:app.py:4", - path="app.py", - title="eval of request input", - severity="high", - confidence="medium", - evidence_level="suspicion", - rationale="CWE-95 eval executes attacker input", - line=4, - function_name=None, - reachability_status="unknown", - reachability_evidence=[], - reachability_conditions=[], - tags=[], - ranking_priority=1.0, - ) - ] + def scripted(tags: list[str]): # type: ignore[no-untyped-def] + def scan(root: Path) -> list[StaticFinding]: + if "eval(" not in (root / "app.py").read_text(): + return [] + return [ + StaticFinding( + finding_id="tool-hunter:app.py:4", + path="app.py", + title="eval of request input", + severity="high", + confidence="medium", + evidence_level="suspicion", + rationale="CWE-95 eval executes attacker input", + line=4, + function_name=None, + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=tags, + ranking_priority=1.0, + ) + ] + + return scan case = _case(tmp_path, "hunt", src, "def f():\n return 1\n", row) - result = evaluate_catalog([case], hunter=scripted) - hunter = result.scorers["sast"]["hunter"] - assert hunter.pair_correct == 1 and hunter.detected_vuln == 1 and hunter.silent_fix == 1 - assert not result.degradations + tagged = evaluate_catalog([case], hunter=scripted(["family:injection"])).scorers["sast"]["hunter"] + assert tagged.pair_correct == 1 and tagged.detected_vuln == 1 and tagged.silent_fix == 1 + # the same finding, the same words, no family tag: not a detection, because the rationale is not evidence + untagged = evaluate_catalog([case], hunter=scripted([])).scorers["sast"]["hunter"] + assert untagged.detected_vuln == 0 and untagged.silent_fix == 1 def test_juliet_strcpy_pair_bodies_differ_and_java_hash_is_known_limit() -> None: