"""Plugin manager - main entry point for plugin system.""" from pathlib import Path from typing import Dict, List, Optional from flask import Flask import logging from .base import BasePlugin, PluginMeta from .registry import PluginRegistry, PluginState from .loader import PluginLoader from .migrations import PluginMigrationManager from ..exceptions import PluginError logger = logging.getLogger(__name__) __all__ = [ 'PluginManager', 'BasePlugin', 'PluginMeta', 'PluginRegistry', 'PluginState', 'plugin_manager' ] class PluginManager: """ Central manager for all plugin operations. Usage: plugin_manager = PluginManager() plugin_manager.init_app(app, db) # In CLI: plugin_manager.install_plugin('printers') """ def __init__(self): self.registry: Optional[PluginRegistry] = None self.loader: Optional[PluginLoader] = None self.migration_manager: Optional[PluginMigrationManager] = None self.verifier = None self._app: Optional[Flask] = None self._db = None # API prefixes already claimed by a registered plugin blueprint, to # detect two plugins overlapping on the same /api/... namespace. self._registered_prefixes: set = set() def init_app(self, app: Flask, db) -> None: """Initialize plugin manager with Flask app.""" self._app = app self._db = db # Reset per-app so the prefix-uniqueness guard tracks only this app's # registrations (the manager is a process-wide singleton; tests build # multiple apps from it). self._registered_prefixes = set() # Setup paths instance_path = Path(app.instance_path) plugins_dir = Path(app.root_path).parent / 'plugins' # Initialize components self.registry = PluginRegistry(instance_path / 'plugins.json') self.loader = PluginLoader(plugins_dir, self.registry) self.migration_manager = PluginMigrationManager( plugins_dir, app.config.get('SQLALCHEMY_DATABASE_URI') ) # One trust policy, shared by verify-at-load and verify-at-migrate. from .verification import PluginVerifier verifier = PluginVerifier( plugins_dir, require_signed=app.config.get('PLUGIN_REQUIRE_SIGNED', False), trusted_key_paths=app.config.get('PLUGIN_TRUSTED_KEYS', []), dev_trust_dirs=app.config.get('PLUGIN_DEV_TRUST_DIRS', []), is_dev=bool(app.config.get('DEBUG') or app.config.get('TESTING')), ) self.verifier = verifier self.loader.verifier = verifier self.migration_manager.verifier = verifier # Install (or clear) the import guard that verifies every `plugins.*` # import against a trusted signature. This is what covers the core # request handlers that do `from plugins..models import ...`, which # never pass through the loader (findings #1, #2). from . import importguard if verifier.require_signed: importguard.install(plugins_dir, verifier) else: importguard.uninstall() # Load enabled plugins self._load_enabled_plugins() # Store on app for access app.extensions['plugin_manager'] = self def _load_enabled_plugins(self) -> None: """Load and register all enabled plugins.""" plugins = self.loader.load_enabled_plugins(self._app, self._db) for name, plugin in plugins.items(): self._register_plugin_components(plugin) def upgrade_all_plugins(self) -> Dict[str, str]: """Run pending Alembic migrations for every discovered plugin. Returns {plugin_name: 'ok'|'no-migrations'|}. Skips plugins with no migrations/ directory. Driven by the CLI (`flask plugin upgrade-all`), which every deploy runs after `flask db upgrade`. Each bundled plugin's chain begins with a stamp-only anchor (the core chain already built its tables); later per-plugin migrations extend that chain. See ADR-008. Idempotent. """ results: Dict[str, str] = {} if not self.migration_manager: return results # Only ADOPTED plugins (those in the registry) get migrated. A plugin # folder that merely sits on disk unadopted must not have its DDL run as # a side effect of a deploy - migrations execute with full DB rights, so # "on disk" is not "trusted to run". (The old code probed a # list_installed() method the registry never had, so it always fell # through to discover_plugins() and migrated every folder.) plugin_names = list(self.registry.get_all().keys()) for name in plugin_names: if not self.migration_manager.has_pending_migrations(name): results[name] = 'no-migrations' continue try: ok = self.migration_manager.run_plugin_migrations(name) results[name] = 'ok' if ok else 'failed' except Exception as ex: results[name] = f'error: {ex}' return results def _register_plugin_components(self, plugin: BasePlugin) -> None: """Register plugin's blueprint, models, CLI commands, etc.""" # Register blueprint blueprint = plugin.get_blueprint() if blueprint: prefix = plugin.meta.api_prefix # Guard against two plugins claiming the same API prefix; Flask only # rejects duplicate blueprint names, not overlapping url_prefixes, so # an overlap would silently shadow routes. if prefix in self._registered_prefixes: raise ValueError( f"Plugin {plugin.meta.name} api_prefix '{prefix}' is already " f"claimed by another blueprint" ) self._app.register_blueprint(blueprint, url_prefix=prefix) self._registered_prefixes.add(prefix) logger.debug(f"Registered blueprint: {prefix}") # Register CLI commands for cmd in plugin.get_cli_commands(): self._app.cli.add_command(cmd) def discover_available(self) -> List[Dict]: """ Get list of all available plugins (installed or not). Returns list of plugin info dicts. """ available = [] for name in self.loader.discover_plugins(): # Under enforcement load_plugin_class refuses an unverified plugin # (it will not import it). Skip such a plugin from the listing # instead of failing the whole call. try: plugin_class = self.loader.load_plugin_class(name) except PluginError as e: logger.warning(f"Skipping plugin {name} in listing: {e}") continue if plugin_class: try: temp = plugin_class() meta = temp.meta state = self.registry.get(name) try: config_schema = temp.get_config_schema() except Exception: config_schema = [] try: provisioning_note = temp.get_provisioning_note() except Exception: provisioning_note = None manifest = self.loader.load_manifest(name) available.append({ 'name': meta.name, # Human label for UIs. Manifest display_name wins; else # title-case the machine name (right for most, e.g. # "computers" -> "Computers"). 'displayname': (manifest.get('display_name') or meta.name.replace('_', ' ').title()), 'version': meta.version, 'description': meta.description, 'author': meta.author, 'dependencies': meta.dependencies, 'installed': state is not None, 'enabled': state.enabled if state else False, 'installedat': state.installed_at if state else None, 'config_schema': config_schema, 'provisioning_note': provisioning_note, 'default_enabled': manifest.get('default_enabled', True), }) except Exception as e: logger.warning(f"Error inspecting plugin {name}: {e}") return available def install_plugin(self, name: str, run_migrations: bool = True) -> bool: """ Install a plugin. Steps: 1. Verify plugin exists 2. Check dependencies 3. Run database migrations 4. Register in registry 5. Call plugin's on_install hook """ # Check if already installed if self.registry.is_installed(name): logger.warning(f"Plugin {name} is already installed") return False # Read metadata from the manifest (single source of truth) instead of # instantiating the plugin class just to inspect deps/version. manifest = self.loader.load_manifest(name) if not manifest: logger.error(f"Plugin {name} not found") return False manifest_version = manifest.get('version') # Check dependencies for dep in manifest.get('dependencies', []): if not self.registry.is_installed(dep): logger.error( f"Plugin {name} requires {dep} to be installed first" ) return False # Run migrations if run_migrations: success = self.migration_manager.run_plugin_migrations(name) if not success: logger.error(f"Failed to run migrations for {name}") return False # Register plugin # Plugins that provision extra tables install disabled until a site # opts in (manifest default_enabled=false). self.registry.register(name, manifest_version, enabled=manifest.get('default_enabled', True)) # Load the plugin plugin = self.loader.load_plugin(name, self._app, self._db) if plugin: self._register_plugin_components(plugin) self._seed_plugin_permissions(plugin) plugin.on_install(self._app) logger.info(f"Installed plugin: {name} v{manifest_version}") return True def _seed_plugin_permissions(self, plugin: BasePlugin) -> None: """Idempotently create Permission rows for a plugin's declared perms. Runs at install and enable so a fresh plugin's RBAC permissions exist without a separate `flask seed permissions` pass. Best-effort: a plugin that raises must not abort the lifecycle.""" try: entries = plugin.get_permissions() or [] except Exception: logger.exception( "get_permissions failed for %s", plugin.meta.name) return if not entries: return from shopdb.core.models import Permission with self._app.app_context(): created = Permission.seed_entries(entries) if created: self._db.session.commit() logger.info( "Seeded %d permission(s) for plugin %s", created, plugin.meta.name) def _is_core_tier(self, name: str) -> bool: """True when the plugin's manifest marks it tier=core (mandatory). A core-tier plugin refuses uninstall/disable so a site cannot remove a plugin its deployment depends on. No plugin ships tier=core today; the guard makes that a manifest edit, not a code change (ADR-013). """ try: return self.loader.load_manifest(name).get('tier') == 'core' except PluginError: return False def _installed_dependents(self, name: str, enabled_only: bool = False) -> List[str]: """Installed plugins that declare `name` as a hard dependency. Reads each dependency list from the manifest on DISK, not from a loaded plugin instance, so a dependent that is installed-but-not-loaded (e.g. disabled, or failed to load) is still counted. enabled_only restricts to currently-enabled dependents (what matters when disabling, since a disabled dependent is not running). """ dependents = [] for other_name, state in self.registry.get_all().items(): if other_name == name: continue if enabled_only and not state.enabled: continue try: manifest = self.loader.load_manifest(other_name) except PluginError: continue if name in manifest.get('dependencies', []): dependents.append(other_name) return dependents def uninstall_plugin(self, name: str, remove_data: bool = False) -> bool: """ Uninstall a plugin. Args: name: Plugin name remove_data: If True, run downgrade migrations to remove tables """ if not self.registry.is_installed(name): logger.warning(f"Plugin {name} is not installed") return False if self._is_core_tier(name): logger.error(f"Cannot uninstall {name}: it is a core-tier plugin") return False # Refuse if any INSTALLED plugin (enabled or not) depends on this one. dependents = self._installed_dependents(name) if dependents: logger.error( f"Cannot uninstall {name}: {', '.join(dependents)} depend(s) on it" ) return False # Get plugin instance plugin = self.loader.get_loaded_plugin(name) # Call on_uninstall hook if plugin: plugin.on_uninstall(self._app) # Optionally remove data if remove_data: self.migration_manager.downgrade_plugin(name) # Unregister self.registry.unregister(name) logger.info(f"Uninstalled plugin: {name}") return True def enable_plugin(self, name: str) -> bool: """Enable a disabled plugin.""" if not self.registry.is_installed(name): logger.error(f"Plugin {name} is not installed") return False if self.registry.is_enabled(name): logger.info(f"Plugin {name} is already enabled") return True # Check dependencies are enabled. Read deps from the manifest, not by # instantiating the plugin class (manifest is the single source of # truth; instantiating fires __init__ side effects unnecessarily). manifest = self.loader.load_manifest(name) for dep in manifest.get('dependencies', []): if not self.registry.is_enabled(dep): logger.error(f"Cannot enable {name}: {dep} is not enabled") return False self.registry.enable(name) # Surface (do not auto-apply) a plugin chain that is ahead of the DB, so # an operator enabling a plugin knows to run `flask plugin upgrade-all`. try: if self.migration_manager and \ self.migration_manager.has_unapplied_migrations(name): logger.warning( f"Plugin {name} has unapplied migrations; " f"run 'flask plugin upgrade-all'" ) except Exception: logger.debug(f"Could not check migration state for {name}") # Fire the on_enable hook best-effort. Do NOT register the blueprint # here: Flask forbids register_blueprint after the first request, so # routes/nav for a re-enabled plugin take effect on the next restart # (symmetric with disable). try: plugin = self.loader.load_plugin(name, self._app, self._db) if plugin: self._seed_plugin_permissions(plugin) plugin.on_enable(self._app) except Exception: logger.exception(f"on_enable hook failed for plugin {name}") logger.info(f"Enabled plugin: {name}") return True def disable_plugin(self, name: str) -> bool: """Disable an enabled plugin.""" if not self.registry.is_enabled(name): logger.info(f"Plugin {name} is already disabled") return True if self._is_core_tier(name): logger.error(f"Cannot disable {name}: it is a core-tier plugin") return False # Refuse if any ENABLED plugin depends on this one (a disabled dependent # is not running, so it does not block). Manifests read from disk. dependents = self._installed_dependents(name, enabled_only=True) if dependents: logger.error( f"Cannot disable {name}: {', '.join(dependents)} depend(s) on it" ) return False plugin = self.loader.get_loaded_plugin(name) if plugin: plugin.on_disable(self._app) self.registry.disable(name) logger.info(f"Disabled plugin: {name}") return True def _dependency_closure(self, names: List[str]) -> List[str]: """Chosen plugins plus their hard-dependency closure, dependencies first. Post-order DFS over manifest `dependencies`, so the returned order is safe to install/enable top to bottom (a dependency always precedes the plugin that needs it). Deduplicated. """ closure: List[str] = [] seen = set() def add(name): if name in seen: return seen.add(name) try: deps = self.loader.load_manifest(name).get('dependencies', []) except PluginError: deps = [] for dep in deps: add(dep) closure.append(name) for name in names: add(name) return closure def apply_profile(self, plugins: List[str], locked: Optional[List[str]] = None) -> Dict[str, List[str]]: """Install AND enable exactly the chosen plugins plus their dependency closure, in dependency order. Declarative replacement for the hand-ordered install/enable sequences in the deploy runbooks (which could enable a plugin that was never installed). Idempotent: already-installed/enabled plugins are left as-is. This does NOT remove anything absent from the list - removal stays an explicit, separate operation. `locked` is accepted for forward compatibility (a future guard against removing a site-mandated plugin) and is currently informational. Raises PluginError if the profile names a plugin (or pulls in a dependency) that does not exist on disk. Returns {installed: [...], enabled: [...], already: [...]}. """ available = set(self.loader.discover_plugins()) unknown = [p for p in plugins if p not in available] if unknown: raise PluginError( f"Profile names unknown plugins: {', '.join(sorted(unknown))}") order = self._dependency_closure(plugins) missing_deps = [p for p in order if p not in available] if missing_deps: raise PluginError( f"Missing dependency plugins: {', '.join(sorted(missing_deps))}") result: Dict[str, List[str]] = { 'installed': [], 'enabled': [], 'already': []} for name in order: changed = False if not self.registry.is_installed(name): if self.install_plugin(name): result['installed'].append(name) changed = True if not self.registry.is_enabled(name): if self.enable_plugin(name): result['enabled'].append(name) changed = True if not changed: result['already'].append(name) return result def get_plugin(self, name: str) -> Optional[BasePlugin]: """Get a loaded plugin instance.""" return self.loader.get_loaded_plugin(name) def get_all_plugins(self) -> Dict[str, BasePlugin]: """Get all loaded plugins.""" return self.loader.get_all_loaded() def get_service(self, name: str): """Resolve a service exposed by an enabled plugin via get_services(). Consumer for the BasePlugin.get_services hook: searches enabled plugins for one that registers `name` and returns the registered value (a service class or factory). Returns None if no enabled plugin provides it. This is how one plugin obtains another's service (e.g. the Zabbix service). """ for plugin_name, plugin in self.get_all_plugins().items(): if not self.registry.is_enabled(plugin_name): continue try: services = plugin.get_services() or {} except Exception: continue if name in services: return services[name] return None # Global plugin manager instance plugin_manager = PluginManager()