From ad16e22c85550aeec2f9c3a299efeab6fa3ea3d8 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 3 Sep 2026 20:48:03 +0200 Subject: [PATCH] feat(three-stage-scan): overlay complexity ledger without hiding sev-5 --- .kiro/specs/three-stage-scan/tasks.md | 2 +- src/openultrasast/complexity/__init__.py | 18 ++ src/openultrasast/complexity/ledger.py | 175 +++++++++++++++ src/openultrasast/complexity/map.py | 4 + tests/test_complexity_ledger.py | 260 +++++++++++++++++++++++ 5 files changed, 458 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/complexity/ledger.py create mode 100644 tests/test_complexity_ledger.py diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 44a6375..375129c 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -31,7 +31,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex ## Phase 2 — Core: Map, hunter tools, sandbox (parallel after foundation) -- [ ] 2. Build the heuristic complexity map and test-tuning hints +- [x] 2. Build the heuristic complexity map and test-tuning hints - [x] 2.1 (P) Compute per-file and per-function complexity signals - Signals include size, nesting, surface tags, reachability, inventory-hit density as a feature not the score, and whether an adjacent test appears to exist. - Function identity uses entry-point names when present and otherwise remains empty. diff --git a/src/openultrasast/complexity/__init__.py b/src/openultrasast/complexity/__init__.py index 88e1d52..c21894b 100644 --- a/src/openultrasast/complexity/__init__.py +++ b/src/openultrasast/complexity/__init__.py @@ -1,6 +1,16 @@ """Heuristic complexity map for stage 2.""" from .hints import TestHint, attach_test_hints +from .ledger import ( + LedgerEntry, + apply_overlay, + hotspot_key, + load_ledger, + must_keep_as_candidate, + record_verdict, + select_forced_candidates, + write_ledger, +) from .map import ComplexityMap, Hotspot, build_complexity_map, write_complexity_map from .signals import ComplexitySignals, collect_signals @@ -8,9 +18,17 @@ __all__ = [ "ComplexityMap", "ComplexitySignals", "Hotspot", + "LedgerEntry", "TestHint", + "apply_overlay", "attach_test_hints", "build_complexity_map", "collect_signals", + "hotspot_key", + "load_ledger", + "must_keep_as_candidate", + "record_verdict", + "select_forced_candidates", "write_complexity_map", + "write_ledger", ] diff --git a/src/openultrasast/complexity/ledger.py b/src/openultrasast/complexity/ledger.py new file mode 100644 index 0000000..4ad823b --- /dev/null +++ b/src/openultrasast/complexity/ledger.py @@ -0,0 +1,175 @@ +"""Loop-owned complexity ledger: overlay hotspot scores without hiding sev-5 inventory.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass, replace +from pathlib import Path +from typing import TYPE_CHECKING + +from ..findings import StaticFinding + +if TYPE_CHECKING: + from .map import Hotspot + +TRIGGERABLE_DELTA = 1.5 +NOT_TRIGGERABLE_DELTA = -1.5 +VERDICT_DELTAS = { + "triggerable": TRIGGERABLE_DELTA, + "not_triggerable": NOT_TRIGGERABLE_DELTA, +} +FORCED_CANDIDATE_SEVERITY = 5 +FORCED_CANDIDATE_REACHABILITY = frozenset({"reachable", "inferred-file-surface"}) + + +@dataclass(frozen=True) +class LedgerEntry: + score_delta: float + last_verdict: str + round: int = 0 + + +def hotspot_key(path: str, function_name: str | None) -> str: + return f"{path}:{function_name or ''}" + + +def load_ledger(path: Path | None) -> dict[str, LedgerEntry]: + if path is None or not path.exists(): + return {} + payload = json.loads(path.read_text()) + entries: dict[str, LedgerEntry] = {} + for key, value in payload.items(): + entry = _coerce_entry(value) + if entry is not None: + entries[str(key)] = entry + return entries + + +def write_ledger(path: Path, ledger: Mapping[str, LedgerEntry | Mapping[str, object]]) -> None: + payload: dict[str, dict[str, object]] = {} + for key, value in ledger.items(): + entry = _coerce_entry(value) + if entry is None: + continue + payload[str(key)] = asdict(entry) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + + +def record_verdict( + ledger: Mapping[str, LedgerEntry | Mapping[str, object]], + *, + path: str, + function_name: str | None, + verdict: str, +) -> dict[str, LedgerEntry]: + updated: dict[str, LedgerEntry] = {} + for key, value in ledger.items(): + entry = _coerce_entry(value) + if entry is not None: + updated[str(key)] = entry + key = hotspot_key(path, function_name) + previous = updated.get(key) + updated[key] = LedgerEntry( + score_delta=round((previous.score_delta if previous else 0.0) + VERDICT_DELTAS.get(verdict, 0.0), 4), + last_verdict=verdict, + round=(previous.round if previous else 0) + 1, + ) + return updated + + +def apply_overlay( + hotspots: Sequence[Hotspot], + ledger: Mapping[str, LedgerEntry | Mapping[str, object]], +) -> tuple[Hotspot, ...]: + if not ledger: + return tuple(hotspots) + adjusted: list[Hotspot] = [] + for hotspot in hotspots: + entry = _coerce_entry(ledger.get(hotspot_key(hotspot.path, hotspot.function_name))) + if entry is None or entry.score_delta == 0.0: + adjusted.append(hotspot) + continue + score = round(hotspot.score + entry.score_delta, 4) + signals = dict(hotspot.signals) + signals["ledger_delta"] = entry.score_delta + adjusted.append( + replace( + hotspot, + score=score, + band=_band_for(score), + signals=signals, + rationale=( + f"{hotspot.rationale} ledger overlay last_verdict={entry.last_verdict} " + f"score_delta={entry.score_delta} (inventory retained)." + ), + ) + ) + return tuple(sorted(adjusted, key=lambda item: (-item.score, item.path, item.function_name or ""))) + + +def must_keep_as_candidate( + finding: StaticFinding, + policy_severity: int | object, + reachability: str | None = None, +) -> bool: + status = finding.reachability_status if reachability is None else reachability + return _as_severity(policy_severity) >= FORCED_CANDIDATE_SEVERITY and status in FORCED_CANDIDATE_REACHABILITY + + +def select_forced_candidates( + findings: Sequence[StaticFinding], + policy: Mapping[str, object], + *, + rule_cwe: Mapping[str, str] | None = None, +) -> tuple[StaticFinding, ...]: + cwe_by_id = rule_cwe or {} + return tuple( + finding + for finding in findings + if must_keep_as_candidate(finding, _policy_severity_for(finding, policy, cwe_by_id), finding.reachability_status) + ) + + +def _policy_severity_for(finding: StaticFinding, policy: Mapping[str, object], cwe_by_id: Mapping[str, str]) -> int: + if finding.finding_id in policy: + return _as_severity(policy[finding.finding_id]) + cwe = cwe_by_id.get(finding.finding_id) + if cwe is not None and cwe in policy: + return _as_severity(policy[cwe]) + return 0 + + +def _as_severity(value: object) -> int: + if isinstance(value, bool): + return 0 + if isinstance(value, int): + return value + severity = getattr(value, "severity", None) + if isinstance(severity, bool) or not isinstance(severity, int): + return 0 + return severity + + +def _coerce_entry(value: object) -> LedgerEntry | None: + if isinstance(value, LedgerEntry): + return value + if not isinstance(value, Mapping): + return None + raw_delta = value.get("score_delta") + if isinstance(raw_delta, bool) or not isinstance(raw_delta, int | float): + return None + raw_round = value.get("round", 0) + round_n = raw_round if isinstance(raw_round, int) and not isinstance(raw_round, bool) else 0 + return LedgerEntry( + score_delta=float(raw_delta), + last_verdict=str(value.get("last_verdict", "")), + round=round_n, + ) + + +def _band_for(score: float) -> str: + from .map import _band + + return _band(score) diff --git a/src/openultrasast/complexity/map.py b/src/openultrasast/complexity/map.py index 8b8d910..8e3b6b3 100644 --- a/src/openultrasast/complexity/map.py +++ b/src/openultrasast/complexity/map.py @@ -8,6 +8,7 @@ from pathlib import Path from ..findings import StaticFinding from ..preprocess import FileTarget from .hints import TestHint, attach_test_hints +from .ledger import apply_overlay, load_ledger from .signals import ComplexitySignals, collect_signals # Score cutoffs for hotspot bands; inventory density is not the rank driver. @@ -40,11 +41,14 @@ def build_complexity_map( *, repo_files: Iterable[str] = (), sources: Mapping[str, str] | None = None, + ledger_path: Path | None = None, ) -> ComplexityMap: collected = collect_signals(targets, findings, repo_files=repo_files, sources=sources) ids_by_path = _inventory_ids_by_path(findings) hotspots = tuple(_hotspot_from_signals(item, ids_by_path.get(item.path, ())) for item in collected) hotspots = attach_test_hints(hotspots, targets, repo_files=repo_files, sources=sources) + if ledger_path is not None: + hotspots = apply_overlay(hotspots, load_ledger(ledger_path)) complexity_map = ComplexityMap(hotspots=hotspots, heuristic_only=True) write_complexity_map(complexity_map, output_path) return complexity_map diff --git a/tests/test_complexity_ledger.py b/tests/test_complexity_ledger.py new file mode 100644 index 0000000..584bdc3 --- /dev/null +++ b/tests/test_complexity_ledger.py @@ -0,0 +1,260 @@ +import json +from pathlib import Path + +from openultrasast.complexity.ledger import ( + NOT_TRIGGERABLE_DELTA, + TRIGGERABLE_DELTA, + LedgerEntry, + apply_overlay, + hotspot_key, + load_ledger, + must_keep_as_candidate, + record_verdict, + select_forced_candidates, + write_ledger, +) +from openultrasast.complexity.map import Hotspot, build_complexity_map +from openultrasast.findings import StaticFinding +from openultrasast.policy import CwePolicy +from openultrasast.preprocess import FileTarget + + +def _target( + path: str, + *, + loc: int, + tags: list[str] | None = None, + language: str = "python", + hints: list[dict[str, object]] | None = None, + has_fuzz_entry_point: bool = False, +) -> FileTarget: + return FileTarget( + path=path, + absolute_path=f"/repo/{path}", + language=language, + loc=loc, + tags=list(tags or []), + has_fuzz_entry_point=has_fuzz_entry_point, + reachability_hints=list(hints or []), + ) + + +def _hit( + path: str, + finding_id: str, + *, + function_name: str | None = None, + reachability: str = "unknown", + severity: str = "low", + rationale: str = "inventory", +) -> StaticFinding: + return StaticFinding( + finding_id=finding_id, + path=path, + title="pattern hit", + severity=severity, + confidence="medium", + evidence_level="static_corroboration", + rationale=rationale, + 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 test_hotspot_key_joins_path_and_function_or_empty() -> None: + assert hotspot_key("parser.c", "parse_header") == "parser.c:parse_header" + assert hotspot_key("src/app.py", None) == "src/app.py:" + assert hotspot_key("src/app.py", "") == "src/app.py:" + + +def test_triggerable_overlay_raises_later_score() -> None: + hotspot = _hotspot("parser.c", function_name="parse_header", score=4.0, inventory_finding_ids=("hit:1",)) + ledger = record_verdict({}, path="parser.c", function_name="parse_header", verdict="triggerable") + + overlaid = apply_overlay((hotspot,), ledger) + raised = overlaid[0] + + assert ledger[hotspot_key("parser.c", "parse_header")].score_delta == TRIGGERABLE_DELTA + assert TRIGGERABLE_DELTA > 0 + assert raised.score == hotspot.score + TRIGGERABLE_DELTA + assert raised.score > hotspot.score + assert raised.inventory_finding_ids == ("hit:1",) + + +def test_not_triggerable_overlay_drops_score_and_keeps_inventory_hit() -> None: + hotspot = _hotspot("src/auth.py", function_name="login", score=5.0, inventory_finding_ids=("cmd:src/auth.py:12",)) + ledger = { + hotspot_key("src/auth.py", "login"): LedgerEntry( + score_delta=NOT_TRIGGERABLE_DELTA, + last_verdict="not_triggerable", + round=1, + ) + } + + overlaid = apply_overlay((hotspot,), ledger) + demoted = overlaid[0] + + assert NOT_TRIGGERABLE_DELTA < 0 + assert demoted.score == hotspot.score + NOT_TRIGGERABLE_DELTA + assert demoted.score < hotspot.score + assert demoted.inventory_finding_ids == ("cmd:src/auth.py:12",) + assert demoted.path == hotspot.path + assert demoted.function_name == hotspot.function_name + + +def test_not_triggerable_overlay_drops_map_score_and_sev5_reachable_stays_candidate(tmp_path: Path) -> None: + auth = _target( + "src/auth.py", + loc=80, + tags=["auth_boundary"], + hints=[{"kind": "route", "access_level": "public", "function_name": "login"}], + ) + helper = _target("src/helper.py", loc=12, tags=[]) + sev5 = _hit( + "src/auth.py", + "cmd:src/auth.py:12", + function_name="login", + reachability="reachable", + severity="critical", + rationale="Static pattern cmd (CWE-78) matched os.system on line 12; manual verification still required.", + ) + noise = _hit("src/helper.py", "print:src/helper.py:1", reachability="unknown", severity="low") + artifact = tmp_path / "complexity_map.json" + + baseline = build_complexity_map([auth, helper], [sev5, noise], artifact) + original = next(item for item in baseline.hotspots if item.path == "src/auth.py") + assert sev5.finding_id in original.inventory_finding_ids + + ledger_path = tmp_path / ".openultrasast" / "calibration" / "complexity_ledger.json" + write_ledger( + ledger_path, + record_verdict({}, path=original.path, function_name=original.function_name, verdict="not_triggerable"), + ) + + overlaid = build_complexity_map([auth, helper], [sev5, noise], artifact, ledger_path=ledger_path) + demoted = next(item for item in overlaid.hotspots if item.path == "src/auth.py") + + assert demoted.score < original.score + assert demoted.score == original.score + NOT_TRIGGERABLE_DELTA + assert demoted.inventory_finding_ids == original.inventory_finding_ids + assert sev5.finding_id in demoted.inventory_finding_ids + assert sev5.severity == "critical" + assert sev5.rationale == ("Static pattern cmd (CWE-78) matched os.system on line 12; manual verification still required.") + + policy = { + sev5.finding_id: 5, + noise.finding_id: 2, + } + forced = select_forced_candidates([sev5, noise], policy) + assert forced == (sev5,) + assert must_keep_as_candidate(sev5, 5, "reachable") is True + assert must_keep_as_candidate(noise, 2, "unknown") is False + + +def test_must_keep_as_candidate_requires_sev5_and_reachable_surface() -> None: + reachable = _hit("a.py", "a:1", reachability="reachable") + inferred = _hit("b.py", "b:1", reachability="inferred-file-surface") + unknown = _hit("c.py", "c:1", reachability="unknown") + + assert must_keep_as_candidate(reachable, 5, "reachable") is True + assert must_keep_as_candidate(inferred, 5, "inferred-file-surface") is True + assert must_keep_as_candidate(unknown, 5, "unknown") is False + assert must_keep_as_candidate(reachable, 4, "reachable") is False + assert must_keep_as_candidate(reachable, 5, None) is True + + +def test_select_forced_candidates_resolves_policy_severity() -> None: + sev5 = _hit("src/auth.py", "cmd:src/auth.py:12", reachability="reachable") + inferred = _hit("src/upload.py", "sql:src/upload.py:4", reachability="inferred-file-surface") + mild = _hit("src/helper.py", "print:src/helper.py:1", reachability="reachable") + policy = { + "CWE-78": CwePolicy("Command Injection", 5, True, False), + "CWE-89": CwePolicy("SQL Injection", 5, True, False), + "CWE-116": CwePolicy("Encode Output", 2, True, False), + } + + forced = select_forced_candidates( + [mild, sev5, inferred], + policy, + rule_cwe={ + sev5.finding_id: "CWE-78", + inferred.finding_id: "CWE-89", + mild.finding_id: "CWE-116", + }, + ) + + assert forced == (sev5, inferred) + + +def test_ledger_round_trip_matches_design_schema(tmp_path: Path) -> None: + path = tmp_path / "complexity_ledger.json" + key = hotspot_key("parser.c", "parse_header") + write_ledger( + path, + {key: LedgerEntry(score_delta=1.5, last_verdict="triggerable", round=3)}, + ) + payload = json.loads(path.read_text()) + + assert payload == { + key: { + "last_verdict": "triggerable", + "round": 3, + "score_delta": 1.5, + } + } + loaded = load_ledger(path) + assert loaded[key] == LedgerEntry(score_delta=1.5, last_verdict="triggerable", round=3) + assert load_ledger(tmp_path / "missing.json") == {} + assert load_ledger(None) == {} + + +def test_overlay_reorders_and_rebands_without_dropping_hits() -> None: + high = _hotspot("src/auth.py", function_name="login", score=3.2, band="medium", inventory_finding_ids=("sev5:1",)) + other = _hotspot("src/helper.py", score=2.0, band="low", inventory_finding_ids=("low:1",)) + ledger = record_verdict({}, path="src/auth.py", function_name="login", verdict="not_triggerable") + + overlaid = apply_overlay((high, other), ledger) + + assert [item.path for item in overlaid] == ["src/helper.py", "src/auth.py"] + demoted = overlaid[1] + assert demoted.score == round(3.2 + NOT_TRIGGERABLE_DELTA, 4) + assert demoted.band == "low" + assert demoted.inventory_finding_ids == ("sev5:1",) + assert overlaid[0].inventory_finding_ids == ("low:1",) + + +def test_build_complexity_map_without_ledger_path_skips_overlay(tmp_path: Path) -> None: + target = _target("src/app.py", loc=20) + finding = _hit("src/app.py", "rule:src/app.py:1") + artifact = tmp_path / "complexity_map.json" + + first = build_complexity_map([target], [finding], artifact) + second = build_complexity_map([target], [finding], artifact) + + assert first.hotspots[0].score == second.hotspots[0].score + assert first.hotspots[0].inventory_finding_ids == (finding.finding_id,)