From 8d5599ae654fa739c15cb3c2c71613d948d9d598 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 3 Sep 2026 20:33:48 +0200 Subject: [PATCH] feat(three-stage-scan): attach test-gap hints to hotspots --- .kiro/specs/three-stage-scan/tasks.md | 2 +- src/openultrasast/complexity/__init__.py | 3 + src/openultrasast/complexity/hints.py | 219 ++++++++++++++++++++++ src/openultrasast/complexity/map.py | 11 +- tests/test_complexity_hints.py | 227 +++++++++++++++++++++++ tests/test_complexity_map.py | 20 +- 6 files changed, 470 insertions(+), 12 deletions(-) create mode 100644 src/openultrasast/complexity/hints.py create mode 100644 tests/test_complexity_hints.py diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index a480f9d..44a6375 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -46,7 +46,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Boundary: ComplexityMapBuilder_ - _Depends: 2.1_ -- [ ] 2.3 Attach test-gap hints without writing into the scanned tree +- [x] 2.3 Attach test-gap hints without writing into the scanned tree - Each hotspot gets a gap of no adjacent test, no function reference, no fuzz entry, or covered. - Recommended test kind is one of unit, property, sanitizer, http-contract, or fuzz-harness, and is omitted when the gap is covered. - Observable completion: a covered hotspot emits `covered` and no test kind; a fuzzable C parser without a harness emits `fuzz-harness`; the scanned tree is unchanged. diff --git a/src/openultrasast/complexity/__init__.py b/src/openultrasast/complexity/__init__.py index beeb9c8..88e1d52 100644 --- a/src/openultrasast/complexity/__init__.py +++ b/src/openultrasast/complexity/__init__.py @@ -1,5 +1,6 @@ """Heuristic complexity map for stage 2.""" +from .hints import TestHint, attach_test_hints from .map import ComplexityMap, Hotspot, build_complexity_map, write_complexity_map from .signals import ComplexitySignals, collect_signals @@ -7,6 +8,8 @@ __all__ = [ "ComplexityMap", "ComplexitySignals", "Hotspot", + "TestHint", + "attach_test_hints", "build_complexity_map", "collect_signals", "write_complexity_map", diff --git a/src/openultrasast/complexity/hints.py b/src/openultrasast/complexity/hints.py new file mode 100644 index 0000000..f7d93ce --- /dev/null +++ b/src/openultrasast/complexity/hints.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +import re +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass, replace +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING + +from ..preprocess import FileTarget +from .signals import _is_test_filename + +if TYPE_CHECKING: + from .map import Hotspot + +GAP_NO_ADJACENT_TEST = "no_adjacent_test" +GAP_NO_FUNCTION_REFERENCE = "no_function_reference" +GAP_NO_FUZZ_ENTRY = "no_fuzz_entry" +GAP_COVERED = "covered" +GAPS = frozenset({GAP_NO_ADJACENT_TEST, GAP_NO_FUNCTION_REFERENCE, GAP_NO_FUZZ_ENTRY, GAP_COVERED}) + +KIND_UNIT = "unit" +KIND_PROPERTY = "property" +KIND_SANITIZER = "sanitizer" +KIND_HTTP_CONTRACT = "http-contract" +KIND_FUZZ_HARNESS = "fuzz-harness" +TEST_KINDS = frozenset({KIND_UNIT, KIND_PROPERTY, KIND_SANITIZER, KIND_HTTP_CONTRACT, KIND_FUZZ_HARNESS}) + +_MEMORY_UNSAFE_LANGUAGES = frozenset({"c", "cpp"}) + + +@dataclass(frozen=True) +class TestHint: + path: str + function_name: str | None + gap: str # no_adjacent_test | no_function_reference | no_fuzz_entry | covered + test_kind: str | None + reason: str + + def __post_init__(self) -> None: + if self.gap not in GAPS: + raise ValueError(f"unknown test gap: {self.gap}") + if self.test_kind is not None and self.test_kind not in TEST_KINDS: + raise ValueError(f"unknown test kind: {self.test_kind}") + if self.gap == GAP_COVERED: + if self.test_kind is not None: + raise ValueError("covered gap must omit test_kind") + elif self.test_kind is None: + raise ValueError(f"{self.gap} gap requires a test_kind") + + +def attach_test_hints( + hotspots: Sequence[Hotspot], + targets: Sequence[FileTarget] = (), + *, + repo_files: Iterable[str] = (), + sources: Mapping[str, str] | None = None, +) -> tuple[Hotspot, ...]: + targets_by_path = {target.path: target for target in targets} + repo_paths = {str(path).replace("\\", "/") for path in repo_files} + repo_root = _repo_root(targets) + test_files = tuple(path for path in repo_paths if _is_test_filename(PurePosixPath(path).name)) + advised: list[Hotspot] = [] + for hotspot in hotspots: + target = targets_by_path.get(hotspot.path) + hint = advise_test_hint( + path=hotspot.path, + function_name=hotspot.function_name, + has_adjacent_test=bool(hotspot.signals.get("has_adjacent_test")), + language=_language(target, hotspot.path), + tags=_tags(target, hotspot.signals), + has_fuzz_entry_point=_has_fuzz_harness(target, hotspot.path, sources, repo_root), + function_referenced=_function_referenced(hotspot.function_name, test_files, sources, repo_root), + ) + advised.append(replace(hotspot, test_hint=hint)) + return tuple(advised) + + +def advise_test_hint( + *, + path: str, + function_name: str | None, + has_adjacent_test: bool, + language: str, + tags: Sequence[str], + has_fuzz_entry_point: bool, + function_referenced: bool, +) -> TestHint: + tag_set = tuple(tags) + if _needs_fuzz_or_sanitizer(language, tag_set) and not has_fuzz_entry_point: + gap, test_kind = GAP_NO_FUZZ_ENTRY, (KIND_FUZZ_HARNESS if "parser" in tag_set else KIND_SANITIZER) + elif not has_adjacent_test: + gap, test_kind = GAP_NO_ADJACENT_TEST, _kind_for_surface(tag_set) + elif function_name and not function_referenced: + gap, test_kind = GAP_NO_FUNCTION_REFERENCE, _kind_for_surface(tag_set) + else: + gap, test_kind = GAP_COVERED, None + return TestHint( + path=path, + function_name=function_name, + gap=gap, + test_kind=test_kind, + reason=_reason( + gap, + test_kind, + path=path, + function_name=function_name, + language=language, + tags=tag_set, + has_adjacent_test=has_adjacent_test, + ), + ) + + +def _needs_fuzz_or_sanitizer(language: str, tags: Sequence[str]) -> bool: + if language not in _MEMORY_UNSAFE_LANGUAGES: + return False + return "parser" in tags or "memory_unsafe" in tags + + +def _kind_for_surface(tags: Sequence[str]) -> str: + if "network_entry" in tags: + return KIND_HTTP_CONTRACT + if "parser" in tags or "deserialization" in tags: + return KIND_PROPERTY + return KIND_UNIT + + +def _reason( + gap: str, + test_kind: str | None, + *, + path: str, + function_name: str | None, + language: str, + tags: Sequence[str], + has_adjacent_test: bool, +) -> str: + identity = f"{path}::{function_name}" if function_name else path + tag_text = ",".join(tags) or "none" + if gap == GAP_COVERED: + return f"covering tests already exist for {identity}; has_adjacent_test=true." + if gap == GAP_NO_FUZZ_ENTRY: + return f"{language} tags={tag_text} surface {identity} has no sanitizer or fuzz harness; recommend {test_kind}." + if gap == GAP_NO_FUNCTION_REFERENCE: + return f"adjacent tests do not reference {function_name}; recommend {test_kind}." + return f"no adjacent test file for {identity}; has_adjacent_test={str(has_adjacent_test).lower()}; recommend {test_kind}." + + +def _language(target: FileTarget | None, path: str) -> str: + if target is not None: + return target.language + suffix = PurePosixPath(path).suffix.lower() + if suffix in {".c", ".h"}: + return "c" + if suffix in {".cc", ".cpp", ".cxx", ".hpp"}: + return "cpp" + return "unknown" + + +def _tags(target: FileTarget | None, signals: Mapping[str, float | int | bool | str]) -> tuple[str, ...]: + if target is not None: + return tuple(target.tags) + raw = signals.get("tags", "") + if not isinstance(raw, str) or not raw: + return () + return tuple(tag for tag in raw.split(",") if tag) + + +def _has_fuzz_harness( + target: FileTarget | None, + path: str, + sources: Mapping[str, str] | None, + repo_root: Path | None, +) -> bool: + if target is not None and target.has_fuzz_entry_point: + return True + return "LLVMFuzzerTestOneInput" in _lookup_source(path, sources, repo_root) + + +def _function_referenced( + function_name: str | None, + test_files: Sequence[str], + sources: Mapping[str, str] | None, + repo_root: Path | None, +) -> bool: + if not function_name: + return True + pattern = re.compile(rf"\b{re.escape(function_name)}\b") + return any(pattern.search(_lookup_source(path, sources, repo_root)) for path in test_files) + + +def _lookup_source(path: str, sources: Mapping[str, str] | None, repo_root: Path | None) -> str: + normalized = path.replace("\\", "/") + if sources: + if normalized in sources: + return sources[normalized] + if path in sources: + return sources[path] + if repo_root is None: + return "" + candidate = repo_root / normalized + if not candidate.is_file(): + return "" + try: + return candidate.read_text(errors="ignore") + except OSError: + return "" + + +def _repo_root(targets: Sequence[FileTarget]) -> Path | None: + for target in targets: + absolute = Path(target.absolute_path).as_posix() + relative = target.path.replace("\\", "/") + if not relative or not absolute.endswith(relative): + continue + root = Path(absolute[: -len(relative)].rstrip("/")) + if root.exists(): + return root + return None diff --git a/src/openultrasast/complexity/map.py b/src/openultrasast/complexity/map.py index 89a63cd..8b8d910 100644 --- a/src/openultrasast/complexity/map.py +++ b/src/openultrasast/complexity/map.py @@ -7,6 +7,7 @@ from pathlib import Path from ..findings import StaticFinding from ..preprocess import FileTarget +from .hints import TestHint, attach_test_hints from .signals import ComplexitySignals, collect_signals # Score cutoffs for hotspot bands; inventory density is not the rank driver. @@ -14,15 +15,6 @@ _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 @@ -52,6 +44,7 @@ def build_complexity_map( 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) complexity_map = ComplexityMap(hotspots=hotspots, heuristic_only=True) write_complexity_map(complexity_map, output_path) return complexity_map diff --git a/tests/test_complexity_hints.py b/tests/test_complexity_hints.py new file mode 100644 index 0000000..5bc466e --- /dev/null +++ b/tests/test_complexity_hints.py @@ -0,0 +1,227 @@ +import json +from pathlib import Path + +from openultrasast.complexity.hints import GAP_COVERED, KIND_FUZZ_HARNESS +from openultrasast.complexity.map import build_complexity_map +from openultrasast.findings import StaticFinding +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, + absolute_path: str | None = None, +) -> FileTarget: + return FileTarget( + path=path, + absolute_path=absolute_path or 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 = "rule:helper:1") -> 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 _tree_snapshot(root: Path) -> dict[str, tuple[int, bytes]]: + snapshot: dict[str, tuple[int, bytes]] = {} + for path in sorted(root.rglob("*")): + if path.is_file(): + snapshot[path.relative_to(root).as_posix()] = (path.stat().st_mtime_ns, path.read_bytes()) + return snapshot + + +def test_covered_hotspot_emits_covered_and_no_test_kind(tmp_path: Path) -> None: + widget = _target( + "src/widget.py", + loc=12, + hints=[{"kind": "route", "access_level": "public", "function_name": "render"}], + ) + artifact = tmp_path / "complexity_map.json" + + complexity_map = build_complexity_map( + [widget], + [_hit("src/widget.py")], + artifact, + repo_files=("src/widget.py", "tests/test_widget.py"), + sources={ + "src/widget.py": "def render():\n return 1\n", + "tests/test_widget.py": "from widget import render\n\ndef test_render():\n assert render() == 1\n", + }, + ) + + hotspot = complexity_map.hotspots[0] + assert hotspot.test_hint is not None + assert hotspot.test_hint.gap == GAP_COVERED + assert hotspot.test_hint.gap == "covered" + assert hotspot.test_hint.test_kind is None + assert hotspot.test_hint.path == "src/widget.py" + assert hotspot.test_hint.function_name == "render" + payload = json.loads(artifact.read_text()) + hint = payload["hotspots"][0]["test_hint"] + assert hint["gap"] == "covered" + assert hint["test_kind"] is None + + +def test_fuzzable_c_parser_without_harness_emits_fuzz_harness(tmp_path: Path) -> None: + parser = _target( + "src/parser.c", + loc=400, + language="c", + tags=["parser", "memory_unsafe"], + hints=[{"kind": "parser", "access_level": "public", "function_name": "parse_header"}], + has_fuzz_entry_point=False, + ) + artifact = tmp_path / "complexity_map.json" + + complexity_map = build_complexity_map( + [parser], + [], + artifact, + repo_files=("src/parser.c", "tests/test_parser.c"), + sources={ + "src/parser.c": "int parse_header(const char *buf) { return 0; }\n", + "tests/test_parser.c": 'int test_parse_header(void) { return parse_header(""); }\n', + }, + ) + + hotspot = complexity_map.hotspots[0] + assert hotspot.test_hint is not None + assert hotspot.test_hint.gap == "no_fuzz_entry" + assert hotspot.test_hint.test_kind == KIND_FUZZ_HARNESS + assert hotspot.test_hint.test_kind == "fuzz-harness" + assert hotspot.test_hint.path == "src/parser.c" + assert hotspot.test_hint.function_name == "parse_header" + assert hotspot.test_hint.reason + payload = json.loads(artifact.read_text()) + hint = payload["hotspots"][0]["test_hint"] + assert hint["gap"] == "no_fuzz_entry" + assert hint["test_kind"] == "fuzz-harness" + + +def test_attach_test_hints_leaves_scanned_tree_unchanged(tmp_path: Path) -> None: + repo = tmp_path / "repo" + src = repo / "src" + tests_dir = repo / "tests" + src.mkdir(parents=True) + tests_dir.mkdir() + parser = src / "parser.c" + helper = src / "helper.py" + parser_test = tests_dir / "test_parser.c" + parser.write_text("int parse_header(const char *buf) {\n return buf[0];\n}\n") + helper.write_text("def ping():\n return 1\n") + parser_test.write_text("int test_parse_header(void) { return 0; }\n") + before = _tree_snapshot(repo) + + artifact = tmp_path / "out" / "complexity_map.json" + targets = [ + _target( + "src/parser.c", + loc=3, + language="c", + tags=["parser", "memory_unsafe"], + hints=[{"kind": "parser", "access_level": "public", "function_name": "parse_header"}], + absolute_path=str(parser), + ), + _target("src/helper.py", loc=2, tags=[], absolute_path=str(helper)), + ] + + complexity_map = build_complexity_map( + targets, + [_hit("src/helper.py")], + artifact, + repo_files=("src/parser.c", "src/helper.py", "tests/test_parser.c"), + sources={ + "src/parser.c": parser.read_text(), + "src/helper.py": helper.read_text(), + "tests/test_parser.c": parser_test.read_text(), + }, + ) + + assert complexity_map.hotspots + assert all(hotspot.test_hint is not None for hotspot in complexity_map.hotspots) + assert _tree_snapshot(repo) == before + assert set(_tree_snapshot(repo)) == {"src/parser.c", "src/helper.py", "tests/test_parser.c"} + assert not any(path.as_posix().endswith(".py") and "test" in path.name for path in src.rglob("*")) + assert artifact.is_file() + assert artifact.is_relative_to(tmp_path / "out") + + +def test_no_adjacent_test_recommends_closed_test_kind(tmp_path: Path) -> None: + helper = _target("src/helper.py", loc=4, tags=[]) + route = _target( + "src/api.py", + loc=40, + tags=["network_entry"], + hints=[{"kind": "route", "access_level": "public", "function_name": "handle"}], + ) + artifact = tmp_path / "complexity_map.json" + + complexity_map = build_complexity_map( + [helper, route], + [], + artifact, + repo_files=("src/helper.py", "src/api.py"), + sources={ + "src/helper.py": "def ping():\n return 1\n", + "src/api.py": "def handle():\n return {}\n", + }, + ) + by_path = {hotspot.path: hotspot for hotspot in complexity_map.hotspots} + + assert by_path["src/helper.py"].test_hint is not None + assert by_path["src/helper.py"].test_hint.gap == "no_adjacent_test" + assert by_path["src/helper.py"].test_hint.test_kind == "unit" + assert by_path["src/api.py"].test_hint is not None + assert by_path["src/api.py"].test_hint.gap == "no_adjacent_test" + assert by_path["src/api.py"].test_hint.test_kind == "http-contract" + + +def test_no_function_reference_when_adjacent_test_omits_function(tmp_path: Path) -> None: + widget = _target( + "src/widget.py", + loc=12, + hints=[{"kind": "route", "access_level": "public", "function_name": "render"}], + ) + artifact = tmp_path / "complexity_map.json" + + complexity_map = build_complexity_map( + [widget], + [], + artifact, + repo_files=("src/widget.py", "tests/test_widget.py"), + sources={ + "src/widget.py": "def render():\n return 1\n", + "tests/test_widget.py": "def test_placeholder():\n assert True\n", + }, + ) + + hint = complexity_map.hotspots[0].test_hint + assert hint is not None + assert hint.gap == "no_function_reference" + assert hint.test_kind == "unit" + assert hint.function_name == "render" diff --git a/tests/test_complexity_map.py b/tests/test_complexity_map.py index f8b12e6..2c55e43 100644 --- a/tests/test_complexity_map.py +++ b/tests/test_complexity_map.py @@ -12,6 +12,7 @@ _HOTSPOT_FIELDS = { "band", "signals", "rationale", + "test_hint", "inventory_finding_ids", } @@ -96,7 +97,12 @@ def test_map_order_differs_from_inventory_hit_count(tmp_path: Path) -> None: assert parser.band == "high" assert helper.band == "low" assert complexity_map.heuristic_only is True - assert parser.test_hint is None + assert parser.test_hint is not None + assert parser.test_hint.gap == "no_fuzz_entry" + assert parser.test_hint.test_kind == "fuzz-harness" + assert helper.test_hint is not None + assert helper.test_hint.gap == "no_adjacent_test" + assert helper.test_hint.test_kind == "unit" assert "verified" not in parser.rationale.lower() assert "worth fixing" not in parser.rationale.lower() assert "worth_fixing" not in parser.rationale.lower() @@ -121,12 +127,22 @@ def test_complexity_map_json_has_required_fields(tmp_path: Path) -> None: 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 + hint = hotspot["test_hint"] + assert isinstance(hint, dict) + assert hint["gap"] in {"no_adjacent_test", "no_function_reference", "no_fuzz_entry", "covered"} + if hint["gap"] == "covered": + assert hint["test_kind"] is None + else: + assert hint["test_kind"] in {"unit", "property", "sanitizer", "http-contract", "fuzz-harness"} 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 parser["test_hint"]["gap"] == "no_fuzz_entry" + assert parser["test_hint"]["test_kind"] == "fuzz-harness" + assert helper["test_hint"]["gap"] == "no_adjacent_test" + assert helper["test_hint"]["test_kind"] == "unit" assert helper["inventory_finding_ids"] == [f"rule:helper:{index}" for index in range(8)] assert parser["inventory_finding_ids"] == []