fix(plugins): prevent root-owned files from blocking plugin updates (#242)

* fix(web): unify operation history tracking for monorepo plugin operations

The operation history UI was reading from the wrong data source
(operation_queue instead of operation_history), install/update records
lacked version details, toggle operations used a type name that didn't
match UI filters, and the Clear History button was non-functional.

- Switch GET /plugins/operation/history to read from OperationHistory
  audit log with return type hint and targeted exception handling
- Add DELETE /plugins/operation/history endpoint; wire up Clear button
- Add _get_plugin_version helper with specific exception handling
  (FileNotFoundError, PermissionError, json.JSONDecodeError) and
  structured logging with plugin_id/path context
- Record plugin version, branch, and commit details on install/update
- Record install failures in the direct (non-queue) code path
- Replace "toggle" operation type with "enable"/"disable"
- Add normalizeStatus() in JS to map completed→success, error→failed
  so status filter works regardless of server-side convention
- Truncate commit SHAs to 7 chars in details display
- Fix HTML filter options, operation type colors, duplicate JS init

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

* fix(plugins): prevent root-owned files from blocking plugin updates

The root ledmatrix service creates __pycache__ and data cache files
owned by root inside plugin directories. The web service (non-root)
cannot delete these when updating or uninstalling plugins, causing
operations to fail with "Permission denied".

Defense in depth with three layers:
- Prevent: PYTHONDONTWRITEBYTECODE=1 in systemd service + run.py
- Fallback: sudoers rules for rm on plugin directories
- Code: _safe_remove_directory() now uses sudo as last resort,
  and all bare shutil.rmtree() calls routed through it

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

* fix(security): harden sudo removal with path-validated helper script

Address code review findings:

- Replace raw rm/find sudoers wildcards with a vetted helper script
  (safe_plugin_rm.sh) that resolves symlinks and validates the target
  is a strict child of plugin-repos/ or plugins/ before deletion
- Add allow-list validation in sudo_remove_directory() that checks
  resolved paths against allowed bases before invoking sudo
- Check _safe_remove_directory() return value before shutil.move()
  in the manifest ID rename path
- Move stat import to module level in store_manager.py
- Use stat.S_IRWXU instead of 0o777 in chmod fallback stage
- Add ignore_errors=True to temp dir cleanup in finally block
- Use command -v instead of which in configure_web_sudo.sh

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

* fix(security): address code review round 2 — harden paths and error handling

- safe_plugin_rm.sh: use realpath --canonicalize-missing for ALLOWED_BASES
  so the script doesn't fail under set -e when dirs don't exist yet
- safe_plugin_rm.sh: add -- before path in rm -rf to prevent flag injection
- permission_utils.py: use shutil.which('bash') instead of hardcoded /bin/bash
  to match whatever path the sudoers BASH_PATH resolves to
- store_manager.py: check _safe_remove_directory() return before shutil.move()
  in _install_from_monorepo_zip to prevent moving into a non-removed target
- store_manager.py: catch OSError instead of PermissionError in Stage 1 removal
  to handle both EACCES and EPERM error codes
- store_manager.py: hoist sudo_remove_directory import to module level
- configure_web_sudo.sh: harden safe_plugin_rm.sh to root-owned 755 so
  the web user cannot modify the vetted helper script

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

* fix(security): validate command paths in sudoers config and use resolved paths

- configure_web_sudo.sh: validate that required commands (systemctl, bash,
  python3) resolve to non-empty paths before generating sudoers entries;
  abort with clear error if any are missing; skip optional commands
  (reboot, poweroff, journalctl) with a warning instead of emitting
  malformed NOPASSWD lines; validate helper script exists on disk
- permission_utils.py: pass the already-resolved path to the subprocess
  call and use it for the post-removal exists() check, eliminating a
  TOCTOU window between Python-side validation and shell-side execution

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

---------

Co-authored-by: Chuck <chuck@example.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-02-12 19:28:05 -05:00
committed by GitHub
parent 9a72adbde1
commit 158e07c82b
6 changed files with 306 additions and 101 deletions

View File

@@ -7,6 +7,7 @@ from both the official registry and custom GitHub repositories.
import os
import json
import stat
import subprocess
import shutil
import zipfile
@@ -18,6 +19,8 @@ from pathlib import Path
from typing import List, Dict, Optional, Any
import logging
from src.common.permission_utils import sudo_remove_directory
try:
import jsonschema
from jsonschema import Draft7Validator, ValidationError
@@ -814,7 +817,7 @@ class PluginStoreManager:
manifest_path = plugin_path / "manifest.json"
if not manifest_path.exists():
self.logger.error(f"No manifest.json found in plugin: {plugin_id}")
shutil.rmtree(plugin_path, ignore_errors=True)
self._safe_remove_directory(plugin_path)
return False
try:
@@ -825,7 +828,7 @@ class PluginStoreManager:
manifest_plugin_id = manifest.get('id')
if not manifest_plugin_id:
self.logger.error(f"Plugin manifest missing 'id' field")
shutil.rmtree(plugin_path, ignore_errors=True)
self._safe_remove_directory(plugin_path)
return False
# If manifest ID doesn't match directory name, rename directory to match manifest
@@ -837,7 +840,9 @@ class PluginStoreManager:
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")
shutil.rmtree(correct_path)
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"
@@ -865,7 +870,7 @@ class PluginStoreManager:
if missing:
self.logger.error(f"Plugin manifest missing required fields for {plugin_id}: {', '.join(missing)}")
shutil.rmtree(plugin_path, ignore_errors=True)
self._safe_remove_directory(plugin_path)
return False
if 'entry_point' not in manifest:
@@ -879,7 +884,7 @@ class PluginStoreManager:
except Exception as manifest_error:
self.logger.error(f"Failed to read/validate manifest for {plugin_id}: {manifest_error}")
shutil.rmtree(plugin_path, ignore_errors=True)
self._safe_remove_directory(plugin_path)
return False
if not self._install_dependencies(plugin_path):
@@ -892,7 +897,7 @@ class PluginStoreManager:
except Exception as e:
self.logger.error(f"Error installing plugin {plugin_id}: {e}", exc_info=True)
if plugin_path.exists():
shutil.rmtree(plugin_path, ignore_errors=True)
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]:
@@ -1053,8 +1058,8 @@ class PluginStoreManager:
finally:
# Cleanup temp directory if it still exists
if temp_dir and temp_dir.exists():
shutil.rmtree(temp_dir)
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.
@@ -1110,7 +1115,7 @@ class PluginStoreManager:
last_error = e
self.logger.debug(f"Git clone failed for branch {try_branch}: {e}")
if target_path.exists():
shutil.rmtree(target_path)
self._safe_remove_directory(target_path)
# Try default branch (Git's configured default) as last resort
try:
@@ -1127,7 +1132,7 @@ class PluginStoreManager:
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError) as e:
last_error = e
if target_path.exists():
shutil.rmtree(target_path)
self._safe_remove_directory(target_path)
self.logger.error(f"Git clone failed for all attempted branches: {last_error}")
return None
@@ -1157,7 +1162,7 @@ class PluginStoreManager:
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():
shutil.rmtree(target_path, ignore_errors=True)
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)
@@ -1277,7 +1282,7 @@ class PluginStoreManager:
f"Path traversal detected: {entry['path']!r} resolves outside target directory"
)
if target_path.exists():
shutil.rmtree(target_path, ignore_errors=True)
self._safe_remove_directory(target_path)
return False
# Create parent directories
@@ -1290,7 +1295,7 @@ class PluginStoreManager:
self.logger.error(f"Failed to download {entry['path']}: HTTP {file_response.status_code}")
# Clean up partial download
if target_path.exists():
shutil.rmtree(target_path, ignore_errors=True)
self._safe_remove_directory(target_path)
return False
dest_file.write_bytes(file_response.content)
@@ -1302,7 +1307,7 @@ class PluginStoreManager:
self.logger.debug(f"API-based monorepo install failed: {e}")
# Clean up partial download
if target_path.exists():
shutil.rmtree(target_path, ignore_errors=True)
self._safe_remove_directory(target_path)
return False
def _install_from_monorepo_zip(self, download_url: str, plugin_subpath: str, target_path: Path) -> bool:
@@ -1363,7 +1368,9 @@ class PluginStoreManager:
ensure_directory_permissions(target_path.parent, get_plugin_dir_mode())
# Ensure target doesn't exist to prevent shutil.move nesting
if target_path.exists():
shutil.rmtree(target_path, ignore_errors=True)
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
@@ -1577,70 +1584,58 @@ class PluginStoreManager:
def _safe_remove_directory(self, path: Path) -> bool:
"""
Safely remove a directory, handling permission errors for __pycache__ directories.
This function attempts to remove a directory and handles permission errors
gracefully, especially for __pycache__ directories that may have been created
by Python with different permissions.
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:
# First, try normal removal
shutil.rmtree(path)
return True
except PermissionError as e:
# Handle permission errors, especially for __pycache__ directories
self.logger.warning(f"Permission error removing {path}: {e}. Attempting to fix permissions...")
try:
# Try to fix permissions on __pycache__ directories recursively
import stat
for root, dirs, files in os.walk(path):
root_path = Path(root)
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:
# Make directory writable
os.chmod(root_path, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
os.chmod(root_path / file, stat.S_IRWXU)
except (OSError, PermissionError):
pass
# Fix file permissions
for file in files:
file_path = root_path / file
try:
os.chmod(file_path, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
except (OSError, PermissionError):
pass
# Try removal again after fixing permissions
shutil.rmtree(path)
self.logger.info(f"Successfully removed {path} after fixing permissions")
return True
except Exception as e2:
self.logger.error(f"Failed to remove {path} even after fixing permissions: {e2}")
# Last resort: try with ignore_errors
try:
shutil.rmtree(path, ignore_errors=True)
# Check if it actually got removed
if not path.exists():
self.logger.warning(f"Removed {path} with ignore_errors=True (some files may remain)")
return True
else:
self.logger.error(f"Could not remove {path} even with ignore_errors")
return False
except Exception as e3:
self.logger.error(f"Final removal attempt failed for {path}: {e3}")
return False
except Exception as e:
self.logger.error(f"Unexpected error removing {path}: {e}")
return False
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]:
"""