leocrm/app/services/plugin_install_service.py

"""Plugin install service — business logic for plugin installation from ZIP/URL."""

from __future__ import annotations

import importlib
import logging
import re
import shutil
import tempfile
import zipfile
from pathlib import Path
from typing import Any

from app.plugins.base import BasePlugin
from app.services.plugin_service import get_plugin_service

logger = logging.getLogger(__name__)


class PluginInstallService:
    """Service layer for plugin installation operations.

    Handles ZIP extraction, security validation, and plugin registration.
    """

    @staticmethod
    def validate_manifest_name(name: str) -> str:
        """Validate plugin name is alphanumeric with underscores only."""
        if not re.match(r"^[a-zA-Z][a-zA-Z0-9_]*$", name):
            raise ValueError(
                f"Invalid plugin name '{name}': must start with a letter and contain only "
                f"alphanumeric characters and underscores"
            )
        return name

    @staticmethod
    def check_dangerous_imports(source_code: str) -> list[str]:
        """Check plugin source for dangerous imports/patterns.

        Returns a list of dangerous patterns found (empty if safe).
        """
        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"),
        ]
        found: list[str] = []
        for pattern, description in dangerous_patterns:
            if re.search(pattern, source_code):
                found.append(description)
        return found

    @staticmethod
    def check_migration_sql(sql_content: str) -> list[str]:
        """Basic SQL validation for migration files.

        Returns a list of issues found (empty if OK).
        """
        issues: list[str] = []
        lines = sql_content.strip().split("\n")
        for i, line in enumerate(lines, 1):
            stripped = line.strip()
            if not stripped or stripped.startswith("--"):
                continue
            if stripped.count("(") != stripped.count(")"):
                issues.append(f"Line {i}: unbalanced parentheses")
            if re.search(r"\bDROP\s+TABLE\b", stripped, re.IGNORECASE):
                issues.append(f"Line {i}: DROP TABLE is not allowed in plugin migrations")
        return issues

    @staticmethod
    def find_plugin_class_in_module(module: Any) -> type[BasePlugin] | None:
        """Find a BasePlugin subclass in a module."""
        for attr_name in dir(module):
            attr = getattr(module, attr_name)
            if (
                isinstance(attr, type)
                and issubclass(attr, BasePlugin)
                and attr is not BasePlugin
            ):
                return attr
        return None

    @staticmethod
    def extract_plugin_from_zip(zip_path: str) -> tuple[Path, str, type[BasePlugin]]:
        """Extract a ZIP file and find the plugin class.

        Returns (extract_dir, plugin_name, plugin_class).
        """
        extract_dir = Path(tempfile.mkdtemp(prefix="plugin_upload_"))
        try:
            with zipfile.ZipFile(zip_path, "r") as zf:
                bad_files = [f for f in zf.namelist() if f.startswith("..") or f.startswith("/")]
                if bad_files:
                    raise ValueError(f"ZIP contains files with unsafe paths: {bad_files}")
                zf.extractall(extract_dir)

            plugin_py_path: Path | None = None
            for fpath in extract_dir.rglob("plugin.py"):
                plugin_py_path = fpath
                break

            if plugin_py_path is None:
                raise ValueError("ZIP does not contain a plugin.py file")

            source_code = plugin_py_path.read_text(encoding="utf-8")
            dangerous = PluginInstallService.check_dangerous_imports(source_code)
            if dangerous:
                raise ValueError(
                    f"Plugin contains dangerous patterns: {', '.join(dangerous)}"
                )

            spec = importlib.util.spec_from_file_location(
                "uploaded_plugin", plugin_py_path
            )
            if spec is None or spec.loader is None:
                raise ValueError("Could not load plugin.py module")
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

            plugin_class = PluginInstallService.find_plugin_class_in_module(module)
            if plugin_class is None:
                raise ValueError(
                    "plugin.py does not contain a BasePlugin subclass"
                )

            plugin_instance = plugin_class()
            plugin_name = plugin_instance.name

            manifest = plugin_instance.manifest
            if not manifest.name or not manifest.version or not manifest.display_name:
                raise ValueError(
                    "Plugin manifest must include name, version, and display_name"
                )

            PluginInstallService.validate_manifest_name(manifest.name)

            migrations_dir = plugin_py_path.parent / "migrations"
            if migrations_dir.exists():
                for sql_file in sorted(migrations_dir.glob("*.sql")):
                    sql_content = sql_file.read_text(encoding="utf-8")
                    issues = PluginInstallService.check_migration_sql(sql_content)
                    if issues:
                        raise ValueError(
                            f"Migration file {sql_file.name} has issues: {'; '.join(issues)}"
                        )

            return extract_dir, plugin_name, plugin_class

        except Exception:
            shutil.rmtree(extract_dir, ignore_errors=True)
            raise

    @staticmethod
    def install_plugin_from_dir(
        extract_dir: Path,
        plugin_name: str,
        plugin_class: type[BasePlugin],
    ) -> None:
        """Copy plugin directory to builtins and register it.

        Copies the extracted plugin directory to app/plugins/builtins/{plugin_name}/.
        """
        builtins_dir = Path(__file__).parent.parent / "plugins" / "builtins" / plugin_name
        builtins_dir.mkdir(parents=True, exist_ok=True)

        for item in extract_dir.iterdir():
            dest = builtins_dir / item.name
            if item.is_dir():
                if dest.exists():
                    shutil.rmtree(dest)
                shutil.copytree(item, dest)
            else:
                shutil.copy2(item, dest)

        registry = get_plugin_service().registry
        instance = plugin_class()
        registry.register_plugin(instance)

        logger.info(
            "Installed plugin '%s' from uploaded ZIP to %s",
            plugin_name,
            builtins_dir,
        )