diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index c6098e7..2667a31 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -93,7 +93,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Boundary: DockerCliRunner_ - _Depends: 4.1_ -- [ ] 4.3 Reject unsafe regression snippets before execution +- [x] 4.3 Reject unsafe regression snippets before execution - Snippets that refer to the Docker socket, host network, or writes under the source mount are not executed and become `inconclusive`. - Observable completion: each forbidden shape is rejected in a unit test and no runner job is created. - _Requirements: 5.8, 6.6_ @@ -103,7 +103,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex ## Phase 3 — Core: Regression verdicts - [ ] 5. Produce worth-fixing verdicts from sandboxed candidates -- [ ] 5.1 Select a bounded candidate set from hotspots and severe reachable inventory +- [x] 5.1 Select a bounded candidate set from hotspots and severe reachable inventory - Selection honors max-candidates and always includes reachable severity-5 inventory even when the ledger demoted the hotspot. - Observable completion: a unit test with six hotspots and max-candidates three returns three, plus the sev-5 reachable extra when present. - _Requirements: 5.1, 7.4, 9.5_ diff --git a/src/openultrasast/regress/__init__.py b/src/openultrasast/regress/__init__.py index 754702a..745bb94 100644 --- a/src/openultrasast/regress/__init__.py +++ b/src/openultrasast/regress/__init__.py @@ -1,6 +1,6 @@ """Isolated regression snippets and verdicts.""" -from .candidate import INCONCLUSIVE, SAFETY_REJECTED, RegressionRunner, RegressionVerdict +from .candidate import INCONCLUSIVE, SAFETY_REJECTED, RegressionRunner, RegressionVerdict, select_candidates from .safety import UnsafeSnippetError, check_snippet_safety __all__ = [ @@ -10,4 +10,5 @@ __all__ = [ "RegressionVerdict", "UnsafeSnippetError", "check_snippet_safety", + "select_candidates", ] diff --git a/src/openultrasast/regress/candidate.py b/src/openultrasast/regress/candidate.py index fa362cf..6928b0e 100644 --- a/src/openultrasast/regress/candidate.py +++ b/src/openultrasast/regress/candidate.py @@ -1,9 +1,13 @@ -"""Safety-gated regression execution (task 4.3).""" +"""Candidate selection and safety-gated regression execution.""" from __future__ import annotations +from collections.abc import Mapping, Sequence from dataclasses import dataclass +from ..complexity.ledger import hotspot_key, select_forced_candidates +from ..complexity.map import Hotspot +from ..findings import StaticFinding from ..sandbox import SandboxJob, SandboxResult, SandboxRunner from .safety import UnsafeSnippetError, check_snippet_safety @@ -13,6 +17,58 @@ TRIGGERABLE = "triggerable" NOT_TRIGGERABLE = "not_triggerable" +def select_candidates( + hotspots: Sequence[Hotspot], + findings: Sequence[StaticFinding], + *, + max_candidates: int, + policy_severity_by_id: Mapping[str, object] | None = None, + rule_cwe: Mapping[str, str] | None = None, +) -> tuple[Hotspot, ...]: + """Select min(max_candidates, ranked hotspots) ∪ sev-5 reachable inventory. + + The cap applies to the ranked hotspot slice. Forced policy-severity-5 + reachable (or inferred-file-surface) inventory stays in the set even when a + ledger overlay demoted its hotspot out of that slice. + """ + ranked = tuple(sorted(hotspots, key=lambda item: (-item.score, item.path, item.function_name or ""))) + selected = list(ranked[: max(max_candidates, 0)]) + seen = {hotspot_key(item.path, item.function_name) for item in selected} + extras: list[Hotspot] = [] + for finding in select_forced_candidates(findings, policy_severity_by_id or {}, rule_cwe=rule_cwe): + hotspot = _hotspot_for_finding(finding, ranked) or _hotspot_from_forced_finding(finding) + key = hotspot_key(hotspot.path, hotspot.function_name) + if key in seen: + continue + seen.add(key) + extras.append(hotspot) + return tuple(selected + extras) + + +def _hotspot_for_finding(finding: StaticFinding, hotspots: Sequence[Hotspot]) -> Hotspot | None: + for hotspot in hotspots: + if finding.finding_id in hotspot.inventory_finding_ids: + return hotspot + key = hotspot_key(finding.path, finding.function_name) + for hotspot in hotspots: + if hotspot_key(hotspot.path, hotspot.function_name) == key: + return hotspot + return None + + +def _hotspot_from_forced_finding(finding: StaticFinding) -> Hotspot: + return Hotspot( + path=finding.path, + function_name=finding.function_name, + score=0.0, + band="low", + signals={}, + rationale="forced reachable severity-5 inventory", + test_hint=None, + inventory_finding_ids=(finding.finding_id,), + ) + + @dataclass(frozen=True) class RegressionVerdict: verdict: str diff --git a/tests/test_regress_candidates.py b/tests/test_regress_candidates.py new file mode 100644 index 0000000..0cb9fb5 --- /dev/null +++ b/tests/test_regress_candidates.py @@ -0,0 +1,159 @@ +"""Bounded regression candidate selection from hotspots and severe inventory (task 5.1).""" + +from openultrasast.complexity.ledger import NOT_TRIGGERABLE_DELTA, apply_overlay, record_verdict +from openultrasast.complexity.map import Hotspot +from openultrasast.findings import StaticFinding +from openultrasast.regress.candidate import select_candidates + + +def _finding( + path: str, + finding_id: str, + *, + function_name: str | None = None, + reachability: str = "unknown", + severity: str = "low", +) -> StaticFinding: + return StaticFinding( + finding_id=finding_id, + path=path, + title="pattern hit", + severity=severity, + confidence="medium", + evidence_level="static_corroboration", + rationale="inventory", + line=1, + function_name=function_name, + reachability_status=reachability, + reachability_evidence=[], + reachability_conditions=[], + tags=[], + ranking_priority=1.0, + ) + + +def _hotspot( + path: str, + *, + function_name: str | None = None, + score: float = 4.0, + band: str = "medium", + inventory_finding_ids: tuple[str, ...] = (), +) -> Hotspot: + return Hotspot( + path=path, + function_name=function_name, + score=score, + band=band, + signals={"loc": 10, "inventory_hit_count": len(inventory_finding_ids)}, + rationale=f"{band} band: score {score}", + test_hint=None, + inventory_finding_ids=inventory_finding_ids, + ) + + +def _six_hotspots(*, sev5_id: str | None = None, sev5_score: float = 4.0) -> tuple[Hotspot, ...]: + return ( + _hotspot("a.py", function_name="a", score=5.0), + _hotspot("b.py", function_name="b", score=4.8), + _hotspot("c.py", function_name="c", score=4.6), + _hotspot("d.py", function_name="d", score=4.4), + _hotspot("e.py", function_name="e", score=4.2), + _hotspot( + "f.py", + function_name="f", + score=sev5_score, + inventory_finding_ids=(sev5_id,) if sev5_id else (), + ), + ) + + +def test_six_hotspots_max_candidates_three_returns_three() -> None: + noise = _finding("f.py", "print:f.py:1", function_name="f", reachability="reachable") + selected = select_candidates( + _six_hotspots(), + [noise], + max_candidates=3, + policy_severity_by_id={noise.finding_id: 2}, + ) + + assert len(selected) == 3 + assert [item.path for item in selected] == ["a.py", "b.py", "c.py"] + + +def test_sev5_reachable_extra_included_when_ledger_demoted_hotspot() -> None: + sev5_id = "cmd:f.py:1" + hotspots = _six_hotspots(sev5_id=sev5_id, sev5_score=5.2) + ledger = record_verdict({}, path="f.py", function_name="f", verdict="not_triggerable") + demoted = apply_overlay(hotspots, ledger) + finding = _finding( + "f.py", + sev5_id, + function_name="f", + reachability="reachable", + severity="critical", + ) + + selected = select_candidates( + demoted, + [finding], + max_candidates=3, + policy_severity_by_id={sev5_id: 5}, + ) + + demoted_f = next(item for item in demoted if item.path == "f.py") + assert demoted_f.score == round(5.2 + NOT_TRIGGERABLE_DELTA, 4) + assert [item.path for item in demoted[:3]] == ["a.py", "b.py", "c.py"] + assert demoted_f not in demoted[:3] + assert len(selected) == 4 + assert [item.path for item in selected[:3]] == ["a.py", "b.py", "c.py"] + assert selected[3] is demoted_f + assert sev5_id in selected[3].inventory_finding_ids + + +def test_sev5_already_in_cap_is_not_duplicated() -> None: + sev5_id = "cmd:a.py:1" + hotspots = ( + _hotspot("a.py", function_name="a", score=5.0, inventory_finding_ids=(sev5_id,)), + _hotspot("b.py", function_name="b", score=4.8), + _hotspot("c.py", function_name="c", score=4.6), + _hotspot("d.py", function_name="d", score=4.4), + _hotspot("e.py", function_name="e", score=4.2), + _hotspot("f.py", function_name="f", score=4.0), + ) + finding = _finding("a.py", sev5_id, function_name="a", reachability="reachable", severity="critical") + + selected = select_candidates( + hotspots, + [finding], + max_candidates=3, + policy_severity_by_id={sev5_id: 5}, + ) + + assert len(selected) == 3 + assert [item.path for item in selected] == ["a.py", "b.py", "c.py"] + assert sev5_id in selected[0].inventory_finding_ids + + +def test_sev5_reachable_inventory_without_hotspot_is_still_selected() -> None: + finding = _finding( + "z.py", + "cmd:z.py:1", + function_name="z", + reachability="reachable", + severity="critical", + ) + + selected = select_candidates( + _six_hotspots(), + [finding], + max_candidates=3, + policy_severity_by_id={finding.finding_id: 5}, + ) + + assert len(selected) == 4 + assert [item.path for item in selected[:3]] == ["a.py", "b.py", "c.py"] + extra = selected[3] + assert extra.path == "z.py" + assert extra.function_name == "z" + assert extra.inventory_finding_ids == (finding.finding_id,)