696 lines
25 KiB
Python
696 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import fnmatch
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from collections import Counter
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from time import perf_counter
|
|
from uuid import uuid4
|
|
|
|
from .architecture import build_architecture
|
|
from .git_analysis import analyze_git
|
|
from .models import (
|
|
Finding,
|
|
RepositoryMetrics,
|
|
RepositoryPolicy,
|
|
ScanCoverage,
|
|
ScanLimits,
|
|
ScanReport,
|
|
ScanSource,
|
|
Severity,
|
|
finding_fingerprint,
|
|
)
|
|
from .policy import load_policy
|
|
from .release_gate import build_release_gate
|
|
from .rules import enrich_findings
|
|
from .signals import analyze_dependencies, detect_test_signals
|
|
|
|
LANGUAGES = {
|
|
".py": "Python",
|
|
".ts": "TypeScript",
|
|
".tsx": "TypeScript",
|
|
".js": "JavaScript",
|
|
".jsx": "JavaScript",
|
|
".mjs": "JavaScript",
|
|
".cjs": "JavaScript",
|
|
".cs": "C#",
|
|
".java": "Java",
|
|
".go": "Go",
|
|
".rs": "Rust",
|
|
".cpp": "C++",
|
|
".cc": "C++",
|
|
".c": "C",
|
|
".h": "C/C++ Header",
|
|
".hpp": "C++ Header",
|
|
".php": "PHP",
|
|
".rb": "Ruby",
|
|
".kt": "Kotlin",
|
|
".swift": "Swift",
|
|
".vue": "Vue",
|
|
".svelte": "Svelte",
|
|
".html": "HTML",
|
|
".css": "CSS",
|
|
".scss": "SCSS",
|
|
".sh": "Shell",
|
|
".ps1": "PowerShell",
|
|
".sql": "SQL",
|
|
}
|
|
|
|
IGNORED_DIRS = {
|
|
".git",
|
|
".idea",
|
|
".vscode",
|
|
".venv",
|
|
"venv",
|
|
"node_modules",
|
|
"dist",
|
|
"build",
|
|
"coverage",
|
|
".next",
|
|
".nuxt",
|
|
".pytest_cache",
|
|
"__pycache__",
|
|
"vendor",
|
|
"target",
|
|
"bin",
|
|
"obj",
|
|
}
|
|
|
|
MAX_TEXT_BYTES = 2_000_000
|
|
MAX_SECRET_FINDINGS = 20
|
|
|
|
DEBT_PATTERN = re.compile(r"\b(TODO|FIXME|HACK|XXX)\b")
|
|
SECRET_PATTERN = re.compile(
|
|
r"""(?ix)
|
|
(?:
|
|
^\s*(?:export\s+)?(?:(?:const|let|var|final|static|private|public|protected|internal)\s+)?(?:[A-Za-z_]\w*\.)*
|
|
|[,{]\s*
|
|
)
|
|
[\"']?(?P<name>api[_-]?key|secret|password|passwd|token|access[_-]?key)[\"']?
|
|
\s*[:=]\s*[\"']?(?P<value>[^\s\"']{8,})
|
|
"""
|
|
)
|
|
PLACEHOLDER_SECRET_MARKERS = ("example", "placeholder", "dummy", "changeme", "not-a-real", "replace-me", "your-", "fake", "test-")
|
|
SECRET_REFERENCE_MARKERS = (
|
|
"os.environ",
|
|
"os.getenv",
|
|
"getenv",
|
|
"process.env",
|
|
"env[",
|
|
"config.",
|
|
"settings.",
|
|
)
|
|
SECRET_SCAN_CONFIG_SUFFIXES = {".json", ".yaml", ".yml", ".toml", ".ini", ".cfg", ".conf", ".md", ".txt"}
|
|
CATEGORY_NAMES = (
|
|
"Architecture",
|
|
"Testing",
|
|
"Maintainability",
|
|
"Documentation",
|
|
"Dependencies",
|
|
"Repository hygiene",
|
|
"Security basics",
|
|
)
|
|
|
|
|
|
def _ignored(relative: str) -> bool:
|
|
return any(part.casefold() in IGNORED_DIRS for part in Path(relative).parts)
|
|
|
|
|
|
def _is_env_template(name: str) -> bool:
|
|
return name == ".env.example" or (name.startswith(".env.") and name.endswith(".example"))
|
|
|
|
|
|
def _is_secret_scan_candidate(path_obj: Path, relative: str, language: str | None) -> bool:
|
|
return language is not None or path_obj.suffix.casefold() in SECRET_SCAN_CONFIG_SUFFIXES or path_obj.name.casefold().startswith(".env")
|
|
|
|
|
|
def _bounded_env_int(name: str, default: int, *, minimum: int, maximum: int) -> int:
|
|
try:
|
|
value = int(os.getenv(name, str(default)))
|
|
except ValueError:
|
|
value = default
|
|
return min(maximum, max(minimum, value))
|
|
|
|
|
|
def scan_limits_from_environment() -> ScanLimits:
|
|
"""Return bounded analysis limits; input cannot turn a scan into an unbounded job."""
|
|
|
|
return ScanLimits(
|
|
max_files=_bounded_env_int("REPOPULSE_SCAN_MAX_FILES", 10_000, minimum=1, maximum=100_000),
|
|
max_text_bytes_per_file=_bounded_env_int(
|
|
"REPOPULSE_SCAN_MAX_TEXT_BYTES", MAX_TEXT_BYTES, minimum=1_024, maximum=10_000_000
|
|
),
|
|
max_total_text_bytes=_bounded_env_int(
|
|
"REPOPULSE_SCAN_MAX_TOTAL_TEXT_BYTES", 50_000_000, minimum=1_000_000, maximum=500_000_000
|
|
),
|
|
max_findings=_bounded_env_int("REPOPULSE_SCAN_MAX_FINDINGS", 2_000, minimum=10, maximum=20_000),
|
|
max_secret_findings=_bounded_env_int(
|
|
"REPOPULSE_SCAN_MAX_SECRET_FINDINGS", MAX_SECRET_FINDINGS, minimum=1, maximum=500
|
|
),
|
|
)
|
|
|
|
|
|
class _GitIgnoreMatcher:
|
|
"""Bounded root/nested .gitignore support for non-Git directory scans."""
|
|
|
|
def __init__(self, root: Path) -> None:
|
|
self.rules: list[tuple[str, str, bool, bool, bool]] = []
|
|
for current, dirs, names in os.walk(root, followlinks=False):
|
|
dirs[:] = [
|
|
name
|
|
for name in dirs
|
|
if name.casefold() not in IGNORED_DIRS and not (Path(current) / name).is_symlink()
|
|
]
|
|
if ".gitignore" not in names:
|
|
continue
|
|
ignore_file = Path(current) / ".gitignore"
|
|
try:
|
|
lines = ignore_file.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
base_path = Path(current).relative_to(root)
|
|
base = "" if str(base_path) == "." else base_path.as_posix()
|
|
except (OSError, ValueError):
|
|
continue
|
|
for raw in lines:
|
|
pattern = raw.strip()
|
|
if not pattern or pattern.startswith("#"):
|
|
continue
|
|
negated = pattern.startswith("!")
|
|
if negated:
|
|
pattern = pattern[1:]
|
|
pattern = pattern.replace("\\", "/")
|
|
anchored = pattern.startswith("/")
|
|
pattern = pattern.lstrip("/")
|
|
dir_only = pattern.endswith("/")
|
|
pattern = pattern.rstrip("/")
|
|
if pattern:
|
|
self.rules.append((base, pattern, negated, dir_only, anchored))
|
|
|
|
def is_ignored(self, relative: str, *, is_dir: bool) -> bool:
|
|
normalized = relative.replace("\\", "/").strip("/")
|
|
ignored = False
|
|
for base, pattern, negated, dir_only, anchored in self.rules:
|
|
if dir_only and not is_dir:
|
|
continue
|
|
if base:
|
|
if normalized == base:
|
|
target = ""
|
|
elif normalized.startswith(f"{base}/"):
|
|
target = normalized[len(base) + 1 :]
|
|
else:
|
|
continue
|
|
else:
|
|
target = normalized
|
|
if not target:
|
|
continue
|
|
if "/" not in pattern:
|
|
matched = (
|
|
fnmatch.fnmatch(target, pattern)
|
|
if anchored
|
|
else any(fnmatch.fnmatch(part, pattern) for part in target.split("/")
|
|
)
|
|
)
|
|
else:
|
|
matched = fnmatch.fnmatch(target, pattern)
|
|
if not matched and pattern.startswith("**/"):
|
|
matched = fnmatch.fnmatch(target, pattern[3:])
|
|
if matched:
|
|
ignored = not negated
|
|
return ignored
|
|
|
|
|
|
def _git_worktree_root(root: Path) -> Path | None:
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "rev-parse", "--show-toplevel"],
|
|
check=True,
|
|
capture_output=True,
|
|
timeout=5,
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
try:
|
|
return Path(result.stdout.decode("utf-8", errors="surrogateescape").strip()).resolve()
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def _safe_file(root: Path, candidate: Path) -> Path | None:
|
|
"""Return a resolved regular file only when every path component is safe."""
|
|
try:
|
|
relative = candidate.relative_to(root)
|
|
if _ignored(str(relative)):
|
|
return None
|
|
current = root
|
|
for part in Path(relative).parts:
|
|
current /= part
|
|
if current.is_symlink():
|
|
return None
|
|
resolved = candidate.resolve(strict=True)
|
|
root_resolved = root.resolve()
|
|
if resolved == root_resolved or root_resolved not in resolved.parents:
|
|
return None
|
|
if not resolved.is_file():
|
|
return None
|
|
except (OSError, RuntimeError, ValueError):
|
|
return None
|
|
return resolved
|
|
|
|
|
|
def _tracked_files(root: Path) -> list[Path] | None:
|
|
if _git_worktree_root(root) != root:
|
|
return None
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "ls-files", "-z"],
|
|
check=True,
|
|
capture_output=True,
|
|
timeout=5,
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
|
|
paths: list[Path] = []
|
|
for raw in result.stdout.split(b"\0"):
|
|
if not raw:
|
|
continue
|
|
relative = raw.decode("utf-8", errors="surrogateescape")
|
|
relative_path = Path(relative)
|
|
if relative_path.is_absolute() or _ignored(relative):
|
|
continue
|
|
safe = _safe_file(root, root / relative_path)
|
|
if safe is not None:
|
|
paths.append(safe)
|
|
return paths
|
|
|
|
|
|
def _walk_files(root: Path) -> list[Path]:
|
|
files: list[Path] = []
|
|
gitignore = _GitIgnoreMatcher(root)
|
|
for current, dirs, names in os.walk(root, followlinks=False):
|
|
current_path = Path(current)
|
|
dirs[:] = [
|
|
name
|
|
for name in dirs
|
|
if name.casefold() not in IGNORED_DIRS
|
|
and not (current_path / name).is_symlink()
|
|
and not gitignore.is_ignored(
|
|
str((current_path / name).relative_to(root)),
|
|
is_dir=True,
|
|
)
|
|
]
|
|
for name in names:
|
|
candidate = current_path / name
|
|
try:
|
|
relative = str(candidate.relative_to(root))
|
|
except (OSError, ValueError):
|
|
continue
|
|
if _ignored(relative):
|
|
continue
|
|
if gitignore.is_ignored(relative, is_dir=False):
|
|
continue
|
|
safe = _safe_file(root, candidate)
|
|
if safe is not None:
|
|
files.append(safe)
|
|
return files
|
|
|
|
|
|
def _read_text(path: Path, max_bytes: int = MAX_TEXT_BYTES) -> str | None:
|
|
try:
|
|
if path.stat().st_size > max_bytes:
|
|
return None
|
|
raw = path.read_bytes()
|
|
except OSError:
|
|
return None
|
|
if b"\0" in raw[:8192]:
|
|
return None
|
|
return raw.decode("utf-8", errors="replace")
|
|
|
|
|
|
def _is_test_file(relative: str) -> bool:
|
|
normalized = relative.replace("\\", "/").lower()
|
|
name = Path(normalized).name
|
|
return (
|
|
"/tests/" in f"/{normalized}"
|
|
or "/test/" in f"/{normalized}"
|
|
or name.startswith("test_")
|
|
or ".test." in name
|
|
or ".spec." in name
|
|
or name.endswith(("tests.cs", "test.cs", "_test.go"))
|
|
)
|
|
|
|
|
|
def _python_function_findings(relative: str, text: str) -> list[Finding]:
|
|
try:
|
|
tree = ast.parse(text)
|
|
except SyntaxError:
|
|
return []
|
|
findings: list[Finding] = []
|
|
branch_types = (ast.If, ast.For, ast.AsyncFor, ast.While, ast.Try, ast.Match, ast.BoolOp)
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
continue
|
|
end_line = getattr(node, "end_lineno", node.lineno)
|
|
length = max(1, end_line - node.lineno + 1)
|
|
branches = sum(1 for child in ast.walk(node) if isinstance(child, branch_types))
|
|
if length >= 140:
|
|
findings.append(
|
|
Finding(
|
|
code="maintainability.long-function",
|
|
category="Maintainability",
|
|
severity=Severity.WARNING,
|
|
message=f"Function {node.name} spans {length} lines",
|
|
path=relative,
|
|
line=node.lineno,
|
|
deduction=4,
|
|
)
|
|
)
|
|
if branches >= 20:
|
|
findings.append(
|
|
Finding(
|
|
code="maintainability.branch-heavy-function",
|
|
category="Maintainability",
|
|
severity=Severity.WARNING,
|
|
message=f"Function {node.name} contains {branches} branch constructs",
|
|
path=relative,
|
|
line=node.lineno,
|
|
deduction=4,
|
|
)
|
|
)
|
|
return findings[:12]
|
|
|
|
|
|
def _is_placeholder_secret(match: re.Match[str]) -> bool:
|
|
value = match.group("value").casefold().strip("`$(){}[]<>")
|
|
exact_placeholder = value in {"", "none", "null", "undefined", "xxx", "xxxxx"}
|
|
obvious_placeholder = any(value.startswith(marker) for marker in PLACEHOLDER_SECRET_MARKERS)
|
|
return exact_placeholder or obvious_placeholder or any(
|
|
value.startswith(marker) for marker in SECRET_REFERENCE_MARKERS
|
|
)
|
|
|
|
|
|
def scan_repository(
|
|
path: str,
|
|
*,
|
|
source: ScanSource | None = None,
|
|
display_name: str | None = None,
|
|
policy: RepositoryPolicy | None = None,
|
|
limits: ScanLimits | None = None,
|
|
) -> ScanReport:
|
|
started = perf_counter()
|
|
root = Path(path).expanduser().resolve()
|
|
if not root.exists() or not root.is_dir():
|
|
raise ValueError("Repository path does not exist or is not a directory")
|
|
|
|
limits = limits or scan_limits_from_environment()
|
|
tracked_files = _tracked_files(root)
|
|
all_files = _walk_files(root) if tracked_files is None else tracked_files
|
|
all_files.sort(key=lambda item: str(item.relative_to(root)).replace("\\", "/").casefold())
|
|
files = all_files[: limits.max_files]
|
|
coverage = ScanCoverage(
|
|
limits=limits,
|
|
discovered_files=len(all_files),
|
|
skipped_files_by_budget=max(0, len(all_files) - len(files)),
|
|
truncated=len(all_files) > len(files),
|
|
notes=(
|
|
[f"File analysis limited to {limits.max_files} of {len(all_files)} discovered safe files."]
|
|
if len(all_files) > len(files)
|
|
else []
|
|
),
|
|
)
|
|
languages: Counter[str] = Counter()
|
|
findings: list[Finding] = []
|
|
metrics = RepositoryMetrics(files=len(all_files))
|
|
files_by_name: dict[str, list[Path]] = {}
|
|
root_entries = {entry.name.lower() for entry in root.iterdir()}
|
|
text_by_path: dict[str, str] = {}
|
|
source_texts: dict[str, str] = {}
|
|
language_by_path: dict[str, str] = {}
|
|
loc_by_path: dict[str, int] = {}
|
|
debt_by_path: dict[str, int] = {}
|
|
secret_findings = 0
|
|
secret_findings_truncated = False
|
|
text_budget_exhausted = False
|
|
|
|
for path_obj in files:
|
|
try:
|
|
relative = str(path_obj.relative_to(root)).replace("\\", "/")
|
|
size = path_obj.stat().st_size
|
|
except (OSError, ValueError):
|
|
continue
|
|
|
|
files_by_name.setdefault(path_obj.name, []).append(path_obj)
|
|
suffix = path_obj.suffix.lower()
|
|
language = LANGUAGES.get(suffix)
|
|
|
|
if size >= 1_000_000:
|
|
metrics.large_files += 1
|
|
findings.append(
|
|
Finding(
|
|
code="hygiene.large-file",
|
|
category="Repository hygiene",
|
|
severity=Severity.INFO,
|
|
message=f"Large tracked file ({size / 1_000_000:.1f} MB)",
|
|
path=relative,
|
|
deduction=1,
|
|
)
|
|
)
|
|
|
|
name = path_obj.name.casefold()
|
|
if name == ".env" or (name.startswith(".env.") and not _is_env_template(name)):
|
|
findings.append(
|
|
Finding(
|
|
code="security.env-committed",
|
|
category="Security basics",
|
|
severity=Severity.BLOCKER,
|
|
message="Environment file is tracked; review it for credentials",
|
|
path=relative,
|
|
deduction=30,
|
|
)
|
|
)
|
|
|
|
if size > limits.max_text_bytes_per_file:
|
|
coverage.skipped_files_by_size += 1
|
|
continue
|
|
if coverage.text_bytes_scanned + size > limits.max_total_text_bytes:
|
|
coverage.skipped_files_by_text_budget += 1
|
|
text_budget_exhausted = True
|
|
continue
|
|
text = _read_text(path_obj, limits.max_text_bytes_per_file)
|
|
if text is None:
|
|
continue
|
|
coverage.scanned_files += 1
|
|
coverage.text_bytes_scanned += len(text.encode("utf-8", errors="replace"))
|
|
text_by_path[relative] = text
|
|
|
|
if language:
|
|
loc = sum(1 for line in text.splitlines() if line.strip())
|
|
metrics.source_files += 1
|
|
metrics.lines_of_code += loc
|
|
languages[language] += loc
|
|
source_texts[relative] = text
|
|
language_by_path[relative] = language
|
|
loc_by_path[relative] = loc
|
|
if loc > 1000:
|
|
metrics.large_source_files += 1
|
|
findings.append(
|
|
Finding(
|
|
code="maintainability.large-source",
|
|
category="Maintainability",
|
|
severity=Severity.WARNING,
|
|
message=f"Source file has {loc} non-empty lines",
|
|
path=relative,
|
|
deduction=5,
|
|
)
|
|
)
|
|
if language == "Python":
|
|
findings.extend(_python_function_findings(relative, text))
|
|
|
|
if _is_test_file(relative):
|
|
metrics.test_files += 1
|
|
|
|
debt = Counter(DEBT_PATTERN.findall(text))
|
|
metrics.todo += debt["TODO"]
|
|
metrics.fixme += debt["FIXME"]
|
|
metrics.hack += debt["HACK"]
|
|
metrics.xxx += debt["XXX"]
|
|
debt_by_path[relative] = sum(debt.values())
|
|
|
|
if _is_secret_scan_candidate(path_obj, relative, language) and secret_findings < limits.max_secret_findings:
|
|
for line_number, line in enumerate(text.splitlines(), start=1):
|
|
match = SECRET_PATTERN.search(line)
|
|
if match is None or _is_placeholder_secret(match):
|
|
continue
|
|
findings.append(
|
|
Finding(
|
|
code="security.secret-assignment",
|
|
category="Security basics",
|
|
severity=Severity.BLOCKER,
|
|
message="Potential credential assignment detected; value intentionally redacted",
|
|
path=relative,
|
|
line=line_number,
|
|
deduction=20,
|
|
)
|
|
)
|
|
secret_findings += 1
|
|
if secret_findings >= limits.max_secret_findings:
|
|
secret_findings_truncated = True
|
|
break
|
|
|
|
if text_budget_exhausted:
|
|
coverage.truncated = True
|
|
coverage.notes.append(
|
|
f"Text analysis limited to {limits.max_total_text_bytes} bytes across repository files."
|
|
)
|
|
|
|
if secret_findings_truncated:
|
|
findings.append(
|
|
Finding(
|
|
code="security.secret-findings-truncated",
|
|
category="Security basics",
|
|
severity=Severity.INFO,
|
|
message=f"Potential credential findings are capped at {limits.max_secret_findings}",
|
|
deduction=0,
|
|
)
|
|
)
|
|
|
|
if not any(name.startswith("readme") for name in root_entries):
|
|
findings.append(Finding(code="docs.missing-readme", category="Documentation", severity=Severity.WARNING, message="No root README detected", deduction=20))
|
|
if not any(name.startswith(("license", "copying")) for name in root_entries):
|
|
findings.append(Finding(code="docs.missing-license", category="Documentation", severity=Severity.INFO, message="No root license file detected", deduction=5))
|
|
if ".gitignore" not in root_entries:
|
|
findings.append(Finding(code="hygiene.missing-gitignore", category="Repository hygiene", severity=Severity.WARNING, message="No root .gitignore detected", deduction=15))
|
|
if not any(_is_env_template(name) for name in root_entries) and any(name.startswith(".env") for name in root_entries):
|
|
findings.append(Finding(code="hygiene.missing-env-example", category="Repository hygiene", severity=Severity.INFO, message="Environment configuration exists without a root .env.example", deduction=3))
|
|
|
|
test_signals = detect_test_signals(
|
|
[str(path.relative_to(root)).replace("\\", "/") for path in files],
|
|
metrics.source_files,
|
|
metrics.test_files,
|
|
text_by_path,
|
|
)
|
|
if metrics.source_files and metrics.test_files == 0:
|
|
findings.append(Finding(code="testing.none-detected", category="Testing", severity=Severity.WARNING, message="No test files detected", deduction=30))
|
|
if not test_signals.ci_workflows:
|
|
findings.append(Finding(code="testing.no-ci", category="Testing", severity=Severity.WARNING, message="No recognized CI workflow detected", deduction=15))
|
|
if metrics.test_files and not test_signals.coverage_artifacts:
|
|
findings.append(Finding(code="testing.no-coverage-artifact", category="Testing", severity=Severity.INFO, message="Tests are present but no coverage artifact is tracked", deduction=2))
|
|
|
|
common_source_roots = {"src", "app", "backend", "frontend", "lib", "packages"}
|
|
if metrics.source_files >= 10 and not (root_entries & common_source_roots):
|
|
findings.append(Finding(code="architecture.flat-layout", category="Architecture", severity=Severity.INFO, message="No conventional source root detected in a non-trivial repository", deduction=8))
|
|
|
|
debt_total = metrics.todo + metrics.fixme + metrics.hack + metrics.xxx
|
|
if debt_total:
|
|
findings.append(Finding(code="maintainability.debt-markers", category="Maintainability", severity=Severity.INFO, message=f"Detected {debt_total} TODO/FIXME/HACK/XXX markers", deduction=min(20, max(1, debt_total // 5))))
|
|
|
|
dependency_signals, dependency_findings = analyze_dependencies(root, files_by_name)
|
|
metrics.dependency_manifests = len(dependency_signals.manifests)
|
|
metrics.declared_dependencies = dependency_signals.declared_dependencies
|
|
findings.extend(dependency_findings)
|
|
|
|
git_signals, churn_by_path = analyze_git(root)
|
|
architecture, hotspots = build_architecture(
|
|
root,
|
|
source_texts,
|
|
language_by_path,
|
|
loc_by_path,
|
|
debt_by_path,
|
|
churn_by_path,
|
|
)
|
|
for hotspot in hotspots[:5]:
|
|
if hotspot.score >= 70:
|
|
findings.append(
|
|
Finding(
|
|
code="maintainability.hotspot",
|
|
category="Maintainability",
|
|
severity=Severity.WARNING,
|
|
message=f"Engineering hotspot score {hotspot.score}/100: {', '.join(hotspot.reasons) or 'combined risk signals'}",
|
|
path=hotspot.path,
|
|
deduction=4,
|
|
)
|
|
)
|
|
|
|
if architecture.truncated:
|
|
findings.append(
|
|
Finding(
|
|
code="architecture.graph-truncated",
|
|
category="Architecture",
|
|
severity=Severity.INFO,
|
|
message="Architecture graph was bounded to keep analysis responsive",
|
|
deduction=0,
|
|
)
|
|
)
|
|
|
|
severity_rank = {Severity.BLOCKER: 0, Severity.WARNING: 1, Severity.INFO: 2}
|
|
if len(findings) > limits.max_findings:
|
|
discarded = len(findings) - limits.max_findings
|
|
findings.sort(key=lambda item: (severity_rank[item.severity], item.category, item.path or "", item.line or 0))
|
|
findings = findings[: limits.max_findings]
|
|
findings.append(
|
|
Finding(
|
|
code="analysis.findings-truncated",
|
|
category="Repository hygiene",
|
|
severity=Severity.INFO,
|
|
message=f"{discarded} lower-priority findings were omitted by the analysis finding budget",
|
|
deduction=0,
|
|
)
|
|
)
|
|
coverage.truncated = True
|
|
coverage.notes.append(f"Finding output limited to {limits.max_findings} deterministic signals.")
|
|
|
|
policy = policy or load_policy(root)
|
|
suppressed_codes = set(policy.suppressions)
|
|
findings = [
|
|
item.model_copy(update={"suppressed": item.suppressed or item.code in suppressed_codes})
|
|
for item in findings
|
|
]
|
|
scores = {category: 100 for category in CATEGORY_NAMES}
|
|
for finding in findings:
|
|
if not finding.suppressed:
|
|
scores[finding.category] = max(0, scores[finding.category] - finding.deduction)
|
|
|
|
health_score = round(sum(scores.values()) / len(scores))
|
|
release_gate = build_release_gate(
|
|
health_score=health_score,
|
|
findings=findings,
|
|
metrics=metrics,
|
|
tests=test_signals,
|
|
dependencies=dependency_signals,
|
|
coverage=coverage,
|
|
policy=policy,
|
|
)
|
|
|
|
findings.sort(key=lambda item: (severity_rank[item.severity], item.category, item.path or "", item.line or 0))
|
|
findings = [
|
|
item.model_copy(update={"fingerprint": item.fingerprint or finding_fingerprint(item)})
|
|
for item in findings
|
|
]
|
|
findings = enrich_findings(findings)
|
|
|
|
return ScanReport(
|
|
id=str(uuid4()),
|
|
repository_path=str(root),
|
|
repository_name=display_name or root.name,
|
|
source=source or ScanSource(),
|
|
policy=policy,
|
|
scanned_at=datetime.now(UTC),
|
|
duration_ms=round((perf_counter() - started) * 1000, 2),
|
|
metrics=metrics,
|
|
languages=dict(languages.most_common()),
|
|
category_scores=scores,
|
|
health_score=health_score,
|
|
release_readiness=release_gate.status,
|
|
findings=findings,
|
|
tests=test_signals,
|
|
git=git_signals,
|
|
dependencies=dependency_signals,
|
|
architecture=architecture,
|
|
hotspots=hotspots,
|
|
release_gate=release_gate,
|
|
coverage=coverage,
|
|
)
|