leocrm/app/plugins/quarantine.py

"""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
    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
        manifest_info = _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():
            shutil.rmtree(temp_dir, ignore_errors=True)