"""Plugin store: installing plugins (registry, URL, git, monorepo ZIP or Trees API) and their Python dependencies. Part of PluginStoreManager (store_manager.py), which mixes this class in; methods reach shared state and helpers through ``self``. """ import errno import os import re import json import subprocess # nosec B404 - list-form argv only, no shell # nosemgrep import shutil import zipfile import tempfile from pathlib import Path from typing import List, Dict, Optional, Any from src.common.permission_utils import ( ensure_directory_permissions, get_plugin_dir_mode, install_requirements_file, ) from src.plugin_system.plugin_loader import ( contained_plugin_dir, requirements_to_install, ) from src.plugin_system.plugin_dirs import BACKUP_MARKER from src.plugin_system.repo_urls import ( USER_AGENT, github_api_headers, github_owner_repo, normalize_repo_url, ) class _InstallMixin: """PluginStoreManager methods: see the module docstring.""" 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( # nosec B603 - list-form argv, no shell # nosemgrep 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( # nosec B603 - list-form argv, no shell # nosemgrep 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