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* ) [\"']?(?Papi[_-]?key|secret|password|passwd|token|access[_-]?key)[\"']? \s*[:=]\s*[\"']?(?P[^\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, )