"""Plugin quarantine — extract, validate, and install external plugins safely. Workflow: 1. Extract ZIP to a temporary directory 2. Validate manifest exists and is valid 3. Check for dangerous imports 4. Validate migration SQL 5. Verify signature (if provided) 6. If all checks pass: move to plugins/ directory 7. If any check fails: delete temp directory and raise error Usage:: from app.plugins.quarantine import quarantine_plugin plugin_dir = await quarantine_plugin( zip_path=Path("plugin.zip"), signature=b"...", public_key=b"...", ) """ from __future__ import annotations import logging import os import re import shutil import tempfile import zipfile from pathlib import Path from app.plugins.signature import PluginSignature logger = logging.getLogger(__name__) # Maximum plugin ZIP size (50 MB) MAX_PLUGIN_SIZE = 50 * 1024 * 1024 # Dangerous patterns in plugin source code DANGEROUS_PATTERNS = [ (r"\bos\.system\b", "os.system call"), (r"\bsubprocess\.", "subprocess module"), (r"\beval\s*\(", "eval() call"), (r"\bexec\s*\(", "exec() call"), (r"\b__import__\s*\(", "__import__() call"), (r"\bcompile\s*\(", "compile() call"), (r"\bopen\s*\([^)]*['\"]w['\"]", "file write outside DMS"), ] class QuarantineError(Exception): """Raised when a plugin fails quarantine validation.""" def _validate_manifest(plugin_dir: Path) -> dict: """Validate that the plugin has a valid manifest. Returns the parsed manifest data. """ plugin_py = plugin_dir / "plugin.py" init_py = plugin_dir / "__init__.py" if not plugin_py.exists() and not init_py.exists(): raise QuarantineError("Plugin must have plugin.py or __init__.py") # Read source and look for manifest source_file = plugin_py if plugin_py.exists() else init_py source = source_file.read_text(encoding="utf-8") if "PluginManifest" not in source: raise QuarantineError("Plugin source must define a PluginManifest") if "BasePlugin" not in source: raise QuarantineError("Plugin source must inherit from BasePlugin") return {"source_file": str(source_file), "has_manifest": True} def _check_dangerous_imports(plugin_dir: Path) -> list[str]: """Check plugin source for dangerous imports/patterns. Returns a list of dangerous patterns found (empty if safe). """ found: list[str] = [] for py_file in plugin_dir.rglob("*.py"): source = py_file.read_text(encoding="utf-8") for pattern, description in DANGEROUS_PATTERNS: if re.search(pattern, source): found.append(f"{py_file.name}: {description}") return found def _check_migration_sql(plugin_dir: Path) -> list[str]: """Validate migration SQL files in the plugin. Returns a list of issues found (empty if OK). """ issues: list[str] = [] migrations_dir = plugin_dir / "migrations" if not migrations_dir.exists(): return issues # No migrations is OK for sql_file in migrations_dir.glob("*.sql"): content = sql_file.read_text(encoding="utf-8") # Check for tenant_id in CREATE TABLE if "CREATE TABLE" in content.upper() and "tenant_id" not in content.lower(): issues.append( f"{sql_file.name}: CREATE TABLE without tenant_id column" ) # Check for DROP DATABASE / DROP SCHEMA if "DROP DATABASE" in content.upper() or "DROP SCHEMA" in content.upper(): issues.append(f"{sql_file.name}: Contains DROP DATABASE/SCHEMA") return issues async def quarantine_plugin( zip_path: Path, signature: bytes | None = None, public_key: bytes | None = None, plugins_dir: Path | None = None, ) -> Path: """Extract, validate, and install a plugin from a ZIP file. Args: zip_path: Path to the plugin ZIP file. signature: Optional Ed25519 signature bytes. public_key: Optional Ed25519 public key bytes. plugins_dir: Target directory for external plugins (default: plugins/). Returns: Path to the installed plugin directory. Raises: QuarantineError: If any validation check fails. """ # Check file size file_size = zip_path.stat().st_size # noqa: ASYNC240 if file_size > MAX_PLUGIN_SIZE: raise QuarantineError( f"Plugin ZIP too large: {file_size} bytes (max {MAX_PLUGIN_SIZE})" ) # Verify signature if provided if signature and public_key: if not PluginSignature.verify_signature(zip_path, signature, public_key): raise QuarantineError("Signature verification failed") # Create temp directory for extraction temp_dir = Path(tempfile.mkdtemp(prefix="plugin_quarantine_")) try: # Extract ZIP with zipfile.ZipFile(zip_path, "r") as zf: # Check for path traversal in ZIP entries for entry in zf.namelist(): if entry.startswith("/") or ".." in entry: raise QuarantineError(f"Unsafe ZIP entry: {entry}") zf.extractall(temp_dir) # Find the plugin directory (might be nested) plugin_dir = temp_dir if not (plugin_dir / "plugin.py").exists() and not (plugin_dir / "__init__.py").exists(): # Look for a single subdirectory subdirs = [d for d in plugin_dir.iterdir() if d.is_dir() and not d.name.startswith("_")] if len(subdirs) == 1: plugin_dir = subdirs[0] else: raise QuarantineError("Could not find plugin root directory in ZIP") # 1. Validate manifest _validate_manifest(plugin_dir) logger.info("Manifest validated for plugin in %s", plugin_dir.name) # 2. Check dangerous imports dangerous = _check_dangerous_imports(plugin_dir) if dangerous: raise QuarantineError( f"Dangerous patterns found in plugin: {', '.join(dangerous)}" ) # 3. Check migration SQL sql_issues = _check_migration_sql(plugin_dir) if sql_issues: raise QuarantineError( f"Migration SQL issues: {', '.join(sql_issues)}" ) # 4. All checks passed — move to plugins directory target_dir = plugins_dir or Path(os.environ.get("EXTERNAL_PLUGINS_PATH", "plugins")) target_dir.mkdir(parents=True, exist_ok=True) plugin_name = plugin_dir.name final_dir = target_dir / plugin_name if final_dir.exists(): raise QuarantineError(f"Plugin directory already exists: {final_dir}") shutil.copytree(plugin_dir, final_dir) logger.info("Plugin installed to %s", final_dir) return final_dir except Exception: # Clean up temp directory on any error shutil.rmtree(temp_dir, ignore_errors=True) raise finally: # Always clean up temp directory if temp_dir.exists(): # noqa: ASYNC240 shutil.rmtree(temp_dir, ignore_errors=True)