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])