"""Verify the repository's source-level production dependency contract. The verifier deliberately uses only Git metadata and Python's standard-library parser. It never imports analyzed modules. """ from __future__ import annotations import argparse import ast import json import re import subprocess import sys import tokenize from collections import defaultdict from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass from datetime import date from pathlib import Path, PurePosixPath from typing import Any POLICY_PATH = "tests/architecture_dependency_policy.json" MAX_FINDINGS = 50 _TOP_LEVEL_KEYS = { "schema_version", "review", "tracked_roots", "domains", "allowed_dependencies", "compatibility_exceptions", "facade_contracts", "accepted_cycles", "dynamic_imports", } _REVIEW_KEYS = { "owner", "reviewed_at", "next_review_by", "static_analysis_policy_schema", } _EXCEPTION_KEYS = { "importer", "imported", "owner", "rationale", "review_condition", } _FACADE_KEYS = { "facade", "implementation", "owner", "rationale", "review_condition", } _CYCLE_KEYS = {"modules", "owner", "rationale", "review_condition"} _DYNAMIC_KEYS = { "path", "scope", "callee", "target_kind", "target", "owner", "rationale", "review_condition", } _METADATA_KEYS = ("owner", "rationale", "review_condition") _DOMAIN_RE = re.compile(r"^[a-z][a-z0-9_-]*$") @dataclass(frozen=True, order=True) class Finding: """A deterministic, content-free policy finding.""" rule_id: str path: str line: int = 0 subject: str = "" @property def code(self) -> str: """Compatibility alias for callers using diagnostic terminology.""" return self.rule_id @property def identity(self) -> str: """Return the bounded identity without exposing source content.""" return self.subject def render(self) -> str: return render_findings((self,)) @dataclass(frozen=True, order=True) class DynamicImport: path: str scope: str callee: str target_kind: str target: str line: int = 0 @property def identity(self) -> tuple[str, str, str, str, str]: return ( self.path, self.scope, self.callee, self.target_kind, self.target, ) @dataclass(frozen=True) class Analysis: owned_paths: tuple[str, ...] static_edges: tuple[tuple[str, str], ...] dynamic_imports: tuple[DynamicImport, ...] cycles: tuple[tuple[str, ...], ...] findings: tuple[Finding, ...] @dataclass class _PolicyContext: tracked_files: set[str] owned_paths: set[str] path_domains: dict[str, str] path_modules: dict[str, str] module_paths: dict[str, str] allowed_dependencies: dict[str, set[str]] compatibility_exceptions: set[tuple[str, str]] facade_contracts: set[tuple[str, str]] accepted_cycles: set[frozenset[str]] dynamic_imports: dict[tuple[str, str, str, str, str], Mapping[str, Any]] def _finding( rule_id: str, path: str = ".", *, line: int = 0, subject: str = "" ) -> Finding: return Finding(rule_id=rule_id, path=path, line=line, subject=subject) def _safe_relative_path(value: Any) -> bool: if not isinstance(value, str) or not value or "\\" in value: return False if re.match(r"^[A-Za-z]:", value): return False path = PurePosixPath(value) return ( not path.is_absolute() and value == path.as_posix() and "." not in path.parts and ".." not in path.parts ) def _within_root(path: str, root: str) -> bool: return path == root or path.startswith(f"{root.rstrip('/')}/") def _module_name(path: str) -> str: parts = list(PurePosixPath(path).with_suffix("").parts) if parts[-1] == "__init__": parts.pop() return ".".join(parts) or "__init__" def _tracked_python_files(repo_root: Path) -> tuple[set[str], list[Finding]]: result = subprocess.run( ["git", "ls-files", "--cached", "--", "*.py"], cwd=repo_root, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, shell=False, ) if result.returncode != 0: return set(), [_finding("TRACKED_DISCOVERY_FAILED")] return ( { line.strip().replace("\\", "/") for line in result.stdout.splitlines() if line.strip() }, [], ) def _validate_review_metadata( entry: Mapping[str, Any], *, path: str, findings: list[Finding], ) -> None: if any( not isinstance(entry.get(key), str) or not str(entry.get(key)).strip() for key in _METADATA_KEYS ): findings.append(_finding("POLICY_REVIEW_METADATA", subject=path)) def _validate_policy( repo_root: Path, policy: Mapping[str, Any], tracked_files: Iterable[str] | None, ) -> tuple[_PolicyContext, list[Finding]]: findings: list[Finding] = [] unknown_keys = set(policy) - _TOP_LEVEL_KEYS for key in sorted(unknown_keys): findings.append(_finding("POLICY_UNKNOWN_KEY", subject=key)) if policy.get("schema_version") != 1: findings.append(_finding("POLICY_SCHEMA_VERSION")) review = policy.get("review") if not isinstance(review, Mapping): findings.append(_finding("POLICY_REVIEW_METADATA", subject="review")) else: for key in sorted(set(review) - _REVIEW_KEYS): findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"review.{key}")) if not isinstance(review.get("owner"), str) or not review["owner"].strip(): findings.append(_finding("POLICY_REVIEW_METADATA", subject="review.owner")) parsed_dates: dict[str, date] = {} for key in ("reviewed_at", "next_review_by"): try: parsed_dates[key] = date.fromisoformat(str(review.get(key, ""))) except ValueError: findings.append( _finding("POLICY_REVIEW_METADATA", subject=f"review.{key}") ) if ( len(parsed_dates) == 2 and parsed_dates["next_review_by"] < parsed_dates["reviewed_at"] ): findings.append( _finding("POLICY_REVIEW_METADATA", subject="review.date_order") ) if tracked_files is None: discovered, discovery_findings = _tracked_python_files(repo_root) findings.extend(discovery_findings) else: discovered = { str(path).replace("\\", "/") for path in tracked_files if str(path).endswith(".py") } roots_value = policy.get("tracked_roots") roots = roots_value if isinstance(roots_value, list) else [] if not isinstance(roots_value, list) or not roots: findings.append(_finding("ROOTS_INVALID")) valid_roots: list[str] = [] seen_roots: set[str] = set() for index, value in enumerate(roots): subject = f"tracked_roots[{index}]" if not _safe_relative_path(value): findings.append(_finding("PATH_UNSAFE", subject=subject)) continue root = str(value) if root in seen_roots: findings.append(_finding("ROOT_DUPLICATE", path=root)) continue seen_roots.add(root) valid_roots.append(root) candidate = repo_root / root try: candidate.resolve().relative_to(repo_root.resolve()) except ValueError: findings.append(_finding("PATH_UNSAFE", path=root)) continue if not candidate.exists(): findings.append(_finding("ROOT_MISSING", path=root)) domains_value = policy.get("domains") domains = domains_value if isinstance(domains_value, Mapping) else {} if not domains: findings.append(_finding("DOMAINS_INVALID")) valid_domain_names = { str(name) for name in domains if isinstance(name, str) and _DOMAIN_RE.fullmatch(name) } for name in domains: if name not in valid_domain_names: findings.append(_finding("DOMAIN_UNKNOWN", subject=str(name))) owned_paths: set[str] = set() path_domains: dict[str, str] = {} path_modules: dict[str, str] = {} module_paths: dict[str, str] = {} for domain_name, entries in domains.items(): if domain_name not in valid_domain_names: continue if not isinstance(entries, list): findings.append(_finding("OWNERSHIP_INVALID", subject=str(domain_name))) continue for index, value in enumerate(entries): subject = f"domains.{domain_name}[{index}]" if not _safe_relative_path(value): findings.append(_finding("PATH_UNSAFE", subject=subject)) continue path = str(value) if not path.endswith(".py"): findings.append(_finding("OWNERSHIP_INVALID", path=path)) continue if path in path_domains: findings.append(_finding("OWN_DUPLICATE", path=path)) continue path_domains[path] = str(domain_name) owned_paths.add(path) if not any(_within_root(path, root) for root in valid_roots): findings.append(_finding("OWN_OUTSIDE_ROOT", path=path)) if path not in discovered: findings.append(_finding("OWN_NOT_TRACKED", path=path)) candidate = repo_root / path try: candidate.resolve().relative_to(repo_root.resolve()) except ValueError: findings.append(_finding("PATH_UNSAFE", path=path)) continue if not candidate.is_file(): findings.append(_finding("OWN_MISSING", path=path)) module = _module_name(path) if module in module_paths: findings.append( _finding("OWN_MODULE_COLLISION", path=path, subject=module) ) else: path_modules[path] = module module_paths[module] = path tracked_in_roots = { path for path in discovered if any(_within_root(path, root) for root in valid_roots) } for path in sorted(tracked_in_roots - owned_paths): findings.append(_finding("OWN_UNOWNED_MODULE", path=path)) allowed_value = policy.get("allowed_dependencies") allowed_raw = allowed_value if isinstance(allowed_value, Mapping) else {} if not isinstance(allowed_value, Mapping): findings.append(_finding("DEPENDENCIES_INVALID")) for domain in sorted(valid_domain_names - set(allowed_raw)): findings.append(_finding("DOMAIN_DIRECTION_MISSING", subject=domain)) for domain in sorted(set(allowed_raw) - valid_domain_names): findings.append(_finding("DOMAIN_UNKNOWN", subject=str(domain))) allowed_dependencies: dict[str, set[str]] = {} for domain in sorted(valid_domain_names): values = allowed_raw.get(domain, []) if not isinstance(values, list): findings.append(_finding("DEPENDENCIES_INVALID", subject=domain)) values = [] accepted: set[str] = set() for target in values: if target not in valid_domain_names: findings.append( _finding( "DOMAIN_UNKNOWN", subject=f"{domain}->{target}", ) ) else: accepted.add(str(target)) allowed_dependencies[domain] = accepted compatibility_exceptions: set[tuple[str, str]] = set() exception_entries = policy.get("compatibility_exceptions") if not isinstance(exception_entries, list): findings.append(_finding("EXCEPTIONS_INVALID")) exception_entries = [] for index, entry in enumerate(exception_entries): subject = f"compatibility_exceptions[{index}]" if not isinstance(entry, Mapping): findings.append(_finding("EXCEPTIONS_INVALID", subject=subject)) continue for key in sorted(set(entry) - _EXCEPTION_KEYS): findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"{subject}.{key}")) _validate_review_metadata(entry, path=subject, findings=findings) edge = (str(entry.get("importer", "")), str(entry.get("imported", ""))) if edge in compatibility_exceptions: findings.append(_finding("DEP_DUPLICATE_EXCEPTION", subject=subject)) compatibility_exceptions.add(edge) if edge[0] not in module_paths or edge[1] not in module_paths: findings.append(_finding("DEP_EXCEPTION_MODULE_UNKNOWN", subject=subject)) facade_contracts: set[tuple[str, str]] = set() facade_entries = policy.get("facade_contracts", []) if not isinstance(facade_entries, list): findings.append(_finding("FACADES_INVALID")) facade_entries = [] for index, entry in enumerate(facade_entries): subject = f"facade_contracts[{index}]" if not isinstance(entry, Mapping): findings.append(_finding("FACADES_INVALID", subject=subject)) continue for key in sorted(set(entry) - _FACADE_KEYS): findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"{subject}.{key}")) _validate_review_metadata(entry, path=subject, findings=findings) edge = ( str(entry.get("facade", "")), str(entry.get("implementation", "")), ) if not edge[0] or not edge[1] or edge[0] == edge[1]: findings.append(_finding("FACADES_INVALID", subject=subject)) continue if edge in facade_contracts: findings.append(_finding("FACADE_DUPLICATE", subject=subject)) facade_contracts.add(edge) if edge[0] not in module_paths or edge[1] not in module_paths: findings.append(_finding("FACADE_MODULE_UNKNOWN", subject=subject)) accepted_cycles: set[frozenset[str]] = set() cycle_entries = policy.get("accepted_cycles") if not isinstance(cycle_entries, list): findings.append(_finding("CYCLES_INVALID")) cycle_entries = [] for index, entry in enumerate(cycle_entries): subject = f"accepted_cycles[{index}]" if not isinstance(entry, Mapping): findings.append(_finding("CYCLES_INVALID", subject=subject)) continue for key in sorted(set(entry) - _CYCLE_KEYS): findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"{subject}.{key}")) _validate_review_metadata(entry, path=subject, findings=findings) modules = entry.get("modules") if ( not isinstance(modules, list) or len(modules) < 2 or any(not isinstance(module, str) for module in modules) ): findings.append(_finding("CYCLES_INVALID", subject=subject)) continue cycle = frozenset(modules) if len(cycle) != len(modules): findings.append(_finding("CYCLE_DUPLICATE_MODULE", subject=subject)) if cycle in accepted_cycles: findings.append(_finding("CYCLE_DUPLICATE_BASELINE", subject=subject)) accepted_cycles.add(cycle) if any(module not in module_paths for module in cycle): findings.append(_finding("CYCLE_MODULE_UNKNOWN", subject=subject)) dynamic_imports: dict[tuple[str, str, str, str, str], Mapping[str, Any]] = {} dynamic_entries = policy.get("dynamic_imports") if not isinstance(dynamic_entries, list): findings.append(_finding("DYNAMIC_INVALID")) dynamic_entries = [] for index, entry in enumerate(dynamic_entries): subject = f"dynamic_imports[{index}]" if not isinstance(entry, Mapping): findings.append(_finding("DYNAMIC_INVALID", subject=subject)) continue for key in sorted(set(entry) - _DYNAMIC_KEYS): findings.append(_finding("POLICY_UNKNOWN_KEY", subject=f"{subject}.{key}")) _validate_review_metadata(entry, path=subject, findings=findings) path_value = entry.get("path") if not _safe_relative_path(path_value): findings.append(_finding("PATH_UNSAFE", subject=subject)) continue dynamic_path = str(path_value) if dynamic_path not in owned_paths: findings.append(_finding("DYNAMIC_PATH_UNOWNED", path=dynamic_path)) target_kind = entry.get("target_kind") identity = ( dynamic_path, str(entry.get("scope", "")), str(entry.get("callee", "")), str(target_kind), str(entry.get("target", "")), ) if ( not identity[1] or identity[2] not in {"__import__", "importlib.import_module", "import_module"} or target_kind not in {"literal", "expression"} or not identity[4] ): findings.append(_finding("DYNAMIC_INVALID", path=dynamic_path)) if identity in dynamic_imports: findings.append(_finding("DYNAMIC_DUPLICATE", path=dynamic_path)) dynamic_imports[identity] = entry context = _PolicyContext( tracked_files=discovered, owned_paths=owned_paths, path_domains=path_domains, path_modules=path_modules, module_paths=module_paths, allowed_dependencies=allowed_dependencies, compatibility_exceptions=compatibility_exceptions, facade_contracts=facade_contracts, accepted_cycles=accepted_cycles, dynamic_imports=dynamic_imports, ) return context, findings def _resolve_relative_import( current_module: str, current_path: str, node: ast.ImportFrom, ) -> str: if not node.level: return node.module or "" is_package = current_path.endswith("/__init__.py") or current_path == "__init__.py" if current_path == "__init__.py": package_parts: list[str] = [] else: package_parts = ( current_module.split(".") if is_package else current_module.split(".")[:-1] ) ascend = node.level - 1 if ascend > len(package_parts): prefix: list[str] = [] elif ascend: prefix = package_parts[:-ascend] else: prefix = package_parts if node.module: prefix.extend(node.module.split(".")) return ".".join(prefix) class _SourceVisitor(ast.NodeVisitor): def __init__( self, *, path: str, module: str, module_paths: Mapping[str, str], ) -> None: self.path = path self.module = module self.module_paths = module_paths self.edges: set[tuple[str, str]] = set() self.dynamic_imports: list[DynamicImport] = [] self.scope: list[str] = [] self.builtins_aliases: set[str] = {"builtins"} self.builtin_import_aliases: set[str] = {"__import__"} self.importlib_aliases: set[str] = {"importlib"} self.import_module_aliases: set[str] = set() def _add_edge(self, imported: str) -> None: # IMPORTANT: require an exact owned module. Falling back to the nearest # package turns missing optional submodules into false dependency edges. target = imported if imported in self.module_paths else None if target and target != self.module: self.edges.add((self.module, target)) def visit_Import(self, node: ast.Import) -> None: for alias in node.names: if alias.name == "builtins": self.builtins_aliases.add(alias.asname or alias.name) if alias.name == "importlib": self.importlib_aliases.add(alias.asname or alias.name) self._add_edge(alias.name) def visit_ImportFrom(self, node: ast.ImportFrom) -> None: base = _resolve_relative_import(self.module, self.path, node) if node.level == 0 and node.module == "importlib": for alias in node.names: if alias.name == "import_module": self.import_module_aliases.add(alias.asname or alias.name) if node.level == 0 and node.module == "builtins": for alias in node.names: if alias.name == "__import__": self.builtin_import_aliases.add(alias.asname or alias.name) exact_children: list[str] = [] for alias in node.names: candidate = f"{base}.{alias.name}" if base else alias.name if candidate in self.module_paths: exact_children.append(candidate) if exact_children: for candidate in exact_children: self._add_edge(candidate) elif base: self._add_edge(base) def _visit_scoped( self, node: ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef, ) -> None: self.scope.append(node.name) self.generic_visit(node) self.scope.pop() def visit_FunctionDef(self, node: ast.FunctionDef) -> None: self._visit_scoped(node) def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: self._visit_scoped(node) def visit_ClassDef(self, node: ast.ClassDef) -> None: self._visit_scoped(node) def visit_Call(self, node: ast.Call) -> None: callee = "" if isinstance(node.func, ast.Name): if node.func.id in self.builtin_import_aliases: callee = "__import__" elif node.func.id in self.import_module_aliases: callee = "import_module" elif ( isinstance(node.func, ast.Attribute) and node.func.attr == "__import__" and isinstance(node.func.value, ast.Name) and node.func.value.id in self.builtins_aliases ): callee = "__import__" elif ( isinstance(node.func, ast.Attribute) and node.func.attr == "import_module" and isinstance(node.func.value, ast.Name) and node.func.value.id in self.importlib_aliases ): callee = "importlib.import_module" if callee: if ( node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str) ): target_kind = "literal" target = node.args[0].value elif node.args: target_kind = "expression" argument = node.args[0] target = ( argument.id if isinstance(argument, ast.Name) else f"<{type(argument).__name__}>" ) else: target_kind = "expression" target = "" self.dynamic_imports.append( DynamicImport( path=self.path, scope=".".join(self.scope) or "", callee=callee, target_kind=target_kind, target=target, line=node.lineno, ) ) self.generic_visit(node) def _strongly_connected_components( modules: Iterable[str], edges: Iterable[tuple[str, str]], ) -> tuple[tuple[str, ...], ...]: adjacency: dict[str, set[str]] = defaultdict(set) for importer, imported in edges: adjacency[importer].add(imported) next_index = 0 indices: dict[str, int] = {} low_links: dict[str, int] = {} stack: list[str] = [] on_stack: set[str] = set() components: list[tuple[str, ...]] = [] def visit(module: str) -> None: nonlocal next_index indices[module] = next_index low_links[module] = next_index next_index += 1 stack.append(module) on_stack.add(module) for imported in sorted(adjacency[module]): if imported not in indices: visit(imported) low_links[module] = min(low_links[module], low_links[imported]) elif imported in on_stack: low_links[module] = min(low_links[module], indices[imported]) if low_links[module] != indices[module]: return component: list[str] = [] while True: member = stack.pop() on_stack.remove(member) component.append(member) if member == module: break if len(component) > 1: components.append(tuple(sorted(component))) for module in sorted(modules): if module not in indices: visit(module) return tuple(sorted(components)) def analyze_repository( repo_root: Path, policy: Mapping[str, Any], *, tracked_files: Iterable[str] | None = None, ) -> Analysis: """Analyze a repository without importing or executing its source modules.""" repo_root = repo_root.resolve() context, findings = _validate_policy(repo_root, policy, tracked_files) edges: set[tuple[str, str]] = set() dynamic_imports: list[DynamicImport] = [] for path in sorted(context.owned_paths): source_path = repo_root / path if not source_path.is_file() or path not in context.path_modules: continue try: # IMPORTANT: tokenize.open handles encoding cookies and existing UTF-8 BOMs # without rewriting source or importing production modules. with tokenize.open(source_path) as source_file: tree = ast.parse(source_file.read(), filename=path) except (OSError, SyntaxError, UnicodeError) as exc: findings.append( _finding("SOURCE_PARSE", path=path, subject=type(exc).__name__) ) continue visitor = _SourceVisitor( path=path, module=context.path_modules[path], module_paths=context.module_paths, ) visitor.visit(tree) edges.update(visitor.edges) dynamic_imports.extend(visitor.dynamic_imports) for importer, imported in sorted(edges): importer_path = context.module_paths.get(importer, ".") importer_domain = context.path_domains.get(importer_path) imported_path = context.module_paths.get(imported, ".") imported_domain = context.path_domains.get(imported_path) if not importer_domain or not imported_domain: continue allowed = imported_domain in context.allowed_dependencies.get( importer_domain, set() ) exception = (importer, imported) in context.compatibility_exceptions if not allowed and not exception: findings.append( _finding( "DEP_FORBIDDEN_DIRECTION", path=importer_path, subject=f"{importer}->{imported}", ) ) for importer, imported in sorted(context.compatibility_exceptions): if (importer, imported) not in edges: path = context.module_paths.get(importer, ".") findings.append( _finding( "DEP_STALE_EXCEPTION", path=path, subject=f"{importer}->{imported}", ) ) for facade, implementation in sorted(context.facade_contracts): if (facade, implementation) not in edges: path = context.module_paths.get(facade, ".") findings.append( _finding( "FACADE_STALE", path=path, subject=f"{facade}->{implementation}", ) ) if (implementation, facade) in edges: path = context.module_paths.get(implementation, ".") findings.append( _finding( "FACADE_REVERSE_DEPENDENCY", path=path, subject=f"{implementation}->{facade}", ) ) cycles = _strongly_connected_components(context.module_paths, edges) current_cycle_sets = {frozenset(cycle) for cycle in cycles} for cycle in cycles: if frozenset(cycle) not in context.accepted_cycles: path = context.module_paths.get(cycle[0], ".") findings.append(_finding("CYCLE_NEW", path=path, subject="|".join(cycle))) for accepted_cycle in sorted( context.accepted_cycles, key=lambda item: sorted(item) ): if accepted_cycle not in current_cycle_sets: first = sorted(accepted_cycle)[0] if accepted_cycle else "" path = context.module_paths.get(first, ".") findings.append( _finding( "CYCLE_STALE", path=path, subject="|".join(sorted(accepted_cycle)), ) ) current_dynamic = {site.identity: site for site in dynamic_imports} for identity, site in sorted(current_dynamic.items()): if identity not in context.dynamic_imports: rule_id = ( "DYNAMIC_UNREGISTERED_LITERAL" if site.target_kind == "literal" else "DYNAMIC_UNREGISTERED_EXPRESSION" ) findings.append( _finding( rule_id, path=site.path, line=site.line, subject=f"{site.scope}:{site.callee}", ) ) for identity in sorted(context.dynamic_imports): if identity not in current_dynamic: path, scope, callee, _, _ = identity findings.append( _finding( "DYNAMIC_STALE", path=path, subject=f"{scope}:{callee}", ) ) return Analysis( owned_paths=tuple(sorted(context.owned_paths)), static_edges=tuple(sorted(edges)), dynamic_imports=tuple(sorted(dynamic_imports)), cycles=cycles, findings=tuple(sorted(set(findings))), ) def verify_repository( repo_root: Path, policy: Mapping[str, Any], *, tracked_files: Iterable[str] | None = None, ) -> tuple[Finding, ...]: return analyze_repository( repo_root, policy, tracked_files=tracked_files, ).findings def evaluate_repository( repo_root: Path, policy: Mapping[str, Any], *, tracked_files: Iterable[str] | None = None, ) -> list[Finding]: """Compatibility facade returning the deterministic findings as a list.""" return list(verify_repository(repo_root, policy, tracked_files=tracked_files)) def render_findings(findings: Sequence[Finding]) -> str: lines: list[str] = [] for finding in sorted(findings): location = finding.path if finding.line: location = f"{location}:{finding.line}" # Security boundary: CLI output is limited to rule IDs and repository-relative # locations. Internal graph identities remain available to in-process tests. lines.append(f"{finding.rule_id} {location}") return "\n".join(lines) def _load_policy(path: Path) -> Mapping[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, Mapping): raise ValueError("policy root must be an object") return value def run_cli( repo_root: Path, policy_path: Path, *, max_findings: int = MAX_FINDINGS, ) -> tuple[int, list[str]]: """Run the bounded CLI contract without printing or leaking host paths.""" repo_root = repo_root.resolve() policy_path = policy_path if policy_path.is_absolute() else repo_root / policy_path try: policy_path.resolve().relative_to(repo_root) except ValueError: return 2, ["POLICY_PATH_OUTSIDE ."] try: policy = _load_policy(policy_path) except (OSError, UnicodeError, json.JSONDecodeError, ValueError): return 2, [f"POLICY_JSON_INVALID {POLICY_PATH}"] findings = verify_repository(repo_root, policy) if not findings: return 0, ["DEPENDENCY_POLICY_PASS ."] limit = max(1, min(int(max_findings), MAX_FINDINGS)) visible = findings[:limit] lines = [finding.render() for finding in visible] omitted = len(findings) - len(visible) if omitted: lines.append(f"FINDINGS_TRUNCATED - {omitted} omitted") return 1, lines def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path(__file__).parents[1]) parser.add_argument("--policy", type=Path) parser.add_argument("--max-findings", type=int, default=MAX_FINDINGS) args = parser.parse_args(argv) repo_root = args.repo_root.resolve() policy_path = args.policy or (repo_root / POLICY_PATH) exit_code, lines = run_cli( repo_root, policy_path, max_findings=args.max_findings, ) for line in lines: print(line) return exit_code if __name__ == "__main__": sys.exit(main())