""" Plugin Loader Handles plugin module imports, dependency installation, and class instantiation. Extracted from PluginManager to improve separation of concerns. """ import importlib import importlib.metadata import importlib.util import json import os import sys import subprocess import threading from pathlib import Path from typing import Dict, Any, Optional, Tuple, Type import logging from packaging.requirements import InvalidRequirement, Requirement from src.exceptions import PluginError from src.logging_config import get_logger def requirements_has_real_deps(requirements_file: str) -> bool: """ Check whether a requirements.txt actually specifies anything to install. Plugins that ship all their dependencies with LEDMatrix core often keep a requirements.txt where every line is commented out, for documentation purposes only. Running pip against such a file still pays the full subprocess/resolver cost for zero effect, so callers should skip the install step entirely when this returns False. """ try: with open(requirements_file, 'r', encoding='utf-8') as fh: for line in fh: line = line.strip() if line and not line.startswith('#'): return True except OSError: # Let the caller's own file handling report the error. return True return False def requirements_are_satisfied(requirements_file: str) -> bool: """ Check whether every real requirement line in requirements.txt is already satisfied by packages installed in the current interpreter. This replaces marker-file tracking with a direct fact check, so it's immune to stale/missing/corrupted markers: it looks at what's actually importable right now rather than trusting a hash comparison from a previous run. Anything ambiguous (pip options, unparseable lines, extras, unresolvable versions) conservatively returns False so the caller falls through to running pip — this check only ever saves work, never masks a real install. """ try: with open(requirements_file, 'r', encoding='utf-8') as fh: lines = fh.readlines() except OSError: return False for raw_line in lines: line = raw_line.strip() if not line or line.startswith('#'): continue if line.startswith('-'): return False # pip option (-r, --index-url, ...), can't verify try: req = Requirement(line) except InvalidRequirement: return False if req.extras: return False # verifying extras' sub-dependencies isn't worth it here if req.marker is not None and not req.marker.evaluate(): continue # not applicable on this platform/interpreter try: installed_version = importlib.metadata.version(req.name) except importlib.metadata.PackageNotFoundError: return False if req.specifier and not req.specifier.contains(installed_version, prereleases=True): return False return True def find_trusted_subdir(trusted_dir: str, name: str) -> Optional[str]: """Return `name` if it names an actual subdirectory of trusted_dir, else None. Used as a containment check for a directory name derived from untrusted input (a manifest-declared plugin id, an externally-supplied plugin path): the returned value always comes from enumerating trusted_dir itself via os.scandir(), so a caller that builds a path by joining trusted_dir with this return value is joining against a name the filesystem produced under a trusted root -- not the caller's original string, which could otherwise smuggle a traversal sequence through. """ try: with os.scandir(trusted_dir) as entries: for entry in entries: if entry.name == name and entry.is_dir(): return entry.name except OSError: pass return None class PluginLoader: """Handles plugin module loading and class instantiation.""" def __init__(self, logger: Optional[logging.Logger] = None) -> None: """ Initialize the plugin loader. Args: logger: Optional logger instance """ self.logger = logger or get_logger(__name__) self._loaded_modules: Dict[str, Any] = {} self._plugin_module_registry: Dict[str, set] = {} # Maps plugin_id to set of module names # Lock to serialize module loading when plugins share module names # (e.g., scroll_display.py, game_renderer.py across sport plugins). # During exec_module, bare-name sub-modules temporarily appear in # sys.modules; the lock prevents concurrent plugins from seeing each # other's entries. After exec_module, _namespace_plugin_modules # moves those bare names to namespaced keys (e.g. # _plg_basketball_scoreboard_scroll_display) so they never collide. self._module_load_lock = threading.Lock() def find_plugin_directory( self, plugin_id: str, plugins_dir: Path, plugin_directories: Optional[Dict[str, Path]] = None ) -> Optional[Path]: """ Find the plugin directory for a given plugin ID. Tries multiple strategies: 1. Use plugin_directories mapping if available 2. Direct path matching 3. Case-insensitive directory matching 4. Manifest-based search Args: plugin_id: Plugin identifier plugins_dir: Base plugins directory plugin_directories: Optional mapping of plugin_id to directory Returns: Path to plugin directory or None if not found """ # Sanitize plugin_id — os.path.basename is a CodeQL-recognized path sanitizer plugin_id = os.path.basename(plugin_id or '') if not plugin_id: return None # Strategy 1: Use mapping from discovery if plugin_directories and plugin_id in plugin_directories: plugin_dir = plugin_directories[plugin_id] if plugin_dir.exists(): self.logger.debug("Using plugin directory from discovery mapping: %s", plugin_dir) return plugin_dir # Strategy 2: Direct paths — resolve and validate they stay within plugins_dir plugins_dir_resolved = plugins_dir.resolve() for _candidate_name in (plugin_id, f"ledmatrix-{plugin_id}"): _candidate = (plugins_dir_resolved / _candidate_name).resolve() try: _candidate.relative_to(plugins_dir_resolved) except ValueError: continue if _candidate.exists(): return _candidate # Strategy 3: Case-insensitive search normalized_id = plugin_id.lower() for item in plugins_dir.iterdir(): if not item.is_dir(): continue item_name = item.name if item_name.lower() == normalized_id: return item if item_name.lower() == f"ledmatrix-{plugin_id}".lower(): return item # Strategy 4: Manifest-based search self.logger.debug("Directory name search failed for %s, searching by manifest...", plugin_id) for item in plugins_dir.iterdir(): if not item.is_dir(): continue # Skip if already checked if item.name.lower() == normalized_id or item.name.lower() == f"ledmatrix-{plugin_id}".lower(): continue manifest_path = item / "manifest.json" if manifest_path.exists(): try: with open(manifest_path, 'r', encoding='utf-8') as f: item_manifest = json.load(f) item_manifest_id = item_manifest.get('id') if item_manifest_id == plugin_id: self.logger.info( "Found plugin %s in directory %s (manifest ID matches)", plugin_id, item.name ) return item except (json.JSONDecodeError, Exception) as e: self.logger.debug("Skipping %s due to manifest error: %s", item.name, e) continue return None def install_dependencies( self, plugin_dir: Path, plugin_id: str, plugins_dir: Path, timeout: int = 300 ) -> bool: """ Install plugin dependencies from requirements.txt. Args: plugin_dir: Plugin directory path plugin_id: Plugin identifier plugins_dir: Trusted base plugins directory for path containment check. Required (not optional) so every caller reconstructs the plugin path through the sanitiser below rather than trusting plugin_dir directly -- CodeQL's path-injection query (and a malicious manifest/plugin_id in practice) can't tell a legitimate plugin_dir from one crafted to traverse outside plugins_dir. timeout: Installation timeout in seconds Returns: True if dependencies installed or not needed, False on error """ plugin_id = os.path.basename(plugin_id or '') if not plugin_id: return False # Resolve to a canonical absolute path (normalises .. and symlinks) plugin_dir_real = os.path.realpath(str(plugin_dir)) plugins_dir_real = os.path.realpath(str(plugins_dir)) requested_name = os.path.basename(plugin_dir_real) # Match the requested directory against an entry actually enumerated # from the trusted plugins_dir, and build the path from that entry -- # not from requested_name. A name that came out of os.scandir() on a # trusted root carries no taint regardless of what the caller asked # for, so this is a real containment guarantee (an allowlist check # against a trusted source), not a string-sanitisation of untrusted # input that a static analyzer has to trust blindly. matched_name = find_trusted_subdir(plugins_dir_real, requested_name) if matched_name is None: self.logger.error( "Plugin directory for %s not found inside plugins dir", plugin_id ) return False safe_plugin_dir = os.path.join(plugins_dir_real, matched_name) requirements_file = os.path.join(safe_plugin_dir, "requirements.txt") if not os.path.isfile(requirements_file): return True # No dependencies needed if not requirements_has_real_deps(requirements_file): self.logger.debug( "requirements.txt for %s has no real dependencies (comments/blank only), skipping pip", plugin_id ) return True if requirements_are_satisfied(requirements_file): self.logger.debug( "Dependencies for %s already satisfied in current environment, skipping pip", plugin_id ) return True try: self.logger.info("Installing dependencies for plugin %s...", plugin_id) result = subprocess.run( [sys.executable, "-m", "pip", "install", "--break-system-packages", "-r", requirements_file], capture_output=True, text=True, timeout=timeout, check=False ) if result.returncode == 0: self.logger.info("Dependencies installed successfully for %s", plugin_id) return True else: stderr = result.stderr or "" # uninstall-no-record-file means a system-managed copy of a package # (e.g. apt's python3-requests, which ships no pip RECORD file) is in # the way of the version this requirements.txt pins. Retry with # --ignore-installed so pip lays the pinned version down alongside # the system copy instead of trying to replace it — matching the # retry already used by install_dependencies_apt.py / safe_pip_install.sh. # Without this retry, the plugin would silently keep running against # whatever version the system happened to ship. if "uninstall-no-record-file" in stderr: self.logger.warning( "Dependencies for %s conflict with a system-managed package " "(no pip RECORD); retrying with --ignore-installed: %s", plugin_id, stderr.strip() ) # Wrapped in its own try/except so a retry timeout is # tolerated the same way as a retry failure, instead of # propagating to the outer handler and returning False # (which would contradict the "assume satisfied" fallback # below). try: # sys.executable is this process's own interpreter (not # attacker-influenced), and requirements_file is a path # built internally by find_plugin_directory, never raw # external input. retry_result = subprocess.run( # nosec B603 - no shell invoked (list-form argv) # nosemgrep [sys.executable, "-m", "pip", "install", "--break-system-packages", "--ignore-installed", "-r", requirements_file], capture_output=True, text=True, timeout=timeout, check=False ) if retry_result.returncode != 0: self.logger.warning( "Retry with --ignore-installed also failed for %s; assuming the " "system-managed version satisfies the requirement: %s", plugin_id, (retry_result.stderr or "").strip() ) except subprocess.TimeoutExpired: self.logger.warning( "Retry with --ignore-installed timed out for %s; assuming the " "system-managed version satisfies the requirement", plugin_id ) return True self.logger.warning( "Dependency installation returned non-zero exit code for %s: %s", plugin_id, stderr ) return False except subprocess.TimeoutExpired: self.logger.error("Dependency installation timed out for %s", plugin_id) return False except FileNotFoundError: self.logger.warning("pip not found. Skipping dependency installation for %s", plugin_id) return True except (BrokenPipeError, OSError) as e: # Handle broken pipe errors (errno 32) which can occur during pip downloads # Often caused by network interruptions or output buffer issues if isinstance(e, OSError) and e.errno == 32: self.logger.error( "Broken pipe error during dependency installation for %s. " "This usually indicates a network interruption or pip output buffer issue. " "Try installing again or check your network connection.", plugin_id ) else: self.logger.error("OS error during dependency installation for %s: %s", plugin_id, e) return False except Exception as e: self.logger.error("Unexpected error installing dependencies for %s: %s", plugin_id, e, exc_info=True) return False @staticmethod def _iter_plugin_bare_modules( plugin_dir: Path, before_keys: set ) -> list: """Return bare-name modules from plugin_dir added after before_keys. Returns a list of (mod_name, module) tuples for modules that: - Were added to sys.modules after before_keys snapshot - Have bare names (no dots) - Have a ``__file__`` inside plugin_dir """ resolved_dir = plugin_dir.resolve() result = [] for key in set(sys.modules.keys()) - before_keys: if "." in key: continue mod = sys.modules.get(key) if mod is None: continue mod_file = getattr(mod, "__file__", None) if not mod_file: continue try: if Path(mod_file).resolve().is_relative_to(resolved_dir): result.append((key, mod)) except (ValueError, TypeError): continue return result def _evict_stale_bare_modules(self, plugin_dir: Path) -> dict: """Temporarily remove bare-name sys.modules entries from other plugins. Before exec_module, scan the current plugin directory for .py files. For each, if sys.modules has a bare-name entry whose ``__file__`` lives in a *different* directory, remove it so Python's import system will load the current plugin's version instead of reusing the stale cache. Returns: Dict mapping evicted module names to their module objects (for restoration on error). """ resolved_dir = plugin_dir.resolve() evicted: dict = {} for py_file in plugin_dir.glob("*.py"): mod_name = py_file.stem if mod_name.startswith("_"): continue existing = sys.modules.get(mod_name) if existing is None: continue existing_file = getattr(existing, "__file__", None) if not existing_file: continue try: if not Path(existing_file).resolve().is_relative_to(resolved_dir): evicted[mod_name] = sys.modules.pop(mod_name) self.logger.debug( "Evicted stale bare-name module '%s' before loading plugin", mod_name, ) except (ValueError, TypeError): continue return evicted def _namespace_plugin_modules( self, plugin_id: str, plugin_dir: Path, before_keys: set ) -> None: """ Move bare-name plugin modules to namespaced keys in sys.modules. After exec_module loads a plugin's entry point, Python will have added the plugin's local modules (scroll_display, game_renderer, …) to sys.modules under their bare names. This method renames them to ``_plg__`` so they cannot collide with identically- named modules from other plugins. The plugin code keeps working because ``from scroll_display import X`` binds ``X`` to the class *object*, not to the sys.modules entry. Args: plugin_id: Plugin identifier plugin_dir: Plugin directory path before_keys: Snapshot of sys.modules keys taken *before* exec_module """ safe_id = plugin_id.replace("-", "_") namespaced_names: set = set() for mod_name, mod in self._iter_plugin_bare_modules(plugin_dir, before_keys): namespaced = f"_plg_{safe_id}_{mod_name}" sys.modules[namespaced] = mod # Remove the bare sys.modules entry. The module object stays # alive via the namespaced key and all existing Python-level # bindings (``from scroll_display import X`` already bound X # to the class object). Leaving bare entries would cause the # NEXT plugin's exec_module to find the cached entry and reuse # it instead of loading its own version. sys.modules.pop(mod_name, None) namespaced_names.add(namespaced) self.logger.debug( "Namespace-isolated module '%s' -> '%s' for plugin %s", mod_name, namespaced, plugin_id, ) # Track for cleanup during unload self._plugin_module_registry[plugin_id] = namespaced_names if namespaced_names: self.logger.info( "Namespace-isolated %d module(s) for plugin %s", len(namespaced_names), plugin_id, ) def unregister_plugin_modules(self, plugin_id: str) -> None: """Remove namespaced sub-modules and cached module for a plugin from sys.modules. Called by PluginManager during unload to clean up all module entries that were created when the plugin was loaded. """ for ns_name in self._plugin_module_registry.pop(plugin_id, set()): sys.modules.pop(ns_name, None) self._loaded_modules.pop(plugin_id, None) def load_module( self, plugin_id: str, plugin_dir: Path, entry_point: str ) -> Optional[Any]: """ Load a plugin module from file. Module loading is serialized via _module_load_lock because plugins are loaded in parallel (ThreadPoolExecutor) and multiple sport plugins share identically-named local modules (scroll_display.py, game_renderer.py, sports.py, etc.). After loading, bare-name modules from the plugin directory are moved to namespaced keys in sys.modules (e.g. ``_plg_basketball_scoreboard_scroll_display``) so they cannot collide with other plugins. Args: plugin_id: Plugin identifier plugin_dir: Plugin directory path entry_point: Entry point filename (e.g., 'manager.py') Returns: Loaded module or None on error """ plugin_id = os.path.basename(plugin_id or '') if not plugin_id: raise PluginError("Invalid plugin ID") try: plugin_dir_resolved = plugin_dir.resolve(strict=True) except OSError: raise PluginError("Plugin directory not found", plugin_id=plugin_id) entry_file = (plugin_dir_resolved / entry_point).resolve() try: entry_file.relative_to(plugin_dir_resolved) except ValueError: raise PluginError("Invalid entry point path", plugin_id=plugin_id) if not entry_file.exists(): error_msg = f"Entry point file not found for plugin {plugin_id}" self.logger.error(error_msg) raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)}) with self._module_load_lock: # Add plugin directory to sys.path if not already there plugin_dir_str = str(plugin_dir) if plugin_dir_str not in sys.path: sys.path.insert(0, plugin_dir_str) self.logger.debug("Added plugin %s's directory to sys.path", plugin_id) # Import the plugin module module_name = f"plugin_{plugin_id.replace('-', '_')}" # Check if already loaded if module_name in sys.modules: self.logger.debug("Module %s already loaded, reusing", module_name) return sys.modules[module_name] spec = importlib.util.spec_from_file_location(module_name, entry_file) if spec is None or spec.loader is None: self.logger.error("Could not create module spec for plugin %s", plugin_id) error_msg = f"Could not create module spec for {entry_file}" raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)}) module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module # Snapshot AFTER inserting the main module so that # _namespace_plugin_modules and error cleanup only target # sub-modules, not the main module entry itself. before_keys = set(sys.modules.keys()) # Evict stale bare-name modules from other plugin directories # so Python's import system loads fresh copies from this plugin. evicted = self._evict_stale_bare_modules(plugin_dir) try: spec.loader.exec_module(module) # Move bare-name plugin modules to namespaced keys so they # cannot collide with identically-named modules from other plugins self._namespace_plugin_modules(plugin_id, plugin_dir, before_keys) except Exception: # Restore evicted modules so other plugins are unaffected for evicted_name, evicted_mod in evicted.items(): if evicted_name not in sys.modules: sys.modules[evicted_name] = evicted_mod # Clean up the partially-initialized main module and any # bare-name sub-modules that were added during exec_module # so they don't leak into subsequent plugin loads. sys.modules.pop(module_name, None) for key, _ in self._iter_plugin_bare_modules(plugin_dir, before_keys): sys.modules.pop(key, None) raise self._loaded_modules[plugin_id] = module self.logger.debug("Loaded module %s for plugin %s", module_name, plugin_id) return module def get_plugin_class( self, plugin_id: str, module: Any, class_name: str ) -> Type[Any]: """ Get the plugin class from a loaded module. Args: plugin_id: Plugin identifier module: Loaded module class_name: Name of the plugin class Returns: Plugin class Raises: PluginError: If class not found """ if not hasattr(module, class_name): error_msg = f"Class {class_name} not found in module for plugin {plugin_id}" self.logger.error(error_msg) raise PluginError( error_msg, plugin_id=plugin_id, context={'class_name': class_name, 'module': module.__name__} ) plugin_class = getattr(module, class_name) # Verify it's a class if not isinstance(plugin_class, type): error_msg = f"{class_name} is not a class in module for plugin {plugin_id}" self.logger.error(error_msg) raise PluginError(error_msg, plugin_id=plugin_id, context={'class_name': class_name}) return plugin_class def instantiate_plugin( self, plugin_id: str, plugin_class: Type[Any], config: Dict[str, Any], display_manager: Any, cache_manager: Any, plugin_manager: Any ) -> Any: """ Instantiate a plugin class. Args: plugin_id: Plugin identifier plugin_class: Plugin class to instantiate config: Plugin configuration display_manager: Display manager instance cache_manager: Cache manager instance plugin_manager: Plugin manager instance Returns: Plugin instance Raises: PluginError: If instantiation fails """ try: plugin_instance = plugin_class( plugin_id=plugin_id, config=config, display_manager=display_manager, cache_manager=cache_manager, plugin_manager=plugin_manager ) self.logger.debug("Instantiated plugin %s", plugin_id) return plugin_instance except Exception as e: error_msg = f"Failed to instantiate plugin {plugin_id}: {e}" self.logger.error(error_msg, exc_info=True) raise PluginError(error_msg, plugin_id=plugin_id) from e @staticmethod def _parse_semver(value: Any) -> Optional[Tuple[int, int, int]]: """Parse 'X.Y.Z' (extra parts/suffixes ignored) into a comparable 3-tuple, or None when unparseable.""" if not isinstance(value, str): return None parts = value.strip().lstrip('v').split('.') try: nums = [int(''.join(ch for ch in p if ch.isdigit()) or 0) for p in parts[:3]] except ValueError: return None while len(nums) < 3: nums.append(0) return tuple(nums) # type: ignore[return-value] def _warn_if_incompatible(self, plugin_id: str, manifest: Dict[str, Any]) -> None: """Log one warning when a plugin declares a minimum LEDMatrix version newer than the running core. Advisory only — never raises — so a plugin that guards optional features with try/except keeps working. """ declared = ( manifest.get('min_ledmatrix_version') or manifest.get('requires', {}).get('min_ledmatrix_version') ) if not declared: versions = manifest.get('versions') or [] if versions and isinstance(versions[0], dict): declared = (versions[0].get('ledmatrix_min_version') or versions[0].get('ledmatrix_min')) needed = self._parse_semver(declared) if needed is None: return from src import __version__ as core_version current = self._parse_semver(core_version) # Anti-spam guard: if the core's own version number is stale (below # the ecosystem floor every shipped plugin declares), comparing would # warn on nearly everything — skip with a debug note instead. if current is None or current < (2, 0, 0): self.logger.debug( "Skipping version compatibility check for %s: core __version__ " "(%s) is below the ecosystem floor", plugin_id, core_version) return if needed > current: self.logger.warning( "Plugin %s declares min LEDMatrix version %s but this core is %s — " "features it relies on may be missing; update the core or expect " "degraded fallbacks", plugin_id, declared, core_version) def load_plugin( self, plugin_id: str, manifest: Dict[str, Any], plugin_dir: Path, config: Dict[str, Any], display_manager: Any, cache_manager: Any, plugin_manager: Any, install_deps: bool = True, plugins_dir: Optional[Path] = None, ) -> Tuple[Any, Any]: """ Complete plugin loading process. Args: plugin_id: Plugin identifier manifest: Plugin manifest plugin_dir: Plugin directory path config: Plugin configuration display_manager: Display manager instance cache_manager: Cache manager instance plugin_manager: Plugin manager instance install_deps: Whether to install dependencies plugins_dir: Trusted base plugins directory forwarded to install_dependencies Returns: Tuple of (plugin_instance, module) Raises: PluginError: If loading fails """ self._warn_if_incompatible(plugin_id, manifest) # Install dependencies if needed if install_deps: if plugins_dir is None: raise PluginError( f"plugins_dir is required to install dependencies for plugin {plugin_id} " "(needed for path containment; pass install_deps=False if the caller " "doesn't have a trusted plugins directory to supply)", plugin_id=plugin_id, context={'plugin_dir': str(plugin_dir)}, ) if not self.install_dependencies(plugin_dir, plugin_id, plugins_dir=plugins_dir): raise PluginError( f"Dependency installation failed for plugin {plugin_id} in {plugin_dir}", plugin_id=plugin_id, context={'plugin_dir': str(plugin_dir)}, ) # Load module entry_point = manifest.get('entry_point', 'manager.py') module = self.load_module(plugin_id, plugin_dir, entry_point) if module is None: raise PluginError(f"Failed to load module for plugin {plugin_id}", plugin_id=plugin_id) # Get plugin class class_name = manifest.get('class_name') if not class_name: raise PluginError(f"No class_name in manifest for plugin {plugin_id}", plugin_id=plugin_id) plugin_class = self.get_plugin_class(plugin_id, module, class_name) # Instantiate plugin plugin_instance = self.instantiate_plugin( plugin_id, plugin_class, config, display_manager, cache_manager, plugin_manager ) return (plugin_instance, module)