aetherra_plugin_discovery.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
# SPDX-FileCopyrightText: 2025 Aetherra Labs and Contributors

"""
[PLUGIN] Aetherra Plugin Discovery Service
====================================
Automatically discovers and registers local plugins with the Aetherra Hub.

This service scans the Aetherra/plugins directory and makes local plugins
visible in the Hub marketplace interface.
"""

# Standard library imports
import asyncio
import base64
import hashlib
import importlib
import importlib.util
import json
import logging
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any  # reduced imports; use built-in generics only

# Third party imports
import requests

# Optional security signing modules (best-effort). We import them here so that
# in-function logic can rely on names existing or handle NameError gracefully.
try:  # pragma: no cover - optional dependency path
    # Aetherra imports
    from Aetherra.security.api_keys import get_key, set_key  # type: ignore
    from Aetherra.security.plugin_signing import (  # type: ignore
        compute_files_hash,
        generate_keypair,
        sign_manifest,
    )

    _SIGNING_AVAILABLE = True
except Exception:  # broad: any failure means signing disabled
    _SIGNING_AVAILABLE = False

logger = logging.getLogger(__name__)


@dataclass
class PluginMetadata:
    """Plugin metadata structure."""

    name: str
    version: str
    description: str
    author: str
    category: str = "utility"
    license: str = "GPL-3.0"
    aetherra_version: str = ">=3.0.0"
    dependencies: dict[str, str] | list[str] | None = None
    keywords: list[str] | None = None
    entry_point: str | None = None
    exports: dict[str, str] | None = None
    repository: str | None = None
    documentation: str | None = None
    homepage: str | None = None
    local_path: str | None = None
    plugin_type: str = "local"  # local, hub, system

    def __post_init__(self):
        if self.dependencies is None:
            self.dependencies = {}
        if self.keywords is None:
            self.keywords = []
        if self.exports is None:
            self.exports = {}


class AetherraPluginDiscovery:
    """
    [SCAN] Plugin Discovery Service

    Discovers and catalogs local plugins, making them available to the Hub.
    """

    def __init__(self, plugins_dir: str | Path | None = None):
        # Normalize to an absolute Path for consistent downstream usage
        self.plugins_dir: Path = (
            Path(plugins_dir) if plugins_dir is not None else Path("Aetherra/plugins")
        ).absolute()
        # Ensure the plugins directory exists so first-run doesn't warn and future installs have a drop-in location
        try:
            self.plugins_dir.mkdir(parents=True, exist_ok=True)
        except Exception as exc:
            logger.debug(
                "[SCAN] Could not create plugins directory %s: %s",
                self.plugins_dir,
                exc,
            )

        self.discovered_plugins: dict[str, PluginMetadata] = {}
        # Respect the Hub URL chosen/started by the launcher; fallback to classic default
        self.hub_url = os.environ.get("AETHERRA_HUB_URL", "http://localhost:3001")

    async def discover_all_plugins(self) -> dict[str, PluginMetadata]:
        """Discover all plugins in the plugins directory."""
        logger.info("[SCAN] Starting plugin discovery...")
        if not self.plugins_dir.exists():
            logger.warning(f"[WARN] Plugins directory not found: {self.plugins_dir}")
            return {}
        # Discover different types of plugins
        await self._discover_aetherplug_plugins()
        await self._discover_python_plugins()
        await self._discover_sample_plugins()
        logger.info(f"[OK] Discovered {len(self.discovered_plugins)} plugins")
        # Compute hashes for integrity (best-effort)
        self._hash_all_plugins()
        return self.discovered_plugins

    async def _discover_aetherplug_plugins(self):
        """Discover .aetherplug format plugins."""
        logger.info("[SCAN] Scanning for .aetherplug plugins...")

        # Look for aetherra-plugin.json files
        for plugin_json in self.plugins_dir.rglob("aetherra-plugin.json"):
            try:
                await self._process_aetherplug_manifest(plugin_json)
            except Exception as e:
                logger.error(f"[ERROR] Error processing {plugin_json}: {e}")

    async def _process_aetherplug_manifest(self, manifest_path: Path):
        """Process an aetherra-plugin.json manifest file."""
        try:
            with open(manifest_path, encoding="utf-8") as f:
                manifest = json.load(f)

            plugin_dir = manifest_path.parent

            metadata = PluginMetadata(
                name=manifest.get("name", plugin_dir.name),
                version=manifest.get("version", "1.0.0"),
                description=manifest.get("description", "No description provided"),
                author=manifest.get("author", "Unknown"),
                category=manifest.get("category", "utility"),
                license=manifest.get("license", "GPL-3.0"),
                aetherra_version=manifest.get("aetherra_version", ">=3.0.0"),
                dependencies=manifest.get("dependencies", {}),
                keywords=manifest.get("keywords", []),
                entry_point=manifest.get("entry_point"),
                exports=manifest.get("exports", {}),
                repository=manifest.get("repository"),
                documentation=manifest.get("documentation"),
                homepage=manifest.get("homepage"),
                local_path=str(plugin_dir),
                plugin_type="aetherplug",
            )

            self.discovered_plugins[metadata.name] = metadata
            logger.info(
                f"[OK] Discovered .aetherplug: {metadata.name} v{metadata.version}"
            )

        except Exception as e:
            logger.error(f"[ERROR] Error processing manifest {manifest_path}: {e}")

    async def _discover_python_plugins(self):
        """Discover Python-based plugins."""
        logger.info("[SCAN] Scanning for Python plugins...")

        # Look for Python files that appear to be plugins
        for py_file in self.plugins_dir.rglob("*.py"):
            if py_file.name.startswith("__") or py_file.name in [
                "setup.py",
                "conftest.py",
            ]:
                continue

            try:
                await self._analyze_python_plugin(py_file)
            except Exception as e:
                logger.error(f"[ERROR] Error analyzing {py_file}: {e}")

    async def _analyze_python_plugin(self, py_file: Path):
        """Analyze a Python file to determine if it's a plugin."""
        try:
            # Read the file to look for plugin indicators
            with open(py_file, encoding="utf-8") as f:
                content = f.read()

            # Look for plugin class or plugin_data
            if "class " in content and (
                "Plugin" in content or "plugin" in content.lower()
            ):
                await self._extract_python_plugin_metadata(py_file, content)
            elif "plugin_data" in content:
                await self._extract_plugin_data_metadata(py_file, content)

        except Exception as e:
            logger.error(f"[ERROR] Error reading {py_file}: {e}")

    async def _extract_python_plugin_metadata(self, py_file: Path, content: str):
        """Extract metadata from a Python plugin class.

        Enhanced: soft-skip GUI plugins ( *_gui.py ) when PySide6 not installed.
        """
        try:
            # Optional explicit skip via env toggle
            if os.getenv(
                "AETHERRA_SKIP_GUI_PLUGINS", "0"
            ) == "1" and py_file.name.endswith("_gui.py"):
                logger.info(
                    f"[SKIP] GUI plugin {py_file.name} skipped via AETHERRA_SKIP_GUI_PLUGINS=1"
                )
                return
            if (
                py_file.name.endswith("_gui.py")
                and importlib.util.find_spec("PySide6") is None
            ):
                logger.info(
                    f"[SKIP] GUI plugin {py_file.name} skipped (PySide6 not installed)"
                )
                return
            spec = importlib.util.spec_from_file_location("plugin_module", py_file)
            if spec and spec.loader:
                module = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(module)
                for attr_name in dir(module):
                    attr = getattr(module, attr_name)
                    if hasattr(attr, "name") and hasattr(attr, "description"):
                        metadata = PluginMetadata(
                            name=getattr(attr, "name", py_file.stem),
                            version=getattr(attr, "version", "1.0.0"),
                            description=getattr(attr, "description", "No description"),
                            author=getattr(
                                attr, "created_by", getattr(attr, "author", "Unknown")
                            ),
                            category=getattr(attr, "category", "utility"),
                            local_path=str(py_file),
                            plugin_type="python",
                            keywords=["python", "local"],
                        )
                        self.discovered_plugins[metadata.name] = metadata
                        logger.info(f"[OK] Discovered Python plugin: {metadata.name}")
                        break
        except Exception as e:
            err_txt = str(e)
            if py_file.name.endswith("_gui.py") and any(
                token in err_txt
                for token in ("PySide6", "QWidget", "QMainWindow", "pyqtSignal")
            ):
                logger.info(
                    f"[SKIP] GUI plugin {py_file.name} not loaded (Qt dependency issue): {e}"
                )
            elif "attempted relative import with no known parent package" in err_txt:
                logger.info(
                    f"[SKIP] Python plugin {py_file.name} requires package import context"
                )
            else:
                logger.error(f"[ERROR] Error importing {py_file}: {e}")

    async def _extract_plugin_data_metadata(self, py_file: Path, content: str):
        """Extract metadata from plugin_data dictionary."""
        try:
            # Try to import and get plugin_data
            spec = importlib.util.spec_from_file_location("plugin_module", py_file)
            if spec and spec.loader:
                module = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(module)

                if hasattr(module, "plugin_data"):
                    data = module.plugin_data
                    metadata = PluginMetadata(
                        name=data.get("name", py_file.stem),
                        version=data.get("version", "1.0.0"),
                        description=data.get("description", "No description"),
                        author=data.get("author", "Unknown"),
                        category=data.get("category", "utility"),
                        local_path=str(py_file),
                        plugin_type="python",
                        keywords=["python", "local"],
                    )

                    self.discovered_plugins[metadata.name] = metadata
                    logger.info(f"[OK] Discovered plugin_data: {metadata.name}")

        except Exception as e:
            if "attempted relative import with no known parent package" in str(e):
                logger.info(
                    f"[SKIP] plugin_data module {py_file.name} requires package import context"
                )
            else:
                logger.error(
                    f"[ERROR] Error extracting plugin_data from {py_file}: {e}"
                )

    async def _discover_sample_plugins(self):
        """Discover sample plugins specifically."""
        logger.info("[SCAN] Scanning for sample plugins...")

        # Look for files matching sample_plugin_*.py pattern
        for sample_file in self.plugins_dir.glob("sample_plugin_*.py"):
            try:
                await self._process_sample_plugin(sample_file)
            except Exception as e:
                logger.error(
                    f"[ERROR] Error processing sample plugin {sample_file}: {e}"
                )

    async def _process_sample_plugin(self, sample_file: Path):
        """Process a sample plugin file."""
        try:
            # Just read to ensure file is accessible; content not used
            with open(sample_file, encoding="utf-8"):
                pass

            # Extract basic info from the sample plugin
            plugin_num = sample_file.stem.split("_")[-1]

            metadata = PluginMetadata(
                name=f"sample_plugin_{plugin_num}",
                version="1.0.0",
                description=f"Sample plugin {plugin_num} for testing and demonstration",
                author="Aetherra Development Team",
                category="sample",
                local_path=str(sample_file),
                plugin_type="sample",
                keywords=["sample", "demo", "testing"],
            )

            self.discovered_plugins[metadata.name] = metadata
            logger.info(f"[OK] Discovered sample plugin: {metadata.name}")

        except Exception as e:
            logger.error(f"[ERROR] Error processing sample plugin {sample_file}: {e}")

    async def register_with_hub(self, plugin_metadata: PluginMetadata) -> bool:
        try:
            entry_point: str | None = plugin_metadata.entry_point
            # Infer a reasonable entry_point when missing
            if not entry_point:
                lp = plugin_metadata.local_path or ""
                if lp and lp.endswith(".py"):
                    entry_point = lp
                elif plugin_metadata.plugin_type == "aetherplug":
                    # Default to conventional entry point for packaged plugin
                    entry_point = "main.py"
                else:
                    # Fallback: non-empty placeholder acceptable to schema
                    entry_point = lp or f"{plugin_metadata.name}.py"

            # Normalize dependencies to list[str]
            deps_raw = plugin_metadata.dependencies
            if isinstance(deps_raw, dict):
                # Use package names only; schema only requires list[str]
                dependencies = list(deps_raw.keys())
            elif isinstance(deps_raw, list):
                dependencies = [str(x) for x in deps_raw]
            else:
                dependencies = []

            hub_plugin = {
                "name": plugin_metadata.name,
                "version": plugin_metadata.version,
                "description": plugin_metadata.description,
                "author": plugin_metadata.author,
                "category": plugin_metadata.category,
                "license": plugin_metadata.license,
                "aetherra_version": plugin_metadata.aetherra_version,
                "entry_point": entry_point,
                "dependencies": dependencies,
                "keywords": plugin_metadata.keywords or [],
                # Integrity hash (if computed)
                "content_hash": self._plugin_hash_cache.get(plugin_metadata.name),
                # Additional, non-schema fields are accepted by the Hub
                "local_path": plugin_metadata.local_path,
                "plugin_type": plugin_metadata.plugin_type,
                "featured": plugin_metadata.plugin_type == "aetherplug",
                "downloads": 0,
                "rating": 5.0 if plugin_metadata.plugin_type == "aetherplug" else 4.5,
                "created_at": "2025-08-02T14:00:00Z",
                "updated_at": "2025-08-02T14:00:00Z",
            }

            # Optional: sign manifest (skip entirely when dev unsigned override set)
            try:
                allow_unsigned_override = (
                    os.environ.get("AETHERRA_ALLOW_UNSIGNED_DEV", "0") == "1"
                )
                if (
                    _SIGNING_AVAILABLE
                    and os.environ.get("AETHERRA_SIGN_PLUGINS") == "1"
                    and not allow_unsigned_override
                ):
                    secret = get_key("plugin_signing_secret")  # type: ignore[name-defined]
                    # Treat invalid base64 secrets as missing so we can regenerate.
                    if secret:
                        try:
                            base64.b64decode(secret)
                        except Exception:
                            logger.warning(
                                "[SIGN] Invalid signing secret format; regenerating ephemeral key"
                            )
                            secret = ""  # force fallback path
                    if not secret:
                        # Fallback: derive ephemeral in-memory secret (dev only) so strict mode can proceed
                        # This does NOT persist and should be replaced with managed key storage in production.
                        if os.environ.get("AETHERRA_SIGNING_STRICT", "0") == "1":
                            try:
                                _pub, _secret = generate_keypair(None)  # type: ignore[arg-type]
                                secret = _secret
                                try:
                                    set_key("plugin_signing_secret", secret)  # type: ignore[name-defined]
                                    logger.info(
                                        "[SIGN] Persisted new ephemeral signing secret to key store"
                                    )
                                except Exception as _persist_exc:
                                    logger.debug(
                                        "[SIGN] Could not persist ephemeral secret: %s",
                                        _persist_exc,
                                    )
                                logger.info(
                                    "[SIGN] Generated ephemeral signing secret for plugin discovery"
                                )
                            except Exception as _eph_exc:
                                logger.warning(
                                    "[SIGN] Ephemeral key generation failed: %s",
                                    _eph_exc,
                                )
                    if secret:
                        # Provide code integrity details (best-effort)
                        code_files: list[str] = []
                        lp = plugin_metadata.local_path or ""
                        if lp and os.path.isdir(lp):
                            for root, _dirs, files in os.walk(lp):
                                for f in files:
                                    if f.endswith(".py"):
                                        code_files.append(str(Path(root) / f))
                        elif lp and os.path.isfile(lp) and lp.endswith(".py"):
                            code_files.append(lp)
                        if code_files:
                            try:
                                hub_plugin["code_files"] = code_files
                                hub_plugin["code_hash"] = compute_files_hash(code_files)  # type: ignore[name-defined]
                            except Exception as _hash_exc:
                                logger.debug(
                                    "[SIGN] code_hash computation failed: %s", _hash_exc
                                )
                        hub_plugin = sign_manifest(hub_plugin, secret)  # type: ignore[name-defined]
                else:
                    if allow_unsigned_override:
                        logger.debug(
                            f"[DEV] Unsigned override active; skipping signing for {plugin_metadata.name}"
                        )
            except Exception as signing_exc:  # best-effort; log at debug level
                logger.debug(f"[SIGN] Plugin signing skipped: {signing_exc}")

            # Dev override: strip any signature/pubkey fields if still present so Hub never attempts verification
            if os.environ.get("AETHERRA_ALLOW_UNSIGNED_DEV", "0") == "1" and (
                hub_plugin.get("signature") or hub_plugin.get("pubkey")
            ):
                hub_plugin.pop("signature", None)
                hub_plugin.pop("pubkey", None)
                logger.debug(
                    f"[DEV] Stripped signature/pubkey fields for {plugin_metadata.name} under unsigned override"
                )

            # Try to register with Hub API
            try:
                headers = {}
                if os.environ.get("AETHERRA_ALLOW_UNSIGNED_DEV", "0") == "1":
                    headers["X-Aeth-Allow-Unsigned"] = "1"
                response = requests.post(
                    f"{self.hub_url}/api/plugins/register",
                    json=hub_plugin,
                    headers=headers or None,
                    timeout=5,
                )
                if response.status_code in [200, 201]:
                    logger.info(f"[OK] Registered {plugin_metadata.name} with Hub")
                    return True
                # Capture server-provided error details if present
                detail = None
                try:
                    detail = response.json()
                except Exception:
                    detail = (response.text or "").strip()
                # Graceful fallback: if strict signing attempted but Hub reports signature verification unavailable,
                # retry once without signature/pubkey fields so plugin can still register (development posture).
                if (
                    response.status_code == 400
                    and isinstance(detail, dict)
                    and str(detail.get("error", "")).lower()
                    == "signature verification unavailable"
                    and hub_plugin.get("signature")
                    and hub_plugin.get("pubkey")
                ):
                    try:
                        unsigned = dict(hub_plugin)
                        unsigned.pop("signature", None)
                        unsigned.pop("pubkey", None)
                        # Hint override for future attempts in this process
                        os.environ.setdefault("AETHERRA_ALLOW_UNSIGNED_DEV", "1")
                        r2 = requests.post(
                            f"{self.hub_url}/api/plugins/register",
                            json=unsigned,
                            headers={"X-Aeth-Allow-Unsigned": "1"},
                            timeout=5,
                        )
                        if r2.status_code in [200, 201]:
                            logger.info(
                                f"[OK] Registered {plugin_metadata.name} without signature (verification unavailable)"
                            )
                            return True
                        try:
                            d2 = r2.json()
                        except Exception:
                            d2 = (r2.text or "").strip()
                        logger.warning(
                            f"[WARN] Fallback unsigned registration failed for {plugin_metadata.name}: {r2.status_code} details={d2}"
                        )
                    except Exception as _fallback_exc:
                        logger.debug(
                            "[SIGN] Fallback unsigned registration error for %s: %s",
                            plugin_metadata.name,
                            _fallback_exc,
                        )
                else:
                    logger.warning(
                        f"[WARN] Hub registration failed for {plugin_metadata.name}: {response.status_code} details={detail}"
                    )
            except requests.exceptions.RequestException:
                logger.warning(
                    f"[WARN] Hub not available for {plugin_metadata.name} registration"
                )

            return False

        except Exception as e:
            logger.error(
                f"[ERROR] Error registering {plugin_metadata.name} with Hub: {e}"
            )
            return False

    async def sync_all_with_hub(self):
        """Sync all discovered plugins with the Hub."""
        logger.info("[LOOP] Syncing all plugins with Aetherra Hub...")

        await self.discover_all_plugins()

        success_count = 0
        for _plugin_name, metadata in self.discovered_plugins.items():
            if await self.register_with_hub(metadata):
                success_count += 1

        logger.info(
            f"[OK] Successfully synced {success_count}/{len(self.discovered_plugins)} plugins with Hub"
        )
        return success_count

    def get_plugin_summary(self) -> dict[str, Any]:
        """Get a summary of discovered plugins."""
        by_type = {}
        by_category = {}

        for plugin in self.discovered_plugins.values():
            # Count by type
            by_type[plugin.plugin_type] = by_type.get(plugin.plugin_type, 0) + 1

            # Count by category
            by_category[plugin.category] = by_category.get(plugin.category, 0) + 1

        return {
            "total_plugins": len(self.discovered_plugins),
            "by_type": by_type,
            "by_category": by_category,
            "plugin_names": list(self.discovered_plugins),
            "hashes": {
                k: self._plugin_hash_cache.get(k) for k in self.discovered_plugins
            },
        }

    _plugin_hash_cache: dict[str, str] = {}

    def _hash_all_plugins(self):
        for _name, meta in self.discovered_plugins.items():
            path = meta.local_path
            if not path or not os.path.exists(path):
                continue
            try:
                if os.path.isdir(path):
                    h = hashlib.sha256()
                    for root, _dirs, files in os.walk(path):
                        for f in sorted(files):
                            fp = os.path.join(root, f)
                            try:
                                with open(fp, "rb") as fh:
                                    for chunk in iter(lambda: fh.read(65536), b""):
                                        h.update(chunk)
                            except Exception as exc:
                                logger.debug("[HASH] Skipped file %s: %s", fp, exc)
                                continue
                    self._plugin_hash_cache[_name] = h.hexdigest()
                else:
                    with open(path, "rb") as fh:
                        data = fh.read()
                    self._plugin_hash_cache[_name] = hashlib.sha256(data).hexdigest()
            except Exception as exc:
                logger.debug("[HASH] Failed hashing %s: %s", path, exc)
                continue


async def main():
    """Main function for testing plugin discovery."""
    logging.basicConfig(level=logging.INFO)

    discovery = AetherraPluginDiscovery()
    await discovery.discover_all_plugins()

    summary = discovery.get_plugin_summary()
    print("\n[SCAN] Plugin Discovery Summary:")
    print(f"Total plugins: {summary['total_plugins']}")
    print(f"By type: {summary['by_type']}")
    print(f"By category: {summary['by_category']}")
    print(f"Plugin names: {summary['plugin_names']}")

    # Try to sync with Hub
    await discovery.sync_all_with_hub()


if __name__ == "__main__":
    asyncio.run(main())