diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 0d7485e..a480f9d 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -39,7 +39,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Requirements: 3.1, 3.2_ - _Boundary: ComplexityMapBuilder_ -- [ ] 2.2 (P) Emit the ordered hotspot map artifact +- [x] 2.2 (P) Emit the ordered hotspot map artifact - Each hotspot carries score, band, rationale, signals, and linked inventory ids; the map records whether it is heuristic-only. - Observable completion: building a map writes a JSON artifact whose hotspot order is not identical to sorting by inventory-hit count on the same fixture. - _Requirements: 3.2, 3.3, 3.5, 8.6_ @@ -197,3 +197,4 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - `RegressConfig.images` is a sorted tuple of `(language, image)` pairs so frozen config stays hashable. - `scan_exit_code` duck-types `worth_fixing` on verdict objects; `_run_scan` still passes no verdicts until task 6.5. Dict-shaped verdicts would not fail the gate. - Map `top_k` default is 20 (not named in design.md); `max_hunter_hotspots=8` and `max_candidates=5` follow design. +- `collect_signals` adjacent-test detection is false unless the caller also passes `repo_files`; task 2.2 must pass enumerated repo paths. diff --git a/src/openultrasast/complexity/__init__.py b/src/openultrasast/complexity/__init__.py index a49a7fa..beeb9c8 100644 --- a/src/openultrasast/complexity/__init__.py +++ b/src/openultrasast/complexity/__init__.py @@ -1,5 +1,13 @@ -"""Heuristic complexity signals for the stage-2 map.""" +"""Heuristic complexity map for stage 2.""" +from .map import ComplexityMap, Hotspot, build_complexity_map, write_complexity_map from .signals import ComplexitySignals, collect_signals -__all__ = ["ComplexitySignals", "collect_signals"] +__all__ = [ + "ComplexityMap", + "ComplexitySignals", + "Hotspot", + "build_complexity_map", + "collect_signals", + "write_complexity_map", +] diff --git a/src/openultrasast/complexity/map.py b/src/openultrasast/complexity/map.py new file mode 100644 index 0000000..89a63cd --- /dev/null +++ b/src/openultrasast/complexity/map.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import json +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import asdict, dataclass +from pathlib import Path + +from ..findings import StaticFinding +from ..preprocess import FileTarget +from .signals import ComplexitySignals, collect_signals + +# Score cutoffs for hotspot bands; inventory density is not the rank driver. +_HIGH_SCORE = 6.0 +_MEDIUM_SCORE = 3.0 + + +@dataclass(frozen=True) +class TestHint: + path: str + function_name: str | None + gap: str + test_kind: str | None + reason: str + + +@dataclass(frozen=True) +class Hotspot: + path: str + function_name: str | None + score: float + band: str # high | medium | low + signals: dict[str, float | int | bool | str] + rationale: str + test_hint: TestHint | None + inventory_finding_ids: tuple[str, ...] + + +@dataclass(frozen=True) +class ComplexityMap: + hotspots: tuple[Hotspot, ...] + heuristic_only: bool + + +def build_complexity_map( + targets: Sequence[FileTarget], + findings: Sequence[StaticFinding], + output_path: Path, + *, + repo_files: Iterable[str] = (), + sources: Mapping[str, str] | 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) + complexity_map = ComplexityMap(hotspots=hotspots, heuristic_only=True) + write_complexity_map(complexity_map, output_path) + return complexity_map + + +def write_complexity_map(complexity_map: ComplexityMap, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(asdict(complexity_map), indent=2, sort_keys=True) + "\n") + + +def _hotspot_from_signals(item: ComplexitySignals, inventory_finding_ids: tuple[str, ...]) -> Hotspot: + band = _band(item.score) + return Hotspot( + path=item.path, + function_name=item.function_name, + score=item.score, + band=band, + signals=_signal_payload(item), + rationale=_rationale(item, band), + test_hint=None, + inventory_finding_ids=inventory_finding_ids, + ) + + +def _band(score: float) -> str: + if score >= _HIGH_SCORE: + return "high" + if score >= _MEDIUM_SCORE: + return "medium" + return "low" + + +def _signal_payload(item: ComplexitySignals) -> dict[str, float | int | bool | str]: + return { + "loc": item.loc, + "nesting": item.nesting, + "reachability": item.reachability, + "inventory_hit_count": item.inventory_hit_count, + "has_adjacent_test": item.has_adjacent_test, + "tags": ",".join(item.tags), + } + + +def _rationale(item: ComplexitySignals, band: str) -> str: + tags = ",".join(item.tags) or "none" + return ( + f"{band} band: score {item.score} against cutoffs high>={_HIGH_SCORE}, medium>={_MEDIUM_SCORE}; " + f"loc={item.loc}, nesting={item.nesting}, tags={tags}, reachability={item.reachability}, " + f"inventory_hit_count={item.inventory_hit_count} (density is a feature not the rank driver), " + f"has_adjacent_test={str(item.has_adjacent_test).lower()}." + ) + + +def _inventory_ids_by_path(findings: Sequence[StaticFinding]) -> dict[str, tuple[str, ...]]: + grouped: dict[str, list[str]] = {} + for finding in findings: + grouped.setdefault(finding.path, []).append(finding.finding_id) + return {path: tuple(ids) for path, ids in grouped.items()} diff --git a/tests/test_complexity_map.py b/tests/test_complexity_map.py new file mode 100644 index 0000000..f8b12e6 --- /dev/null +++ b/tests/test_complexity_map.py @@ -0,0 +1,132 @@ +import json +from pathlib import Path + +from openultrasast.complexity.map import build_complexity_map +from openultrasast.findings import StaticFinding +from openultrasast.preprocess import FileTarget + +_HOTSPOT_FIELDS = { + "path", + "function_name", + "score", + "band", + "signals", + "rationale", + "inventory_finding_ids", +} + + +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) -> StaticFinding: + return StaticFinding( + finding_id=finding_id, + path=path, + title="pattern hit", + severity="low", + confidence="medium", + evidence_level="static_corroboration", + rationale="inventory", + line=1, + function_name=None, + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=[], + ranking_priority=1.0, + ) + + +def _parser_helper_fixture() -> tuple[list[FileTarget], list[StaticFinding], tuple[str, ...]]: + parser = _target( + "src/parser.c", + loc=400, + language="c", + tags=["parser", "memory_unsafe"], + hints=[{"kind": "parser", "access_level": "public", "function_name": "parse_header"}], + ) + helper = _target("src/helper.py", loc=1, tags=[]) + findings = [_hit("src/helper.py", f"rule:helper:{index}") for index in range(8)] + repo_files = ("src/parser.c", "src/helper.py", "tests/test_parser.c") + return [parser, helper], findings, repo_files + + +def test_map_order_differs_from_inventory_hit_count(tmp_path: Path) -> None: + targets, findings, repo_files = _parser_helper_fixture() + artifact = tmp_path / "complexity_map.json" + + complexity_map = build_complexity_map(targets, findings, artifact, repo_files=repo_files) + + map_paths = [hotspot.path for hotspot in complexity_map.hotspots] + hit_count_paths = [ + hotspot.path + for hotspot in sorted( + complexity_map.hotspots, + key=lambda item: (-len(item.inventory_finding_ids), item.path), + ) + ] + parser = next(hotspot for hotspot in complexity_map.hotspots if hotspot.path == "src/parser.c") + helper = next(hotspot for hotspot in complexity_map.hotspots if hotspot.path == "src/helper.py") + + assert map_paths != hit_count_paths + assert map_paths[0] == "src/parser.c" + assert parser.inventory_finding_ids == () + assert len(helper.inventory_finding_ids) == 8 + assert helper.inventory_finding_ids == tuple(f"rule:helper:{index}" for index in range(8)) + assert parser.score > helper.score + assert parser.function_name == "parse_header" + assert parser.band == "high" + assert helper.band == "low" + assert complexity_map.heuristic_only is True + assert parser.test_hint is None + assert "verified" not in parser.rationale.lower() + assert "worth fixing" not in parser.rationale.lower() + assert "worth_fixing" not in parser.rationale.lower() + + +def test_complexity_map_json_has_required_fields(tmp_path: Path) -> None: + targets, findings, repo_files = _parser_helper_fixture() + artifact = tmp_path / "runs" / "complexity_map.json" + + complexity_map = build_complexity_map(targets, findings, artifact, repo_files=repo_files) + raw = artifact.read_text() + payload = json.loads(raw) + + assert raw == json.dumps(payload, indent=2, sort_keys=True) + "\n" + assert payload["heuristic_only"] is True + assert [item["path"] for item in payload["hotspots"]] == [hotspot.path for hotspot in complexity_map.hotspots] + for hotspot in payload["hotspots"]: + assert set(hotspot) >= _HOTSPOT_FIELDS + assert hotspot["band"] in {"high", "medium", "low"} + assert isinstance(hotspot["score"], float) + assert isinstance(hotspot["rationale"], str) and hotspot["rationale"] + assert isinstance(hotspot["signals"], dict) + assert set(hotspot["signals"]) >= {"loc", "nesting", "reachability", "inventory_hit_count", "has_adjacent_test"} + assert isinstance(hotspot["inventory_finding_ids"], list) + assert hotspot["test_hint"] is None + assert "6.0" in hotspot["rationale"] + assert "3.0" in hotspot["rationale"] + parser = next(item for item in payload["hotspots"] if item["path"] == "src/parser.c") + helper = next(item for item in payload["hotspots"] if item["path"] == "src/helper.py") + assert parser["signals"]["has_adjacent_test"] is True + assert helper["signals"]["has_adjacent_test"] is False + assert helper["inventory_finding_ids"] == [f"rule:helper:{index}" for index in range(8)] + assert parser["inventory_finding_ids"] == []