abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
274 lines
8.7 KiB
Python
274 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Check for forbidden direct cross-plugin imports.
|
|
|
|
This script enforces that plugins communicate only through contracts,
|
|
not by importing internal modules from each other.
|
|
|
|
Allowed:
|
|
from app.plugins.builtins.contracts import get_contract
|
|
from app.plugins.builtins.<name>.contracts import ...
|
|
from app.plugins.builtins.<name> import <PluginClass> (in __init__.py only)
|
|
|
|
Forbidden:
|
|
from app.plugins.builtins.<name>.services import ...
|
|
from app.plugins.builtins.<name>.models import ...
|
|
from app.plugins.builtins.<name>.routes import ...
|
|
|
|
Exceptions (files that are allowed to import anything):
|
|
- */contracts.py — contracts import from internal modules
|
|
- */__init__.py — plugin discovery
|
|
- app/plugins/registry.py — registry manages all plugins
|
|
- app/plugins/builtins/__init__.py — builtin discovery
|
|
- tests/* — test files
|
|
- conftest.py — test fixtures
|
|
|
|
Usage:
|
|
python scripts/check_cross_plugin_imports.py
|
|
python scripts/check_cross_plugin_imports.py --path app/plugins/builtins
|
|
|
|
Exit codes:
|
|
0 — no violations
|
|
1 — violations found
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# ─── Configuration ───
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
BUILTINS_DIR = PROJECT_ROOT / "app" / "plugins" / "builtins"
|
|
|
|
# Files that are exempt from the rule
|
|
EXEMPT_FILES = {
|
|
"contracts.py",
|
|
"__init__.py",
|
|
"conftest.py",
|
|
}
|
|
|
|
# Directories that are exempt
|
|
EXEMPT_DIRS = {
|
|
"tests",
|
|
"__pycache__",
|
|
"migrations",
|
|
}
|
|
|
|
# Files that are exempt by path
|
|
EXEMPT_PATHS = {
|
|
PROJECT_ROOT / "app" / "plugins" / "registry.py",
|
|
PROJECT_ROOT / "app" / "plugins" / "builtins" / "__init__.py",
|
|
PROJECT_ROOT / "app" / "plugins" / "base.py",
|
|
PROJECT_ROOT / "app" / "plugins" / "manifest.py",
|
|
PROJECT_ROOT / "app" / "plugins" / "migration_runner.py",
|
|
}
|
|
|
|
# Pattern for cross-plugin imports
|
|
IMPORT_PATTERN = re.compile(
|
|
r"^\s*(?:from|import)\s+app\.plugins\.builtins\.([^.]+)\.(.+?)\s+import\s+(.+)$"
|
|
)
|
|
|
|
# Pattern for allowed contract imports
|
|
CONTRACT_IMPORT_PATTERN = re.compile(
|
|
r"^\s*from\s+app\.plugins\.builtins\.(?:contracts|[^.]+\.contracts)\s+import\s+(.+)$"
|
|
)
|
|
|
|
# Pattern for __init__.py plugin class imports (allowed in __init__.py only)
|
|
PLUGIN_CLASS_IMPORT_PATTERN = re.compile(
|
|
r"^\s*from\s+app\.plugins\.builtins\.([^.]+)\s+import\s+([A-Z]\w*Plugin)\s*$"
|
|
)
|
|
|
|
|
|
def is_exempt(filepath: Path) -> bool:
|
|
"""Check if a file is exempt from the cross-plugin import rule."""
|
|
# Exempt by filename
|
|
if filepath.name in EXEMPT_FILES:
|
|
return True
|
|
|
|
# Exempt by path
|
|
if filepath in EXEMPT_PATHS:
|
|
return True
|
|
|
|
# Exempt test directories
|
|
parts = filepath.parts
|
|
for exempt_dir in EXEMPT_DIRS:
|
|
if exempt_dir in parts:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def check_file(filepath: Path) -> list[str]:
|
|
"""Check a single file for forbidden cross-plugin imports.
|
|
|
|
Returns a list of violation messages (empty if clean).
|
|
"""
|
|
if is_exempt(filepath):
|
|
return []
|
|
|
|
violations: list[str] = []
|
|
rel_path = filepath.relative_to(PROJECT_ROOT)
|
|
|
|
# Determine the source plugin from the file path
|
|
try:
|
|
parts = filepath.relative_to(BUILTINS_DIR).parts
|
|
src_plugin = parts[0] if parts else ""
|
|
except ValueError:
|
|
src_plugin = ""
|
|
|
|
# Core files (outside builtins) cannot declare plugin dependencies —
|
|
# any cross-plugin import from Core is always a violation.
|
|
is_core_file = src_plugin == ""
|
|
|
|
with open(filepath, encoding="utf-8") as f:
|
|
for line_num, line in enumerate(f, 1):
|
|
line_stripped = line.strip()
|
|
|
|
# Skip comments and empty lines
|
|
if not line_stripped or line_stripped.startswith("#"):
|
|
continue
|
|
|
|
# Check for cross-plugin import
|
|
match = IMPORT_PATTERN.match(line)
|
|
if not match:
|
|
continue
|
|
|
|
tgt_plugin = match.group(1)
|
|
tgt_module = match.group(2)
|
|
|
|
# Skip if importing from contracts
|
|
if tgt_module == "contracts":
|
|
continue
|
|
|
|
# Skip if same plugin (INTRA-Plugin import)
|
|
if tgt_plugin == src_plugin:
|
|
continue
|
|
|
|
# Core files cannot import plugin internals — always a violation
|
|
# regardless of any 'Declared dependency' annotation.
|
|
if is_core_file:
|
|
violations.append(
|
|
f"{rel_path}:{line_num}: {line_stripped}\n"
|
|
f" → Core→Plugin import forbidden: '{tgt_plugin}.{tgt_module}'. "
|
|
f"Use contracts: get_contract(\"{tgt_plugin}\")"
|
|
)
|
|
continue
|
|
|
|
# Plugin→Plugin: check if declared in manifest.dependencies
|
|
if not is_core_file and src_plugin:
|
|
import re as _re
|
|
manifest_path = BUILTINS_DIR / src_plugin / "plugin.py"
|
|
declared_deps: set[str] = set()
|
|
if manifest_path.exists():
|
|
with open(manifest_path, encoding="utf-8") as mf:
|
|
manifest_content = mf.read()
|
|
dep_match = _re.search(r'dependencies=\[([^\]]*)\]', manifest_content)
|
|
if dep_match:
|
|
declared_deps = set(_re.findall(r'"([^"]+)"', dep_match.group(1)))
|
|
if tgt_plugin in declared_deps:
|
|
continue # Declared dependency — allowed
|
|
|
|
# Skip if it's a contract import via the central registry
|
|
if CONTRACT_IMPORT_PATTERN.match(line):
|
|
continue
|
|
|
|
# This is a forbidden cross-plugin import
|
|
violations.append(
|
|
f"{rel_path}:{line_num}: {line_stripped}\n"
|
|
f" → Forbidden cross-plugin import: '{tgt_plugin}.{tgt_module}'. "
|
|
f"Use contracts instead: get_contract(\"{tgt_plugin}\")"
|
|
)
|
|
|
|
return violations
|
|
|
|
|
|
def find_python_files(search_path: Path | None = None) -> list[Path]:
|
|
"""Find all Python files in the search path.
|
|
|
|
Default: scans both app/plugins/builtins/ (Plugin→Plugin) and
|
|
app/core/, app/services/, app/routes/, app/commands/, app/ai/ (Core→Plugin).
|
|
"""
|
|
if search_path is not None:
|
|
files: list[Path] = []
|
|
for root, dirs, fnames in os.walk(search_path):
|
|
dirs[:] = [d for d in dirs if d not in EXEMPT_DIRS]
|
|
for fname in fnames:
|
|
if fname.endswith(".py"):
|
|
files.append(Path(root) / fname)
|
|
return sorted(files)
|
|
|
|
# Default: scan both plugin and core directories
|
|
files: list[Path] = []
|
|
scan_dirs = [
|
|
BUILTINS_DIR,
|
|
PROJECT_ROOT / "app" / "core",
|
|
PROJECT_ROOT / "app" / "services",
|
|
PROJECT_ROOT / "app" / "routes",
|
|
PROJECT_ROOT / "app" / "commands",
|
|
PROJECT_ROOT / "app" / "ai",
|
|
]
|
|
for scan_dir in scan_dirs:
|
|
if not scan_dir.exists():
|
|
continue
|
|
for root, dirs, fnames in os.walk(scan_dir):
|
|
dirs[:] = [d for d in dirs if d not in EXEMPT_DIRS]
|
|
for fname in fnames:
|
|
if fname.endswith(".py"):
|
|
files.append(Path(root) / fname)
|
|
return sorted(files)
|
|
|
|
|
|
def main() -> int:
|
|
"""Run the cross-plugin import checker."""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Check for forbidden cross-plugin imports.")
|
|
parser.add_argument(
|
|
"--path",
|
|
type=Path,
|
|
default=BUILTINS_DIR,
|
|
help="Path to check (default: app/plugins/builtins)",
|
|
)
|
|
parser.add_argument(
|
|
"--verbose",
|
|
action="store_true",
|
|
help="Print checked files even if clean.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
files = find_python_files(args.path)
|
|
all_violations: list[str] = []
|
|
checked = 0
|
|
|
|
for filepath in files:
|
|
checked += 1
|
|
violations = check_file(filepath)
|
|
if violations:
|
|
all_violations.extend(violations)
|
|
elif args.verbose:
|
|
print(f" ✅ {filepath.relative_to(PROJECT_ROOT)}")
|
|
|
|
print(f"\nChecked {checked} files.")
|
|
|
|
if all_violations:
|
|
print(f"\n❌ Found {len(all_violations)} violation(s):\n")
|
|
for v in all_violations:
|
|
print(f" {v}")
|
|
print(
|
|
"\nFix: Replace direct imports with contract-based access:\n"
|
|
" from app.plugins.builtins.contracts import get_contract\n"
|
|
" contract = get_contract(\"plugin_name\")\n"
|
|
" if contract:\n result = await contract.some_function(...)"
|
|
)
|
|
return 1
|
|
else:
|
|
print("✅ No forbidden cross-plugin imports found.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|