From 3d643040bd0d60ce781f035f14598917723a6c52 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 3 Sep 2026 20:05:13 +0200 Subject: [PATCH] feat(three-stage-scan): compute complexity signals for hotspot ranking --- .kiro/specs/three-stage-scan/tasks.md | 2 +- src/openultrasast/complexity/__init__.py | 5 + src/openultrasast/complexity/signals.py | 232 +++++++++++++++++++++++ tests/test_complexity_signals.py | 171 +++++++++++++++++ 4 files changed, 409 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/complexity/__init__.py create mode 100644 src/openultrasast/complexity/signals.py create mode 100644 tests/test_complexity_signals.py diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index d72c30a..0d7485e 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -32,7 +32,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 -- [ ] 2.1 (P) Compute per-file and per-function complexity signals +- [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. - Observable completion: a parser file with no pattern hit scores higher than a one-line helper that matches a pattern, proven by a unit test on a constructed target list. diff --git a/src/openultrasast/complexity/__init__.py b/src/openultrasast/complexity/__init__.py new file mode 100644 index 0000000..a49a7fa --- /dev/null +++ b/src/openultrasast/complexity/__init__.py @@ -0,0 +1,5 @@ +"""Heuristic complexity signals for the stage-2 map.""" + +from .signals import ComplexitySignals, collect_signals + +__all__ = ["ComplexitySignals", "collect_signals"] diff --git a/src/openultrasast/complexity/signals.py b/src/openultrasast/complexity/signals.py new file mode 100644 index 0000000..1bcafa1 --- /dev/null +++ b/src/openultrasast/complexity/signals.py @@ -0,0 +1,232 @@ +from __future__ import annotations + +import math +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path, PurePosixPath + +from ..findings import StaticFinding +from ..preprocess import FileTarget + +# Inventory density is a recorded feature and a capped rationale term, not the rank driver. +_LOC_COEFF = 0.85 +_NESTING_COEFF = 0.2 +_REACHABILITY_COEFF = 0.55 +_TEST_GAP_COEFF = 0.35 +_INVENTORY_COEFF = 0.12 +_TAG_WEIGHTS: dict[str, float] = { + "parser": 4.0, + "deserialization": 3.5, + "network_entry": 1.75, + "auth_boundary": 1.5, + "fuzzable": 1.5, + "memory_unsafe": 1.25, + "crypto": 1.25, + "syscall_entry": 1.25, + "filesystem_entry": 1.0, +} +_TEST_ROOT_PARTS = frozenset({"tests", "test", "__tests__", "spec"}) + + +@dataclass(frozen=True) +class ComplexitySignals: + path: str + function_name: str | None + loc: int + nesting: int + tags: tuple[str, ...] + reachability: int + inventory_hit_count: int + has_adjacent_test: bool + score: float + + +def collect_signals( + targets: Sequence[FileTarget], + findings: Sequence[StaticFinding] = (), + repo_files: Iterable[str] = (), + sources: Mapping[str, str] | None = None, +) -> tuple[ComplexitySignals, ...]: + inventory = _inventory_counts(findings) + repo_paths = {str(path).replace("\\", "/") for path in repo_files} + collected: list[ComplexitySignals] = [] + for target in targets: + text = _source_text(target, sources) + nesting = measure_nesting(text) + hit_count = inventory.get(target.path, 0) + adjacent = _has_adjacent_test(target.path, repo_paths) + tags = tuple(target.tags) + names = _function_names(target.reachability_hints) + identities: tuple[str | None, ...] = names or (None,) + for function_name in identities: + reachability = _reachability(target.reachability_hints, function_name) + collected.append( + ComplexitySignals( + path=target.path, + function_name=function_name, + loc=target.loc, + nesting=nesting, + tags=tags, + reachability=reachability, + inventory_hit_count=hit_count, + has_adjacent_test=adjacent, + score=_score( + loc=target.loc, + nesting=nesting, + tags=tags, + reachability=reachability, + inventory_hit_count=hit_count, + has_adjacent_test=adjacent, + ), + ) + ) + return tuple(sorted(collected, key=lambda item: (-item.score, item.path, item.function_name or ""))) + + +def measure_nesting(text: str) -> int: + if not text: + return 0 + max_indent = 0 + max_braces = 0 + braces = 0 + for raw in text.splitlines(): + if not raw.strip(): + continue + expanded = raw.expandtabs(4) + indent = len(expanded) - len(expanded.lstrip(" ")) + max_indent = max(max_indent, indent // 4) + for char in raw: + if char == "{": + braces += 1 + max_braces = max(max_braces, braces) + elif char == "}": + braces = max(0, braces - 1) + return max(max_indent, max_braces) + + +def _score( + *, + loc: int, + nesting: int, + tags: Sequence[str], + reachability: int, + inventory_hit_count: int, + has_adjacent_test: bool, +) -> float: + loc_term = math.log1p(max(loc, 0)) * _LOC_COEFF + nest_term = min(max(nesting, 0), 16) * _NESTING_COEFF + tag_term = sum(_TAG_WEIGHTS.get(tag, 0.0) for tag in dict.fromkeys(tags)) + reach_term = max(reachability, 0) * _REACHABILITY_COEFF + gap_term = 0.0 if has_adjacent_test else _TEST_GAP_COEFF + density = inventory_hit_count / max(loc, 1) + inventory_term = min(max(density, 0.0), 1.0) * _INVENTORY_COEFF + return round(loc_term + nest_term + tag_term + reach_term + gap_term + inventory_term, 4) + + +def _source_text(target: FileTarget, sources: Mapping[str, str] | None) -> str: + if sources: + if target.path in sources: + return sources[target.path] + if target.absolute_path in sources: + return sources[target.absolute_path] + path = Path(target.absolute_path) + if not path.is_file(): + return "" + try: + return path.read_text(errors="ignore") + except OSError: + return "" + + +def _inventory_counts(findings: Sequence[StaticFinding]) -> dict[str, int]: + counts: dict[str, int] = {} + for finding in findings: + counts[finding.path] = counts.get(finding.path, 0) + 1 + return counts + + +def _function_names(hints: Sequence[Mapping[str, object]]) -> tuple[str, ...]: + names: list[str] = [] + seen: set[str] = set() + for hint in hints: + name = _hint_function_name(hint) + if name is not None and name not in seen: + names.append(name) + seen.add(name) + return tuple(names) + + +def _hint_function_name(hint: Mapping[str, object]) -> str | None: + if "function_name" in hint: + value = hint.get("function_name") + return value if isinstance(value, str) and value else None + value = hint.get("name") + return value if isinstance(value, str) and value else None + + +def _reachability(hints: Sequence[Mapping[str, object]], function_name: str | None) -> int: + selected: Sequence[Mapping[str, object]] = hints + if function_name is not None: + matching = [hint for hint in hints if _hint_function_name(hint) == function_name] + if matching: + selected = matching + if not selected: + return 0 + score = 2 + for hint in selected: + access_level = hint.get("access_level") + kind = hint.get("kind") + if access_level == "public": + score = max(score, 5 if kind in {"route", "parser", "fuzz"} else 4) + elif access_level in {"authenticated", "contract-only/callback"}: + score = max(score, 4) + elif access_level == "role-restricted": + score = max(score, 3) + else: + score = max(score, 2) + return score + + +def _has_adjacent_test(path: str, repo_files: set[str]) -> bool: + source = PurePosixPath(path) + source_stem = source.stem.lower() + source_parent = _parent(source) + for candidate in repo_files: + name = PurePosixPath(candidate).name + if not _is_test_filename(name): + continue + if _tested_stem(name).lower() != source_stem: + continue + parent = _parent(PurePosixPath(candidate)) + if parent == source_parent or _under_common_test_root(candidate): + return True + return False + + +def _parent(path: PurePosixPath) -> str: + parent = path.parent.as_posix() + return "" if parent == "." else parent + + +def _is_test_filename(name: str) -> bool: + if name.startswith("test_") or name.endswith("Test.java"): + return True + return PurePosixPath(name).stem.endswith("_test") + + +def _tested_stem(filename: str) -> str: + if filename.endswith("Test.java"): + return filename[: -len("Test.java")] + stem = PurePosixPath(filename).stem + if stem.startswith("test_"): + return stem[len("test_") :] + if stem.endswith("_test"): + return stem[: -len("_test")] + return stem + + +def _under_common_test_root(path: str) -> bool: + parts = PurePosixPath(path).parts + if len(parts) >= 2 and parts[0] == "src" and parts[1] == "test": + return True + return any(part in _TEST_ROOT_PARTS for part in parts[:-1]) diff --git a/tests/test_complexity_signals.py b/tests/test_complexity_signals.py new file mode 100644 index 0000000..4acb5f0 --- /dev/null +++ b/tests/test_complexity_signals.py @@ -0,0 +1,171 @@ +from pathlib import Path + +from openultrasast.complexity import ComplexitySignals, collect_signals +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 _by_path(signals: tuple[ComplexitySignals, ...]) -> dict[str, ComplexitySignals]: + return {signal.path: signal for signal in signals} + + +def test_parser_without_pattern_hit_outranks_one_line_helper_with_hit() -> None: + 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")] + + signals = collect_signals([parser, helper], findings) + by_path = _by_path(signals) + + assert by_path["src/parser.c"].inventory_hit_count == 0 + assert by_path["src/helper.py"].inventory_hit_count == 1 + assert by_path["src/parser.c"].score > by_path["src/helper.py"].score + assert signals[0].path == "src/parser.c" + + +def test_function_identity_uses_entry_point_names_or_stays_empty() -> None: + named = _target( + "src/parser.c", + loc=40, + language="c", + tags=["parser"], + hints=[{"function_name": "parse_header", "kind": "parser", "access_level": "public"}], + ) + named_alias = _target( + "src/decode.c", + loc=20, + language="c", + tags=["parser"], + hints=[{"name": "decode_frame", "kind": "parser", "access_level": "public"}], + ) + tag_only = _target( + "src/App.java", + loc=80, + language="java", + tags=["parser"], + hints=[{"function_name": None, "name": "parser_input", "kind": "parser", "line": None}], + ) + bare = _target("src/util.py", loc=4, tags=[]) + + signals = collect_signals([named, named_alias, tag_only, bare]) + by_path = _by_path(signals) + + assert by_path["src/parser.c"].function_name == "parse_header" + assert by_path["src/decode.c"].function_name == "decode_frame" + assert by_path["src/App.java"].function_name is None + assert by_path["src/util.py"].function_name is None + + +def test_nesting_from_source_text_otherwise_zero() -> None: + nested = _target("src/nested.py", loc=4, tags=[]) + constructed = _target("src/flat.py", loc=12, tags=["parser"]) + source = "def outer():\n if True:\n while True:\n return 1\n" + + with_text = collect_signals([nested], sources={"src/nested.py": source}) + without_text = collect_signals([constructed]) + + assert with_text[0].nesting >= 3 + assert without_text[0].nesting == 0 + assert without_text[0].loc == 12 + + +def test_nesting_reads_absolute_path_when_file_exists(tmp_path: Path) -> None: + path = tmp_path / "braces.c" + path.write_text("int f(void) {\n if (1) {\n while (1) {\n return 0;\n }\n }\n}\n") + target = _target("braces.c", loc=7, language="c", tags=[], absolute_path=str(path)) + + assert collect_signals([target])[0].nesting >= 3 + + +def test_adjacent_test_from_repo_relative_paths() -> None: + parser = _target("src/parser.c", loc=40, language="c", tags=["parser"]) + helper = _target("pkg/helper.go", loc=8, language="go", tags=[]) + widget = _target("src/main/java/com/acme/Widget.java", loc=30, language="java", tags=[]) + lonely = _target("src/lonely.py", loc=5, tags=[]) + + signals = collect_signals( + [parser, helper, widget, lonely], + repo_files=( + "src/parser.c", + "tests/test_parser.c", + "pkg/helper.go", + "pkg/helper_test.go", + "src/main/java/com/acme/Widget.java", + "src/test/java/com/acme/WidgetTest.java", + "src/lonely.py", + "tests/test_other.py", + ), + ) + by_path = _by_path(signals) + + assert by_path["src/parser.c"].has_adjacent_test is True + assert by_path["pkg/helper.go"].has_adjacent_test is True + assert by_path["src/main/java/com/acme/Widget.java"].has_adjacent_test is True + assert by_path["src/lonely.py"].has_adjacent_test is False + + +def test_inventory_density_is_recorded_but_cannot_dominate_score() -> None: + parser = _target( + "src/parser.c", + loc=250, + language="c", + tags=["parser"], + 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(12)] + + signals = collect_signals([parser, helper], findings) + by_path = _by_path(signals) + + assert by_path["src/helper.py"].inventory_hit_count == 12 + assert by_path["src/parser.c"].inventory_hit_count == 0 + assert by_path["src/parser.c"].score > by_path["src/helper.py"].score + assert "parser" in by_path["src/parser.c"].tags + assert by_path["src/parser.c"].reachability > by_path["src/helper.py"].reachability