mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
Chuck
32d637a446
* fix(store): read the core version from disk, not from a stale import
Updating the core to 3.3.0 and then updating plugins refused all eight sports
scoreboards:
Refusing to install nrl-scoreboard: NRL Scoreboard supports LEDMatrix
>=3.3.0, but this system is running 3.2.0.
while src/__init__.py on that machine read 3.3.0. Observed on hardware, not
theorised.
The gate ran `from src import __version__ as core_version`, which binds
whatever the process loaded at start. The plugin store's gate lives in the web
UI, a long-lived service of its own, and the update route deliberately restarts
nothing -- it replaces files on disk and asks the user to restart. Its prompt
named only the *display* service, so a user who followed it left the web
process holding the previous number.
Stale by exactly one release is the case that bites: every plugin flooring on
the release you just installed is refused, blaming a core version that is
already correct on disk. It reads as a broken plugin store. 3.3.0 is the first
release where this hits a whole family at once, since all eight scoreboards
floor there.
compatibility.current_core_version() reads the version from the file instead,
falling back to the imported value on any failure -- so it can only ever be as
correct as before, never worse. All four gate call sites use it: three in
store_manager (install, the git-pull update path, install_from_url) and one in
plugin_loader's advisory warning.
The restart prompt now names both services.
Twelve tests, including the hardware failure itself: a process holding 3.2.0
while disk says 3.3.0 refuses hockey, and reading fresh allows it. The inverse
is asserted too -- a genuinely old core still refuses, so the gate has not
become permissive. One test greps both modules for the old import-bound read;
reintroducing that line fails it, which is what stops this coming back.
Not changed: web_interface/__init__.py also imports __version__, but for
display rather than gating, and the API endpoint already reports a fresh
git describe.
* fix: drop the unused os import
Left over from a first draft that joined paths by hand before this used
pathlib. Flagged by CodeRabbit on #518; confirmed dead -- no os. reference
remains in the module.
871 lines
36 KiB
Python
871 lines
36 KiB
Python
"""
|
|
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, List, 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 _extra_dependencies(dist_name: str, extras) -> Optional[List[Requirement]]:
|
|
"""Dependencies a distribution declares *only* behind the given extras.
|
|
|
|
Returns None when the installed metadata cannot be read or parsed, so the
|
|
caller can fall back to running pip rather than assuming anything.
|
|
"""
|
|
try:
|
|
meta = importlib.metadata.metadata(dist_name)
|
|
except importlib.metadata.PackageNotFoundError:
|
|
return None
|
|
|
|
gated: List[Requirement] = []
|
|
for raw in meta.get_all('Requires-Dist') or []:
|
|
try:
|
|
dep = Requirement(raw)
|
|
except InvalidRequirement:
|
|
return None
|
|
if dep.marker is None:
|
|
continue
|
|
# Keep only what the distribution gates behind an extra we asked for:
|
|
# satisfied when `extra` is that name, but not when no extra is
|
|
# requested. A marker that holds either way (python_version, sys_platform)
|
|
# belongs to the base install and is already covered by the version check.
|
|
if dep.marker.evaluate({'extra': ''}):
|
|
continue
|
|
if any(dep.marker.evaluate({'extra': extra}) for extra in extras):
|
|
gated.append(dep)
|
|
return gated
|
|
|
|
|
|
def _extras_are_satisfied(req: Requirement, _visited: Optional[set] = None) -> bool:
|
|
"""Check the dependencies pulled in by req's extras are installed.
|
|
|
|
Follows extras through nested extras. A gated dependency can itself request
|
|
one (`requests[socks]`), and checking only that `requests` is installed at
|
|
an acceptable version says nothing about whether the socks extra's own
|
|
dependency is there -- so the caller would skip pip and the plugin would
|
|
fail at import instead. Plain dependencies are still checked one level
|
|
deep, which is all that is needed to tell "the extra was installed" from
|
|
"the extra was never installed".
|
|
|
|
`_visited` carries the (distribution, extras) pairs already seen, so a
|
|
dependency cycle between extras terminates instead of recursing forever.
|
|
Anything unreadable returns False, so the caller still falls through to pip.
|
|
"""
|
|
if _visited is None:
|
|
_visited = set()
|
|
marker = (req.name.lower(), frozenset(e.lower() for e in req.extras))
|
|
if marker in _visited:
|
|
# Already accounted for higher up the chain; treating a cycle as
|
|
# satisfied here is safe because the outer frame still has to pass.
|
|
return True
|
|
_visited.add(marker)
|
|
|
|
gated = _extra_dependencies(req.name, req.extras)
|
|
if gated is None:
|
|
return False
|
|
|
|
for dep in gated:
|
|
try:
|
|
dep_version = importlib.metadata.version(dep.name)
|
|
except importlib.metadata.PackageNotFoundError:
|
|
return False
|
|
if dep.specifier and not dep.specifier.contains(dep_version, prereleases=True):
|
|
return False
|
|
if dep.extras and not _extras_are_satisfied(dep, _visited):
|
|
return False
|
|
return True
|
|
|
|
|
|
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.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
|
|
|
|
if req.extras and not _extras_are_satisfied(req):
|
|
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_<plugin_id>_<module>`` 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.
|
|
"""
|
|
from src.plugin_system import compatibility
|
|
core_version = compatibility.current_core_version()
|
|
|
|
compatible, _reason = compatibility.check(manifest, core_version)
|
|
if compatible:
|
|
# Distinguish "fine" from "couldn't tell" for anyone reading logs:
|
|
# a core below the trustworthy floor is skipped, not cleared.
|
|
current = compatibility.parse_semver(core_version)
|
|
if current is None or current < compatibility.TRUSTWORTHY_FLOOR:
|
|
self.logger.debug(
|
|
"Skipping version compatibility check for %s: core __version__ "
|
|
"(%s) is below the ecosystem floor", plugin_id, core_version)
|
|
return
|
|
|
|
declared = compatibility.declared_min_version(manifest)
|
|
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)
|
|
|