""" Filename: plugin_manager.py Description: Manager file for anything plugin related Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ # Built-in imports import io import ssl import sys import json import shutil import zipfile import platform import urllib.request import importlib.util from pathlib import Path from typing import Any, cast # External library imports import certifi # Provides up-to-date CA certificates (security fix for S004) from PySide6.QtCore import QObject, QThread, QUrl, Signal from PySide6.QtGui import QDesktopServices from PySide6.QtWidgets import QMainWindow, QMenu, QMessageBox # Security utilities - prevents path traversal and unsafe operations (S002) from src.shared.security_utils import ( validate_plugin_path, safe_join_path, sanitize_filename, validate_download_path, safe_extract_archive ) # Authentication utilities - S003 fix: add auth to plugin API calls from src.shared.auth_utils import AuthManager, get_default_auth_manager # Logging infrastructure from src.shared.logging_config import get_logger # Initialize module logger _logger = get_logger(__name__) from src.shared.shareddata import APP_NAME, CURRENT_VERSION, PLATFORM_NAME def get_ssl_context() -> ssl.SSLContext: """Returns a cross-platform SSL context configured with certifi's CA bundle.""" return ssl.create_default_context(cafile=certifi.where()) def get_current_platform_id() -> str: """Returns standardized platform identifier (win_x64, darwin_arm64, etc.).""" sys_name = sys.platform arch = platform.machine().lower() if sys_name == "win32": return "win_x64" if "64" in arch or "amd64" in arch else "win_x86" elif sys_name == "darwin": return "darwin_arm64" if "arm" in arch or "aarch64" in arch else "darwin_x64" elif sys_name.startswith("linux"): return "linux_x64" if "64" in arch else "linux_x86" return sys_name def parse_version(v_str: str) -> tuple[int, ...]: """Parses a version string into a comparable integer tuple.""" try: return tuple(int(x) for x in v_str.strip().lstrip("v").split(".")) except ValueError: return (0, 0, 0) class PluginManager(QObject): """ Handles plugin discovery, dynamic loading, menu building, and plugin lifecycle (install, toggle, uninstall). """ plugins_changed = Signal() def __init__(self, main_window: QMainWindow) -> None: super().__init__() self.main_window: QMainWindow = main_window self.plugins_dir: Path = self._resolve_plugins_dir() self.loaded_plugins: list[Any] = [] self.current_platform: str = get_current_platform_id() self.current_app_version: tuple[int, ...] = parse_version(str(CURRENT_VERSION)) def _resolve_plugins_dir(self) -> Path: """Determines the local plugins directory based on execution context.""" if PLATFORM_NAME == "darwin": base_dir = Path(sys.executable).parent / "../../.." else: base_dir = Path.cwd() plugins_path = (base_dir / "plugins").resolve() plugins_path.mkdir(parents=True, exist_ok=True) return plugins_path def reload_plugins(self) -> list[Any]: """Scans directory, imports enabled plugins, and updates the main application menu.""" self.unload_plugins() plugins_dir_str = str(self.plugins_dir) # SECURITY FIX (S002): Validate plugins directory path before use valid, msg = validate_plugin_path(self.plugins_dir, self.plugins_dir) if not valid: print(f"[PluginManager] Invalid plugins directory: {msg}") return [] if plugins_dir_str not in sys.path: sys.path.insert(0, plugins_dir_str) failed_plugins: list[tuple[str, str]] = [] for entry in sorted(self.plugins_dir.iterdir(), key=lambda p: p.name.lower()): # Skip hidden files/dirs and disabled plugins if entry.name.startswith((".", "__")) or entry.name.endswith(".disabled"): continue # SECURITY FIX (S002): Validate each plugin path to prevent traversal valid, msg = validate_plugin_path(entry, self.plugins_dir) if not valid: print(f"[PluginManager] Skipping unsafe plugin: {msg}") failed_plugins.append((entry.name, msg)) continue module_name = "" init_file: Path | None = None if entry.is_dir(): init_file = entry / "__init__.py" module_name = entry.name elif entry.is_file() and entry.suffix == ".py": init_file = entry module_name = entry.stem if not init_file or not init_file.exists(): continue try: instance = self._import_and_instantiate(module_name, init_file) if instance is not None: self.loaded_plugins.append(instance) except Exception as e: failed_plugins.append((module_name, str(e))) # Report failures after all plugins have attempted to load if failed_plugins: error_details = "\n".join(f"• {name}: {err}" for name, err in failed_plugins) summary_msg = f"One or more plugins failed to load:\n\n{error_details}" print(f"[PluginManager] {summary_msg}") parent_widget = getattr(self, "main_window", None) QMessageBox.warning( parent_widget, "Plugin Load Failures", summary_msg ) self.build_plugins_menu() self.plugins_changed.emit() return self.loaded_plugins def _import_and_instantiate(self, module_name: str, file_path: Path) -> Any | None: """Dynamically imports a plugin module, attaches its metadata, and returns the Plugin instance.""" spec = importlib.util.spec_from_file_location(module_name, file_path) if spec is None or spec.loader is None: raise ImportError(f"Invalid or missing spec for file: {file_path}") module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) if hasattr(module, "Plugin") and isinstance(module.Plugin, type): instance = module.Plugin(self.main_window) # Attach manifest metadata directly to the plugin instance if plugin.json exists manifest_path = file_path.parent / "plugin.json" if manifest_path.exists(): try: with open(manifest_path, "r", encoding="utf-8") as f: instance.metadata = json.load(f) # Override class name property if specified in manifest if "name" in instance.metadata: instance.name = instance.metadata["name"] except Exception as e: print(f"[PluginManager] Could not attach metadata to '{module_name}': {e}") return instance return None def build_plugins_menu(self) -> None: """Constructs or refreshes submenus under the top-level 'Plugins' menu item.""" menubar = self.main_window.menuBar() # Dynamically locate the existing 'Plugins' menu on the menubar plugins_menu: QMenu | None = None for action in menubar.actions(): clean_text = action.text().replace("&", "").strip().lower() if clean_text == "plugins": menu = action.menu() if isinstance(menu, QMenu): plugins_menu = menu break # If it doesn't exist yet, create it if plugins_menu is None: plugins_menu = menubar.addMenu("&Plugins") plugins_menu.clear() # 1. Populate loaded plugin submenus if not self.loaded_plugins: disabled_action = plugins_menu.addAction("No active plugins") disabled_action.setEnabled(False) else: for plugin in self.loaded_plugins: plugin_name = getattr(plugin, "name", "Unnamed Plugin") plugin_submenu = plugins_menu.addMenu(plugin_name) if hasattr(plugin, "register_menu") and callable(plugin.register_menu): plugin.register_menu(plugin_submenu) # 2. Append entry to open Plugin Manager GUI plugins_menu.addSeparator() manager_action = plugins_menu.addAction("Manage Plugins...") plugins_gui_func = getattr(self.main_window, "plugins_gui", None) if callable(plugins_gui_func): manager_action.triggered.connect(plugins_gui_func) def unload_plugins(self) -> None: """Clears current active plugin instances.""" self.loaded_plugins.clear() def get_installed_plugins_info(self) -> list[dict[str, Any]]: """Returns metadata for all local plugins in the plugins folder by reading their plugin.json.""" results: list[dict[str, Any]] = [] for entry in sorted(self.plugins_dir.iterdir(), key=lambda p: p.name.lower()): if entry.name.startswith((".", "__")): continue is_disabled = entry.name.endswith(".disabled") clean_name = entry.name[:-9] if is_disabled else entry.name if entry.is_dir() or (entry.is_file() and (entry.suffix == ".py" or entry.name.endswith(".py.disabled"))): if entry.is_file() and entry.stem in ("__init__", "__init__.py"): continue # Load local plugin.json if present manifest: dict[str, Any] = {} manifest_path = (entry / "plugin.json") if entry.is_dir() else None if manifest_path and manifest_path.exists(): try: with open(manifest_path, "r", encoding="utf-8") as f: manifest = json.load(f) except Exception as e: print(f"[PluginManager] Warning: Failed to read manifest for '{entry.name}': {e}") display_name = manifest.get("name", clean_name.removesuffix(".py")) version = manifest.get("version", "1.0.0") description = manifest.get("description", "") author = manifest.get("author", "Unknown") results.append({ "id": manifest.get("id", clean_name), "name": display_name, "version": version, "description": description, "author": author, "path": entry, "is_disabled": is_disabled, "manifest": manifest, }) return results def toggle_plugin_state(self, plugin_path: Path) -> None: """Swaps a plugin between enabled and disabled by renaming with/without '.disabled'.""" if not plugin_path.exists(): return if plugin_path.name.endswith(".disabled"): new_path = plugin_path.with_name(plugin_path.name.removesuffix(".disabled")) else: new_path = plugin_path.with_name(f"{plugin_path.name}.disabled") plugin_path.rename(new_path) self.reload_plugins() def uninstall_plugin(self, plugin_path: Path) -> None: """Deletes a plugin directory or file from disk.""" if not plugin_path.exists(): return if plugin_path.is_dir(): shutil.rmtree(plugin_path) else: plugin_path.unlink() self.reload_plugins() def fetch_remote_repositories(self, repo_urls: list[str]) -> tuple[list[dict[str, Any]], bool]: """Fetches remote plugin metadata from repository URLs.""" aggregated: list[dict[str, Any]] = [] failed_count = 0 for url in repo_urls: try: req = urllib.request.Request( url, headers={"User-Agent": f"{APP_NAME}-PluginManager"}, ) with urllib.request.urlopen(req, timeout=5, context=get_ssl_context()) as response: if response.status == 200: raw_data = response.read().decode("utf-8") data = json.loads(raw_data) if isinstance(data, list): aggregated.extend(cast(list[Any], data)) else: failed_count += 1 except Exception: failed_count += 1 all_failed = failed_count == len(repo_urls) and len(repo_urls) > 0 return aggregated, all_failed def install_plugin_from_url(self, download_url: str, plugin_id: str = "") -> Path: """ Downloads a single-folder ZIP archive and extracts it directly into plugins/. SECURITY FIX (S002): Validates URL, sanitizes paths, prevents zip slip attacks. """ # SECURITY FIX (S004): Validate download URL before fetching valid, msg = validate_download_path(download_url, self.plugins_dir) if not valid: raise RuntimeError(f"Download validation failed: {msg}") req = urllib.request.Request( download_url, headers={"User-Agent": f"{APP_NAME}-PluginManager"}, ) # SECURITY FIX (S004): Use SSL context with certifi CA bundle with urllib.request.urlopen(req, timeout=15, context=get_ssl_context()) as response: if response.status != 200: raise RuntimeError(f"Download failed with HTTP status code {response.status}") zip_bytes = response.read() # SECURITY FIX (S002): Write to temp file first, then validate and extract import tempfile with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp: tmp.write(zip_bytes) tmp_path = Path(tmp.name) try: # SECURITY FIX (S002): Safe extraction prevents zip slip attacks success, extracted_paths = safe_extract_archive(tmp_path, self.plugins_dir) if not success: raise RuntimeError("Archive extraction failed due to security validation") self.reload_plugins() return self.plugins_dir finally: # Cleanup temp file try: tmp_path.unlink() except OSError: pass # Already removed or inaccessible def open_plugins_directory(self) -> bool: """Opens the local plugins folder in the operating system's file manager.""" if self.plugins_dir.exists(): return QDesktopServices.openUrl(QUrl.fromLocalFile(str(self.plugins_dir))) return False def check_for_updates(self, repo_urls: list[str]) -> list[dict[str, Any]]: """ Fetches remote repositories and compares server plugin versions against installed plugins, returning the installed plugins list enriched with update metadata. """ installed_plugins = self.get_installed_plugins_info() remote_data, _ = self.fetch_remote_repositories(repo_urls) # Index remote plugins by their ID (fallback to name if id is missing) remote_map: dict[str, dict[str, Any]] = {} for remote_plugin in remote_data: plugin_id = remote_plugin.get("id") or remote_plugin.get("name") if plugin_id: remote_map[str(plugin_id)] = remote_plugin # Compare versions and annotate each installed plugin dictionary for plugin in installed_plugins: plugin_id = str(plugin.get("id", "")) local_version_str = str(plugin.get("version", "0.0.0")) local_tuple = parse_version(local_version_str) remote_plugin = remote_map.get(plugin_id) if remote_plugin: remote_version_str = str(remote_plugin.get("version", "0.0.0")) remote_tuple = parse_version(remote_version_str) if remote_tuple > local_tuple: plugin["update_available"] = True plugin["remote_version"] = remote_version_str plugin["download_url"] = str(remote_plugin.get("download_url", "")) else: plugin["update_available"] = False plugin["remote_version"] = remote_version_str plugin["download_url"] = "" else: plugin["update_available"] = False plugin["remote_version"] = None plugin["download_url"] = "" return installed_plugins class StartupUpdateWorker(QThread): updates_checked = Signal(list) def __init__(self, manager: Any, repo_urls: list[str]) -> None: super().__init__() self.manager = manager self.repo_urls = repo_urls def run(self) -> None: try: print("[StartupWorker] Starting background update check...") installed_plugins = self.manager.get_installed_plugins_info() print(f"[StartupWorker] Found {len(installed_plugins)} installed local plugins.") remote_data, all_failed = self.manager.fetch_remote_repositories(self.repo_urls) print(f"[StartupWorker] Fetch remote complete. Failed: {all_failed}, Items fetched: {len(remote_data)}") if all_failed or not remote_data: print("[StartupWorker] Aborting: Remote fetch failed or returned empty data.") return remote_map: dict[str, dict[str, Any]] = {} for raw_plugin in remote_data: if isinstance(raw_plugin, dict): remote_plugin = cast(dict[str, Any], raw_plugin) plugin_id = remote_plugin.get("id") or remote_plugin.get("name") if isinstance(plugin_id, str): remote_map[plugin_id] = remote_plugin outdated_plugins: list[dict[str, Any]] = [] for plugin in installed_plugins: plugin_id = str(plugin.get("id", "")) local_version_str = str(plugin.get("version", "0.0.0")) local_tuple = parse_version(local_version_str) remote_plugin = remote_map.get(plugin_id) if remote_plugin: remote_version_str = str(remote_plugin.get("version", "0.0.0")) remote_tuple = parse_version(remote_version_str) print(f"[StartupWorker] Comparing '{plugin.get('name')}': Local v{local_version_str} vs Remote v{remote_version_str}") if remote_tuple > local_tuple: plugin["remote_version"] = remote_version_str plugin["download_url"] = str(remote_plugin.get("download_url", "")) outdated_plugins.append(plugin) print(f"[StartupWorker] Outdated plugins found: {len(outdated_plugins)}") self.updates_checked.emit(outdated_plugins) except Exception as e: print(f"[StartupWorker] Error during startup update check: {e}")