mirror of
https://github.com/rookiestar28/ComfyUI-OpenClaw.git
synced 2026-08-14 08:52:45 +00:00
feat(architecture): enforce production dependency boundaries
This commit is contained in:
rookiestar28
@@ -0,0 +1,877 @@
|
||||
"""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",
|
||||
"accepted_cycles",
|
||||
"dynamic_imports",
|
||||
}
|
||||
_REVIEW_KEYS = {
|
||||
"owner",
|
||||
"reviewed_at",
|
||||
"next_review_by",
|
||||
"static_analysis_policy_schema",
|
||||
}
|
||||
_EXCEPTION_KEYS = {
|
||||
"importer",
|
||||
"imported",
|
||||
"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]]
|
||||
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))
|
||||
|
||||
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,
|
||||
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 = "<missing>"
|
||||
self.dynamic_imports.append(
|
||||
DynamicImport(
|
||||
path=self.path,
|
||||
scope=".".join(self.scope) or "<module>",
|
||||
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}",
|
||||
)
|
||||
)
|
||||
|
||||
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())
|
||||
Reference in New Issue
Block a user