rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-04 14:25:08 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
f7bdb247729c96abffd21adce2acb66ba4d556a3
LEDMatrix/src/plugin_system/store_manager.py
T
ChuckandClaude Opus 5.5 c00bf5e8e6 fix(plugin-system): unload/update race, failed-load cleanup, limits validation, schema lookup, install rollback (#653) 
* fix(plugin-system): unload/update race, failed-load module cleanup, limits validation, schema lookup, install rollback, op-queue dedupe

- unload_plugin takes the per-plugin lock (5s bounded) before cleanup(),
  and an update() that finishes after its plugin was unloaded no longer
  sets the state back to ENABLED.
- A load that fails after import drops plugin_<id> and its submodules
  and forgets its manager fonts, so a fixed plugin reloads new code.
- Resource limits are validated as non-negative numbers: 400 at
  POST /plugins/limits, bad cached records ignored with one warning.
  Route docstrings note health/metrics reset and limits only change the
  web process's view.
- SchemaManager.get_schema_path resolves each search dir via
  resolve_plugin_dir (manifest id, ledmatrix-<id>) before the literal
  paths; plugins/ still before plugin-repos/. Misses cached 30s and
  logged once at DEBUG.
- install_from_url sets an existing copy aside and restores it if the
  move fails, under the per-plugin reinstall lock.
- Operation queue refuses a second pending op for a plugin and trims
  _operations with history.
- get_vegas_render_width reads display_manager.width first.
- get_logger in store/schema/health/resource/saved_repositories;
  UTF-8 reads in store_manager and state_manager.
- Docs: update_interval precedence (manifest over config) stated where
  users are told to set it in config.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(web): build the limits 400 message from the field name, not an exception

CodeQL flagged str(e) flowing into the response. invalid_limit_field()
returns the offending field without raising, and limits_from_dict uses it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-28 10:41:40 -04:00

2978 lines
145 KiB
Python
Raw Blame History

"""
Plugin Store Manager for LEDMatrix
Handles plugin discovery, installation, updates, and uninstallation
from both the official registry and custom GitHub repositories.
"""
import errno
import os
import re
import json
import stat
import subprocess
import shutil
import threading
import zipfile
import tempfile
import requests
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional, Any, Tuple, Set
from jsonschema import Draft7Validator, ValidationError
from src.logging_config import get_logger
from src.common.permission_utils import (
ensure_directory_permissions, get_plugin_dir_mode, install_requirements_file,
sudo_remove_directory,
)
from src.plugin_system.plugin_loader import contained_plugin_dir, requirements_to_install
from src.plugin_system.plugin_dirs import (
BACKUP_MARKER, PluginDirectoryIndex, resolve_plugin_dir, store_search_dirs,
)
from src.plugin_system.repo_urls import (
USER_AGENT, github_api_headers, github_owner_repo, normalize_repo_url, same_repo,
)
class PluginStoreManager:
"""
Manages plugin discovery, installation, and updates from GitHub.
Supports two installation methods:
1. From official registry (curated plugins)
2. From custom GitHub URL (any repo)
"""
REGISTRY_URL = "https://raw.githubusercontent.com/ChuckBuilds/ledmatrix-plugins/main/plugins.json"
# A valid plugin id is a single path component: starts alphanumeric, then
# alphanumerics / dot / dash / underscore. Used to keep the uninstall
# registry from ever turning a corrupt or hand-edited entry (e.g. "",
# "..", "../x") into a filesystem path that purge_uninstalled_plugins
# would delete — an empty id resolves to the plugins root itself.
_PLUGIN_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
@staticmethod
def is_plugin_entry(entry) -> bool:
"""Whether a registry entry is a plugin core can install.
A missing ``type`` means plugin. Anything else (registries used to
carry ``"type": "skin"`` entries, and a custom registry still can) is
hidden from the store and refused at install, rather than being
unpacked into the plugins directory as if it were a plugin.
"""
return isinstance(entry, dict) and (entry.get('type') or 'plugin') == 'plugin'
def __init__(self, plugins_dir: str = "plugins",
uninstalled_registry_path: Optional[str] = None):
"""
Initialize the plugin store manager.
Args:
plugins_dir: Directory where plugins are installed
uninstalled_registry_path: Path to the JSON file recording plugins
the user has uninstalled. Defaults to
``config/uninstalled_plugins.json`` under the project root.
"""
self.plugins_dir = Path(plugins_dir)
self.logger = get_logger(__name__)
self.registry_cache = None
self.registry_cache_time = None # Timestamp of when registry was cached
self.github_cache = {} # Cache for GitHub API responses
self.cache_timeout = 3600 # 1 hour cache timeout (repo info: stars, default_branch)
# 15 minutes for registry cache. Long enough that the plugin list
# endpoint on a warm cache never hits the network, short enough that
# new plugins show up within a reasonable window. See also the
# stale-cache fallback in fetch_registry for transient network
# failures.
self.registry_cache_timeout = 900
self.commit_info_cache = {} # Cache for latest commit info: {key: (timestamp, data)}
# 30 minutes for commit/manifest caches. Plugin Store users browse
# the catalog via /plugins/store/list, which fetches commit info per
# plugin; with a 5-minute TTL nearly every browse on a Pi4 paid for
# an HTTP request per plugin again. 30 minutes keeps the cache warm
# across a realistic session while still picking up upstream updates
# within a reasonable window.
self.commit_cache_timeout = 1800
self.manifest_cache = {} # Cache for GitHub manifest fetches: {key: (timestamp, data)}
self.manifest_cache_timeout = 1800
self.github_token = self._load_github_token()
self._token_validation_cache = {} # Cache for token validation results: {token: (is_valid, timestamp, error_message)}
self._token_validation_cache_timeout = 300 # 5 minutes cache for token validation
# Persistent record of plugins the user has uninstalled. It survives
# restarts so that a core ``git pull`` update cannot resurrect a
# built-in plugin the user removed. Built-in plugins (e.g.
# ``web-ui-info``, ``starlark-apps``) are committed into the repo under
# ``plugin-repos/``, so a plain ``git pull`` restores their files even
# after the user deleted them. ``purge_uninstalled_plugins`` re-removes
# any such resurrected directory; ``install_plugin`` clears the record
# when the user deliberately reinstalls. The file is gitignored.
if uninstalled_registry_path is not None:
self._uninstalled_registry_path = Path(uninstalled_registry_path)
else:
self._uninstalled_registry_path = (
Path(__file__).parent.parent.parent / "config" / "uninstalled_plugins.json"
)
# Serializes read-modify-write of the registry file so concurrent
# install/uninstall requests can't lose updates.
self._uninstalled_registry_lock = threading.Lock()
# Cache for _get_local_git_info: {plugin_path_str: (signature, data)}
# where ``signature`` is a tuple of (head_mtime, resolved_ref_mtime,
# head_contents) so a fast-forward update to the current branch
# (which touches .git/refs/heads/<branch> but NOT .git/HEAD) still
# invalidates the cache. Without it every /plugins/installed request
# runs a git subprocess per plugin, which adds up on a Pi4 with a
# dozen plugins. The cached
# ``data`` dict is the same shape returned by ``_get_local_git_info``
# itself (sha / short_sha / branch / optional remote_url, date_iso,
# date) — all string-keyed strings.
self._git_info_cache: Dict[str, Tuple[Tuple, Dict[str, str]]] = {}
# How long to wait before re-attempting a failed GitHub metadata
# fetch after we've already served a stale cache hit. Without this,
# a single expired-TTL + network-error would cause every subsequent
# request to re-hit the network (and fail again) until the network
# actually came back — amplifying the failure and blocking request
# handlers. Bumping the cached-entry timestamp on failure serves
# the stale payload cheaply until the backoff expires.
self._failure_backoff_seconds = 60
# Prevents concurrent callers from each firing a network request when
# the registry cache expires. Only one thread fetches; others wait and
# then get the result from the warm cache (double-checked locking).
self._registry_fetch_lock = threading.Lock()
# Per-plugin locks for _reinstall_with_rollback: the web UI runs
# Flask with threaded=True, so two overlapping requests for the
# same plugin_id (double-click, two browser tabs) would otherwise
# both rename the same directory aside — one succeeds, and the
# loser can end up renaming the winner's in-progress install aside
# mid-download, stealing its own rollback safety net. Keyed by
# plugin_id so unrelated plugins still update concurrently.
# Reentrant: install_plugin takes this lock, and _reinstall_with_rollback
# holds it across its call to install_plugin. A plain Lock would
# self-deadlock on that nesting.
self._reinstall_locks: Dict[str, "threading.RLock"] = {}
self._reinstall_locks_guard = threading.Lock()
# Ensure plugins directory exists
self.plugins_dir.mkdir(exist_ok=True)
def _get_reinstall_lock(self, plugin_id: str):
"""Lazily create (or fetch) the per-plugin reinstall lock.
Reentrant by necessity: `install_plugin` acquires it to protect its
set-aside/restore, and `_reinstall_with_rollback` holds it across its
own call to `install_plugin`. With a plain `Lock` that nesting
deadlocks the request thread.
"""
with self._reinstall_locks_guard:
lock = self._reinstall_locks.get(plugin_id)
if lock is None:
lock = threading.RLock()
self._reinstall_locks[plugin_id] = lock
return lock
def _record_cache_backoff(self, cache_dict: Dict, cache_key: str,
cache_timeout: int, payload: Any) -> None:
"""Bump a cache entry's timestamp so subsequent lookups hit the
cache rather than re-failing over the network.
Used by the stale-on-error fallbacks in the GitHub metadata fetch
paths. Without this, a cache entry whose TTL just expired would
cause every subsequent request to re-hit the network and fail
again until the network actually came back. We write a synthetic
timestamp ``(now + backoff - cache_timeout)`` so the cache-valid
check ``(now - ts) < cache_timeout`` succeeds for another
``backoff`` seconds.
"""
synthetic_ts = time.time() + self._failure_backoff_seconds - cache_timeout
cache_dict[cache_key] = (synthetic_ts, payload)
def _is_valid_plugin_id(self, plugin_id: Any) -> bool:
"""Return True if ``plugin_id`` is a safe single-component plugin id.
Rejects empty strings, anything with a path separator, and traversal
sequences like ``..`` so a registry entry can never escape (or target
the root of) ``self.plugins_dir`` during a purge.
"""
return isinstance(plugin_id, str) and bool(self._PLUGIN_ID_RE.match(plugin_id))
def _read_uninstalled_registry(self) -> Set[str]:
"""Read the persistent set of uninstalled plugin IDs.
Returns an empty set if the file is missing, unreadable, or corrupt —
a broken registry must never block normal plugin operations. Invalid
ids are dropped here so callers never turn them into paths.
"""
try:
if not self._uninstalled_registry_path.exists():
return set()
with open(self._uninstalled_registry_path, 'r', encoding='utf-8') as f:
data = json.load(f)
if not isinstance(data, list):
self.logger.warning(
"Uninstalled-plugin registry at %s is not a list; ignoring it",
self._uninstalled_registry_path,
)
return set()
valid: Set[str] = set()
for pid in data:
if self._is_valid_plugin_id(pid):
valid.add(pid)
else:
self.logger.warning(
"Ignoring invalid plugin id in uninstall registry: %r", pid
)
return valid
except (OSError, ValueError) as e:
self.logger.warning(
"Could not read uninstalled-plugin registry at %s: %s",
self._uninstalled_registry_path, e,
)
return set()
def _write_uninstalled_registry(self, plugin_ids: Set[str]) -> None:
"""Persist the set of uninstalled plugin IDs (sorted, atomically)."""
path = self._uninstalled_registry_path
try:
path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = path.with_suffix(path.suffix + ".tmp")
with open(tmp_path, 'w', encoding='utf-8') as f:
json.dump(sorted(plugin_ids), f, indent=2)
os.replace(tmp_path, path)
except OSError as e:
self.logger.error(
"Failed to write uninstalled-plugin registry at %s: %s", path, e
)
def record_uninstalled_plugin(self, plugin_id: str) -> None:
"""Persistently record that the user uninstalled ``plugin_id``.
Survives restarts so a core update cannot resurrect the plugin.
"""
if not self._is_valid_plugin_id(plugin_id):
self.logger.error("Refusing to record invalid plugin id: %r", plugin_id)
return
with self._uninstalled_registry_lock:
recorded = self._read_uninstalled_registry()
if plugin_id not in recorded:
recorded.add(plugin_id)
self._write_uninstalled_registry(recorded)
self.logger.info("Recorded %s as uninstalled (persistent)", plugin_id)
def forget_uninstalled_plugin(self, *plugin_ids: str) -> None:
"""Drop ``plugin_ids`` from the persistent uninstall registry.
Called when a plugin is deliberately (re)installed so future updates
keep it.
"""
with self._uninstalled_registry_lock:
recorded = self._read_uninstalled_registry()
to_remove = {pid for pid in plugin_ids if pid in recorded}
if to_remove:
self._write_uninstalled_registry(recorded - to_remove)
self.logger.info(
"Cleared uninstall record for %s", ", ".join(sorted(to_remove))
)
def get_uninstalled_plugins(self) -> Set[str]:
"""Return the persistent set of user-uninstalled plugin IDs."""
return self._read_uninstalled_registry()
def is_plugin_uninstalled(self, plugin_id: str) -> bool:
"""Return True if ``plugin_id`` is in the persistent uninstall registry."""
return plugin_id in self._read_uninstalled_registry()
def purge_uninstalled_plugins(self) -> List[str]:
"""Remove on-disk directories for plugins the user has uninstalled.
Built-in plugins committed into the repo are restored on disk by a
core ``git pull``; this re-removes any that the user previously
uninstalled. The registry entries are kept so the purge is idempotent
across every future update (until the user reinstalls). Returns the
list of plugin IDs whose directories were actually removed.
"""
removed: List[str] = []
plugins_root = self.plugins_dir.resolve()
for plugin_id in sorted(self._read_uninstalled_registry()):
plugin_path = self.plugins_dir / plugin_id
# Defense in depth: ids are already validated on read, but never
# remove anything that isn't a direct child of the plugins root.
resolved = plugin_path.resolve()
if resolved == plugins_root or resolved.parent != plugins_root:
self.logger.error(
"Refusing to purge unsafe plugin path for id %r", plugin_id
)
continue
if not plugin_path.exists():
continue
self.logger.info(
"Purging resurrected uninstalled plugin: %s", plugin_id
)
if self._safe_remove_directory(plugin_path):
removed.append(plugin_id)
else:
self.logger.error(
"Failed to purge resurrected plugin directory: %s", plugin_path
)
return removed
def _load_github_token(self) -> Optional[str]:
"""
Load GitHub API token from config_secrets.json if available.
Returns:
GitHub token or None if not configured
"""
try:
config_path = Path(__file__).parent.parent.parent / "config" / "config_secrets.json"
if config_path.exists():
with open(config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
token = config.get('github', {}).get('api_token', '').strip()
if token and token != "YOUR_GITHUB_PERSONAL_ACCESS_TOKEN":
return token
except Exception as e:
self.logger.debug(f"Could not load GitHub token: {e}")
return None
def _validate_github_token(self, token: str) -> tuple[bool, Optional[str]]:
"""
Validate a GitHub token by making a lightweight API call.
Args:
token: GitHub personal access token to validate
Returns:
Tuple of (is_valid, error_message)
- is_valid: True if token is valid, False otherwise
- error_message: None if valid, error description if invalid
"""
if not token:
return (False, "No token provided")
# Check cache first
cache_key = token[:10] # Use first 10 chars as cache key for privacy
if cache_key in self._token_validation_cache:
cached_valid, cached_time, cached_error = self._token_validation_cache[cache_key]
if time.time() - cached_time < self._token_validation_cache_timeout:
return (cached_valid, cached_error)
# Validate token by making a lightweight API call to /user endpoint
try:
api_url = "https://api.github.com/user"
response = requests.get(api_url, headers=github_api_headers(token), timeout=5)
if response.status_code == 200:
# Token is valid
result = (True, None)
self._token_validation_cache[cache_key] = (True, time.time(), None)
return result
elif response.status_code == 401:
# Token is invalid or expired
error_msg = "Token is invalid or expired"
result = (False, error_msg)
self._token_validation_cache[cache_key] = (False, time.time(), error_msg)
return result
elif response.status_code == 403:
# Rate limit or forbidden (but token might be valid)
# Check if it's a rate limit issue
if 'rate limit' in response.text.lower():
# Rate limit: return error but don't cache (rate limits are temporary)
error_msg = "Rate limit exceeded"
result = (False, error_msg)
return result
else:
# Token lacks permissions: cache the result (permissions don't change)
error_msg = "Token lacks required permissions"
result = (False, error_msg)
self._token_validation_cache[cache_key] = (False, time.time(), error_msg)
return result
else:
# Other error
error_msg = f"GitHub API error: {response.status_code}"
result = (False, error_msg)
self._token_validation_cache[cache_key] = (False, time.time(), error_msg)
return result
except requests.exceptions.Timeout:
error_msg = "GitHub API request timed out"
result = (False, error_msg)
# Don't cache timeout errors
return result
except requests.exceptions.RequestException as e:
error_msg = f"Network error: {str(e)}"
result = (False, error_msg)
# Don't cache network errors
return result
except Exception as e:
error_msg = f"Unexpected error: {str(e)}"
result = (False, error_msg)
# Don't cache unexpected errors
return result
@staticmethod
def _iso_to_date(iso_timestamp: str) -> str:
"""Convert an ISO timestamp to YYYY-MM-DD string."""
if not iso_timestamp:
return ""
try:
dt = datetime.fromisoformat(iso_timestamp.replace('Z', '+00:00'))
return dt.strftime('%Y-%m-%d')
except Exception:
return ""
@staticmethod
def _distinct_sequence(values: List[str]) -> List[str]:
"""Return list preserving order while removing duplicates and falsey entries."""
seen = set()
ordered = []
for value in values:
if not value:
continue
if value in seen:
continue
seen.add(value)
ordered.append(value)
return ordered
def _validate_manifest_version_fields(self, manifest: Dict[str, Any]) -> List[str]:
"""
Validate version-related fields in manifest for consistency.
Checks:
- compatible_versions is present and is an array
- Standardized field names are used (min_ledmatrix_version, max_ledmatrix_version)
- Deprecated fields are not used (ledmatrix_version)
- versions array entries use ledmatrix_min_version instead of ledmatrix_min
Args:
manifest: Manifest dictionary to validate
Returns:
List of validation error/warning messages (empty if valid)
"""
errors = []
# Check compatible_versions is an array
if 'compatible_versions' in manifest:
if not isinstance(manifest['compatible_versions'], list):
errors.append("compatible_versions must be an array")
elif len(manifest['compatible_versions']) == 0:
errors.append("compatible_versions array cannot be empty")
# Warn about deprecated ledmatrix_version field
if 'ledmatrix_version' in manifest:
errors.append("ledmatrix_version is deprecated, use compatible_versions instead")
# Check versions array entries use standardized field names
if 'versions' in manifest and isinstance(manifest['versions'], list):
for i, version_entry in enumerate(manifest['versions']):
if not isinstance(version_entry, dict):
continue
# Check for old ledmatrix_min field
if 'ledmatrix_min' in version_entry and 'ledmatrix_min_version' not in version_entry:
errors.append(f"versions[{i}] uses deprecated 'ledmatrix_min', should use 'ledmatrix_min_version'")
return errors
def _validate_manifest_schema(self, manifest: Dict[str, Any], plugin_id: str) -> List[str]:
"""
Validate manifest against JSON schema if available.
Args:
manifest: Manifest dictionary to validate
plugin_id: Plugin ID for error messages
Returns:
List of validation error messages (empty if valid or schema unavailable)
"""
try:
# Load manifest schema
schema_path = Path(__file__).parent.parent.parent / "schema" / "manifest_schema.json"
if not schema_path.exists():
return [] # Schema not available, skip validation
with open(schema_path, 'r', encoding='utf-8') as f:
schema = json.load(f)
# Validate schema itself
Draft7Validator.check_schema(schema)
# Validate manifest against schema
validator = Draft7Validator(schema)
errors = []
for error in validator.iter_errors(manifest):
error_path = '.'.join(str(p) for p in error.path)
errors.append(f"{error_path}: {error.message}")
return errors
except json.JSONDecodeError as e:
self.logger.warning(f"Could not parse manifest schema: {e}")
return []
except ValidationError as e:
self.logger.warning(f"Manifest schema is invalid: {e}")
return []
except Exception as e:
self.logger.debug(f"Error validating manifest schema for {plugin_id}: {e}")
return []
_EMPTY_REPO_INFO: Dict[str, Any] = {
'stars': 0,
'forks': 0,
'open_issues': 0,
'updated_at_iso': '',
'last_commit_iso': '',
'last_commit_date': '',
'language': '',
'license': '',
'default_branch': 'main',
}
def _get_github_repo_info(self, repo_url: str) -> Dict[str, Any]:
"""GitHub metadata for a repository (stars, default branch, last push).
Returns zeroed defaults (``_EMPTY_REPO_INFO``) for a non-GitHub URL or
when GitHub cannot be asked and nothing is cached.
"""
try:
owner_repo = github_owner_repo(repo_url)
if owner_repo is None:
return dict(self._EMPTY_REPO_INFO)
owner, repo = owner_repo
cache_key = f"{owner}/{repo}"
if cache_key in self.github_cache:
cached_time, cached_data = self.github_cache[cache_key]
if time.time() - cached_time < self.cache_timeout:
return cached_data
api_url = f"https://api.github.com/repos/{owner}/{repo}"
try:
response = requests.get(
api_url, headers=github_api_headers(self.github_token), timeout=10)
except requests.RequestException as req_err:
# Network error: prefer a stale cache hit over an empty
# default so the UI keeps working on a flaky Pi WiFi link.
# Bump the cached entry's timestamp into a short backoff
# window so subsequent requests serve the stale payload
# cheaply instead of re-hitting the network on every request.
if cache_key in self.github_cache:
_, stale = self.github_cache[cache_key]
self._record_cache_backoff(self.github_cache, cache_key, self.cache_timeout, stale)
self.logger.warning(
"GitHub repo info fetch failed for %s (%s); serving stale cache.",
cache_key, req_err,
)
return stale
raise
if response.status_code == 200:
data = response.json()
pushed_at = data.get('pushed_at', '') or data.get('updated_at', '')
repo_info = {
'stars': data.get('stargazers_count', 0),
'forks': data.get('forks_count', 0),
'open_issues': data.get('open_issues_count', 0),
'updated_at_iso': data.get('updated_at', ''),
'last_commit_iso': pushed_at,
'last_commit_date': self._iso_to_date(pushed_at),
'language': data.get('language', ''),
'license': data.get('license', {}).get('name', '') if data.get('license') else '',
'default_branch': data.get('default_branch', 'main')
}
self.github_cache[cache_key] = (time.time(), repo_info)
return repo_info
if response.status_code == 403:
# Rate limit or authentication issue. A stale star count is
# better than a reset to zero, and the backoff bump stops the
# store hammering the API while rate-limited.
if cache_key in self.github_cache:
_, stale = self.github_cache[cache_key]
self._record_cache_backoff(self.github_cache, cache_key, self.cache_timeout, stale)
self.logger.warning(
"GitHub API 403 for %s; serving stale cache.", cache_key,
)
return stale
if not self.github_token:
self.logger.warning(
"GitHub API rate limit likely exceeded (403). "
"Add a GitHub personal access token to config/config_secrets.json "
"under 'github.api_token' to increase rate limits from 60 to 5000/hour."
)
else:
self.logger.warning(
f"GitHub API request failed: 403 for {api_url}. "
f"Your token may have insufficient permissions or rate limit exceeded."
)
else:
self.logger.warning(f"GitHub API request failed: {response.status_code} for {api_url}")
if cache_key in self.github_cache:
_, stale = self.github_cache[cache_key]
self._record_cache_backoff(self.github_cache, cache_key, self.cache_timeout, stale)
return stale
return dict(self._EMPTY_REPO_INFO)
except requests.exceptions.RequestException as e:
# Offline, DNS or a timeout reaching GitHub: the listing still
# works without the extra repo info, so this is not an error.
self.logger.warning("GitHub repo info unavailable for %s: %s", repo_url, e)
return dict(self._EMPTY_REPO_INFO)
except Exception as e:
self.logger.error(f"Error fetching GitHub repo info for {repo_url}: {e}")
return dict(self._EMPTY_REPO_INFO)
def _http_get_with_retries(self, url: str, *, timeout: int = 10, stream: bool = False, headers: Dict[str, str] = None, max_retries: int = 3, backoff_sec: float = 0.75):
"""
HTTP GET with simple retry strategy and exponential backoff.
Returns a requests.Response or raises the last exception.
"""
last_exc = None
for attempt in range(1, max_retries + 1):
try:
resp = requests.get(url, timeout=timeout, stream=stream, headers=headers)
return resp
except requests.RequestException as e:
last_exc = e
self.logger.warning(f"HTTP GET failed (attempt {attempt}/{max_retries}) for {url}: {e}")
if attempt < max_retries:
time.sleep(backoff_sec * attempt)
# Exhausted retries
raise last_exc
def fetch_registry_from_url(self, repo_url: str) -> Optional[Dict]:
"""
Fetch a registry-style plugins.json from a custom GitHub repository URL.
This allows users to point to a registry-style monorepo (like the official
ledmatrix-plugins repo) and browse/install plugins from it.
Args:
repo_url: GitHub repository URL (e.g., https://github.com/user/ledmatrix-plugins)
Returns:
Registry dict with plugins list, or None if not found/invalid
"""
try:
repo_url = normalize_repo_url(repo_url)
# plugins.json or registry.json at the root of main, then master.
registry_urls = []
owner_repo = github_owner_repo(repo_url)
if owner_repo is not None:
owner, repo = owner_repo
for branch in ['main', 'master']:
registry_urls.append(f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/plugins.json")
registry_urls.append(f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/registry.json")
for url in registry_urls:
try:
response = self._http_get_with_retries(url, timeout=10)
if response.status_code == 200:
registry = response.json()
# Validate it looks like a registry
if isinstance(registry, dict) and 'plugins' in registry:
self.logger.info(f"Successfully fetched registry from {url}")
return registry
except Exception as e:
self.logger.debug(f"Failed to fetch from {url}: {e}")
continue
self.logger.warning(f"No valid registry found at {repo_url}")
return None
except Exception as e:
self.logger.error(f"Error fetching registry from URL: {e}", exc_info=True)
return None
def fetch_registry(self, force_refresh: bool = False, raise_on_failure: bool = False) -> Dict:
"""
Fetch the plugin registry from GitHub.
Args:
force_refresh: Force refresh even if cached
raise_on_failure: If True, re-raise network / JSON errors instead
of silently falling back to stale cache / empty dict. UI
callers prefer the stale-fallback default so the plugin
list keeps working on flaky WiFi; the state reconciler
needs the explicit failure signal so it can distinguish
"plugin genuinely not in registry" from "I couldn't reach
the registry at all" and not mark everything unrecoverable.
Returns:
Registry data with list of available plugins
Raises:
requests.RequestException / json.JSONDecodeError when
``raise_on_failure`` is True and the fetch fails.
"""
# Check if cache is still valid (within timeout)
current_time = time.time()
if (self.registry_cache and self.registry_cache_time and
not force_refresh and
(current_time - self.registry_cache_time) < self.registry_cache_timeout):
return self.registry_cache
with self._registry_fetch_lock:
# Re-check inside the lock — a concurrent caller that was waiting
# may have already populated the cache while we blocked.
current_time = time.time()
if (self.registry_cache and self.registry_cache_time and
not force_refresh and
(current_time - self.registry_cache_time) < self.registry_cache_timeout):
return self.registry_cache
try:
self.logger.info(f"Fetching plugin registry from {self.REGISTRY_URL}")
response = self._http_get_with_retries(self.REGISTRY_URL, timeout=10)
response.raise_for_status()
self.registry_cache = response.json()
self.registry_cache_time = current_time
self.logger.info(f"Fetched registry with {len(self.registry_cache.get('plugins', []))} plugins")
return self.registry_cache
except requests.RequestException as e:
self.logger.error(f"Error fetching registry: {e}")
if raise_on_failure:
raise
# Prefer stale cache over an empty list so the plugin list UI
# keeps working on a flaky connection (e.g. Pi on WiFi). Bump
# registry_cache_time into a short backoff window so the next
# request serves the stale payload cheaply instead of
# re-hitting the network on every request (matches the
# pattern used by github_cache / commit_info_cache).
if self.registry_cache:
self.logger.warning("Falling back to stale registry cache")
self.registry_cache_time = (
time.time() + self._failure_backoff_seconds - self.registry_cache_timeout
)
return self.registry_cache
return {"plugins": []}
except json.JSONDecodeError as e:
self.logger.error(f"Error parsing registry JSON: {e}")
if raise_on_failure:
raise
if self.registry_cache:
self.registry_cache_time = (
time.time() + self._failure_backoff_seconds - self.registry_cache_timeout
)
return self.registry_cache
return {"plugins": []}
def search_plugins(self, query: str = "", category: str = "", tags: List[str] = None, fetch_commit_info: bool = True, include_saved_repos: bool = True, saved_repositories_manager = None) -> List[Dict]:
"""
Search for plugins in the registry with enhanced metadata.
GitHub supplies live metadata such as stars and last commit
timestamps; the registry supplies descriptive information (name,
description, repo URL, etc.).
Args:
query: Search query string (searches name, description, id, author)
category: Filter by category (e.g., 'sports', 'weather', 'time')
tags: Filter by tags (matches any tag in list)
fetch_commit_info: If True (default), fetch commit metadata from GitHub.
include_saved_repos: If True (default), also search the
registry-style repositories the user saved.
saved_repositories_manager: The SavedRepositoriesManager holding
those repositories; without it only the official registry is
searched.
Returns:
List of matching plugin metadata enriched with GitHub information
"""
if tags is None:
tags = []
# Fetch from official registry
registry = self.fetch_registry()
plugins = registry.get('plugins', []) or []
# Also fetch from saved repositories if enabled
if include_saved_repos and saved_repositories_manager:
saved_repos = saved_repositories_manager.get_registry_repositories()
for repo_info in saved_repos:
repo_url = repo_info.get('url')
if repo_url:
try:
custom_registry = self.fetch_registry_from_url(repo_url)
if custom_registry:
custom_plugins = custom_registry.get('plugins', []) or []
# Mark these as from custom repository
for plugin in custom_plugins:
plugin['_source'] = 'custom_repository'
plugin['_repository_url'] = repo_url
plugin['_repository_name'] = repo_info.get('name', repo_url)
plugins.extend(custom_plugins)
except Exception as e:
self.logger.warning(f"Failed to fetch plugins from saved repository {repo_url}: {e}")
# First pass: apply cheap filters (category/tags/query) so we only
# fetch GitHub metadata for plugins that will actually be returned.
filtered: List[Dict] = []
for plugin in plugins:
if category and plugin.get('category') != category:
continue
if tags and not any(tag in plugin.get('tags', []) for tag in tags):
continue
if query:
query_lower = query.lower()
searchable_text = ' '.join([
plugin.get('name', ''),
plugin.get('description', ''),
plugin.get('id', ''),
plugin.get('author', ''),
]).lower()
if query_lower not in searchable_text:
continue
filtered.append(plugin)
def _enrich(plugin: Dict) -> Dict:
"""Enrich a single plugin with GitHub metadata.
Called concurrently from a ThreadPoolExecutor. Both HTTP helpers
(``_get_github_repo_info`` / ``_get_latest_commit_info``) are
thread-safe -- they use ``requests`` and write their own cache
keys on Python dicts, which is atomic under the GIL for
single-key assignments.
"""
enhanced_plugin = plugin.copy()
repo_url = plugin.get('repo', '')
if not repo_url:
return enhanced_plugin
github_info = self._get_github_repo_info(repo_url)
enhanced_plugin['stars'] = github_info.get('stars', plugin.get('stars', 0))
enhanced_plugin['default_branch'] = github_info.get('default_branch', plugin.get('branch', 'main'))
enhanced_plugin['last_updated_iso'] = github_info.get('last_commit_iso')
enhanced_plugin['last_updated'] = github_info.get('last_commit_date')
if fetch_commit_info:
branch = plugin.get('branch') or github_info.get('default_branch', 'main')
commit_info = self._get_latest_commit_info(repo_url, branch)
if commit_info:
enhanced_plugin['last_commit'] = commit_info.get('short_sha')
enhanced_plugin['last_commit_sha'] = commit_info.get('sha')
enhanced_plugin['last_updated'] = commit_info.get('date') or enhanced_plugin.get('last_updated')
enhanced_plugin['last_updated_iso'] = commit_info.get('date_iso') or enhanced_plugin.get('last_updated_iso')
enhanced_plugin['last_commit_message'] = commit_info.get('message')
enhanced_plugin['last_commit_author'] = commit_info.get('author')
enhanced_plugin['branch'] = commit_info.get('branch', branch)
enhanced_plugin['last_commit_branch'] = commit_info.get('branch')
# Intentionally NO per-plugin manifest.json fetch here.
# The registry's plugins.json already carries ``description``
# (it is generated from each plugin's manifest by
# ``update_registry.py``), and ``last_updated`` is filled in
# from the commit info above. Fetching manifest.json per
# plugin costs one extra HTTPS round trip per result; on a Pi4
# with a flaky WiFi link the tail retries of that one call
# (_http_get_with_retries does 3 attempts with exponential
# backoff) dominate wall time even with the thread pool.
return enhanced_plugin
# Fan out the per-plugin GitHub enrichment. Serially, a Pi4 with ~15
# plugins and a cold cache makes 30+ HTTP requests in strict sequence
# (the "connecting to display" hang users reported). With a thread
# pool, latency is dominated by the slowest request rather than
# their sum. Workers capped at 10 to stay well under the
# unauthenticated GitHub rate limit burst and avoid overwhelming a
# Pi's WiFi link.
if not filtered:
return []
# Not worth the pool overhead for tiny workloads. Parenthesized to
# make Python's default ``and`` > ``or`` precedence explicit: a
# single plugin, OR a small batch where we don't need commit info.
if (len(filtered) == 1) or ((not fetch_commit_info) and (len(filtered) < 4)):
return [_enrich(p) for p in filtered]
max_workers = min(10, len(filtered))
with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix='plugin-search') as executor:
# executor.map preserves input order, which the UI relies on.
return list(executor.map(_enrich, filtered))
def _fetch_manifest_from_github(self, repo_url: str, branch: str = "master", manifest_path: str = "manifest.json", force_refresh: bool = False) -> Optional[Dict]:
"""
Fetch manifest.json directly from a GitHub repository.
Args:
repo_url: GitHub repository URL
branch: Branch name (default: master)
manifest_path: Path to manifest within the repo (default: manifest.json).
For monorepo plugins this will be e.g. "plugins/football-scoreboard/manifest.json".
force_refresh: If True, bypass the cache.
Returns:
Manifest data or None if not found
"""
try:
owner_repo = github_owner_repo(repo_url)
if owner_repo is None:
return None
owner, repo = owner_repo
cache_key = f"{owner}/{repo}:{branch}:{manifest_path}"
if not force_refresh and cache_key in self.manifest_cache:
cached_time, cached_data = self.manifest_cache[cache_key]
if time.time() - cached_time < self.manifest_cache_timeout:
return cached_data
raw_url = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{manifest_path}"
response = self._http_get_with_retries(raw_url, timeout=10)
if response.status_code == 200:
result = response.json()
self.manifest_cache[cache_key] = (time.time(), result)
return result
if response.status_code == 404 and branch != "main":
raw_url = f"https://raw.githubusercontent.com/{owner}/{repo}/main/{manifest_path}"
response = self._http_get_with_retries(raw_url, timeout=10)
if response.status_code == 200:
result = response.json()
self.manifest_cache[cache_key] = (time.time(), result)
return result
# Cache the miss too, so a plugin without a manifest at this path
# is not re-fetched on every browse.
self.manifest_cache[cache_key] = (time.time(), None)
except Exception as e:
self.logger.debug(f"Could not fetch manifest from GitHub for {repo_url}: {e}")
return None
def _get_latest_commit_info(self, repo_url: str, branch: str = "main", force_refresh: bool = False) -> Optional[Dict[str, Any]]:
"""Return metadata about the latest commit on the given branch."""
try:
owner_repo = github_owner_repo(repo_url)
if owner_repo is None:
return None
owner, repo = owner_repo
cache_key = f"{owner}/{repo}:{branch}"
if not force_refresh and cache_key in self.commit_info_cache:
cached_time, cached_data = self.commit_info_cache[cache_key]
if time.time() - cached_time < self.commit_cache_timeout:
return cached_data
branches_to_try = self._distinct_sequence([branch, 'main', 'master'])
headers = github_api_headers(self.github_token)
last_error = None
for branch_name in branches_to_try:
api_url = f"https://api.github.com/repos/{owner}/{repo}/commits/{branch_name}"
try:
response = requests.get(api_url, headers=headers, timeout=10)
except requests.RequestException as req_err:
# Network failure: fall back to a stale cache hit if
# available so the plugin store UI keeps populating
# commit info on a flaky WiFi link. Bump the cached
# timestamp into the backoff window so we don't
# re-retry on every request.
if cache_key in self.commit_info_cache:
_, stale = self.commit_info_cache[cache_key]
if stale is not None:
self._record_cache_backoff(
self.commit_info_cache, cache_key,
self.commit_cache_timeout, stale,
)
self.logger.warning(
"GitHub commit fetch failed for %s (%s); serving stale cache.",
cache_key, req_err,
)
return stale
last_error = str(req_err)
continue
if response.status_code == 200:
commit_data = response.json()
commit_sha_full = commit_data.get('sha', '')
commit_sha_short = commit_sha_full[:7] if commit_sha_full else ''
commit_meta = commit_data.get('commit', {})
commit_author = commit_meta.get('author', {})
commit_date_iso = commit_author.get('date', '')
result = {
'branch': branch_name,
'sha': commit_sha_full,
'short_sha': commit_sha_short,
'date_iso': commit_date_iso,
'date': self._iso_to_date(commit_date_iso),
'author': commit_author.get('name', ''),
'message': commit_meta.get('message', ''),
}
self.commit_info_cache[cache_key] = (time.time(), result)
return result
if response.status_code == 403 and not self.github_token:
self.logger.debug("GitHub commit API rate limited (403). Consider adding a token.")
last_error = response.text
else:
last_error = response.text
if last_error:
self.logger.debug(f"Unable to fetch commit info for {repo_url}: {last_error}")
# All branches returned a non-200 response (e.g. 404 on every
# candidate, or a transient 5xx). If we already had a good
# cached value, prefer serving that — overwriting it with
# None here would wipe out commit info the UI just showed
# on the previous request. Bump the timestamp into the
# backoff window so subsequent lookups hit the cache.
if cache_key in self.commit_info_cache:
_, prior = self.commit_info_cache[cache_key]
if prior is not None:
self._record_cache_backoff(
self.commit_info_cache, cache_key,
self.commit_cache_timeout, prior,
)
return prior
# No prior good value — cache the negative result so we don't
# hammer a plugin that genuinely has no reachable commits.
self.commit_info_cache[cache_key] = (time.time(), None)
except Exception as e:
self.logger.debug(f"Error fetching latest commit metadata for {repo_url}: {e}")
return None
def get_plugin_info(self, plugin_id: str, fetch_latest_from_github: bool = True, force_refresh: bool = False) -> Optional[Dict]:
"""
Get detailed information about a plugin from the registry.
GitHub provides authoritative metadata such as stars and the latest
commit. The registry supplies descriptive information (name, id, repo URL).
Args:
plugin_id: Plugin identifier
fetch_latest_from_github: If True (default), augment with GitHub commit metadata.
force_refresh: If True, bypass caches for commit/manifest data.
Returns:
Plugin metadata or None if not found
"""
registry = self.fetch_registry()
plugins = registry.get('plugins', []) or []
plugin_info = self._match_registry_entry(plugins, plugin_id)
if not plugin_info:
return None
if fetch_latest_from_github:
repo_url = plugin_info.get('repo')
if repo_url:
plugin_info = plugin_info.copy()
github_info = self._get_github_repo_info(repo_url)
branch = plugin_info.get('branch') or github_info.get('default_branch', 'main')
plugin_info['default_branch'] = github_info.get('default_branch', branch)
plugin_info['stars'] = github_info.get('stars', plugin_info.get('stars', 0))
plugin_info['last_updated'] = github_info.get('last_commit_date', plugin_info.get('last_updated'))
plugin_info['last_updated_iso'] = github_info.get('last_commit_iso', plugin_info.get('last_updated_iso'))
commit_info = self._get_latest_commit_info(repo_url, branch, force_refresh=force_refresh)
if commit_info:
plugin_info['last_commit'] = commit_info.get('short_sha')
plugin_info['last_commit_sha'] = commit_info.get('sha')
plugin_info['last_commit_message'] = commit_info.get('message')
plugin_info['last_commit_author'] = commit_info.get('author')
plugin_info['last_updated'] = commit_info.get('date') or plugin_info.get('last_updated')
plugin_info['last_updated_iso'] = commit_info.get('date_iso') or plugin_info.get('last_updated_iso')
plugin_info['branch'] = commit_info.get('branch', branch)
plugin_info['last_commit_branch'] = commit_info.get('branch')
plugin_subpath = plugin_info.get('plugin_path', '')
manifest_rel = f"{plugin_subpath}/manifest.json" if plugin_subpath else "manifest.json"
github_manifest = self._fetch_manifest_from_github(repo_url, branch, manifest_rel, force_refresh=force_refresh)
if github_manifest:
if 'last_updated' in github_manifest and not plugin_info.get('last_updated'):
plugin_info['last_updated'] = github_manifest['last_updated']
if 'description' in github_manifest:
plugin_info['description'] = github_manifest['description']
return plugin_info
@staticmethod
def _match_registry_entry(plugins: List[Dict], plugin_id: str) -> Optional[Dict]:
"""Find a registry entry by its id, or by the directory it installs to.
Four shipped plugins have a registry ``id`` that differs from the ``id``
in their own manifest: ``weather`` installs to ``plugins/ledmatrix-weather``,
and likewise stocks, music and leaderboard. Installation already prefers
the manifest id for the directory name, so on disk, in ``config.json``
and in a backup manifest those plugins are called ``ledmatrix-weather``.
Only the registry calls them ``weather``, and nothing resolved that in
reverse: restoring a backup asked the store for ``ledmatrix-weather``
and got "Plugin not found in registry", silently dropping four enabled
plugins from a restored device.
Matching ``plugin_path`` fixes it without renaming any published id,
which would orphan ``plugin_state.json`` entries keyed on the old ones.
Exact id always wins, so an entry whose *path* happens to collide with
another entry's id cannot shadow it.
"""
if not plugin_id:
return None
exact = next((p for p in plugins if p.get('id') == plugin_id), None)
if exact is not None:
return exact
for entry in plugins:
path = (entry.get('plugin_path') or '').rstrip('/')
if path and path.rsplit('/', 1)[-1] == plugin_id:
return entry
return None
def get_registry_info(self, plugin_id: str) -> Optional[Dict]:
"""
Get plugin information from the registry cache only (no GitHub API calls).
Use this for lightweight lookups where only registry fields are needed
(e.g., verified status, latest_version).
Args:
plugin_id: Plugin identifier
Returns:
Plugin metadata from registry or None if not found
"""
registry = self.fetch_registry()
plugins = registry.get('plugins', []) or []
return self._match_registry_entry(plugins, plugin_id)
def install_plugin(self, plugin_id: str, branch: Optional[str] = None) -> bool:
"""Install a plugin, keeping any existing install until the new one is
known good.
`_install_plugin_impl` deletes the existing directory *before*
downloading, so every failure after that point — a dropped connection, a
malformed manifest, or the compatibility gate refusing the new version —
left the user with no plugin at all. `_reinstall_with_rollback` gives the
*update* path exactly this protection; a direct install had none, and the
compatibility gate added a new way to reach it.
Pass-through when nothing is installed, and when called from
`_reinstall_with_rollback`, which has already moved the old copy aside.
The aside name embeds BACKUP_MARKER ('.standalone-backup-') so every
plugin directory lookup (src/plugin_system/plugin_dirs.py) skips it
even though it still holds a manifest.json.
Held under the per-plugin reinstall lock for the same reason
`_reinstall_with_rollback` is: the web UI runs Flask with
threaded=True, so a double-clicked Install button gives two threads the
same plugin_id. Interleaved, one thread's restore would delete the
other's freshly installed copy. The lock is reentrant because the
rollback path already holds it when it calls in here.
"""
# Before anything touches the filesystem: plugin_id comes from the
# request body, and the set-aside below moves plugins_dir / plugin_id
# -- which for "../x" is a directory outside the plugins directory.
if not self._is_valid_plugin_id(plugin_id):
self.logger.error(f"Refusing to install invalid plugin id: {plugin_id!r}")
return False
with self._get_reinstall_lock(plugin_id):
plugin_path = self.plugins_dir / plugin_id
if not plugin_path.exists():
return self._install_plugin_impl(plugin_id, branch)
backup_path = plugin_path.with_name(
f"{plugin_path.name}{BACKUP_MARKER}preinstall")
problem = self._set_aside(plugin_path, backup_path)
if problem:
# Can't stage a safety net. Attempting the install anyway is
# what callers got before the net existed; refusing would be
# a new failure mode for a direct install.
self.logger.warning(
"Installing %s without a rollback net: %s", plugin_id, problem)
return self._install_plugin_impl(plugin_id, branch)
try:
installed = self._install_plugin_impl(plugin_id, branch)
except Exception:
self._restore_backup(plugin_id, plugin_path, backup_path, "Install")
raise
if installed:
self._discard_backup(plugin_id, backup_path, "install")
return True
self._restore_backup(plugin_id, plugin_path, backup_path, "Install")
return False
def _set_aside(self, plugin_path: Path, backup_path: Path) -> Optional[str]:
"""Rename an installed plugin to ``backup_path`` so a failed
(re)install can put it back.
A stale backup left by a crash is cleared first, since it would block
the rename. Returns None on success, otherwise why it could not.
"""
if backup_path.exists() and not self._safe_remove_directory(backup_path):
return f"could not clear stale backup at {backup_path}"
try:
plugin_path.rename(backup_path)
except OSError as e:
return f"could not set aside {plugin_path}: {e}"
return None
def _discard_backup(self, plugin_id: str, backup_path: Path, action: str) -> None:
"""Remove the set-aside copy after a successful (re)install."""
if not self._safe_remove_directory(backup_path):
self.logger.warning(
"%s of %s succeeded but the previous copy at %s could not be "
"removed; it will be cleared on the next %s",
action.capitalize(), plugin_id, backup_path, action)
def _restore_backup(
self, plugin_id: str, plugin_path: Path, backup_path: Path, action: str
) -> None:
"""Put the set-aside copy back after a failed (re)install."""
self.logger.error(
"%s of %s failed; restoring the previous version", action, plugin_id)
try:
if plugin_path.exists():
# Partial download debris from the failed install.
self._safe_remove_directory(plugin_path)
backup_path.rename(plugin_path)
self.logger.info("Restored previous install of %s", plugin_id)
except OSError as e:
self.logger.error(
"CRITICAL: could not restore %s from %s: %s. The previous "
"install is preserved there — rename it back manually.",
plugin_id, backup_path, e)
def _install_plugin_impl(self, plugin_id: str, branch: Optional[str] = None) -> bool:
"""
Install a plugin from the official registry. Always installs the latest commit
from the repository's default branch (or specified branch).
Args:
plugin_id: Plugin identifier
branch: Optional branch name to install from. If provided, this branch will be
prioritized. If not provided or branch doesn't exist, falls back to
default branch logic.
"""
branch_info = f" (branch: {branch})" if branch else " (latest branch head)"
self.logger.info(f"Installing plugin: {plugin_id}{branch_info}")
# Remember the originally-requested id so we can clear its uninstall
# record on success even if the manifest renames the directory below.
requested_id = plugin_id
plugin_info = self.get_plugin_info(plugin_id, fetch_latest_from_github=True, force_refresh=True)
if not plugin_info:
self.logger.error(f"Plugin not found in registry: {plugin_id}")
return False
if not self.is_plugin_entry(plugin_info):
self.logger.error(f"Not installing {plugin_id}: registry entry type "
f"{plugin_info.get('type')!r} is not a plugin")
return False
repo_url = plugin_info.get('repo')
if not repo_url:
self.logger.error(f"Plugin {plugin_id} missing repository URL")
return False
plugin_subpath = plugin_info.get('plugin_path')
# If branch is provided, prioritize it; otherwise use default logic
branch_candidates = self._distinct_sequence([
branch, # User-specified branch takes highest priority
plugin_info.get('branch'),
plugin_info.get('default_branch'),
plugin_info.get('last_commit_branch'),
'main',
'master'
])
# Use manifest ID for directory name (not registry plugin_id) to ensure consistency
# We'll read the manifest after installation to get the actual ID
# For now, use plugin_id but we'll correct it after reading manifest
plugin_path = self.plugins_dir / plugin_id
if plugin_path.exists():
self.logger.warning(f"Plugin directory already exists: {plugin_id}. Removing it before reinstall.")
if not self._safe_remove_directory(plugin_path):
self.logger.error(f"Failed to remove existing plugin directory: {plugin_path}")
return False
try:
branch_used = None
if plugin_subpath:
self.logger.info(f"Installing from monorepo subdirectory: {plugin_subpath}")
for candidate in branch_candidates:
download_url = f"{repo_url}/archive/refs/heads/{candidate}.zip"
if self._install_from_monorepo(download_url, plugin_subpath, plugin_path):
branch_used = candidate
break
if branch_used is None:
self.logger.error(f"Failed to install plugin from monorepo path {plugin_subpath} for {plugin_id}")
return False
else:
branch_used = self._install_via_git(repo_url, plugin_path, branch_candidates)
if branch_used is None:
self.logger.info("Git not available or clone failed, attempting archive download...")
for candidate in branch_candidates:
download_url = f"{repo_url}/archive/refs/heads/{candidate}.zip"
if self._install_via_download(download_url, plugin_path):
branch_used = candidate
break
if branch_used is None:
self.logger.error(f"Failed to install plugin {plugin_id} via git or archive download")
return False
manifest_path = plugin_path / "manifest.json"
if not manifest_path.exists():
self.logger.error(f"No manifest.json found in plugin: {plugin_id}")
self._safe_remove_directory(plugin_path)
return False
try:
with open(manifest_path, 'r', encoding='utf-8') as mf:
manifest = json.load(mf)
# Get the actual plugin ID from manifest (source of truth)
manifest_plugin_id = manifest.get('id')
if not manifest_plugin_id:
self.logger.error("Plugin manifest missing 'id' field")
self._safe_remove_directory(plugin_path)
return False
# The manifest id becomes a directory name below (and the old
# directory is removed to make room), so a downloaded manifest
# saying "../x" must not steer that outside plugins_dir.
if not self._is_valid_plugin_id(manifest_plugin_id):
self.logger.error(f"Plugin manifest has an invalid 'id': {manifest_plugin_id!r}")
self._safe_remove_directory(plugin_path)
return False
# If manifest ID doesn't match directory name, rename directory to match manifest
if manifest_plugin_id != plugin_id:
self.logger.warning(
f"Manifest ID '{manifest_plugin_id}' doesn't match registry ID '{plugin_id}'. "
f"Renaming directory to match manifest ID."
)
correct_path = self.plugins_dir / manifest_plugin_id
if correct_path.exists():
self.logger.warning(f"Target directory {manifest_plugin_id} already exists, removing it")
if not self._safe_remove_directory(correct_path):
self.logger.error(f"Failed to remove existing directory {correct_path}, cannot rename plugin")
return False
shutil.move(str(plugin_path), str(correct_path))
plugin_path = correct_path
manifest_path = plugin_path / "manifest.json"
# Update plugin_id to match manifest for rest of function
plugin_id = manifest_plugin_id
required_fields = ['id', 'name', 'class_name', 'display_modes']
missing = [field for field in required_fields if field not in manifest]
manifest_modified = False
if 'class_name' in missing:
entry_point = manifest.get('entry_point', 'manager.py')
manager_file = plugin_path / entry_point
if manager_file.exists():
try:
detected_class = self._detect_class_name(manager_file)
if detected_class:
manifest['class_name'] = detected_class
missing.remove('class_name')
manifest_modified = True
self.logger.info(f"Auto-detected class_name '{detected_class}' from {entry_point}")
except Exception as err:
self.logger.warning(f"Could not auto-detect class_name for {plugin_id}: {err}")
if missing:
self.logger.error(f"Plugin manifest missing required fields for {plugin_id}: {', '.join(missing)}")
self._safe_remove_directory(plugin_path)
return False
# Refuse a plugin that needs a newer core than this one. The
# registry carries no compatibility field, so the floor is only
# knowable once the files are down — checking here, before
# dependency installation, is the earliest possible point.
#
# Refusing costs the user nothing: on an update this returns
# False and _reinstall_with_rollback restores the version they
# already had. Allowing it costs them a plugin that raises
# ModuleNotFoundError at load and is reported only as one line
# in the journal. See docs/SPORTS_UNIFICATION.md (phase B4/B6).
from src.plugin_system import compatibility
# On disk, not as imported: this process may predate the core
# update that made the plugin compatible. See
# compatibility.current_core_version.
core_version = compatibility.current_core_version()
compatible, reason = compatibility.check(manifest, core_version)
if not compatible:
self.logger.error(
"Refusing to install %s: %s", plugin_id, reason)
self._safe_remove_directory(plugin_path)
return False
if 'entry_point' not in manifest:
manifest['entry_point'] = 'manager.py'
manifest_modified = True
self.logger.info(f"Added missing entry_point field to {plugin_id} manifest (defaulted to manager.py)")
if manifest_modified:
with open(manifest_path, 'w', encoding='utf-8') as mf:
json.dump(manifest, mf, indent=2)
except Exception as manifest_error:
self.logger.error(f"Failed to read/validate manifest for {plugin_id}: {manifest_error}")
self._safe_remove_directory(plugin_path)
return False
if not self._install_dependencies(plugin_path):
self.logger.warning(f"Some dependencies may not have installed correctly for {plugin_id}")
branch_display = branch_used or plugin_info.get('branch') or plugin_info.get('default_branch', 'unknown')
self.logger.info(f"Successfully installed plugin: {plugin_id} (branch {branch_display})")
# User deliberately (re)installed this plugin — clear any persistent
# uninstall record so future core updates keep it.
self.forget_uninstalled_plugin(requested_id, plugin_id)
return True
except Exception as e:
self.logger.error(f"Error installing plugin {plugin_id}: {e}", exc_info=True)
if plugin_path.exists():
self._safe_remove_directory(plugin_path)
return False
def install_from_url(self, repo_url: str, plugin_id: str = None, plugin_path: str = None, branch: Optional[str] = None) -> Dict[str, Any]:
"""
Install a plugin directly from a GitHub URL.
This allows users to install custom/unverified plugins.
Supports two installation modes:
1. Direct plugin repo: Repository contains a single plugin with manifest.json at root
2. Monorepo with plugin_path: Repository contains multiple plugins, install from subdirectory
Args:
repo_url: GitHub repository URL (e.g., https://github.com/user/repo)
plugin_id: Optional plugin ID (extracted from manifest if not provided)
plugin_path: Optional subdirectory path for monorepo installations (e.g., "plugins/hello-world")
branch: Optional branch name to install from. If provided, this branch will be
prioritized. If not provided or branch doesn't exist, falls back to
default branch logic (main, then master).
Returns:
Dict with status and plugin_id or error message
"""
branch_info = f" (branch: {branch})" if branch else ""
self.logger.info(f"Installing plugin from custom URL: {repo_url}{branch_info}" + (f" (subpath: {plugin_path})" if plugin_path else ""))
repo_url = normalize_repo_url(repo_url)
temp_dir = None
try:
# Create temporary directory
temp_dir = Path(tempfile.mkdtemp(prefix='ledmatrix_plugin_'))
# Build branch candidates list - prioritize user-specified branch
branch_candidates = self._distinct_sequence([branch, 'main', 'master']) if branch else ['main', 'master']
# For monorepo installations, download and extract subdirectory
if plugin_path:
branch_used = None
for candidate in branch_candidates:
download_url = f"{repo_url}/archive/refs/heads/{candidate}.zip"
if self._install_from_monorepo(download_url, plugin_path, temp_dir):
branch_used = candidate
break
if branch_used is None:
return {
'success': False,
'error': f'Failed to download or extract plugin from monorepo subdirectory: {plugin_path}'
}
else:
branch_used = self._install_via_git(repo_url, temp_dir, branch_candidates)
if branch_used is not None:
self.logger.info(f"Cloned via git (branch: {branch_used})")
else:
self.logger.info("Git not available or clone failed, attempting archive download...")
for candidate in branch_candidates:
download_url = f"{repo_url}/archive/refs/heads/{candidate}.zip"
if self._install_via_download(download_url, temp_dir):
branch_used = candidate
break
if branch_used is None:
return {
'success': False,
'error': 'Failed to clone or download repository'
}
# Read manifest to get plugin ID
manifest_path = temp_dir / "manifest.json"
if not manifest_path.exists():
return {
'success': False,
'error': 'No manifest.json found in repository' + (f' at path: {plugin_path}' if plugin_path else '')
}
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
requested_id = plugin_id
plugin_id = plugin_id or manifest.get('id')
if not plugin_id:
return {
'success': False,
'error': 'No plugin ID found in manifest'
}
# plugin_id names the directory that is removed and then replaced
# below, and it comes from the request body or a downloaded
# manifest -- so "../x" would reach outside plugins_dir.
if not self._is_valid_plugin_id(plugin_id):
return {
'success': False,
'error': f'Invalid plugin ID: {plugin_id!r}'
}
# Validate manifest has required fields
required_fields = ['id', 'name', 'class_name', 'display_modes']
missing_fields = [field for field in required_fields if field not in manifest]
if missing_fields:
return {
'success': False,
'error': f'Manifest missing required fields: {", ".join(missing_fields)}'
}
# Refuse a plugin that needs a newer core than this one, exactly as
# _install_plugin_impl does after its download. Sideloading is an
# explicit act rather than an automatic store update, but the floor
# is not advice about intent -- it is a statement that the plugin
# cannot run here, and letting it through produces the same silent
# PluginState.ERROR at load. This was the last of the three routes
# in that skipped the check.
#
# Before the move, so the `finally` below removes the temp tree and
# nothing half-installed is left behind.
from src.plugin_system import compatibility
core_version = compatibility.current_core_version()
compatible, reason = compatibility.check(manifest, core_version)
if not compatible:
self.logger.error(
"Refusing to install %s from %s: %s",
plugin_id, repo_url, reason)
return {'success': False, 'error': reason}
# Validate version fields consistency (warnings only, not required)
validation_errors = self._validate_manifest_version_fields(manifest)
if validation_errors:
self.logger.warning(f"Manifest version field validation warnings for {plugin_id}: {', '.join(validation_errors)}")
# Optional: Full schema validation if available
schema_errors = self._validate_manifest_schema(manifest, plugin_id)
if schema_errors:
self.logger.warning(f"Manifest schema validation warnings for {plugin_id}: {', '.join(schema_errors)}")
# entry_point is optional, default to "manager.py" if not specified
if 'entry_point' not in manifest:
manifest['entry_point'] = 'manager.py'
# Write updated manifest back to file
with open(manifest_path, 'w', encoding='utf-8') as f:
json.dump(manifest, f, indent=2)
self.logger.info(f"Added missing entry_point field to {plugin_id} manifest (defaulted to manager.py)")
# The directory is named for the caller's plugin_id when one was
# given (update_plugin passes the installed id), else for the
# manifest's id -- so it can differ from the manifest id, which
# discovery tolerates by reading the manifest.
final_path = self.plugins_dir / plugin_id
# Set the existing copy aside rather than deleting it, and put it
# back if the move fails: deleting first left the user with no
# plugin at all whenever the move broke part-way. Under the
# per-plugin reinstall lock, as install_plugin() is, so two
# overlapping installs of one id can't interleave their renames.
with self._get_reinstall_lock(plugin_id):
backup_path = None
if final_path.exists():
self.logger.warning(f"Plugin {plugin_id} already exists, replacing existing copy")
backup_path = final_path.with_name(
f"{final_path.name}{BACKUP_MARKER}preinstall")
problem = self._set_aside(final_path, backup_path)
if problem:
return {
'success': False,
'error': f'Failed to replace existing plugin directory: {problem}'
}
try:
shutil.move(str(temp_dir), str(final_path))
except Exception:
if backup_path is not None:
self._restore_backup(plugin_id, final_path, backup_path, "Install")
raise
temp_dir = None # Prevent cleanup since we moved it
if backup_path is not None:
self._discard_backup(plugin_id, backup_path, "install")
# Install dependencies
self._install_dependencies(final_path)
branch_info = f" (branch: {branch_used})" if branch_used else ""
self.logger.info(f"Successfully installed plugin from URL: {plugin_id}{branch_info}")
# User deliberately (re)installed this plugin -- clear any persistent
# uninstall record, exactly as install_plugin() does. Without this the
# id stays in config/uninstalled_plugins.json and
# purge_uninstalled_plugins(), which runs at every web-app startup,
# deletes the directory again: the plugin works for the rest of the
# session and is gone after the next reboot.
self.forget_uninstalled_plugin(
*(pid for pid in (requested_id, plugin_id, manifest.get('id')) if pid)
)
result = {
'success': True,
'plugin_id': plugin_id,
'name': manifest.get('name')
}
if branch_used:
result['branch'] = branch_used
return result
except json.JSONDecodeError as e:
self.logger.error(f"Error parsing manifest JSON: {e}")
return {
'success': False,
'error': f'Invalid manifest.json: {str(e)}'
}
except Exception as e:
self.logger.error(f"Error installing from URL: {e}", exc_info=True)
return {
'success': False,
'error': str(e)
}
finally:
# Cleanup temp directory if it still exists
if temp_dir and temp_dir.exists():
shutil.rmtree(temp_dir, ignore_errors=True)
def _detect_class_name(self, manager_file: Path) -> Optional[str]:
"""
Attempt to auto-detect the plugin class name from the manager file.
Args:
manager_file: Path to the manager.py file
Returns:
Class name if found, None otherwise
"""
try:
with open(manager_file, 'r', encoding='utf-8') as f:
content = f.read()
# Look for class definition that inherits from BasePlugin
pattern = r'class\s+(\w+)\s*\([^)]*BasePlugin[^)]*\)'
match = re.search(pattern, content)
if match:
return match.group(1)
# Fallback: find first class definition
pattern = r'^class\s+(\w+)'
match = re.search(pattern, content, re.MULTILINE)
if match:
return match.group(1)
return None
except Exception as e:
self.logger.warning(f"Error detecting class name from {manager_file}: {e}")
return None
def _install_via_git(self, repo_url: str, target_path: Path, branches: Optional[List[str]] = None) -> Optional[str]:
"""Clone a repository into ``target_path``.
Tries each of ``branches`` (default ``main``, ``master``), then the
repository's own default branch, so a repository whose only branch
is e.g. ``develop`` still installs.
Returns:
The branch that was cloned, or None when every clone failed and
``target_path`` has been removed. After a default-branch clone
this is the branch the clone checked out (``'HEAD'`` if the
remote's HEAD is detached), never None.
"""
branches_to_try = self._distinct_sequence(branches or [])
if not branches_to_try:
branches_to_try = ['main', 'master']
last_error = None
for try_branch in branches_to_try:
try:
cmd = ['git', 'clone', '--depth', '1', '--branch', try_branch, repo_url, str(target_path)]
subprocess.run(
cmd,
check=True,
capture_output=True,
text=True,
timeout=60
)
self.logger.debug(f"Successfully cloned {repo_url} (branch: {try_branch}) to {target_path}")
return try_branch
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError) as e:
last_error = e
self.logger.debug(f"Git clone failed for branch {try_branch}: {e}")
if target_path.exists():
self._safe_remove_directory(target_path)
# Try default branch (Git's configured default) as last resort
try:
cmd = ['git', 'clone', '--depth', '1', repo_url, str(target_path)]
subprocess.run(
cmd,
check=True,
capture_output=True,
text=True,
timeout=60
)
self.logger.debug(f"Successfully cloned {repo_url} (git default branch) to {target_path}")
return self._checked_out_branch(target_path)
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError) as e:
last_error = e
if target_path.exists():
self._safe_remove_directory(target_path)
self.logger.error(f"Git clone failed for all attempted branches: {last_error}")
return None
@staticmethod
def _checked_out_branch(checkout: Path) -> str:
"""The branch a fresh clone has checked out, read from ``.git/HEAD``.
``'HEAD'`` when HEAD is detached or unreadable.
"""
try:
head = (checkout / '.git' / 'HEAD').read_text(encoding='utf-8').strip()
except OSError:
return 'HEAD'
prefix = 'ref: refs/heads/'
return head[len(prefix):] if head.startswith(prefix) else 'HEAD'
def _install_from_monorepo(self, download_url: str, plugin_subpath: str, target_path: Path) -> bool:
"""
Install a plugin from a monorepo by downloading only the target subdirectory.
Uses the GitHub Git Trees API to list files, then downloads each file
individually from raw.githubusercontent.com. Falls back to downloading
the full ZIP archive if the API approach fails.
Args:
download_url: URL to download zip from (used as fallback and to extract repo info)
plugin_subpath: Path within repo (e.g., "plugins/hello-world")
target_path: Target directory for plugin
Returns:
True if successful
"""
# Try the API-based approach first (downloads only the target directory)
repo_url, branch = self._parse_monorepo_download_url(download_url)
if repo_url and branch:
result = self._install_from_monorepo_api(repo_url, branch, plugin_subpath, target_path)
if result:
return True
self.logger.info(f"API-based install failed for {plugin_subpath}, falling back to ZIP download")
# Ensure no partial files remain before ZIP fallback
if target_path.exists():
self._safe_remove_directory(target_path)
# Fallback: download full ZIP and extract subdirectory
return self._install_from_monorepo_zip(download_url, plugin_subpath, target_path)
@staticmethod
def _parse_monorepo_download_url(download_url: str):
"""Extract repo URL and branch from a GitHub archive download URL.
Example: "https://github.com/ChuckBuilds/ledmatrix-plugins/archive/refs/heads/main.zip"
Returns: ("https://github.com/ChuckBuilds/ledmatrix-plugins", "main")
"""
try:
# Pattern: {repo_url}/archive/refs/heads/{branch}.zip
if '/archive/refs/heads/' in download_url:
parts = download_url.split('/archive/refs/heads/')
repo_url = parts[0]
branch = parts[1].removesuffix('.zip')
return repo_url, branch
except (IndexError, AttributeError):
pass
return None, None
def _install_from_monorepo_api(self, repo_url: str, branch: str, plugin_subpath: str, target_path: Path) -> bool:
"""
Install a plugin subdirectory using the GitHub Git Trees API.
Downloads only the files in the target subdirectory (~200KB) instead
of the entire repository ZIP (~5MB+). Uses one API call for the tree
listing, then downloads individual files from raw.githubusercontent.com.
Args:
repo_url: GitHub repository URL (e.g., "https://github.com/owner/repo")
branch: Branch name (e.g., "main")
plugin_subpath: Path within repo (e.g., "plugins/hello-world")
target_path: Target directory for plugin
Returns:
True if successful, False to trigger ZIP fallback
"""
try:
owner_repo = github_owner_repo(repo_url)
if owner_repo is None:
return False
owner, repo = owner_repo
# Step 1: Get the recursive tree listing (1 API call)
api_url = f"https://api.github.com/repos/{owner}/{repo}/git/trees/{branch}?recursive=true"
tree_response = self._http_get_with_retries(
api_url, timeout=15, headers=github_api_headers(self.github_token))
if tree_response.status_code != 200:
self.logger.debug(f"Trees API returned {tree_response.status_code} for {owner}/{repo}")
return False
tree_data = tree_response.json()
if tree_data.get('truncated'):
self.logger.debug(f"Tree response truncated for {owner}/{repo}, falling back to ZIP")
return False
# Step 2: Filter for files in the target subdirectory
prefix = f"{plugin_subpath.strip('/')}/"
file_entries = [
entry for entry in tree_data.get('tree', [])
if entry['path'].startswith(prefix) and entry['type'] == 'blob'
]
if not file_entries:
self.logger.error(f"No files found under '{plugin_subpath}' in tree for {owner}/{repo}")
return False
# Sanity check: refuse unreasonably large plugin directories
max_files = 500
if len(file_entries) > max_files:
self.logger.error(
f"Plugin {plugin_subpath} has {len(file_entries)} files (limit {max_files}), "
f"falling back to ZIP"
)
return False
self.logger.info(f"Downloading {len(file_entries)} files for {plugin_subpath} via API")
# Step 3: Create target directory and download each file
ensure_directory_permissions(target_path.parent, get_plugin_dir_mode())
target_path.mkdir(parents=True, exist_ok=True)
prefix_len = len(prefix)
target_root = target_path.resolve()
for entry in file_entries:
# Relative path within the plugin directory
rel_path = entry['path'][prefix_len:]
dest_file = target_path / rel_path
# Guard against path traversal
if not dest_file.resolve().is_relative_to(target_root):
self.logger.error(
f"Path traversal detected: {entry['path']!r} resolves outside target directory"
)
if target_path.exists():
self._safe_remove_directory(target_path)
return False
# Create parent directories
dest_file.parent.mkdir(parents=True, exist_ok=True)
# Download from raw.githubusercontent.com (no API rate limit cost)
raw_url = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{entry['path']}"
file_response = self._http_get_with_retries(raw_url, timeout=30)
if file_response.status_code != 200:
self.logger.error(f"Failed to download {entry['path']}: HTTP {file_response.status_code}")
# Clean up partial download
if target_path.exists():
self._safe_remove_directory(target_path)
return False
dest_file.write_bytes(file_response.content)
self.logger.info(f"Successfully installed {plugin_subpath} via API ({len(file_entries)} files)")
return True
except Exception as e:
self.logger.debug(f"API-based monorepo install failed: {e}")
# Clean up partial download
if target_path.exists():
self._safe_remove_directory(target_path)
return False
def _install_from_monorepo_zip(self, download_url: str, plugin_subpath: str, target_path: Path) -> bool:
"""
Fallback: install a plugin from a monorepo by downloading the full ZIP.
Used when the API-based approach fails (rate limited, auth issues, etc.).
"""
tmp_zip_path = None
temp_extract = None
try:
self.logger.info(f"Downloading monorepo ZIP from: {download_url}")
response = self._http_get_with_retries(download_url, timeout=60, stream=True)
response.raise_for_status()
# Download to temporary file
with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp_file:
for chunk in response.iter_content(chunk_size=8192):
tmp_file.write(chunk)
tmp_zip_path = tmp_file.name
with zipfile.ZipFile(tmp_zip_path, 'r') as zip_ref:
zip_contents = zip_ref.namelist()
if not zip_contents:
return False
root_dir = zip_contents[0].split('/')[0]
plugin_prefix = f"{root_dir}/{plugin_subpath}/"
# Extract ONLY files under the plugin subdirectory
plugin_members = [m for m in zip_contents if m.startswith(plugin_prefix)]
if not plugin_members:
self.logger.error(f"Plugin path not found in archive: {plugin_subpath}")
return False
temp_extract = Path(tempfile.mkdtemp())
temp_extract_resolved = temp_extract.resolve()
for member in plugin_members:
# Guard against zip-slip (directory traversal)
member_dest = (temp_extract / member).resolve()
if not member_dest.is_relative_to(temp_extract_resolved):
self.logger.error(
f"Zip-slip detected: member {member!r} resolves outside "
f"temp directory, aborting"
)
shutil.rmtree(temp_extract, ignore_errors=True)
return False
zip_ref.extract(member, temp_extract)
source_plugin_dir = temp_extract / root_dir / plugin_subpath
ensure_directory_permissions(target_path.parent, get_plugin_dir_mode())
# Ensure target doesn't exist to prevent shutil.move nesting
if target_path.exists():
if not self._safe_remove_directory(target_path):
self.logger.error(f"Cannot remove existing target {target_path} for monorepo install")
return False
shutil.move(str(source_plugin_dir), str(target_path))
return True
except Exception as e:
self.logger.error(f"Monorepo ZIP download failed: {e}", exc_info=True)
return False
finally:
if tmp_zip_path and os.path.exists(tmp_zip_path):
os.remove(tmp_zip_path)
if temp_extract and temp_extract.exists():
shutil.rmtree(temp_extract, ignore_errors=True)
def _install_via_download(self, download_url: str, target_path: Path) -> bool:
"""
Install plugin by downloading and extracting zip archive.
Args:
download_url: URL to download zip from
target_path: Target directory
Returns:
True if successful
"""
try:
self.logger.info(f"Downloading from: {download_url}")
# Allow redirects (GitHub archive URLs redirect to codeload.github.com)
response = self._http_get_with_retries(download_url, timeout=60, stream=True, headers={'User-Agent': USER_AGENT})
response.raise_for_status()
# Download to temporary file
with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp_file:
for chunk in response.iter_content(chunk_size=8192):
tmp_file.write(chunk)
tmp_zip_path = tmp_file.name
temp_extract = None
try:
# Extract zip
with zipfile.ZipFile(tmp_zip_path, 'r') as zip_ref:
# GitHub zips have a root directory, we need to extract contents
zip_contents = zip_ref.namelist()
if not zip_contents:
return False
# Find the root directory in the zip
root_dir = zip_contents[0].split('/')[0]
# Extract to temp location with zip-slip protection
temp_extract = Path(tempfile.mkdtemp())
temp_extract_resolved = temp_extract.resolve()
for member in zip_ref.namelist():
member_dest = (temp_extract / member).resolve()
if not member_dest.is_relative_to(temp_extract_resolved):
self.logger.error(
f"Zip-slip detected: member {member!r} resolves outside "
f"temp directory, aborting"
)
return False
zip_ref.extractall(temp_extract)
# Move contents from root_dir to target
source_dir = temp_extract / root_dir
if source_dir.exists():
ensure_directory_permissions(target_path.parent, get_plugin_dir_mode())
shutil.move(str(source_dir), str(target_path))
else:
# No root dir, move everything
shutil.move(str(temp_extract), str(target_path))
return True
finally:
# Remove temporary zip file
if os.path.exists(tmp_zip_path):
os.remove(tmp_zip_path)
# Cleanup temp extract dir here rather than on the success
# path, so a failed extract or move doesn't leave a copy of
# the plugin in /tmp on every attempt. (Gone already when the
# whole dir was moved into place.)
if temp_extract is not None and temp_extract.exists():
shutil.rmtree(temp_extract, ignore_errors=True)
except Exception as e:
self.logger.error(f"Download failed: {e}")
return False
def _install_dependencies(self, plugin_path: Path) -> bool:
"""
Install Python dependencies from requirements.txt.
``plugin_path`` is ultimately derived from a plugin-supplied manifest
``id``, so it is only used after contained_plugin_dir() has rebuilt it
from a listing of ``self.plugins_dir``.
Args:
plugin_path: Path to plugin directory
Returns:
True if successful or no requirements file
"""
safe_plugin_dir = contained_plugin_dir(plugin_path, self.plugins_dir)
if safe_plugin_dir is None:
self.logger.error("Plugin directory not found inside plugins dir for dependency install")
return False
requirements_file = requirements_to_install(safe_plugin_dir, self.logger, plugin_path.name)
if requirements_file is None:
return True
try:
self.logger.info(f"Installing dependencies for {plugin_path.name}")
# Routed through the shared root-visible installer (same one the
# web UI's "Reinstall Plugin Deps" tool uses) rather than a bare
# `pip`/`pip3` off PATH: a bare pip binary can silently resolve to
# a different Python installation than the one that actually runs
# ledmatrix.service, so pip reports success while the package
# stays invisible to the running plugin (e.g. missing `astral`
# for the weather plugin even though "install" succeeded).
result = install_requirements_file(Path(requirements_file), timeout=300)
if result.returncode != 0:
self.logger.error(
f"Error installing dependencies for {plugin_path.name}: {result.stderr}"
)
return False
self.logger.info(f"Dependencies installed successfully for {plugin_path.name}")
return True
except subprocess.TimeoutExpired:
self.logger.error("Dependency installation timed out")
return False
except OSError as e:
# A broken pipe (EPIPE) happens when pip's output pipe closes
# mid-download, usually a network interruption.
if e.errno == errno.EPIPE:
self.logger.error(
f"Broken pipe error during dependency installation for {plugin_path.name}. "
f"This usually indicates a network interruption or pip output buffer issue. "
f"Try installing again or check your network connection."
)
else:
self.logger.error(f"OS error during dependency installation: {e}")
return False
except Exception as e:
self.logger.error(f"Unexpected error installing dependencies for {plugin_path.name}: {e}", exc_info=True)
return False
def _git_cache_signature(self, git_dir: Path) -> Optional[Tuple]:
"""Build a cache signature that invalidates on the kind of updates
a plugin user actually cares about.
Caching on ``.git/HEAD`` mtime alone is not enough: a ``git pull``
that fast-forwards the current branch updates
``.git/refs/heads/<branch>`` (or ``.git/packed-refs``) but leaves
HEAD's contents and mtime untouched. And the cached ``result``
dict includes ``remote_url`` — a value read from ``.git/config`` —
so a config-only change (e.g. a monorepo-migration re-pointing
``remote.origin.url``) must also invalidate the cache.
Signature components:
- HEAD contents (catches detach / branch switch)
- HEAD mtime
- if HEAD points at a ref, that ref file's mtime (catches
fast-forward / reset on the current branch)
- packed-refs mtime as a coarse fallback for repos using packed refs
- .git/config contents + mtime (catches remote URL changes and
any other config-only edit that affects what the cached
``remote_url`` field should contain)
Returns ``None`` if HEAD cannot be read at all (caller will skip
the cache and take the slow path).
"""
head_file = git_dir / 'HEAD'
try:
head_mtime = head_file.stat().st_mtime
head_contents = head_file.read_text(encoding='utf-8', errors='replace').strip()
except OSError:
return None
ref_mtime = None
if head_contents.startswith('ref: '):
ref_path = head_contents[len('ref: '):].strip()
# ``ref_path`` looks like ``refs/heads/main``. It lives either
# as a loose file under .git/ or inside .git/packed-refs.
loose_ref = git_dir / ref_path
try:
ref_mtime = loose_ref.stat().st_mtime
except OSError:
ref_mtime = None
packed_refs_mtime = None
if ref_mtime is None:
try:
packed_refs_mtime = (git_dir / 'packed-refs').stat().st_mtime
except OSError:
packed_refs_mtime = None
config_mtime = None
config_contents = None
config_file = git_dir / 'config'
try:
config_mtime = config_file.stat().st_mtime
config_contents = config_file.read_text(encoding='utf-8', errors='replace').strip()
except OSError:
config_mtime = None
config_contents = None
return (
head_contents, head_mtime,
ref_mtime, packed_refs_mtime,
config_contents, config_mtime,
)
def _get_local_git_info(self, plugin_path: Path) -> Optional[Dict[str, str]]:
"""Return local git branch, commit hash, and commit date if the plugin is a git checkout.
Results are cached keyed on a signature that includes HEAD
contents plus the mtime of HEAD AND the resolved ref (or
packed-refs). Repeated calls skip the ``git log`` subprocess when
nothing has changed, and a ``git pull`` that fast-forwards the
branch correctly invalidates the cache.
"""
git_dir = plugin_path / '.git'
if not git_dir.exists():
return None
cache_key = str(plugin_path)
signature = self._git_cache_signature(git_dir)
if signature is not None:
cached = self._git_info_cache.get(cache_key)
if cached is not None and cached[0] == signature:
return cached[1]
try:
# .git may be a file (worktree / submodule) containing "gitdir: <path>".
# Resolve it to the actual git directory before reading any files.
try:
if git_dir.is_file():
pointer = git_dir.read_text(encoding='utf-8', errors='replace').strip()
if pointer.startswith('gitdir:'):
resolved = (plugin_path / pointer[len('gitdir:'):].strip()).resolve()
if resolved.is_dir():
git_dir = resolved
else:
return None
else:
return None
except (OSError, NotADirectoryError):
return None
# Read branch directly from .git/HEAD (no subprocess).
branch = ''
try:
head_text = (git_dir / 'HEAD').read_text(encoding='utf-8', errors='replace').strip()
if head_text.startswith('ref: refs/heads/'):
branch = head_text[len('ref: refs/heads/'):]
elif head_text.startswith('ref: '):
branch = head_text[len('ref: '):]
# else: detached HEAD — branch stays ''
except (OSError, NotADirectoryError):
pass
# Remote URL from .git/config — parse [remote "origin"] url line.
remote_url = None
try:
config_text = (git_dir / 'config').read_text(encoding='utf-8', errors='replace')
in_origin = False
for line in config_text.splitlines():
stripped = line.strip()
if stripped == '[remote "origin"]':
in_origin = True
elif stripped.startswith('['):
in_origin = False
elif in_origin and stripped.startswith('url') and '=' in stripped:
remote_url = stripped.split('=', 1)[1].strip()
break
except (OSError, NotADirectoryError):
pass
# Single subprocess: SHA + commit date in one call.
log_result = subprocess.run(
['git', '-C', str(plugin_path), 'log', '-1', '--format=%H%n%cI', 'HEAD'],
capture_output=True,
text=True,
timeout=10,
check=True
)
lines = log_result.stdout.strip().splitlines()
sha = lines[0] if lines else ''
commit_date_iso = lines[1] if len(lines) > 1 else ''
result = {
'sha': sha,
'short_sha': sha[:7] if sha else '',
'branch': branch,
}
if remote_url:
result['remote_url'] = remote_url
if commit_date_iso:
result['date_iso'] = commit_date_iso
result['date'] = self._iso_to_date(commit_date_iso)
if signature is not None:
self._git_info_cache[cache_key] = (signature, result)
return result
except subprocess.CalledProcessError as err:
self.logger.debug(f"Failed to read git info for {plugin_path.name}: {err}")
except subprocess.TimeoutExpired:
self.logger.debug(f"Timed out reading git info for {plugin_path.name}")
return None
def _safe_remove_directory(self, path: Path) -> bool:
"""
Safely remove a directory, handling permission errors for root-owned files.
Attempts removal in three stages:
1. Normal shutil.rmtree()
2. Fix permissions via os.chmod() then retry (works for same-owner files)
3. Use sudo rm -rf as last resort (works for root-owned __pycache__, etc.)
Args:
path: Path to directory to remove
Returns:
True if directory was removed successfully, False otherwise
"""
if not path.exists():
return True # Already removed
# Stage 1: Try normal removal
try:
shutil.rmtree(path)
return True
except OSError:
self.logger.warning(f"Permission error removing {path}, attempting chmod fix...")
# Stage 2: Try chmod + retry (works when we own the files)
try:
for root, _dirs, files in os.walk(path):
root_path = Path(root)
try:
os.chmod(root_path, stat.S_IRWXU)
except (OSError, PermissionError):
pass
for file in files:
try:
os.chmod(root_path / file, stat.S_IRWXU)
except (OSError, PermissionError):
pass
shutil.rmtree(path)
self.logger.info(f"Removed {path} after fixing permissions")
return True
except (PermissionError, OSError):
self.logger.warning(f"chmod fix failed for {path}, attempting sudo removal...")
# Stage 3: Use sudo rm -rf (for root-owned __pycache__, data/.cache, etc.)
if sudo_remove_directory(path):
return True
# Final check — maybe partial removal got everything
if not path.exists():
return True
self.logger.error(f"All removal strategies failed for {path}")
return False
def _find_plugin_path(self, plugin_id: str) -> Optional[Path]:
"""
Find the plugin path by checking the configured directory and standard plugins directory.
Searches the configured directory, then a sibling ``plugins/`` (the
case where plugins sit in plugins/ but config says plugin-repos/) --
a store-only fallback; discovery scans the configured directory only.
Each directory is searched completely before the next, by the shared
rules in ``src/plugin_system/plugin_dirs.py``: a directory whose
manifest declares the id wins, then a directory named exactly for it.
The manifest match matters because a directory name can differ from
the id its manifest declares (a hand-made or legacy layout such as
`ledmatrix-stocks/` holding id `stocks`); a lookup by directory name
alone reported such a plugin as not installed, so update_plugin()
silently did nothing.
No ``ledmatrix-`` prefix and no case folding here, unlike the loader:
a store operation may delete what this returns, so it only accepts a
directory that names the id exactly or declares it. So a registry id
such as `stocks` does not resolve to an installed `ledmatrix-stocks/`
declaring `ledmatrix-stocks` (the monorepo's leaderboard, music,
stocks and weather); callers pass the installed id, and
update_plugin() maps it back to the registry id itself.
Args:
plugin_id: Plugin identifier
Returns:
Path to plugin directory if found, None otherwise
"""
return resolve_plugin_dir(
plugin_id, self._candidate_plugin_dirs(), prefix=False,
case_insensitive=False)
def _candidate_plugin_dirs(self) -> List[Path]:
"""Directories that may hold installed plugins, configured one first."""
return [d for d in store_search_dirs(self.plugins_dir) if d.exists()]
def uninstall_plugin(self, plugin_id: str) -> bool:
"""
Uninstall a plugin by removing its directory.
Args:
plugin_id: Plugin identifier
Returns:
True if uninstalled successfully (or already not installed)
"""
plugin_path = self._find_plugin_path(plugin_id)
if plugin_path is None or not plugin_path.exists():
self.logger.info(f"Plugin {plugin_id} not found (already uninstalled)")
return True # Already uninstalled, consider this success
try:
self.logger.info(f"Uninstalling plugin: {plugin_id}")
if self._safe_remove_directory(plugin_path):
self.logger.info(f"Successfully uninstalled plugin: {plugin_id}")
return True
else:
self.logger.error(f"Failed to remove plugin directory: {plugin_path}")
return False
except Exception as e:
self.logger.error(f"Error uninstalling plugin {plugin_id}: {e}")
return False
def _gate_pulled_commit(self, plugin_id: str, plugin_path: Path,
previous_sha: Optional[str]) -> bool:
"""Apply the compatibility gate to a commit that arrived via git pull.
Every other route into an installed plugin goes through
``install_plugin``, which gates in ``_install_plugin_impl``. This one
did not: a ``git pull`` could deliver a manifest flooring above this
core and nothing would notice until the plugin failed to load, which
surfaces as one line in the journal and a scoreboard that silently
stopped appearing.
Checked after the pull rather than before it, for the same reason
``_install_plugin_impl`` checks after the download: the registry
carries no compatibility field, so the incoming floor is only knowable
once the new commit is on disk.
Undone with ``git reset --hard`` rather than by removing the directory.
This is a live checkout, the previous commit is still in the object
store, and the reset leaves the user on the exact version they were
already running -- the same promise ``_reinstall_with_rollback`` makes,
reached by the means this path actually has. It is also the gentler
option: no window in which the plugin directory does not exist, and no
``.standalone-backup-`` debris if the process dies mid-way.
A manifest that cannot be read is not evidence of incompatibility, so
it allows. ``compatibility.check`` refuses only on evidence for the
same reason: a wrong refusal breaks a working install, while a wrong
allowance degrades to exactly the behaviour this path had before the
gate existed.
"""
manifest_path = plugin_path / "manifest.json"
try:
with open(manifest_path, 'r', encoding='utf-8') as mf:
manifest = json.load(mf)
except (OSError, ValueError) as e:
self.logger.warning(
"Could not read %s after updating %s (%s); allowing the "
"update, as an unreadable manifest declares no floor",
manifest_path, plugin_id, e)
return True
from src.plugin_system import compatibility
core_version = compatibility.current_core_version()
compatible, reason = compatibility.check(manifest, core_version)
if compatible:
return True
self.logger.error("Refusing the update to %s: %s", plugin_id, reason)
if not previous_sha:
self.logger.error(
"Cannot roll %s back: the commit it was on before the pull is "
"unknown. It is now on a version this core cannot run — "
"reinstall it from the plugin store.", plugin_id)
return False
# Safe by construction: update_plugin returns before pulling unless the
# tree was clean or successfully stashed, so there are no uncommitted
# tracked edits for --hard to discard. The stash is not popped on the
# success path either, so the reset leaves the working tree exactly
# where a successful pull would have. Say "commit", not "changes".
reset = subprocess.run(
['git', '-C', str(plugin_path), 'reset', '--hard', previous_sha],
capture_output=True, text=True, timeout=60, check=False)
if reset.returncode != 0:
self.logger.error(
"CRITICAL: could not roll %s back to commit %s: %s. It is left "
"on a version this core cannot run; "
"`git -C %s reset --hard %s` restores it.",
plugin_id, previous_sha[:7],
(reset.stderr or reset.stdout or '').strip(),
plugin_path, previous_sha)
else:
self.logger.info(
"Rolled %s back to commit %s; it stays on the version it was "
"already running.", plugin_id, previous_sha[:7])
return False
def _reinstall_with_rollback(self, plugin_id: str, plugin_path: Path) -> bool:
"""Replace an installed plugin with a fresh install, atomically.
The old install is renamed aside (not deleted) until the new install
succeeds, then removed; on ANY install failure the old directory is
restored. Deleting first turns a failed download into a destroyed
plugin: during the monorepo migration a Pi with broken DNS lost every
old-remote plugin that way, with none able to be re-downloaded.
The aside name embeds BACKUP_MARKER ('.standalone-backup-') so every
plugin directory lookup (src/plugin_system/plugin_dirs.py) ignores it
even though it still contains a manifest.json.
Held for the whole operation under a per-plugin_id lock: two
overlapping requests for the same plugin (double-click, two
browser tabs — the web UI runs Flask with threaded=True) must not
interleave their renames, or the second could steal the first's
rollback safety net mid-install. Other plugin_ids are unaffected.
"""
with self._get_reinstall_lock(plugin_id):
backup_path = plugin_path.with_name(
f"{plugin_path.name}{BACKUP_MARKER}migrating")
problem = self._set_aside(plugin_path, backup_path)
if problem:
self.logger.error(
"Not updating %s: %s; the installed version is left in place",
plugin_id, problem)
return False
try:
installed = self.install_plugin(plugin_id)
except Exception as e:
self.logger.error(f"Reinstall of {plugin_id} raised: {e}")
installed = False
if installed:
self._discard_backup(plugin_id, backup_path, "update")
return True
# Bad network, registry error...: the user keeps a working plugin.
self._restore_backup(plugin_id, plugin_path, backup_path, "Reinstall")
return False
def update_plugin(self, plugin_id: str) -> bool:
"""
Update a plugin to the latest commit on its upstream branch.
"""
plugin_path = self._find_plugin_path(plugin_id)
if plugin_path is None or not plugin_path.exists():
self.logger.error(f"Plugin not installed: {plugin_id}")
return False
try:
self.logger.info(f"Checking for updates to plugin {plugin_id}")
# Check if this is a bundled/unmanaged plugin (no registry entry, no git remote)
# These are plugins shipped with LEDMatrix itself and updated via LEDMatrix updates.
metadata_path = plugin_path / ".plugin_metadata.json"
if metadata_path.exists():
try:
with open(metadata_path, 'r', encoding='utf-8') as f:
metadata = json.load(f)
if metadata.get('install_type') == 'bundled':
self.logger.info(f"Plugin {plugin_id} is a bundled plugin; updates are delivered via LEDMatrix itself")
return True
except (OSError, ValueError) as e:
self.logger.debug(f"[PluginStore] Could not read metadata for {plugin_id} at {metadata_path}: {e}")
# First check if it's a git repository - if so, we can update directly
git_info = self._get_local_git_info(plugin_path)
if git_info:
# Plugin is a git repository - try to update via git
local_branch = git_info.get('branch') or 'main'
local_sha = git_info.get('sha')
# Try to get remote info from registry (optional)
self.fetch_registry(force_refresh=True)
plugin_info_remote = self.get_plugin_info(plugin_id, fetch_latest_from_github=True, force_refresh=True)
# Try without 'ledmatrix-' prefix (monorepo migration)
resolved_id = plugin_id
if not plugin_info_remote and plugin_id.startswith('ledmatrix-'):
alt_id = plugin_id[len('ledmatrix-'):]
plugin_info_remote = self.get_plugin_info(alt_id, fetch_latest_from_github=True, force_refresh=True)
if plugin_info_remote:
resolved_id = alt_id
self.logger.info(f"Plugin {plugin_id} found in registry as {resolved_id}")
remote_branch = None
remote_sha = None
if plugin_info_remote:
remote_branch = plugin_info_remote.get('branch') or plugin_info_remote.get('default_branch')
remote_sha = plugin_info_remote.get('last_commit_sha')
# Check if the local git remote still matches the registry repo URL.
# After monorepo migration, old clones point to archived individual repos
# while the registry now points to the monorepo. Detect this and reinstall.
registry_repo = plugin_info_remote.get('repo', '')
local_remote = git_info.get('remote_url', '')
if local_remote and registry_repo and not same_repo(local_remote, registry_repo):
self.logger.info(
f"Plugin {resolved_id} git remote ({local_remote}) differs from registry ({registry_repo}). "
f"Reinstalling from registry to migrate to new source."
)
return self._reinstall_with_rollback(resolved_id, plugin_path)
# Check if already up to date
if remote_sha and local_sha and remote_sha.startswith(local_sha):
self.logger.info(f"Plugin {plugin_id} already matches remote commit {remote_sha[:7]}")
return True
# Update via git pull
self.logger.info(f"Updating {plugin_id} via git pull (local branch: {local_branch})...")
try:
# Fetch latest changes first to get all remote branch info
# If fetch fails, we'll still try to pull (might work with existing remote refs)
fetch_result = subprocess.run(
['git', '-C', str(plugin_path), 'fetch', 'origin'],
capture_output=True,
text=True,
timeout=60,
check=False
)
if fetch_result.returncode != 0:
self.logger.warning(f"Git fetch failed for {plugin_id}: {fetch_result.stderr or fetch_result.stdout}. Will still attempt pull.")
else:
self.logger.debug(f"Successfully fetched remote changes for {plugin_id}")
# Determine which remote branch to pull from
# Strategy: Use what the local branch is tracking, or find the best match
remote_pull_branch = None
# First, check what the local branch is tracking
tracking_result = subprocess.run(
['git', '-C', str(plugin_path), 'rev-parse', '--abbrev-ref', '--symbolic-full-name', f'{local_branch}@{{upstream}}'],
capture_output=True,
text=True,
timeout=10,
check=False
)
if tracking_result.returncode == 0 and tracking_result.stdout.strip():
# Local branch is tracking a remote branch
tracking_ref = tracking_result.stdout.strip()
# Extract branch name from refs/remotes/origin/branch-name or origin/branch-name
if tracking_ref.startswith('refs/remotes/origin/'):
remote_pull_branch = tracking_ref.replace('refs/remotes/origin/', '')
self.logger.info(f"Local branch {local_branch} is tracking origin/{remote_pull_branch}")
elif tracking_ref.startswith('origin/'):
remote_pull_branch = tracking_ref.replace('origin/', '')
self.logger.info(f"Local branch {local_branch} is tracking origin/{remote_pull_branch}")
# If not tracking anything, try to find the best remote branch match
if not remote_pull_branch:
# Check if remote branch from registry exists
if remote_branch:
remote_check = subprocess.run(
['git', '-C', str(plugin_path), 'ls-remote', '--heads', 'origin', remote_branch],
capture_output=True,
text=True,
timeout=10,
check=False
)
if remote_check.returncode == 0 and remote_check.stdout.strip():
remote_pull_branch = remote_branch
self.logger.info(f"Using remote branch {remote_branch} from registry")
# If registry branch doesn't exist, check if local branch name exists on remote
if not remote_pull_branch:
local_as_remote_check = subprocess.run(
['git', '-C', str(plugin_path), 'ls-remote', '--heads', 'origin', local_branch],
capture_output=True,
text=True,
timeout=10,
check=False
)
if local_as_remote_check.returncode == 0 and local_as_remote_check.stdout.strip():
remote_pull_branch = local_branch
self.logger.info(f"Using local branch name {local_branch} as remote branch")
# Last resort: try to get remote's default branch
if not remote_pull_branch:
default_branch_result = subprocess.run(
['git', '-C', str(plugin_path), 'symbolic-ref', 'refs/remotes/origin/HEAD'],
capture_output=True,
text=True,
timeout=10,
check=False
)
if default_branch_result.returncode == 0:
default_ref = default_branch_result.stdout.strip()
if default_ref.startswith('refs/remotes/origin/'):
remote_pull_branch = default_ref.replace('refs/remotes/origin/', '')
self.logger.info(f"Using remote default branch {remote_pull_branch}")
# If we still don't have a remote branch, use local branch name (git will handle it)
if not remote_pull_branch:
remote_pull_branch = local_branch
self.logger.info(f"Falling back to local branch name {local_branch} for pull")
# Ensure we're on the local branch
checkout_result = subprocess.run(
['git', '-C', str(plugin_path), 'checkout', local_branch],
capture_output=True,
text=True,
timeout=30,
check=False
)
if checkout_result.returncode != 0:
self.logger.warning(f"Git checkout to {local_branch} failed for {plugin_id}: {checkout_result.stderr or checkout_result.stdout}. Will still attempt pull.")
# Check for local changes and untracked files that might conflict
# First, check for untracked files that would be overwritten
try:
# Check for untracked files
untracked_result = subprocess.run(
['git', '-C', str(plugin_path), 'status', '--porcelain', '--untracked-files=all'],
capture_output=True,
text=True,
timeout=30,
check=False
)
untracked_files = []
if untracked_result.returncode == 0:
for line in untracked_result.stdout.strip().split('\n'):
if line.startswith('??'):
# Untracked file
file_path = line[3:].strip()
untracked_files.append(file_path)
# Check for tracked file changes
status_result = subprocess.run(
['git', '-C', str(plugin_path), 'status', '--porcelain', '--untracked-files=no'],
capture_output=True,
text=True,
timeout=30,
check=False
)
has_changes = bool(status_result.stdout.strip())
# If there are untracked files, stash them
if untracked_files:
self.logger.info(f"Found {len(untracked_files)} untracked files in {plugin_id}, will stash them")
has_changes = True
except subprocess.TimeoutExpired:
# If status check times out, assume there might be changes and proceed
self.logger.warning(f"Git status check timed out for {plugin_id}, proceeding with update")
has_changes = True
stash_info = ""
# Whether the pull can be undone without destroying work.
tree_is_recoverable = not has_changes
if has_changes:
self.logger.info(f"Stashing local changes in {plugin_id} before update")
try:
# Use -u to include untracked files in stash
stash_result = subprocess.run(
['git', '-C', str(plugin_path), 'stash', 'push', '-u', '-m', f'LEDMatrix auto-stash before update {plugin_id}'],
capture_output=True,
text=True,
timeout=30,
check=False
)
if stash_result.returncode == 0:
stash_info = " (local changes were stashed)"
tree_is_recoverable = True
self.logger.info(f"Stashed local changes (including untracked files) for {plugin_id}")
else:
self.logger.warning(f"Failed to stash local changes for {plugin_id}: {stash_result.stderr}")
except subprocess.TimeoutExpired:
self.logger.warning(f"Stash operation timed out for {plugin_id}, proceeding with pull")
# Do not pull what cannot be un-pulled.
#
# The compatibility gate below can refuse the commit this
# pull brings down, and its only way back is `git reset
# --hard`, which discards uncommitted tracked edits. Those
# edits are exactly what the stash above exists to protect,
# so a stash that failed or timed out leaves the rollback
# unable to run without destroying them.
#
# A pull does not necessarily refuse on a dirty tree -- git
# merges happily as long as the incoming commit touches
# different files -- so without this the update would
# succeed, the gate would refuse, and the reset would take
# the user's work with it. Refusing here costs an update in
# a case that already went wrong; the alternative costs
# data.
if not tree_is_recoverable:
self.logger.error(
"Refusing to update %s: it has local changes that could "
"not be stashed, and an incompatible update could then "
"only be rolled back by discarding them. Commit or stash "
"them by hand, then update.", plugin_id)
return False
# Pull from the determined remote branch
self.logger.info(f"Pulling from origin/{remote_pull_branch} for {plugin_id}...")
pull_result = subprocess.run(
['git', '-C', str(plugin_path), 'pull', 'origin', remote_pull_branch],
capture_output=True,
text=True,
timeout=120,
check=True
)
pull_message = pull_result.stdout.strip() or f"Pulled latest changes for {plugin_id}"
if stash_info:
pull_message += stash_info
self.logger.info(pull_message)
updated_git_info = self._get_local_git_info(plugin_path) or {}
updated_sha = updated_git_info.get('sha', '')
if remote_sha and updated_sha and remote_sha.startswith(updated_sha):
self.logger.info(f"Plugin {plugin_id} now at remote commit {remote_sha[:7]}{stash_info}")
elif updated_sha:
self.logger.info(f"Plugin {plugin_id} updated to commit {updated_sha[:7]}{stash_info}")
# The install gate, at the only point on this path where
# it can be answered. Every other route in goes through
# install_plugin, which gates in _install_plugin_impl; this
# one did not, so a pull could deliver a manifest flooring
# above this core and nothing would notice.
if not self._gate_pulled_commit(plugin_id, plugin_path, local_sha):
return False
self._install_dependencies(plugin_path)
return True
except subprocess.CalledProcessError as git_error:
error_output = git_error.stderr or git_error.stdout or "Unknown error"
cmd_str = ' '.join(git_error.cmd)
self.logger.error(f"Git update failed for {plugin_id}")
self.logger.error(f"Command: {cmd_str}")
self.logger.error(f"Return code: {git_error.returncode}")
self.logger.error(f"Error output: {error_output}")
# Check for specific error conditions
error_lower = error_output.lower()
if "would be overwritten" in error_output or "local changes" in error_lower:
self.logger.warning(f"Plugin {plugin_id} has local changes that prevent update. Consider committing or stashing changes manually.")
elif "refusing to merge unrelated histories" in error_lower:
self.logger.error(f"Plugin {plugin_id} has unrelated git histories. Plugin may need to be reinstalled.")
elif "authentication" in error_lower or "permission denied" in error_lower:
self.logger.error(f"Authentication failed for {plugin_id}. Check git credentials or repository permissions.")
elif "not found" in error_lower or "does not exist" in error_lower:
self.logger.error(f"Remote branch or repository not found for {plugin_id}. Check repository URL and branch name.")
elif "conflict" in error_lower:
self.logger.error(f"Merge conflict detected for {plugin_id}. Resolve conflicts manually or reinstall plugin.")
return False
except subprocess.TimeoutExpired:
self.logger.warning(f"Git update timed out for {plugin_id}")
return False
# A plugin with its own .git that _get_local_git_info could not
# read (e.g. no commits yet) may still name a remote to reinstall
# from. Without its own .git, `git -C <plugin>` walks up and finds
# the enclosing LEDMatrix checkout when plugins live in
# plugin-repos/ -- `--local` does not prevent that -- and the
# "plugin's" remote would be LEDMatrix itself.
repo_url = None
if (plugin_path / '.git').exists():
try:
remote_url_result = subprocess.run(
['git', '-C', str(plugin_path), 'config', '--local', '--get', 'remote.origin.url'],
capture_output=True,
text=True,
timeout=10,
check=False
)
if remote_url_result.returncode == 0:
repo_url = remote_url_result.stdout.strip() or None
if repo_url:
self.logger.info(f"Found git remote URL for {plugin_id}: {repo_url}")
except (OSError, subprocess.SubprocessError) as e:
self.logger.debug(f"Could not get git remote URL: {e}")
# Try registry-based update
self.logger.info(f"Plugin {plugin_id} is not a git repository, checking registry...")
self.fetch_registry(force_refresh=True)
plugin_info_remote = self.get_plugin_info(plugin_id, fetch_latest_from_github=True, force_refresh=True)
# If not found, try without 'ledmatrix-' prefix (monorepo migration)
registry_id = plugin_id
if not plugin_info_remote and plugin_id.startswith('ledmatrix-'):
alt_id = plugin_id[len('ledmatrix-'):]
plugin_info_remote = self.get_plugin_info(alt_id, fetch_latest_from_github=True, force_refresh=True)
if plugin_info_remote:
registry_id = alt_id
self.logger.info(f"Plugin {plugin_id} found in registry as {alt_id}")
# If not in registry but we have a repo URL, try reinstalling from that URL
if not plugin_info_remote and repo_url:
self.logger.info(f"Plugin {plugin_id} not in registry but has git remote URL. Reinstalling from {repo_url} to enable updates...")
try:
# Get current branch if possible
branch_result = subprocess.run(
['git', '-C', str(plugin_path), 'rev-parse', '--abbrev-ref', 'HEAD'],
capture_output=True,
text=True,
timeout=10,
check=False
)
branch = branch_result.stdout.strip() if branch_result.returncode == 0 else None
if branch == 'HEAD' or not branch:
branch = 'main'
# Reinstall from URL
result = self.install_from_url(repo_url, plugin_id=plugin_id, branch=branch)
if result.get('success'):
self.logger.info(f"Successfully reinstalled {plugin_id} from {repo_url} as git repository")
return True
else:
self.logger.warning(f"Failed to reinstall {plugin_id} from {repo_url}: {result.get('error')}")
except Exception as e:
self.logger.error(f"Error reinstalling {plugin_id} from URL: {e}")
if not plugin_info_remote:
self.logger.warning(f"Plugin {plugin_id} not found in registry and not a git repository; cannot update automatically")
if not repo_url:
self.logger.warning("Plugin may have been installed via ZIP download. Try reinstalling from GitHub URL to enable updates.")
return False
repo_url = plugin_info_remote.get('repo')
remote_sha = plugin_info_remote.get('last_commit_sha')
remote_branch = plugin_info_remote.get('branch') or plugin_info_remote.get('default_branch')
# Compare local manifest version against registry latest_version
# to avoid unnecessary reinstalls for monorepo plugins. Uses the
# same semantic comparator as the web UI's update badge, so
# equivalent spellings ("v1.2.0" vs "1.2.0") never trigger a
# reinstall and a locally-ahead version is never downgraded.
try:
local_manifest_path = plugin_path / "manifest.json"
if local_manifest_path.exists():
with open(local_manifest_path, 'r', encoding='utf-8') as f:
local_manifest = json.load(f)
local_version = local_manifest.get('version', '')
remote_version = plugin_info_remote.get('latest_version', '')
from src.plugin_system.compatibility import is_update_available
# No truthiness gate: the shared comparator already treats
# a missing version on either side as "no update", and the
# store must agree with the UI badge in that case too. A
# missing manifest (not just a missing version field)
# still falls through to the reinstall recovery path.
if not is_update_available(local_version, remote_version):
self.logger.info(
f"Plugin {plugin_id} already at latest version "
f"(installed {local_version}, registry {remote_version})")
return True
except Exception as e:
self.logger.debug(f"Could not compare versions for {plugin_id}: {e}")
# Plugin is not a git repo but is in registry and has a newer version - reinstall
self.logger.info(f"Plugin {plugin_id} not installed via git; re-installing latest archive (registry id: {registry_id})")
# Reinstall with the old version kept aside until the new
# download succeeds — this is the path every routine store
# update takes, and a mid-update network failure must not
# destroy the user's plugin.
return self._reinstall_with_rollback(registry_id, plugin_path)
except Exception as e:
self.logger.error(f"Error updating plugin {plugin_id}: {e}", exc_info=True)
return False
def list_installed_plugins(self) -> List[str]:
"""
Get list of installed plugin IDs.
One entry per plugin directory in the configured directory that has a
manifest.json, named by the manifest's id (the directory name when
the manifest carries none, e.g. because it does not parse). Backup
and hidden directories are not plugins.
Returns:
List of plugin IDs, sorted
"""
if not self.plugins_dir.exists():
return []
index = PluginDirectoryIndex.scan(self.plugins_dir)
return sorted(index.installed_ids(require_parseable_manifest=False))
Reference in New Issue View Git Blame Copy Permalink