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
224847cebc23a0ac4e8eb18829c84d79cf59caaf
LEDMatrix/web_interface/blueprints/api_v3/plugins.py
T
ChuckandClaude Opus 5.5 7eb7a58d0c fix: web UI and src.common bugs (wifi wrong-password, plugin icon, starlark toggle, API caching, scroll/logo/font helpers) (#646) 
- wifi: keep the "wrong_password:" prefix through the restore/AP fallback so
  the UI's incorrect-password prompt fires again.
- /plugins/installed returns the manifest's icon (string only).
- /starlark/apps/<id>/toggle coerces `enabled` and delegates to
  _toggle_starlark_app (disk before memory, no KeyError, "false" is false).
- /api/v3/ JSON GETs are sent Cache-Control: no-store; non-JSON keeps 5s.
- ScrollHelper.set_scrolling_image converts non-RGB input (alpha onto black);
  create/set_scrolling_image reset last_update_time like reset_scroll.
- LogoHelper backs off a failed download per path for
  MISSING_LOGO_RECHECK_SECONDS; cleared on invalidate/clear_cache.
- refresh_placeholder_timestamp saves atomically.
- FontManager.clear_cache / _clear_plugin_font_cache bump cache_generation.
- Odds manager: per-game logs to DEBUG; JSON decode error caught before
  RequestException (same cooldown).
- element_style mangled continuations; startup validator skips null plugin
  blocks and reuses the controller's discovery.
- src/common/README lists frame_timing, json_body, render_gate.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-28 08:26:05 -04:00

3222 lines
143 KiB
Python
Raw Blame History

"""Plugin install, update, enable/disable, config and store routes.
Routes decorate the shared `api_v3` Blueprint from the package `__init__`,
so their endpoint names are unchanged by living here.
"""
from web_interface.blueprints.api_v3 import (
ErrorCode, OperationType, PROJECT_ROOT, Path, Response,
_CALENDAR_LIST_MAX_PAGES, _RENDERED_SECTION_FIELD, _SKIP_FIELD, _coerce_to_bool,
_do_transactional_uninstall, _enhance_schema_with_core_properties, _non_plugin_id_error,
_filter_config_by_schema, _get_plugin_version, _get_schema_property,
_hidden_array_item_property, _installed_plugin_ids, _is_plugin_update_available,
_plugin_directory,
_parse_form_value_with_schema, _prune_credential_backups,
_run_calendar_registration, _schema_allows_null, _schema_type_is,
_set_missing_booleans_to_false,
_set_nested_value, _starlark_virtual_plugins, _toggle_starlark_app,
api_v3, datetime, deep_merge, describe_exception, error_response,
exception_error_response,
find_secret_fields, hashlib, json, jsonify, logger,
merge_secrets, os, redact_text, remove_empty_secrets, request,
separate_secrets, shutil, stat, subprocess, success_response,
tempfile, uuid, validate_request_json,
)
from src.common.path_safety import (
resolve_under, safe_path_component, safe_relative_parts,
)
from src.web_interface.config_arrays import coerce_array_shapes
from src.web_interface.validators import dedup_unique_arrays
import web_interface.blueprints.api_v3 as _pkg
# Read through the module rather than bound by value: tests patch these
# as module attributes, and a value binding would not see the patch.
# Several are also called from helpers that live in __init__, so the
# package is the only patch point that covers every caller.
@api_v3.route('/plugins/installed', methods=['GET'])
def get_installed_plugins():
"""Get installed plugins"""
if not api_v3.plugin_manager or not api_v3.plugin_store_manager:
return jsonify({'status': 'error', 'message': 'Plugin managers not initialized'}), 500
# Re-discover plugins to ensure we have the latest list
# This handles cases where plugins are added/removed after app startup
api_v3.plugin_manager.discover_plugins()
# Get all installed plugin info from the plugin manager
all_plugin_info = api_v3.plugin_manager.get_all_plugin_info()
# Load config once before the loop (not per-plugin)
full_config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
def _build_plugin_entry(plugin_info):
plugin_id = plugin_info.get('id')
try:
return _build_plugin_entry_inner(plugin_info, plugin_id)
except Exception:
logger.exception("Error building plugin entry for %s — skipping", plugin_id)
return None
def _build_plugin_entry_inner(plugin_info, plugin_id):
# Capture runtime state (state machine + error context) before the
# manifest merge below can shadow the 'state' key. get_all_plugin_info
# attaches this via PluginStateManager.get_state_info(); surfacing it
# lets the UI show *why* a plugin isn't running instead of just
# 'loaded: false'.
state_info = plugin_info.get('state')
plugin_state = None
plugin_error_info = None
if isinstance(state_info, dict):
plugin_state = state_info.get('state')
plugin_error_info = state_info.get('error_info')
# Re-read manifest from disk to ensure we have the latest metadata
manifest_path = Path(api_v3.plugin_manager.plugins_dir) / plugin_id / "manifest.json"
if manifest_path.exists():
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
fresh_manifest = json.load(f)
if isinstance(fresh_manifest, dict):
plugin_info.update(fresh_manifest)
else:
logger.debug("Manifest for %s is not a dict (%s) — skipping merge",
plugin_id, type(fresh_manifest).__name__)
except (FileNotFoundError, PermissionError, json.JSONDecodeError) as e:
logger.debug("Could not read fresh manifest for %s: %s", plugin_id, e)
# Enabled status: config is source of truth, fall back to instance
enabled = None
plugin_config = full_config.get(plugin_id, {})
if 'enabled' in plugin_config:
enabled = bool(plugin_config['enabled'])
# Single get_plugin() call shared for both enabled fallback and Vegas mode
plugin_instance = api_v3.plugin_manager.get_plugin(plugin_id)
if enabled is None:
enabled = plugin_instance.enabled if plugin_instance else True
# Verified + latest published version from registry (no network call)
store_info = api_v3.plugin_store_manager.get_registry_info(plugin_id)
verified = store_info.get('verified', False) if store_info else False
latest_version = store_info.get('latest_version', '') if store_info else ''
installed_version = plugin_info.get('version', '')
update_available = _is_plugin_update_available(installed_version, latest_version)
# Local git info (single subprocess on cache miss, zero on hit)
plugin_path = Path(api_v3.plugin_manager.plugins_dir) / plugin_id
local_git_info = api_v3.plugin_store_manager._get_local_git_info(plugin_path) if plugin_path.exists() else None
if local_git_info:
sha = local_git_info.get('sha', '')
last_commit = local_git_info.get('short_sha') or (sha[:7] if sha else None)
branch = local_git_info.get('branch')
last_updated = local_git_info.get('date_iso') or local_git_info.get('date')
else:
last_updated = plugin_info.get('last_updated')
last_commit = plugin_info.get('last_commit') or plugin_info.get('last_commit_sha')
branch = plugin_info.get('branch')
if store_info:
last_updated = last_updated or store_info.get('last_updated') or store_info.get('last_updated_iso')
last_commit = last_commit or store_info.get('last_commit') or store_info.get('last_commit_sha')
branch = branch or store_info.get('branch') or store_info.get('default_branch')
last_commit_message = plugin_info.get('last_commit_message')
if store_info and not last_commit_message:
last_commit_message = store_info.get('last_commit_message')
# Vegas mode from instance, overridden by explicit config value
vegas_mode = None
vegas_content_type = None
if plugin_instance:
try:
if hasattr(plugin_instance, 'get_vegas_display_mode'):
mode = plugin_instance.get_vegas_display_mode()
vegas_mode = mode.value if hasattr(mode, 'value') else str(mode)
except (AttributeError, TypeError, ValueError) as e:
logger.debug("[%s] Failed to get vegas_display_mode: %s", plugin_id, e)
try:
if hasattr(plugin_instance, 'get_vegas_content_type'):
vegas_content_type = plugin_instance.get_vegas_content_type()
except (AttributeError, TypeError, ValueError) as e:
logger.debug("[%s] Failed to get vegas_content_type: %s", plugin_id, e)
if 'vegas_mode' in plugin_config:
vegas_mode = plugin_config['vegas_mode']
return {
'id': plugin_id,
'name': plugin_info.get('name', plugin_id),
'version': plugin_info.get('version', ''),
'latest_version': latest_version,
'update_available': update_available,
'author': plugin_info.get('author', 'Unknown'),
'category': plugin_info.get('category', 'General'),
'description': plugin_info.get('description', 'No description available'),
'tags': plugin_info.get('tags', []),
# The tab nav uses this as the <i> element's Font Awesome class
# (app-shell.js / app-early.js); only a string can be one.
'icon': plugin_info.get('icon') if isinstance(plugin_info.get('icon'), str) else None,
'enabled': enabled,
'verified': verified,
'loaded': plugin_info.get('loaded', False),
'state': plugin_state,
'error_info': plugin_error_info,
'last_updated': last_updated,
'last_commit': last_commit,
'last_commit_message': last_commit_message,
'branch': branch,
'web_ui_actions': plugin_info.get('web_ui_actions', []),
'vegas_mode': vegas_mode,
'vegas_content_type': vegas_content_type,
}
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=8) as executor:
results = list(executor.map(_build_plugin_entry, all_plugin_info))
plugins = [r for r in results if r is not None]
plugins.extend(_starlark_virtual_plugins())
return jsonify({'status': 'success', 'data': {'plugins': plugins}})
@api_v3.route('/plugins/health', methods=['GET'])
def get_plugin_health():
"""Get health metrics for all plugins"""
if not api_v3.plugin_manager:
return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
if not api_v3.plugin_manager.health_tracker:
return jsonify({
'status': 'success',
'data': {},
'message': 'Health tracking not available'
})
tracker = api_v3.plugin_manager.health_tracker
# Build per-plugin summaries by ID so persisted (cross-process) health
# is included, then fold in any in-memory-only entries.
health_summaries = {}
for pid in _installed_plugin_ids():
try:
# force_reload: this process only reads; bypass the in-memory
# snapshot so each poll reflects the display service's latest
# persisted state.
health_summaries[pid] = tracker.get_health_summary(pid, force_reload=True)
except Exception:
logger.debug('Could not read health summary for %s', pid, exc_info=True)
try:
for pid, summary in tracker.get_all_health_summaries().items():
health_summaries.setdefault(pid, summary)
except Exception:
logger.debug('get_all_health_summaries failed', exc_info=True)
return jsonify({
'status': 'success',
'data': health_summaries
})
@api_v3.route('/plugins/health/<plugin_id>', methods=['GET'])
def get_plugin_health_single(plugin_id):
"""Get health metrics for a specific plugin"""
if not api_v3.plugin_manager:
return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
if not api_v3.plugin_manager.health_tracker:
return jsonify({
'status': 'error',
'message': 'Health tracking not available'
}), 503
# force_reload for the same reason as the list route above.
health_summary = api_v3.plugin_manager.health_tracker.get_health_summary(
plugin_id, force_reload=True)
return jsonify({
'status': 'success',
'data': health_summary
})
@api_v3.route('/plugins/health/<plugin_id>/reset', methods=['POST'])
def reset_plugin_health(plugin_id):
"""Reset health state for a plugin (manual recovery)"""
if not api_v3.plugin_manager:
return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
if not api_v3.plugin_manager.health_tracker:
return jsonify({
'status': 'error',
'message': 'Health tracking not available'
}), 503
# Reset health state
api_v3.plugin_manager.health_tracker.reset_health(plugin_id)
return jsonify({
'status': 'success',
'message': f'Health state reset for plugin {plugin_id}'
})
@api_v3.route('/plugins/metrics', methods=['GET'])
def get_plugin_metrics():
"""Get resource metrics for all plugins"""
if not api_v3.plugin_manager:
return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
if not api_v3.plugin_manager.resource_monitor:
return jsonify({
'status': 'success',
'data': {},
'message': 'Resource monitoring not available'
})
monitor = api_v3.plugin_manager.resource_monitor
# Build per-plugin summaries by ID so persisted (cross-process) metrics
# are included, then fold in any in-memory-only entries.
metrics_summaries = {}
for pid in _installed_plugin_ids():
try:
# force_reload: read-only path — bypass the in-memory snapshot so
# each poll reflects the display service's latest persisted metrics.
metrics_summaries[pid] = monitor.get_metrics_summary(pid, force_reload=True)
except Exception:
logger.debug('Could not read metrics summary for %s', pid, exc_info=True)
try:
for pid, summary in monitor.get_all_metrics_summaries().items():
metrics_summaries.setdefault(pid, summary)
except Exception:
logger.debug('get_all_metrics_summaries failed', exc_info=True)
return jsonify({
'status': 'success',
'data': metrics_summaries
})
@api_v3.route('/plugins/metrics/<plugin_id>', methods=['GET'])
def get_plugin_metrics_single(plugin_id):
"""Get resource metrics for a specific plugin"""
if not api_v3.plugin_manager:
return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
if not api_v3.plugin_manager.resource_monitor:
return jsonify({
'status': 'error',
'message': 'Resource monitoring not available'
}), 503
# force_reload for the same reason as the list route above.
metrics_summary = api_v3.plugin_manager.resource_monitor.get_metrics_summary(
plugin_id, force_reload=True)
return jsonify({
'status': 'success',
'data': metrics_summary
})
@api_v3.route('/plugins/metrics/<plugin_id>/reset', methods=['POST'])
def reset_plugin_metrics(plugin_id):
"""Reset metrics for a plugin"""
if not api_v3.plugin_manager:
return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
if not api_v3.plugin_manager.resource_monitor:
return jsonify({
'status': 'error',
'message': 'Resource monitoring not available'
}), 503
# Reset metrics
api_v3.plugin_manager.resource_monitor.reset_metrics(plugin_id)
return jsonify({
'status': 'success',
'message': f'Metrics reset for plugin {plugin_id}'
})
@api_v3.route('/plugins/limits/<plugin_id>', methods=['GET', 'POST'])
def manage_plugin_limits(plugin_id):
"""Get or set resource limits for a plugin"""
if not api_v3.plugin_manager:
return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
if not api_v3.plugin_manager.resource_monitor:
return jsonify({
'status': 'error',
'message': 'Resource monitoring not available'
}), 503
if request.method == 'GET':
# Get limits
limits = api_v3.plugin_manager.resource_monitor.get_limits(plugin_id)
if limits:
return jsonify({
'status': 'success',
'data': {
'max_memory_mb': limits.max_memory_mb,
'max_cpu_percent': limits.max_cpu_percent,
'max_execution_time': limits.max_execution_time,
'warning_threshold': limits.warning_threshold
}
})
else:
return jsonify({
'status': 'success',
'data': None,
'message': 'No limits configured for this plugin'
})
else:
# POST - Set limits
data = request.get_json(silent=True) or {}
from src.plugin_system.resource_monitor import ResourceLimits
limits = ResourceLimits(
max_memory_mb=data.get('max_memory_mb'),
max_cpu_percent=data.get('max_cpu_percent'),
max_execution_time=data.get('max_execution_time'),
warning_threshold=data.get('warning_threshold', 0.8)
)
api_v3.plugin_manager.resource_monitor.set_limits(plugin_id, limits)
return jsonify({
'status': 'success',
'message': f'Resource limits updated for plugin {plugin_id}'
})
@api_v3.route('/plugins/toggle', methods=['POST'])
def toggle_plugin():
"""Toggle plugin enabled/disabled"""
plugin_id = None
enabled = None
try:
if not api_v3.plugin_manager or not api_v3.config_manager:
return jsonify({'status': 'error', 'message': 'Plugin or config manager not initialized'}), 500
# Support both JSON and form data (for HTMX submissions)
content_type = request.content_type or ''
if 'application/json' in content_type:
data = request.get_json(silent=True)
if not data or 'plugin_id' not in data or 'enabled' not in data:
return jsonify({'status': 'error', 'message': 'plugin_id and enabled required'}), 400
plugin_id = data['plugin_id']
enabled = data['enabled']
else:
# Form data or query string (HTMX submission)
plugin_id = request.args.get('plugin_id') or request.form.get('plugin_id')
if not plugin_id:
return jsonify({'status': 'error', 'message': 'plugin_id required'}), 400
# For checkbox toggle, if form was submitted, the checkbox was checked (enabled)
# If using HTMX with hx-trigger="change", we need to check if checkbox is checked
# The checkbox value or 'enabled' form field indicates the state
enabled_str = request.form.get('enabled', request.args.get('enabled', ''))
# Handle various truthy/falsy values
if enabled_str.lower() in ('true', '1', 'on', 'yes'):
enabled = True
elif enabled_str.lower() in ('false', '0', 'off', 'no', ''):
# Empty string means checkbox was unchecked (toggle off)
enabled = False
else:
# Default: toggle based on current state
config = api_v3.config_manager.load_config()
current_enabled = config.get(plugin_id, {}).get('enabled', False)
enabled = not current_enabled
# A Starlark app is not a plugin in plugin_manager's sense -- it is an
# entry in starlark-apps' own manifest -- so its enable/disable is
# handled here rather than falling through to the check below, which
# would answer "Plugin not found".
if plugin_id.startswith('starlark:'):
return _toggle_starlark_app(plugin_id[len('starlark:'):], enabled)
# Check if plugin exists in manifests (discovered but may not be loaded)
if plugin_id not in _pkg._discovered_plugin_manifests(plugin_id):
return jsonify({'status': 'error', 'message': 'Plugin not found'}), 404
# Update config (this is what the display controller reads)
config = api_v3.config_manager.load_config()
if plugin_id not in config:
config[plugin_id] = {}
config[plugin_id]['enabled'] = enabled
success, error_msg = _pkg._save_config_atomic(api_v3.config_manager, config, create_backup=True)
if not success:
return error_response(
ErrorCode.CONFIG_SAVE_FAILED,
f"Failed to save configuration: {error_msg}",
status_code=500
)
# Update state manager if available
if api_v3.plugin_state_manager:
api_v3.plugin_state_manager.set_plugin_enabled(plugin_id, enabled)
# Log operation
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"enable" if enabled else "disable",
plugin_id=plugin_id,
status="success"
)
# If plugin is loaded, also call its lifecycle methods
# Wrap in try/except to prevent lifecycle errors from failing the toggle
plugin = api_v3.plugin_manager.get_plugin(plugin_id)
if plugin:
try:
if enabled:
if hasattr(plugin, 'on_enable'):
plugin.on_enable()
else:
if hasattr(plugin, 'on_disable'):
plugin.on_disable()
except Exception as lifecycle_error:
# Log the error but don't fail the toggle - config is already saved
logger.warning("Lifecycle method error for %s: %s", plugin_id, lifecycle_error, exc_info=True)
return success_response(
message=f"Plugin {plugin_id} {'enabled' if enabled else 'disabled'} successfully"
)
except Exception as e:
# Not PLUGIN_OPERATION_CONFLICT: that told the user "an operation is
# already in progress" whatever actually went wrong.
logger.error('Error toggling plugin %s', plugin_id, exc_info=True)
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"enable" if enabled else "disable",
plugin_id=plugin_id,
status="failed",
error=str(e)
)
action = 'enable' if enabled else 'disable' if enabled is not None else 'toggle'
return error_response(
ErrorCode.UNKNOWN_ERROR,
f"Failed to {action} plugin {plugin_id or ''}".rstrip(),
details=describe_exception(e),
status_code=500
)
@api_v3.route('/plugins/operation/<operation_id>', methods=['GET'])
def get_operation_status(operation_id):
"""Get status of a plugin operation"""
try:
if not api_v3.operation_queue:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Operation queue not initialized',
status_code=500
)
operation = api_v3.operation_queue.get_operation_status(operation_id)
if not operation:
return error_response(
ErrorCode.PLUGIN_NOT_FOUND,
f'Operation {operation_id} not found',
status_code=404
)
return success_response(data=operation.to_dict())
except Exception as e:
return exception_error_response(e, ErrorCode.SYSTEM_ERROR, with_context=False)
@api_v3.route('/plugins/operation/history', methods=['GET'])
def get_operation_history() -> Response:
"""Get operation history from the audit log."""
if not api_v3.operation_history:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Operation history not initialized',
status_code=500
)
try:
limit = request.args.get('limit', 50, type=int)
plugin_id = request.args.get('plugin_id')
operation_type = request.args.get('operation_type')
except (ValueError, TypeError) as e:
return error_response(ErrorCode.INVALID_INPUT, f'Invalid query parameter: {e}', status_code=400)
try:
history = api_v3.operation_history.get_history(
limit=limit,
plugin_id=plugin_id,
operation_type=operation_type
)
except (AttributeError, RuntimeError) as e:
return exception_error_response(e, ErrorCode.SYSTEM_ERROR, with_context=False)
return success_response(data=[record.to_dict() for record in history])
@api_v3.route('/plugins/operation/history', methods=['DELETE'])
def clear_operation_history() -> Response:
"""Clear operation history."""
if not api_v3.operation_history:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Operation history not initialized',
status_code=500
)
try:
api_v3.operation_history.clear_history()
except (OSError, RuntimeError) as e:
return exception_error_response(e, ErrorCode.SYSTEM_ERROR, with_context=False)
return success_response(message='Operation history cleared')
@api_v3.route('/plugins/state', methods=['GET'])
def get_plugin_state():
"""Get plugin state from state manager"""
try:
if not api_v3.plugin_state_manager:
return error_response(
ErrorCode.SYSTEM_ERROR,
'State manager not initialized',
status_code=500
)
plugin_id = request.args.get('plugin_id')
if plugin_id:
# Get state for specific plugin
state = api_v3.plugin_state_manager.get_plugin_state(plugin_id)
if not state:
return error_response(
ErrorCode.PLUGIN_NOT_FOUND,
f'Plugin {plugin_id} not found in state manager',
context={'plugin_id': plugin_id},
status_code=404
)
return success_response(data=state.to_dict())
else:
# Get all plugin states
all_states = api_v3.plugin_state_manager.get_all_states()
return success_response(data={
plugin_id: state.to_dict()
for plugin_id, state in all_states.items()
})
except Exception as e:
return exception_error_response(e, ErrorCode.SYSTEM_ERROR)
@api_v3.route('/plugins/state/reconcile', methods=['POST'])
def reconcile_plugin_state():
"""Reconcile plugin state across all sources"""
try:
if not api_v3.plugin_state_manager or not api_v3.plugin_manager:
return error_response(
ErrorCode.SYSTEM_ERROR,
'State manager or plugin manager not initialized',
status_code=500
)
from src.plugin_system.state_reconciliation import StateReconciliation
# Parse optional `force` flag from request body, guarding against
# non-dict bodies (bare string, array, null) that would raise AttributeError.
payload = request.get_json(silent=True)
if not isinstance(payload, dict):
payload = {}
force = _coerce_to_bool(payload.get('force', False))
reconciler = StateReconciliation(
state_manager=api_v3.plugin_state_manager,
config_manager=api_v3.config_manager,
plugin_manager=api_v3.plugin_manager,
plugins_dir=Path(api_v3.plugin_manager.plugins_dir)
)
result = reconciler.reconcile_state(force=force)
return success_response(
data={
'inconsistencies_found': len(result.inconsistencies_found),
'inconsistencies_fixed': len(result.inconsistencies_fixed),
'inconsistencies_manual': len(result.inconsistencies_manual),
'inconsistencies': [
{
'plugin_id': inc.plugin_id,
'type': inc.inconsistency_type.value,
'description': inc.description,
'fix_action': inc.fix_action.value
}
for inc in result.inconsistencies_found
],
'fixed': [
{
'plugin_id': inc.plugin_id,
'type': inc.inconsistency_type.value,
'description': inc.description
}
for inc in result.inconsistencies_fixed
],
'manual_fix_required': [
{
'plugin_id': inc.plugin_id,
'type': inc.inconsistency_type.value,
'description': inc.description
}
for inc in result.inconsistencies_manual
]
},
message=result.message
)
except Exception as e:
return exception_error_response(e, ErrorCode.SYSTEM_ERROR)
def _drop_stale_reconciliation_findings(unresolved):
"""Re-check a stored reconciliation verdict against current state.
The verdict is a snapshot written once per run, and a run that could not
apply a fix also refuses to retry -- so a resolved condition was reported
indefinitely. Best-effort: any failure here returns the list untouched,
because showing a stale warning beats failing the endpoint.
Both sets come from the reconciliation module's own extractors rather than
being re-derived here. That matters for correctness, not tidiness: a plain
set(load_config()) also contains system keys, the secrets-file keys merged
in by load_config(), and non-dict values, and any directory holding a
manifest.json would count as installed even if that manifest does not
parse. Either looseness clears findings that are still true -- and a secrets
key read as a plugin is the very bug the filter exists to stop reporting.
"""
try:
from src.plugin_system.state_reconciliation import (
config_plugin_ids, disk_plugin_ids, ignored_config_keys,
still_unresolved,
)
cm = api_v3.config_manager
plugins_dir = getattr(api_v3.plugin_manager, 'plugins_dir', None)
installed = disk_plugin_ids(plugins_dir) if plugins_dir else set()
config_keys = config_plugin_ids(cm.load_config() or {},
ignored_config_keys(cm, installed))
return still_unresolved(unresolved, config_keys, installed)
except Exception:
logger.debug("[Reconciliation] Could not re-check stored findings", exc_info=True)
return unresolved
@api_v3.route('/plugins/reconciliation-status', methods=['GET'])
def get_reconciliation_status():
"""Return the result of the last startup reconciliation from /tmp status file."""
_recon_path = os.path.join(tempfile.gettempdir(), "ledmatrix_reconciliation.json")
try:
st = os.lstat(_recon_path)
except FileNotFoundError:
return jsonify({'status': 'success', 'data': {'done': False, 'unresolved': []}})
if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode):
logger.warning("[Reconciliation] Status file is not a regular file: %s", _recon_path)
return jsonify({'status': 'success', 'data': {'done': False, 'unresolved': []}})
try:
with open(_recon_path) as _f:
data = json.load(_f)
if data.get('unresolved'):
data['unresolved'] = _drop_stale_reconciliation_findings(data['unresolved'])
return jsonify({'status': 'success', 'data': data})
except json.JSONDecodeError:
logger.exception("[Reconciliation] Failed to parse status file: %s", _recon_path)
return jsonify({'status': 'success', 'data': {'done': False, 'unresolved': []}})
except PermissionError:
logger.exception("[Reconciliation] Permission denied reading status file: %s", _recon_path)
return jsonify({'status': 'success', 'data': {'done': False, 'unresolved': []}})
@api_v3.route('/plugins/config', methods=['GET'])
def get_plugin_config():
"""Get plugin configuration"""
try:
if not api_v3.config_manager:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Config manager not initialized',
status_code=500
)
plugin_id = request.args.get('plugin_id')
if not plugin_id:
return error_response(
ErrorCode.INVALID_INPUT,
'plugin_id required',
context={'missing_params': ['plugin_id']},
status_code=400
)
# Get plugin configuration from config manager
main_config = api_v3.config_manager.load_config()
plugin_config = main_config.get(plugin_id, {})
# Merge with defaults from schema so form shows default values for
# missing fields, reading legacy booleans as objects first: what the
# plugin runs with, and what posts back through the JSON save
schema_mgr = api_v3.schema_manager
if schema_mgr:
try:
from src.plugin_system.schema_manager import prepare_plugin_config
defaults = schema_mgr.generate_default_config(plugin_id, use_cache=True)
plugin_config = prepare_plugin_config(
plugin_config, schema_mgr.load_schema(plugin_id, use_cache=True), defaults)
except Exception as e:
# Log but don't fail - defaults merge is best effort
logger.warning("Could not merge defaults for %s: %s", plugin_id, e)
# Special handling for of-the-day plugin: populate uploaded_files and categories from disk
if plugin_id == 'of-the-day' or plugin_id == 'ledmatrix-of-the-day':
# The manifest id is 'of-the-day'; the directory is usually
# 'ledmatrix-of-the-day'.
plugin_dir = (_plugin_directory('ledmatrix-of-the-day')
or _plugin_directory(plugin_id))
if plugin_dir:
data_dir = plugin_dir / 'of_the_day'
if data_dir.exists():
# Scan for JSON files
uploaded_files = []
categories_from_files = {}
for json_file in data_dir.glob('*.json'):
try:
# Get file stats
stat = json_file.stat()
# Read JSON to count entries
with open(json_file, 'r', encoding='utf-8') as f:
json_data = json.load(f)
entry_count = len(json_data) if isinstance(json_data, dict) else 0
# Extract category name from filename
category_name = json_file.stem
filename = json_file.name
# Create file entry
file_entry = {
'id': category_name,
'category_name': category_name,
'filename': filename,
'original_filename': filename,
'path': f'of_the_day/{filename}',
'size': stat.st_size,
'uploaded_at': datetime.fromtimestamp(stat.st_mtime).isoformat() + 'Z',
'entry_count': entry_count
}
uploaded_files.append(file_entry)
# Create/update category entry if not in config
if category_name not in plugin_config.get('categories', {}):
display_name = category_name.replace('_', ' ').title()
categories_from_files[category_name] = {
'enabled': False, # Default to disabled, user can enable
'data_file': f'of_the_day/{filename}',
'display_name': display_name
}
else:
# Update with file info if needed
categories_from_files[category_name] = plugin_config['categories'][category_name]
# Ensure data_file is correct
categories_from_files[category_name]['data_file'] = f'of_the_day/{filename}'
except Exception as e:
logger.debug("Could not read json file: %s", e)
continue
# Update plugin_config with scanned files
if uploaded_files:
plugin_config['uploaded_files'] = uploaded_files
# Merge categories from files with existing config
# Start with existing categories (preserve user settings like enabled/disabled)
existing_categories = plugin_config.get('categories', {}).copy()
# Update existing categories with file info, add new ones from files
for cat_name, cat_data in categories_from_files.items():
if cat_name in existing_categories:
# Preserve existing enabled state and display_name, but update data_file path
existing_categories[cat_name]['data_file'] = cat_data['data_file']
if 'display_name' not in existing_categories[cat_name] or not existing_categories[cat_name]['display_name']:
existing_categories[cat_name]['display_name'] = cat_data['display_name']
else:
# Add new category from file (default to disabled)
existing_categories[cat_name] = cat_data
if existing_categories:
plugin_config['categories'] = existing_categories
# Update category_order to include all categories
category_order = plugin_config.get('category_order', []).copy()
all_category_names = set(existing_categories.keys())
for cat_name in all_category_names:
if cat_name not in category_order:
category_order.append(cat_name)
if category_order:
plugin_config['category_order'] = category_order
# If no config exists, return defaults
if not plugin_config:
plugin_config = {
'enabled': True,
'display_duration': 30
}
return success_response(data=plugin_config)
except Exception as e:
return exception_error_response(e, ErrorCode.CONFIG_LOAD_FAILED)
@api_v3.route('/plugins/update', methods=['POST'])
def update_plugin():
"""Update plugin"""
try:
# Support both JSON and form data
content_type = request.content_type or ''
if 'application/json' in content_type:
# JSON request
data, error = validate_request_json(['plugin_id'])
if error:
logger.debug("[UPDATE] JSON validation failed. Content-Type: %s", content_type)
return error
else:
# Form data or query string
plugin_id = request.args.get('plugin_id') or request.form.get('plugin_id')
if not plugin_id:
logger.debug("[UPDATE] Missing plugin_id. Content-Type: %s", content_type)
return error_response(
ErrorCode.INVALID_INPUT,
'plugin_id required',
status_code=400
)
data = {'plugin_id': plugin_id}
# /plugins/installed lists installed Starlark apps as virtual
# 'starlark:<app_id>' entries. They are not plugin directories, so the
# store manager can only fail to find them -- which used to surface as
# a 500 "plugin not found" for an app that is installed and working.
raw_id = data.get('plugin_id')
if isinstance(raw_id, str) and raw_id.startswith('starlark:'):
return error_response(
ErrorCode.INVALID_INPUT,
f'{raw_id} is a Starlark app, not a plugin; Starlark apps are '
'not updated through the plugin updater',
status_code=400
)
if not api_v3.plugin_store_manager:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Plugin store manager not initialized',
status_code=500
)
# The id names a directory this handler reads a manifest out of and
# hands to the store manager to run git in. It comes from the request
# body, so it is validated before anything is joined to a path, and
# the validated value -- not the raw one -- is what gets used below.
plugin_id = safe_path_component(data['plugin_id'])
if not plugin_id:
return error_response(
ErrorCode.INVALID_INPUT,
'Invalid plugin_id',
status_code=400
)
# Always do direct updates (they're fast git pull operations)
# Operation queue is reserved for longer operations like install/uninstall
plugins_base = Path(api_v3.plugin_store_manager.plugins_dir)
plugin_dir = plugins_base / plugin_id
manifest_path = resolve_under(plugin_dir, "manifest.json")
if manifest_path is None:
return error_response(
ErrorCode.INVALID_INPUT,
'Invalid plugin_id',
status_code=400
)
current_last_updated = None
current_version = None
current_commit = None
current_branch = None
if manifest_path.exists():
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
current_last_updated = manifest.get('last_updated')
current_version = manifest.get('version')
if manifest.get('local_only'):
logger.debug("Skipping update for local-only plugin: %s", plugin_id)
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"update",
plugin_id=plugin_id,
status="skipped",
details={"reason": "local_only"}
)
return success_response(
data={'update_status': 'local_only'},
message=f'Plugin {plugin_id} is managed locally and does not receive registry updates')
except Exception as e:
logger.debug("Could not read local manifest for plugin: %s", e)
if api_v3.plugin_store_manager:
git_info_before = api_v3.plugin_store_manager._get_local_git_info(plugin_dir)
if git_info_before:
current_commit = git_info_before.get('sha')
current_branch = git_info_before.get('branch')
# Check if plugin is a git repo first (for better error messages)
# plugin_id is validated above; reuse the same directory rather
# than rebuilding it from a value that might not match.
plugin_path_dir = plugin_dir
is_git_repo = False
if plugin_path_dir.exists():
git_info = api_v3.plugin_store_manager._get_local_git_info(plugin_path_dir)
is_git_repo = git_info is not None
if is_git_repo:
logger.debug("Plugin is a git repository, will update via git pull")
remote_info = api_v3.plugin_store_manager.get_plugin_info(plugin_id, fetch_latest_from_github=True)
remote_commit = remote_info.get('last_commit_sha') if remote_info else None
remote_branch = remote_info.get('branch') if remote_info else None
# Update the plugin
success = api_v3.plugin_store_manager.update_plugin(plugin_id)
if success:
updated_last_updated = current_last_updated
updated_version = current_version
try:
if manifest_path.exists():
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
updated_last_updated = manifest.get('last_updated', current_last_updated)
updated_version = manifest.get('version', current_version)
except Exception as e:
logger.debug("Could not read updated manifest after update: %s", e)
updated_commit = None
updated_branch = remote_branch or current_branch
if api_v3.plugin_store_manager:
git_info_after = api_v3.plugin_store_manager._get_local_git_info(plugin_dir)
if git_info_after:
updated_commit = git_info_after.get('sha')
updated_branch = git_info_after.get('branch') or updated_branch
# update_plugin() answers True for "nothing to do" as well as for
# a real update (a ZIP-installed monorepo plugin already at the
# registry version, a bundled plugin), so what changed is read off
# the plugin itself: its git commit, else its manifest.
update_status = 'updated'
message = f'Plugin {plugin_id} updated successfully'
if current_commit and updated_commit and current_commit == updated_commit:
update_status = 'up_to_date'
message = f'Plugin {plugin_id} already up to date (commit {updated_commit[:7]})'
elif updated_commit:
message = f'Plugin {plugin_id} updated to commit {updated_commit[:7]}'
if updated_branch:
message += f' on branch {updated_branch}'
elif updated_version and updated_version != current_version:
message = f'Plugin {plugin_id} updated to version {updated_version}'
elif updated_last_updated and updated_last_updated != current_last_updated:
message = f'Plugin {plugin_id} refreshed (Last Updated {updated_last_updated})'
elif not current_commit:
update_status = 'up_to_date'
message = f'Plugin {plugin_id} already up to date'
if updated_version:
message += f' (version {updated_version})'
remote_commit_short = remote_commit[:7] if remote_commit else None
if remote_commit_short and updated_commit and remote_commit_short != updated_commit[:7]:
message += f' (remote latest {remote_commit_short})'
# Invalidate schema cache
if api_v3.schema_manager:
api_v3.schema_manager.invalidate_cache(plugin_id)
# Rediscover plugins
if api_v3.plugin_manager:
api_v3.plugin_manager.discover_plugins()
if plugin_id in api_v3.plugin_manager.plugins:
api_v3.plugin_manager.reload_plugin(plugin_id)
# Update state and history
if api_v3.plugin_state_manager:
api_v3.plugin_state_manager.update_plugin_state(
plugin_id,
{'last_updated': datetime.now()}
)
if api_v3.operation_history:
version = _get_plugin_version(plugin_id)
api_v3.operation_history.record_operation(
"update",
plugin_id=plugin_id,
status="success",
details={
"version": version,
"previous_commit": current_commit[:7] if current_commit else None,
"commit": updated_commit[:7] if updated_commit else None,
"branch": updated_branch,
"update_status": update_status
}
)
return success_response(
data={
'last_updated': updated_last_updated,
'commit': updated_commit,
'update_status': update_status
},
message=message
)
else:
plugin_path_dir = plugin_dir
if not plugin_path_dir.exists():
client_msg = 'Plugin update failed: plugin not found'
else:
git_info = api_v3.plugin_store_manager._get_local_git_info(plugin_path_dir)
if not git_info:
plugin_info = api_v3.plugin_store_manager.get_plugin_info(plugin_id)
if not plugin_info:
client_msg = 'Plugin update failed: not found in registry'
else:
client_msg = 'Plugin update failed; check logs for details'
else:
client_msg = 'Plugin update failed; check logs for details'
logger.error("update_plugin failed for plugin_id=%s: %s", plugin_id, client_msg)
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"update",
plugin_id=plugin_id,
status="failed",
error=client_msg,
details={
"previous_commit": current_commit[:7] if current_commit else None,
"branch": current_branch
}
)
return error_response(
ErrorCode.PLUGIN_UPDATE_FAILED,
client_msg,
status_code=500
)
except Exception as e:
logger.error("Unhandled exception in update endpoint", exc_info=True)
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"update",
plugin_id=data.get('plugin_id') if 'data' in locals() else None,
status="failed",
error=str(e)
)
return exception_error_response(e, ErrorCode.PLUGIN_UPDATE_FAILED)
@api_v3.route('/plugins/uninstall', methods=['POST'])
def uninstall_plugin():
"""Uninstall plugin"""
try:
# Validate request
data, error = validate_request_json(['plugin_id'])
if error:
return error
if not api_v3.plugin_store_manager:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Plugin store manager not initialized',
status_code=500
)
plugin_id = data['plugin_id']
preserve_config = data.get('preserve_config', False)
id_error = _non_plugin_id_error(plugin_id)
if id_error:
return id_error
# Both queued and direct paths use the same transactional helper so
# snapshot/rollback behaviour is consistent regardless of deployment.
if api_v3.operation_queue:
def uninstall_callback(operation):
"""Callback to execute plugin uninstallation via transactional helper."""
success, error_msg = _do_transactional_uninstall(plugin_id, preserve_config)
if not success:
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"uninstall",
plugin_id=plugin_id,
status="failed",
error=error_msg
)
raise Exception(error_msg or f'Failed to uninstall plugin {plugin_id}')
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"uninstall",
plugin_id=plugin_id,
status="success",
details={"preserve_config": preserve_config}
)
return {'success': True, 'message': 'Plugin uninstalled successfully'}
# Enqueue operation
operation_id = api_v3.operation_queue.enqueue_operation(
OperationType.UNINSTALL,
plugin_id,
operation_callback=uninstall_callback
)
return success_response(
data={'operation_id': operation_id},
message='Plugin uninstallation queued'
)
else:
# Direct (non-queued) transactional uninstall
success, error_msg = _do_transactional_uninstall(plugin_id, preserve_config)
if success:
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"uninstall",
plugin_id=plugin_id,
status="success",
details={"preserve_config": preserve_config}
)
return success_response(message='Plugin uninstalled successfully')
else:
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"uninstall",
plugin_id=plugin_id,
status="failed",
error=error_msg
)
return error_response(
ErrorCode.PLUGIN_UNINSTALL_FAILED,
error_msg or 'Plugin uninstall failed',
status_code=500
)
except Exception as e:
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"uninstall",
plugin_id=data.get('plugin_id') if 'data' in locals() else None,
status="failed",
error=str(e)
)
return exception_error_response(e, ErrorCode.PLUGIN_UNINSTALL_FAILED)
@api_v3.route('/plugins/install', methods=['POST'])
def install_plugin():
"""Install plugin from store"""
if not api_v3.plugin_store_manager:
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
data = request.get_json(silent=True)
if not data or 'plugin_id' not in data:
return jsonify({'status': 'error', 'message': 'plugin_id required'}), 400
plugin_id = data['plugin_id']
branch = data.get('branch') # Optional branch parameter
# A registry entry that isn't a plugin (a custom registry can still
# list old "type": "skin" entries) gets a clear refusal, not a failed
# install.
try:
registry_entry = api_v3.plugin_store_manager.get_registry_info(plugin_id)
except Exception:
registry_entry = None
if isinstance(registry_entry, dict) and not api_v3.plugin_store_manager.is_plugin_entry(registry_entry):
return jsonify({'status': 'error',
'message': f"{plugin_id} is a {registry_entry.get('type')!r} entry, not a plugin"}), 400
plugins_dir = api_v3.plugin_store_manager.plugins_dir
logger.info("Installing plugin to directory: %s", plugins_dir)
# Use operation queue if available
if api_v3.operation_queue:
def install_callback(operation):
"""Callback to execute plugin installation."""
success = api_v3.plugin_store_manager.install_plugin(plugin_id, branch=branch)
if success:
# Invalidate schema cache
if api_v3.schema_manager:
api_v3.schema_manager.invalidate_cache(plugin_id)
# Discover and load the new plugin
if api_v3.plugin_manager:
api_v3.plugin_manager.discover_plugins()
api_v3.plugin_manager.load_plugin(plugin_id)
# Update state manager
if api_v3.plugin_state_manager:
api_v3.plugin_state_manager.set_plugin_installed(plugin_id)
# Record in history
if api_v3.operation_history:
version = _get_plugin_version(plugin_id)
api_v3.operation_history.record_operation(
"install",
plugin_id=plugin_id,
status="success",
details={"version": version, "branch": branch}
)
branch_msg = f" (branch: {branch})" if branch else ""
return {'success': True, 'message': f'Plugin {plugin_id} installed successfully{branch_msg}'}
else:
error_msg = f'Failed to install plugin {plugin_id}'
if branch:
error_msg += f' (branch: {branch})'
plugin_info = api_v3.plugin_store_manager.get_plugin_info(plugin_id)
if not plugin_info:
error_msg += ' (plugin not found in registry)'
# Record failure in history
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"install",
plugin_id=plugin_id,
status="failed",
error=error_msg,
details={"branch": branch}
)
raise Exception(error_msg)
# Enqueue operation
operation_id = api_v3.operation_queue.enqueue_operation(
OperationType.INSTALL,
plugin_id,
operation_callback=install_callback
)
branch_msg = f" (branch: {branch})" if branch else ""
return success_response(
data={'operation_id': operation_id},
message=f'Plugin {plugin_id} installation queued{branch_msg}'
)
else:
# Fallback to direct installation
success = api_v3.plugin_store_manager.install_plugin(plugin_id, branch=branch)
if success:
if api_v3.schema_manager:
api_v3.schema_manager.invalidate_cache(plugin_id)
if api_v3.plugin_manager:
api_v3.plugin_manager.discover_plugins()
api_v3.plugin_manager.load_plugin(plugin_id)
if api_v3.plugin_state_manager:
api_v3.plugin_state_manager.set_plugin_installed(plugin_id)
if api_v3.operation_history:
version = _get_plugin_version(plugin_id)
api_v3.operation_history.record_operation(
"install",
plugin_id=plugin_id,
status="success",
details={"version": version, "branch": branch}
)
branch_msg = f" (branch: {branch})" if branch else ""
return success_response(message=f'Plugin installed successfully{branch_msg}')
else:
error_msg = f'Failed to install plugin {plugin_id}'
if branch:
error_msg += f' (branch: {branch})'
plugin_info = api_v3.plugin_store_manager.get_plugin_info(plugin_id)
if not plugin_info:
error_msg += ' (plugin not found in registry)'
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"install",
plugin_id=plugin_id,
status="failed",
error=error_msg,
details={"branch": branch}
)
return error_response(
ErrorCode.PLUGIN_INSTALL_FAILED,
error_msg,
status_code=500
)
@api_v3.route('/plugins/install-from-url', methods=['POST'])
def install_plugin_from_url():
"""Install plugin from custom GitHub URL"""
if not api_v3.plugin_store_manager:
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
data = request.get_json(silent=True)
if not data or 'repo_url' not in data:
return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
# A non-string repo_url is a client mistake, not a server fault:
# .strip() would raise and the catch-all would report it as a 500.
if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
repo_url = data['repo_url'].strip()
plugin_id = data.get('plugin_id') # Optional, for monorepo installations
plugin_path = data.get('plugin_path') # Optional, for monorepo subdirectory
branch = data.get('branch') # Optional branch parameter
# Install the plugin
result = api_v3.plugin_store_manager.install_from_url(
repo_url=repo_url,
plugin_id=plugin_id,
plugin_path=plugin_path,
branch=branch
)
if result.get('success'):
# Invalidate schema cache for the installed plugin
installed_plugin_id = result.get('plugin_id')
if api_v3.schema_manager and installed_plugin_id:
api_v3.schema_manager.invalidate_cache(installed_plugin_id)
# Discover and load the new plugin
if api_v3.plugin_manager and installed_plugin_id:
api_v3.plugin_manager.discover_plugins()
api_v3.plugin_manager.load_plugin(installed_plugin_id)
branch_msg = f" (branch: {result.get('branch', branch)})" if (result.get('branch') or branch) else ""
response_data = {
'status': 'success',
'message': f"Plugin {installed_plugin_id} installed successfully{branch_msg}",
'plugin_id': installed_plugin_id,
'name': result.get('name')
}
if result.get('branch'):
response_data['branch'] = result.get('branch')
return jsonify(response_data)
else:
return jsonify({
'status': 'error',
'message': result.get('error', 'Failed to install plugin from URL')
}), 500
@api_v3.route('/plugins/registry-from-url', methods=['POST'])
def get_registry_from_url():
"""Get plugin list from a registry-style monorepo URL"""
if not api_v3.plugin_store_manager:
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
data = request.get_json(silent=True)
if not data or 'repo_url' not in data:
return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
# A non-string repo_url is a client mistake, not a server fault:
# .strip() would raise and the catch-all would report it as a 500.
if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
repo_url = data['repo_url'].strip()
# Get registry from the URL
registry = api_v3.plugin_store_manager.fetch_registry_from_url(repo_url)
if registry:
return jsonify({
'status': 'success',
'plugins': [p for p in registry.get('plugins', [])
if api_v3.plugin_store_manager.is_plugin_entry(p)],
'registry_url': repo_url
})
else:
return jsonify({
'status': 'error',
'message': 'Failed to fetch registry from URL or URL does not contain a valid registry'
}), 400
@api_v3.route('/plugins/saved-repositories', methods=['GET'])
def get_saved_repositories():
"""Get all saved repositories"""
if not api_v3.saved_repositories_manager:
return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
repositories = api_v3.saved_repositories_manager.get_all()
return jsonify({'status': 'success', 'data': {'repositories': repositories}})
@api_v3.route('/plugins/saved-repositories', methods=['POST'])
def add_saved_repository():
"""Add a repository to saved list"""
if not api_v3.saved_repositories_manager:
return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
data = request.get_json(silent=True)
if not data or 'repo_url' not in data:
return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
# A non-string repo_url is a client mistake, not a server fault:
# .strip() would raise and the catch-all would report it as a 500.
if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
repo_url = data['repo_url'].strip()
name = data.get('name')
success = api_v3.saved_repositories_manager.add(repo_url, name)
if success:
return jsonify({
'status': 'success',
'message': 'Repository saved successfully',
'data': {'repositories': api_v3.saved_repositories_manager.get_all()}
})
else:
return jsonify({
'status': 'error',
'message': 'Repository already exists or failed to save'
}), 400
@api_v3.route('/plugins/saved-repositories', methods=['DELETE'])
def remove_saved_repository():
"""Remove a repository from saved list"""
if not api_v3.saved_repositories_manager:
return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
data = request.get_json(silent=True)
if not data or 'repo_url' not in data:
return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
repo_url = data['repo_url']
success = api_v3.saved_repositories_manager.remove(repo_url)
if success:
return jsonify({
'status': 'success',
'message': 'Repository removed successfully',
'data': {'repositories': api_v3.saved_repositories_manager.get_all()}
})
else:
return jsonify({
'status': 'error',
'message': 'Repository not found'
}), 404
@api_v3.route('/plugins/store/list', methods=['GET'])
def list_plugin_store():
"""Search plugin store"""
if not api_v3.plugin_store_manager:
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
query = request.args.get('query', '')
category = request.args.get('category', '')
tags = request.args.getlist('tags')
# Default to fetching commit metadata to ensure accurate commit timestamps
fetch_commit_param = request.args.get('fetch_commit_info', request.args.get('fetch_latest_versions', '')).lower()
fetch_commit = fetch_commit_param != 'false'
# Search plugins from the registry (including saved repositories)
plugins = api_v3.plugin_store_manager.search_plugins(
query=query,
category=category,
tags=tags,
fetch_commit_info=fetch_commit,
include_saved_repos=True,
saved_repositories_manager=api_v3.saved_repositories_manager
)
# Format plugins for the web interface
formatted_plugins = []
for plugin in plugins:
if not api_v3.plugin_store_manager.is_plugin_entry(plugin):
continue
formatted_plugins.append({
'id': plugin.get('id'),
'name': plugin.get('name'),
'author': plugin.get('author'),
'category': plugin.get('category'),
'description': plugin.get('description'),
'tags': plugin.get('tags', []),
'stars': plugin.get('stars', 0),
'verified': plugin.get('verified', False),
'repo': plugin.get('repo', ''),
'last_updated': plugin.get('last_updated') or plugin.get('last_updated_iso', ''),
'last_updated_iso': plugin.get('last_updated_iso', ''),
'last_commit': plugin.get('last_commit') or plugin.get('last_commit_sha'),
'last_commit_message': plugin.get('last_commit_message'),
'last_commit_author': plugin.get('last_commit_author'),
'version': plugin.get('latest_version') or plugin.get('version', ''),
'branch': plugin.get('branch') or plugin.get('default_branch'),
'default_branch': plugin.get('default_branch'),
'plugin_path': plugin.get('plugin_path', '')
})
return jsonify({'status': 'success', 'data': {'plugins': formatted_plugins}})
@api_v3.route('/plugins/store/github-status', methods=['GET'])
def get_github_auth_status():
"""Check if GitHub authentication is configured and validate token"""
if not api_v3.plugin_store_manager:
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
token = api_v3.plugin_store_manager.github_token
# Check if GitHub token is configured
if not token or len(token) == 0:
return jsonify({
'status': 'success',
'data': {
'token_status': 'none',
'authenticated': False,
'rate_limit': 60,
'message': 'No GitHub token configured',
'error': None
}
})
# Validate the token
is_valid, error_message = api_v3.plugin_store_manager._validate_github_token(token)
if is_valid:
return jsonify({
'status': 'success',
'data': {
'token_status': 'valid',
'authenticated': True,
'rate_limit': 5000,
'message': 'GitHub API authenticated',
'error': None
}
})
else:
return jsonify({
'status': 'success',
'data': {
'token_status': 'invalid',
'authenticated': False,
'rate_limit': 60,
'message': f'GitHub token is invalid: {error_message}' if error_message else 'GitHub token is invalid',
'error': error_message
}
})
@api_v3.route('/plugins/store/refresh', methods=['POST'])
def refresh_plugin_store():
"""Re-download the plugin registry, bypassing its cache.
Takes no body. Answers ``{status, message, plugin_count}``, the count
being the registry's entries. Commit metadata is not refreshed here: the
store list fetches it per plugin when it is shown.
"""
if not api_v3.plugin_store_manager:
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
registry = api_v3.plugin_store_manager.fetch_registry(force_refresh=True)
plugin_count = len(registry.get('plugins', []))
return jsonify({
'status': 'success',
'message': 'Plugin store refreshed',
'plugin_count': plugin_count
})
@api_v3.route('/plugins/config', methods=['POST'])
def save_plugin_config():
"""Save plugin configuration, separating secrets from regular config"""
try:
if not api_v3.config_manager:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Config manager not initialized',
status_code=500
)
# Support both JSON and form data (for HTMX submissions)
content_type = request.content_type or ''
if 'application/json' in content_type:
# JSON request
data, error = validate_request_json(['plugin_id'])
if error:
return error
plugin_id = data['plugin_id']
submitted_config = data.get('config', {})
if not isinstance(submitted_config, dict):
return error_response(
ErrorCode.INVALID_INPUT,
'config must be a JSON object',
status_code=400
)
plugin_config = _merge_onto_stored_plugin_config(plugin_id, submitted_config)
else:
# Form data (HTMX submission)
# plugin_id comes from query string, config from form fields
plugin_id = request.args.get('plugin_id')
if not plugin_id:
return error_response(
ErrorCode.INVALID_INPUT,
'plugin_id required in query string',
status_code=400
)
# Load existing config as base (partial form updates should merge, not replace)
existing_config = {}
if api_v3.config_manager:
full_config = api_v3.config_manager.load_config()
existing_config = full_config.get(plugin_id, {}).copy()
# Get schema manager instance (needed for type conversion)
schema_mgr = api_v3.schema_manager
if not schema_mgr:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Schema manager not initialized',
status_code=500
)
# Load plugin schema BEFORE processing form data (needed for type conversion)
schema = schema_mgr.load_schema(plugin_id, use_cache=False)
# Start with existing config and apply form updates
plugin_config = existing_config
# Convert form data to config dict
# Form fields can use dot notation for nested values (e.g., "transition.type")
form_data = request.form.to_dict()
# Meta fields describe the submission, they are not config paths.
# Unknown keys are otherwise written straight into config.json by
# the non-indexed pass below.
form_data = {k: v for k, v in form_data.items()
if not k.startswith('__')}
# First pass: handle bracket notation array fields (e.g., "field_name[]" from checkbox-group)
# These fields use getlist() to preserve all values, then replace in form_data
# Sentinel empty value ("") allows clearing array to [] when all checkboxes unchecked
bracket_array_fields = {} # Maps base field path to list of values
for key in request.form.keys():
# Check if key ends with "[]" (bracket notation for array fields)
if key.endswith('[]'):
base_path = key[:-2] # Remove "[]" suffix
values = request.form.getlist(key)
# Filter out sentinel empty string - if only sentinel present, array should be []
# If sentinel + values present, use the actual values
filtered_values = [v for v in values if v and v.strip()]
# If no non-empty values but key exists, it means all checkboxes unchecked (empty array)
bracket_array_fields[base_path] = filtered_values
# Remove the bracket notation key from form_data if present
if key in form_data:
del form_data[key]
# Process bracket notation fields and set directly in plugin_config
# Use JSON encoding instead of comma-join to handle values containing commas
for base_path, values in bracket_array_fields.items():
# Get schema property to verify it's an array
base_prop = _get_schema_property(schema, base_path)
if base_prop and base_prop.get('type') == 'array':
# Filter out empty values and sentinel empty strings
filtered_values = [v for v in values if v and v.strip()]
# Set directly in plugin_config (values are already strings, no need to parse)
# Empty array (all unchecked) is represented as []
_set_nested_value(plugin_config, base_path, filtered_values)
logger.debug(f"Processed bracket notation array field {base_path}: {values} -> {filtered_values}")
# Remove from form_data to avoid double processing
if base_path in form_data:
del form_data[base_path]
# Second pass: detect and combine array index fields (e.g., "text_color.0", "text_color.1" -> "text_color" as array)
# This handles cases where forms send array fields as indexed inputs
array_fields = {} # Maps base field path to list of (index, value) tuples
processed_keys = set()
indexed_base_paths = set() # Track which base paths have indexed fields
for key, value in form_data.items():
# Check if this looks like an array index field (ends with .0, .1, .2, etc.)
if '.' in key:
parts = key.rsplit('.', 1) # Split on last dot
if len(parts) == 2:
base_path, last_part = parts
# Check if last part is a numeric string (array index)
if last_part.isdigit():
# Get schema property for the base path to verify it's an array
base_prop = _get_schema_property(schema, base_path)
if base_prop and _schema_type_is(base_prop, 'array'):
# This is an array index field
index = int(last_part)
if base_path not in array_fields:
array_fields[base_path] = []
array_fields[base_path].append((index, value))
processed_keys.add(key)
indexed_base_paths.add(base_path)
continue
# Process combined array fields
for base_path, index_values in array_fields.items():
# Sort by index and extract values
index_values.sort(key=lambda x: x[0])
values = [v for _, v in index_values]
# Every channel blank on a nullable field means "unset", not
# an empty array: joining them would produce ", , ", which
# parses to [] and then fails the minItems the array
# declares. This is how a per-mode colour override says
# "inherit the base colour".
base_prop_for_null = _get_schema_property(schema, base_path)
if (_schema_allows_null(base_prop_for_null)
and all(str(v).strip() == '' for v in values)):
_set_nested_value(plugin_config, base_path, None)
continue
# Combine values into comma-separated string for parsing
combined_value = ', '.join(str(v) for v in values)
# Parse as array using schema
parsed_value = _parse_form_value_with_schema(combined_value, base_path, schema)
# Debug logging
logger.debug(f"Combined indexed array field {base_path}: {values} -> {combined_value} -> {parsed_value}")
# Only set if not skipped
if parsed_value is not _SKIP_FIELD:
_set_nested_value(plugin_config, base_path, parsed_value)
# Process remaining (non-indexed) fields
# Skip any base paths that were processed as indexed arrays
for key, value in form_data.items():
if key not in processed_keys:
# Skip if this key is a base path that was processed as indexed array
# (to avoid overwriting the combined array with a single value)
if key not in indexed_base_paths:
# Parse value using schema to determine correct type
parsed_value = _SKIP_FIELD
decoded = False
# A hidden property inside an array row is carried
# through the form JSON-encoded (a posted row replaces
# the stored item, so it must be posted at all). Decode
# it exactly: the generic parse would turn an id "1"
# into the integer 1 and fail validation.
if _hidden_array_item_property(schema, key) is not None:
try:
parsed_value = json.loads(value)
decoded = True
except (TypeError, ValueError):
pass
if not decoded:
parsed_value = _parse_form_value_with_schema(value, key, schema)
# Debug logging for array fields
if schema:
prop = _get_schema_property(schema, key)
if prop and prop.get('type') == 'array':
logger.debug(f"Array field {key}: form value='{value}' -> parsed={parsed_value}")
# Use helper to set nested values correctly (skips if _SKIP_FIELD)
if parsed_value is not _SKIP_FIELD:
_set_nested_value(plugin_config, key, parsed_value)
# Before the booleans below: that walk replaces anything it
# expects to be a list and finds is not one.
if schema and 'properties' in schema:
coerce_array_shapes(plugin_config, schema['properties'],
short_lists_take_default=True)
# Fix unchecked boolean checkboxes: HTML checkboxes don't submit values
# when unchecked, so the existing config value (potentially True) persists.
# Walk the schema and set any boolean fields missing from form data to False.
if schema and 'properties' in schema:
form_keys = set(request.form.keys())
# The rendered form reports which top-level sections it drew, so
# an unchecked box can be told apart from a field the caller
# never had in front of it. A caller that sends none gets the
# evidence-based fallback in _boolean_is_in_scope.
rendered_sections = set(request.form.getlist(_RENDERED_SECTION_FIELD))
_set_missing_booleans_to_false(
plugin_config, schema['properties'], form_keys,
sections=rendered_sections or None)
# Get schema manager instance (for JSON requests)
schema_mgr = api_v3.schema_manager
if not schema_mgr:
return error_response(
ErrorCode.SYSTEM_ERROR,
'Schema manager not initialized',
status_code=500
)
# Load plugin schema using SchemaManager (force refresh to get latest schema)
# For JSON requests, schema wasn't loaded yet
if 'application/json' in content_type:
schema = schema_mgr.load_schema(plugin_id, use_cache=False)
regular_config, secrets_config, error = _prepare_plugin_config_for_save(
plugin_id, plugin_config, schema, schema_mgr,
is_json='application/json' in content_type)
if error:
return error
# Get current configs
current_config = api_v3.config_manager.load_config()
current_secrets = api_v3.config_manager.get_raw_file_content('secrets')
# Deep merge plugin configuration in main config (preserves nested structures)
if plugin_id not in current_config:
current_config[plugin_id] = {}
# Retired core keys (skin, skin_options) leave the stored section here
from src.plugin_system.schema_manager import drop_retired_plugin_keys
current_config[plugin_id] = deep_merge(
drop_retired_plugin_keys(current_config[plugin_id], schema), regular_config)
# Deep merge plugin secrets in secrets config
if secrets_config:
if plugin_id not in current_secrets:
current_secrets[plugin_id] = {}
# See above -- secrets lists must merge element-wise.
current_secrets[plugin_id] = merge_secrets(
current_secrets[plugin_id], secrets_config)
# Save secrets file
try:
api_v3.config_manager.save_raw_file_content('secrets', current_secrets)
except PermissionError as e:
secrets_path = api_v3.config_manager.secrets_path
secrets_dir = os.path.dirname(secrets_path) if secrets_path else None
# Check permissions
dir_readable = os.access(secrets_dir, os.R_OK) if secrets_dir and os.path.exists(secrets_dir) else False
dir_writable = os.access(secrets_dir, os.W_OK) if secrets_dir and os.path.exists(secrets_dir) else False
file_writable = os.access(secrets_path, os.W_OK) if secrets_path and os.path.exists(secrets_path) else False
logger.error(
f"Permission error saving secrets config for {plugin_id}: {e}\n"
f"Secrets path: {secrets_path}\n"
f"Directory readable: {dir_readable}, writable: {dir_writable}\n"
f"File writable: {file_writable}",
exc_info=True
)
return error_response(
ErrorCode.CONFIG_SAVE_FAILED,
f"Failed to save secrets configuration: Permission denied. Check file permissions on {secrets_path}",
status_code=500
)
except Exception:
secrets_path = api_v3.config_manager.secrets_path
logger.error("Error saving secrets config for %s (path=%s)", plugin_id, secrets_path, exc_info=True)
return error_response(
ErrorCode.CONFIG_SAVE_FAILED,
"Failed to save secrets configuration; see logs for details",
status_code=500
)
# Save the updated main config using atomic save
success, error_msg = _pkg._save_config_atomic(api_v3.config_manager, current_config, create_backup=True)
if not success:
return error_response(
ErrorCode.CONFIG_SAVE_FAILED,
f"Failed to save configuration: {error_msg}",
status_code=500
)
# If the plugin is loaded, notify it of the config change with merged config
try:
if api_v3.plugin_manager:
plugin_instance = api_v3.plugin_manager.get_plugin(plugin_id)
if plugin_instance:
# Reload merged config (includes secrets) and pass the plugin-specific section
merged_config = api_v3.config_manager.load_config()
plugin_full_config = _pkg._prepared_plugin_config(
plugin_id, merged_config.get(plugin_id, {}))
if hasattr(plugin_instance, 'on_config_change'):
plugin_instance.on_config_change(plugin_full_config)
# Update plugin state manager and call lifecycle methods based on enabled state
# This ensures the plugin state is synchronized with the config
enabled = plugin_full_config.get('enabled', plugin_instance.enabled)
# Update state manager if available
if api_v3.plugin_state_manager:
api_v3.plugin_state_manager.set_plugin_enabled(plugin_id, enabled)
# Call lifecycle methods to ensure plugin state matches config
try:
if enabled:
if hasattr(plugin_instance, 'on_enable'):
plugin_instance.on_enable()
else:
if hasattr(plugin_instance, 'on_disable'):
plugin_instance.on_disable()
except Exception as lifecycle_error:
# Log the error but don't fail the save - config is already saved
logger.warning("Lifecycle method error for %s: %s", plugin_id, lifecycle_error, exc_info=True)
except Exception as hook_err:
# Do not fail the save if hook fails; just log
logger.warning("on_config_change failed: %s", hook_err)
secret_count = len(secrets_config)
message = f'Plugin {plugin_id} configuration saved successfully'
if secret_count > 0:
message += f' ({secret_count} secret field(s) saved to config_secrets.json)'
return success_response(message=message)
except Exception as e:
if api_v3.operation_history:
api_v3.operation_history.record_operation(
"configure",
plugin_id=data.get('plugin_id') if 'data' in locals() else None,
status="failed",
error=str(e)
)
return exception_error_response(e, ErrorCode.CONFIG_SAVE_FAILED)
def _merge_onto_stored_plugin_config(plugin_id, submitted_config, current_config=None):
"""A JSON plugin-config body merged onto the plugin's stored section.
A JSON body carries the settings being changed, as a form post does: the
rest keep their stored values. Built from defaults alone, a body such as
``{"enabled": true}`` reset every unsent setting of the plugin.
``current_config`` is the loaded main config when the caller has one.
"""
import copy
if current_config is None:
current_config = api_v3.config_manager.load_config()
stored = current_config.get(plugin_id)
base = copy.deepcopy(stored) if isinstance(stored, dict) else {}
return deep_merge(base, submitted_config)
def _prepare_plugin_config_for_save(plugin_id, plugin_config, schema, schema_mgr, is_json):
"""Turn a submitted plugin config into what gets stored.
Fixes array shapes, keeps the enabled state, reads legacy booleans and
applies schema defaults, normalizes value types, filters to the schema plus
the core-owned per-plugin properties, validates, and splits out secrets.
Shared by POST /plugins/config and plugin sections posted to
POST /config/main, so both store the same thing.
Returns ``(regular_config, secrets_config, None)``, or
``(None, None, error_response)`` when validation fails.
"""
# The form path has already done this, before its checkbox pass.
if is_json and schema and 'properties' in schema:
coerce_array_shapes(plugin_config, schema['properties'])
# PRE-PROCESSING: Preserve 'enabled' state if not in request
# This prevents overwriting the enabled state when saving config from a form that doesn't include the toggle
if 'enabled' not in plugin_config:
try:
current_config = api_v3.config_manager.load_config()
if plugin_id in current_config and 'enabled' in current_config[plugin_id]:
plugin_config['enabled'] = current_config[plugin_id]['enabled']
elif api_v3.plugin_manager:
# Fallback to plugin instance if config doesn't have it
plugin_instance = api_v3.plugin_manager.get_plugin(plugin_id)
if plugin_instance:
plugin_config['enabled'] = plugin_instance.enabled
# Final fallback: default to True if plugin is loaded (matches BasePlugin default)
if 'enabled' not in plugin_config:
plugin_config['enabled'] = True
except Exception as e:
logger.debug("Error preserving enabled state: %s", e)
# Default to True on error to avoid disabling plugins
plugin_config['enabled'] = True
# Find secret fields (supports nested schemas and array-item secrets)
secret_fields = set()
if schema and 'properties' in schema:
secret_fields = find_secret_fields(schema['properties'])
# Apply defaults from schema to config BEFORE validation
# This ensures required fields with defaults are present before validation
# Store preserved enabled value before merge to protect it from defaults
preserved_enabled = None
if 'enabled' in plugin_config:
preserved_enabled = plugin_config['enabled']
if schema:
# Legacy booleans read as {"enabled": ...} objects first (#588), the
# same preparation loading applies, so a config the device reads
# without complaint also saves.
from src.plugin_system.schema_manager import prepare_plugin_config
defaults = schema_mgr.generate_default_config(plugin_id, use_cache=True)
plugin_config = prepare_plugin_config(plugin_config, schema, defaults)
# The defaults merge replaces a None only where the schema has a default,
# so an array the client sent as None, or left out, can still be one here.
def _fix_none_arrays(cfg, props):
for k, pschema in props.items():
if pschema.get('type') == 'array':
if isinstance(cfg, dict) and (k not in cfg or cfg[k] is None):
cfg[k] = pschema.get('default', [])
elif pschema.get('type') == 'object' and 'properties' in pschema:
if isinstance(cfg, dict) and isinstance(cfg.get(k), dict):
_fix_none_arrays(cfg[k], pschema['properties'])
if schema and 'properties' in schema and isinstance(plugin_config, dict):
_fix_none_arrays(plugin_config, schema['properties'])
# Ensure enabled state is preserved after defaults merge
# Defaults should not overwrite an explicitly preserved enabled value
if preserved_enabled is not None:
# Restore preserved value if it was changed by defaults merge
if plugin_config.get('enabled') != preserved_enabled:
plugin_config['enabled'] = preserved_enabled
# Normalize config data: convert string numbers to integers/floats where schema expects numbers
# This handles form data which sends everything as strings
def normalize_config_values(config, schema_props, prefix=''):
"""Recursively normalize config values based on schema types"""
if not isinstance(config, dict) or not isinstance(schema_props, dict):
return config
normalized = {}
for key, value in config.items():
field_path = f"{prefix}.{key}" if prefix else key
if key not in schema_props:
# Field not in schema, keep as-is (will be caught by additionalProperties check if needed)
normalized[key] = value
continue
prop_schema = schema_props[key]
prop_type = prop_schema.get('type')
# Handle union types (e.g., ["integer", "null"])
if isinstance(prop_type, list):
# Check if null is allowed and value is empty/null
if 'null' in prop_type:
# Handle various representations of null/empty
if value is None:
normalized[key] = None
continue
elif isinstance(value, str):
# Strip whitespace and check for null representations
value_stripped = value.strip()
if value_stripped == '' or value_stripped.lower() in ('null', 'none', 'undefined'):
normalized[key] = None
continue
# Try to normalize based on non-null types in the union
# Check integer first (more specific than number)
if 'integer' in prop_type:
if isinstance(value, str):
try:
normalized[key] = int(value.strip())
continue
except (ValueError, TypeError, OverflowError):
pass
elif isinstance(value, (int, float)):
normalized[key] = int(value)
continue
# Check number (less specific, but handles floats)
if 'number' in prop_type:
if isinstance(value, str):
try:
normalized[key] = float(value.strip())
continue
except (ValueError, TypeError, OverflowError):
pass
elif isinstance(value, (int, float)):
normalized[key] = float(value)
continue
# Check boolean
if 'boolean' in prop_type:
if isinstance(value, str):
normalized[key] = value.strip().lower() in ('true', '1', 'on', 'yes')
continue
# The scalar conversions above are the only ones this branch
# knows, so a union naming a structural or string type fell
# through here even when the value already matched it --
# customization.modes.<mode> makes every override nullable
# (see element_style._nullable), so every per-mode colour is
# ['array', 'null'] and warned on a perfectly valid [r, g, b].
# Worse than the noise: `continue` skipped the single-type
# handling below, so a nullable array never had its items
# normalized and form-posted ["0", "249", "0"] stayed strings
# where a plain 'array' field would have become ints. Re-enter
# that handling with the matched member instead.
if isinstance(value, list) and 'array' in prop_type:
prop_type = 'array'
elif isinstance(value, dict) and 'object' in prop_type:
prop_type = 'object'
elif isinstance(value, str) and 'string' in prop_type:
normalized[key] = value
continue
else:
# Nothing converted: keep the value for validation to report.
logger.warning(f"Could not normalize field {field_path}: value={repr(value)}, type={type(value)}, schema_type={prop_type}")
normalized[key] = value
continue
if isinstance(value, dict) and prop_type == 'object' and 'properties' in prop_schema:
# Recursively normalize nested objects
normalized[key] = normalize_config_values(value, prop_schema['properties'], field_path)
elif isinstance(value, list) and prop_type == 'array' and 'items' in prop_schema:
# Normalize array items
items_schema = prop_schema['items']
item_type = items_schema.get('type')
# Handle union types in array items
if isinstance(item_type, list):
normalized_array = []
for v in value:
# Check if null is allowed
if 'null' in item_type:
if v is None or v == '' or (isinstance(v, str) and v.lower() in ('null', 'none')):
normalized_array.append(None)
continue
# Try to normalize based on non-null types
if 'integer' in item_type:
if isinstance(v, str):
try:
normalized_array.append(int(v))
continue
except (ValueError, TypeError, OverflowError):
pass
elif isinstance(v, (int, float)):
# Only a genuinely integral value converts.
# int(2.5) == 2 would store a silently
# corrected number where the client sent a
# wrong one; leaving it lets the validator
# reject it. A whole float (2.0 out of JSON)
# is integral and still converts.
if isinstance(v, int) or float(v).is_integer():
normalized_array.append(int(v))
else:
normalized_array.append(v)
continue
elif 'number' in item_type:
if isinstance(v, str):
try:
normalized_array.append(float(v))
continue
except (ValueError, TypeError, OverflowError):
pass
elif isinstance(v, (int, float)):
normalized_array.append(float(v))
continue
# If no conversion worked, keep original value
normalized_array.append(v)
normalized[key] = normalized_array
elif item_type == 'integer':
# Convert string numbers to integers
normalized_array = []
for v in value:
if isinstance(v, str):
try:
normalized_array.append(int(v))
except (ValueError, TypeError, OverflowError):
normalized_array.append(v)
elif isinstance(v, (int, float)):
# Integral only -- see the union branch above.
if isinstance(v, int) or float(v).is_integer():
normalized_array.append(int(v))
else:
normalized_array.append(v)
else:
normalized_array.append(v)
normalized[key] = normalized_array
elif item_type == 'number':
# Convert string numbers to floats
normalized_array = []
for v in value:
if isinstance(v, str):
try:
normalized_array.append(float(v))
except (ValueError, TypeError, OverflowError):
normalized_array.append(v)
else:
normalized_array.append(v)
normalized[key] = normalized_array
elif item_type == 'object' and 'properties' in items_schema:
# Recursively normalize array of objects
normalized_array = []
for v in value:
if isinstance(v, dict):
normalized_array.append(
normalize_config_values(v, items_schema['properties'], f"{field_path}[]")
)
else:
normalized_array.append(v)
normalized[key] = normalized_array
else:
normalized[key] = value
elif prop_type == 'integer':
# Convert string to integer
if isinstance(value, str):
try:
normalized[key] = int(value)
except (ValueError, TypeError, OverflowError):
normalized[key] = value
else:
normalized[key] = value
elif prop_type == 'number':
# Convert string to float
if isinstance(value, str):
try:
normalized[key] = float(value)
except (ValueError, TypeError, OverflowError):
normalized[key] = value
else:
normalized[key] = value
elif prop_type == 'boolean':
# Convert string booleans
if isinstance(value, str):
normalized[key] = value.lower() in ('true', '1', 'on', 'yes')
else:
normalized[key] = value
else:
normalized[key] = value
return normalized
# Normalize config before validation
if schema and 'properties' in schema:
plugin_config = normalize_config_values(plugin_config, schema['properties'])
# Filter config to only include schema-defined fields (important when additionalProperties is false)
# Use enhanced schema with core properties to ensure core properties are preserved during filtering
if schema and 'properties' in schema:
enhanced_schema_for_filtering = _enhance_schema_with_core_properties(schema)
plugin_config = _filter_config_by_schema(plugin_config, enhanced_schema_for_filtering)
# A uniqueItems array can arrive with a repeat -- the form merges onto
# the stored list, so a stock symbol already saved and submitted again
# appears twice -- and validation would refuse the whole save for it.
if schema:
dedup_unique_arrays(plugin_config, schema)
if schema:
is_valid, validation_errors = schema_mgr.validate_config_against_schema(
plugin_config, schema, plugin_id
)
if not is_valid:
# Schema keys including the injected core properties, for the error
enhanced_schema = _enhance_schema_with_core_properties(schema)
# Keys, never values: plugin_config still holds the submitted
# secrets here (separate_secrets runs below), and logging it wrote
# live credentials to the journal.
logger.warning("Config validation failed for %s: %s (config keys: %s)",
plugin_id, validation_errors, list(plugin_config.keys()))
return None, None, error_response(
ErrorCode.CONFIG_VALIDATION_FAILED,
'Configuration validation failed',
details='; '.join(validation_errors) if validation_errors else 'Unknown validation error',
context={
'plugin_id': plugin_id,
'validation_errors': validation_errors,
'config_keys': list(plugin_config.keys()),
'schema_keys': list(enhanced_schema.get('properties', {}).keys())
},
suggested_fixes=[
'Review validation errors above',
'Check config against schema',
'Verify all required fields are present'
],
status_code=400
)
# Separate secrets from regular config (handles nested configs and
# array-item secrets — see src/web_interface/secret_helpers.py)
regular_config, secrets_config = separate_secrets(plugin_config, secret_fields)
# The config form renders secrets masked, so every save posts
# them back blank. Without this the blank is merged over the
# stored value and the credential is destroyed by the act of
# changing an unrelated setting. A blank means "unchanged".
secrets_config = remove_empty_secrets(secrets_config)
return regular_config, secrets_config, None
@api_v3.route('/plugins/schema', methods=['GET'])
def get_plugin_schema():
"""Get plugin configuration schema"""
plugin_id = request.args.get('plugin_id')
if not plugin_id:
return jsonify({'status': 'error', 'message': 'plugin_id required'}), 400
# Get schema manager instance
schema_mgr = api_v3.schema_manager
if not schema_mgr:
return jsonify({'status': 'error', 'message': 'Schema manager not initialized'}), 500
# Load schema using SchemaManager (uses caching)
schema = schema_mgr.load_schema(plugin_id, use_cache=True)
if schema:
return jsonify({'status': 'success', 'data': {'schema': schema}})
# Return a simple default schema if file not found
default_schema = {
'type': 'object',
'properties': {
'enabled': {
'type': 'boolean',
'title': 'Enable Plugin',
'description': 'Enable or disable this plugin',
'default': True
},
'display_duration': {
'type': 'integer',
'title': 'Display Duration',
'description': 'How long to show content (seconds)',
'minimum': 5,
'maximum': 300,
'default': 30
}
}
}
return jsonify({'status': 'success', 'data': {'schema': default_schema}})
@api_v3.route('/plugins/config/reset', methods=['POST'])
def reset_plugin_config():
"""Reset plugin configuration to schema defaults"""
if not api_v3.config_manager:
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
data = request.get_json(silent=True) or {}
plugin_id = data.get('plugin_id')
preserve_secrets = data.get('preserve_secrets', True)
if not plugin_id:
return jsonify({'status': 'error', 'message': 'plugin_id required'}), 400
id_error = _non_plugin_id_error(plugin_id)
if id_error:
return id_error
# Get schema manager instance
schema_mgr = api_v3.schema_manager
if not schema_mgr:
return jsonify({'status': 'error', 'message': 'Schema manager not initialized'}), 500
# Generate defaults from schema
defaults = schema_mgr.generate_default_config(plugin_id, use_cache=True)
# Get current configs
current_config = api_v3.config_manager.load_config()
current_secrets = api_v3.config_manager.get_raw_file_content('secrets')
# Load schema to identify secret fields
schema = schema_mgr.load_schema(plugin_id, use_cache=True)
secret_fields = set()
if schema and 'properties' in schema:
secret_fields = find_secret_fields(schema['properties'])
# Separate defaults into regular and secret configs
default_regular, default_secrets = separate_secrets(defaults, secret_fields)
# Update main config with defaults
current_config[plugin_id] = default_regular
# Update secrets config (preserve existing secrets if preserve_secrets=True)
if preserve_secrets:
# Keep existing secrets for this plugin
if plugin_id in current_secrets:
# Merge defaults with existing secrets
existing_secrets = current_secrets[plugin_id]
for key, value in default_secrets.items():
if key not in existing_secrets or not existing_secrets[key]:
existing_secrets[key] = value
else:
current_secrets[plugin_id] = default_secrets
else:
# Replace all secrets with defaults
current_secrets[plugin_id] = default_secrets
success, error_msg = _pkg._save_config_atomic(api_v3.config_manager, current_config, create_backup=True)
if not success:
return error_response(
ErrorCode.CONFIG_SAVE_FAILED,
f"Failed to save configuration: {error_msg}",
status_code=500
)
if default_secrets or not preserve_secrets:
api_v3.config_manager.save_raw_file_content('secrets', current_secrets)
# Notify plugin of config change if loaded
try:
if api_v3.plugin_manager:
plugin_instance = api_v3.plugin_manager.get_plugin(plugin_id)
if plugin_instance:
merged_config = api_v3.config_manager.load_config()
plugin_full_config = _pkg._prepared_plugin_config(
plugin_id, merged_config.get(plugin_id, {}))
if hasattr(plugin_instance, 'on_config_change'):
plugin_instance.on_config_change(plugin_full_config)
except Exception as hook_err:
logger.warning("on_config_change failed: %s", hook_err)
return jsonify({
'status': 'success',
'message': f'Plugin {plugin_id} configuration reset to defaults',
'data': {'config': defaults}
})
@api_v3.route('/plugins/action', methods=['POST'])
def execute_plugin_action():
"""Execute a plugin-defined action (e.g., authentication)"""
try:
# Try to get JSON data, with better error handling
try:
data = request.get_json(force=True) or {}
except Exception as e:
# The module logger, not a local one: binding `logger` anywhere in
# this function made every other `logger.error` here raise
# UnboundLocalError, so the step-1 handler below reported that
# instead of the plugin script's real failure.
logger.error(f"Error parsing JSON in execute_plugin_action: {e}")
return jsonify({
'status': 'error',
'message': 'Invalid JSON in request body',
'content_type': request.content_type }), 400
plugin_id = data.get('plugin_id')
action_id = data.get('action_id')
action_params = data.get('params', {})
if not plugin_id or not action_id:
return jsonify({
'status': 'error',
'message': 'plugin_id and action_id required',
'received': {'plugin_id': plugin_id, 'action_id': action_id, 'has_params': bool(action_params)}
}), 400
# plugin_id comes from the request body and picks the directory whose
# manifest names the script to run, so it must be a plain name.
plugin_id = safe_path_component(plugin_id)
if plugin_id is None:
return jsonify({'status': 'error', 'message': 'Invalid plugin_id'}), 400
plugin_dir = _plugin_directory(plugin_id)
if not plugin_dir:
return jsonify({'status': 'error', 'message': 'Plugin not found'}), 404
# Load manifest to get action definition
manifest_path = Path(plugin_dir) / 'manifest.json'
if not manifest_path.exists():
return jsonify({'status': 'error', 'message': 'Plugin manifest not found'}), 404
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
web_ui_actions = manifest.get('web_ui_actions', [])
action_def = None
for action in web_ui_actions:
if action.get('id') == action_id:
action_def = action
break
if not action_def:
return jsonify({'status': 'error', 'message': f'Action {action_id} not found in plugin manifest'}), 404
# Set LEDMATRIX_ROOT environment variable
env = os.environ.copy()
env['LEDMATRIX_ROOT'] = str(PROJECT_ROOT)
# Execute action based on type
action_type = action_def.get('type', 'script')
if action_type == 'script':
# Execute a Python script
script_path = action_def.get('script')
if not script_path:
return jsonify({'status': 'error', 'message': 'Script path not defined for action'}), 400
script_file = Path(plugin_dir) / script_path
if not script_file.exists():
return jsonify({'status': 'error', 'message': f'Script not found: {script_path}'}), 404
# Handle multi-step actions (like Spotify OAuth)
step = action_params.get('step')
if step == '2' and action_params.get('redirect_url'):
# Step 2: Complete authentication with redirect URL
redirect_url = action_params.get('redirect_url')
import tempfile
import json as json_lib
redirect_url_escaped = json_lib.dumps(redirect_url)
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as wrapper:
wrapper.write(f'''import sys
import subprocess
import os
# Set LEDMATRIX_ROOT
os.environ['LEDMATRIX_ROOT'] = r"{PROJECT_ROOT}"
# Run the script and provide redirect URL
proc = subprocess.Popen(
[sys.executable, r"{script_file}"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
env=os.environ
)
# Send redirect URL to stdin
redirect_url = {redirect_url_escaped}
stdout, _ = proc.communicate(input=redirect_url + "\\n", timeout=120)
print(stdout)
sys.exit(proc.returncode)
''')
wrapper_path = wrapper.name
try:
result = subprocess.run(
['python3', wrapper_path],
capture_output=True,
text=True,
timeout=120,
env=env
)
os.unlink(wrapper_path)
if result.returncode == 0:
return jsonify({
'status': 'success',
'message': action_def.get('success_message', 'Action completed successfully'),
'output': result.stdout
})
else:
return jsonify({
'status': 'error',
'message': action_def.get('error_message', 'Action failed'),
'output': result.stdout + result.stderr
}), 400
except subprocess.TimeoutExpired:
if os.path.exists(wrapper_path):
os.unlink(wrapper_path)
return jsonify({'status': 'error', 'message': 'Action timed out'}), 408
else:
# Regular script execution - pass params via stdin if provided
if action_params:
# Pass params as JSON via stdin
import tempfile
import json as json_lib
params_json = json_lib.dumps(action_params)
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as wrapper:
wrapper.write(f'''import sys
import subprocess
import os
import json
# Set LEDMATRIX_ROOT
os.environ['LEDMATRIX_ROOT'] = r"{PROJECT_ROOT}"
# Run the script and provide params as JSON via stdin
proc = subprocess.Popen(
[sys.executable, r"{script_file}"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
env=os.environ
)
# Send params as JSON to stdin
params = {params_json}
stdout, _ = proc.communicate(input=json.dumps(params), timeout=120)
print(stdout)
sys.exit(proc.returncode)
''')
wrapper_path = wrapper.name
try:
result = subprocess.run(
['python3', wrapper_path],
capture_output=True,
text=True,
timeout=120,
env=env
)
os.unlink(wrapper_path)
# Try to parse output as JSON
try:
output_data = json.loads(result.stdout)
if result.returncode == 0:
return jsonify(output_data)
else:
return jsonify({
'status': 'error',
'message': output_data.get('message', action_def.get('error_message', 'Action failed')),
'output': result.stdout + result.stderr
}), 400
except json.JSONDecodeError:
# Output is not JSON, return as text
if result.returncode == 0:
return jsonify({
'status': 'success',
'message': action_def.get('success_message', 'Action completed successfully'),
'output': result.stdout
})
else:
return jsonify({
'status': 'error',
'message': action_def.get('error_message', 'Action failed'),
'output': result.stdout + result.stderr
}), 400
except subprocess.TimeoutExpired:
if os.path.exists(wrapper_path):
os.unlink(wrapper_path)
return jsonify({'status': 'error', 'message': 'Action timed out'}), 408
else:
# No params - check for OAuth flow first, then run script normally
# Step 1: Get initial data (like auth URL)
# For OAuth flows, we might need to import the script as a module
if action_def.get('oauth_flow'):
# Import script as module to get auth URL
import sys
import importlib.util
spec = importlib.util.spec_from_file_location("plugin_action", script_file)
action_module = importlib.util.module_from_spec(spec)
sys.modules["plugin_action"] = action_module
try:
spec.loader.exec_module(action_module)
# Try to get auth URL using common patterns
auth_url = None
if hasattr(action_module, 'get_auth_url'):
auth_url = action_module.get_auth_url()
elif hasattr(action_module, 'load_spotify_credentials'):
# Spotify-specific pattern
client_id, client_secret, redirect_uri = action_module.load_spotify_credentials()
if all([client_id, client_secret, redirect_uri]):
from spotipy.oauth2 import SpotifyOAuth
sp_oauth = SpotifyOAuth(
client_id=client_id,
client_secret=client_secret,
redirect_uri=redirect_uri,
scope=getattr(action_module, 'SCOPE', ''),
cache_path=getattr(action_module, 'SPOTIFY_AUTH_CACHE_PATH', None),
open_browser=False
)
auth_url = sp_oauth.get_authorize_url()
if auth_url:
return jsonify({
'status': 'success',
'message': action_def.get('step1_message', 'Authorization URL generated'),
'auth_url': auth_url,
'requires_step2': True
})
else:
return jsonify({
'status': 'error',
'message': 'Could not generate authorization URL'
}), 400
except Exception as e:
# Not a copy of the blueprint handler: without it, a
# TimeoutExpired from the plugin's script would reach
# this route's own `except subprocess.TimeoutExpired`
# and be answered as a 408 "Action timed out".
logger.error("Error executing action step 1", exc_info=True)
return jsonify({
'status': 'error',
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
}), 500
else:
# Simple script execution
result = subprocess.run(
['python3', str(script_file)],
capture_output=True,
text=True,
timeout=60,
env=env
)
# Try to parse output as JSON
try:
import json as json_module
output_data = json_module.loads(result.stdout)
if result.returncode == 0:
return jsonify(output_data)
else:
return jsonify({
'status': 'error',
'message': output_data.get('message', action_def.get('error_message', 'Action failed')),
'output': result.stdout + result.stderr
}), 400
except json.JSONDecodeError:
# Output is not JSON, return as text
if result.returncode == 0:
return jsonify({
'status': 'success',
'message': action_def.get('success_message', 'Action completed successfully'),
'output': result.stdout
})
else:
return jsonify({
'status': 'error',
'message': action_def.get('error_message', 'Action failed'),
'output': result.stdout + result.stderr
}), 400
elif action_type == 'endpoint':
# Call a plugin-defined HTTP endpoint (future feature)
return jsonify({'status': 'error', 'message': 'Endpoint actions not yet implemented'}), 501
else:
return jsonify({'status': 'error', 'message': f'Unknown action type: {action_type}'}), 400
except subprocess.TimeoutExpired:
return jsonify({'status': 'error', 'message': 'Action timed out'}), 408
def _plugin_uploads_dir(plugin_id):
"""assets/plugins/<plugin_id>/uploads for a request-supplied id, or None.
Same guard as the serving route in app.py: one plain path segment,
resolved and contained under assets/plugins.
"""
return resolve_under(PROJECT_ROOT / 'assets' / 'plugins', plugin_id, 'uploads')
@api_v3.route('/plugins/assets/upload', methods=['POST'])
def upload_plugin_asset():
"""Upload asset files for a plugin"""
plugin_id = request.form.get('plugin_id')
if not plugin_id:
return jsonify({'status': 'error', 'message': 'plugin_id is required'}), 400
if 'files' not in request.files:
return jsonify({'status': 'error', 'message': 'No files provided'}), 400
files = request.files.getlist('files')
if not files or all(not f.filename for f in files):
return jsonify({'status': 'error', 'message': 'No files provided'}), 400
# Validate file count
if len(files) > 10:
return jsonify({'status': 'error', 'message': 'Maximum 10 files per upload'}), 400
# Setup plugin assets directory. plugin_id is a form field: without
# the guard '../../config' created, listed and wrote into directories
# outside assets/plugins (the serving route was fixed in #561).
assets_dir = _plugin_uploads_dir(plugin_id)
if assets_dir is None:
return jsonify({'status': 'error', 'message': 'Invalid plugin_id'}), 400
plugin_id = safe_path_component(plugin_id)
assets_dir.mkdir(parents=True, exist_ok=True)
# Load metadata file
metadata_file = assets_dir / '.metadata.json'
if metadata_file.exists():
with open(metadata_file, 'r') as f:
metadata = json.load(f)
else:
metadata = {}
uploaded_files = []
total_size = 0
max_size_per_file = 5 * 1024 * 1024 # 5MB
max_total_size = 50 * 1024 * 1024 # 50MB
# Calculate current total size
for entry in metadata.values():
if 'size' in entry:
total_size += entry.get('size', 0)
for file in files:
if not file.filename:
continue
# Validate file type
allowed_extensions = ['.png', '.jpg', '.jpeg', '.bmp', '.gif']
file_ext = '.' + file.filename.lower().split('.')[-1]
if file_ext not in allowed_extensions:
return jsonify({
'status': 'error',
'message': f'Invalid file type: {file_ext}. Allowed: {allowed_extensions}'
}), 400
# Read file to check size and validate
file.seek(0, os.SEEK_END)
file_size = file.tell()
file.seek(0)
if file_size > max_size_per_file:
return jsonify({
'status': 'error',
'message': f'File {file.filename} exceeds 5MB limit'
}), 400
if total_size + file_size > max_total_size:
return jsonify({
'status': 'error',
'message': 'Upload would exceed 50MB total storage limit'
}), 400
# Validate file is actually an image (check magic bytes)
file_content = file.read(8)
file.seek(0)
is_valid_image = False
if file_content.startswith(b'\x89PNG\r\n\x1a\n'): # PNG
is_valid_image = True
elif file_content[:2] == b'\xff\xd8': # JPEG
is_valid_image = True
elif file_content[:2] == b'BM': # BMP
is_valid_image = True
elif file_content[:6] in [b'GIF87a', b'GIF89a']: # GIF
is_valid_image = True
if not is_valid_image:
return jsonify({
'status': 'error',
'message': f'File {file.filename} is not a valid image file'
}), 400
# Generate unique filename
timestamp = int(_pkg.time.time())
file_hash = hashlib.md5(file_content + file.filename.encode()).hexdigest()[:8]
safe_filename = f"image_{timestamp}_{file_hash}{file_ext}"
file_path = assets_dir / safe_filename
# Ensure filename is unique
counter = 1
while file_path.exists():
safe_filename = f"image_{timestamp}_{file_hash}_{counter}{file_ext}"
file_path = assets_dir / safe_filename
counter += 1
# Save file
file.save(str(file_path))
# Make file readable
os.chmod(file_path, 0o644)
# Generate unique ID
image_id = str(uuid.uuid4())
# Store metadata
relative_path = f"assets/plugins/{plugin_id}/uploads/{safe_filename}"
metadata[image_id] = {
'id': image_id,
'filename': safe_filename,
'path': relative_path,
'size': file_size,
'uploaded_at': datetime.utcnow().isoformat() + 'Z',
'original_filename': file.filename
}
uploaded_files.append({
'id': image_id,
'filename': safe_filename,
'path': relative_path,
'size': file_size,
'uploaded_at': metadata[image_id]['uploaded_at']
})
total_size += file_size
# Save metadata
with open(metadata_file, 'w') as f:
json.dump(metadata, f, indent=2)
return jsonify({
'status': 'success',
'uploaded_files': uploaded_files,
'total_files': len(metadata)
})
@api_v3.route('/plugins/<plugin_id>/static/<path:file_path>', methods=['GET'])
def serve_plugin_static(plugin_id, file_path):
"""Serve static files from plugin directory.
Both URL parts are validated before anything is opened. This handler used
to read whatever the path resolved to as long as ``str(file).startswith``
the plugin directory, which let two different things through:
* ``plugin_id`` of ``..`` -- Flask's default converter forbids a slash but
not dots, and ``get_plugin_directory('..')`` happily returned the parent
of the plugins directory because it exists. Every file under the project
root then "started with" that directory, ``config/config_secrets.json``
included.
* a sibling directory sharing a prefix: with the plugin directory
``plugin-repos/foo``, ``../foo-evil/x`` resolves to
``plugin-repos/foo-evil/x``, whose string does start with
``plugin-repos/foo``.
"""
safe_plugin_id = safe_path_component(plugin_id)
if not safe_plugin_id:
return jsonify({'status': 'error', 'message': 'Invalid plugin ID'}), 400
safe_parts = safe_relative_parts(file_path)
if not safe_parts:
return jsonify({'status': 'error', 'message': 'Invalid file path'}), 400
plugin_dir = _plugin_directory(safe_plugin_id)
if not plugin_dir:
return jsonify({'status': 'error', 'message': 'Plugin not found'}), 404
# Containment is still checked after resolving: name validation cannot
# see a symlink inside the plugin directory that points out of it.
requested_file = resolve_under(plugin_dir, *safe_parts)
if requested_file is None:
return jsonify({'status': 'error', 'message': 'Invalid file path'}), 403
# Check if file exists
if not requested_file.exists() or not requested_file.is_file():
return jsonify({'status': 'error', 'message': 'File not found'}), 404
# Determine content type
content_type = 'text/plain'
name = requested_file.name
if name.endswith('.html'):
content_type = 'text/html'
elif name.endswith('.js'):
content_type = 'application/javascript'
elif name.endswith('.css'):
content_type = 'text/css'
elif name.endswith('.json'):
content_type = 'application/json'
# Read and return file
with open(requested_file, 'r', encoding='utf-8') as f:
content = f.read()
return Response(content, mimetype=content_type)
@api_v3.route('/plugins/calendar/upload-credentials', methods=['POST'])
def upload_calendar_credentials():
"""Upload credentials.json file for calendar plugin"""
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file provided'}), 400
file = request.files['file']
if not file or not file.filename:
return jsonify({'status': 'error', 'message': 'No file provided'}), 400
# Validate file extension
if not file.filename.lower().endswith('.json'):
return jsonify({'status': 'error', 'message': 'File must be a JSON file (.json)'}), 400
# Validate file size (max 1MB for credentials)
file.seek(0, os.SEEK_END)
file_size = file.tell()
file.seek(0)
if file_size > 1024 * 1024: # 1MB
return jsonify({'status': 'error', 'message': 'File exceeds 1MB limit'}), 400
# Validate it's valid JSON
try:
file_content = file.read()
file.seek(0)
creds_data = json.loads(file_content)
except json.JSONDecodeError:
return jsonify({'status': 'error', 'message': 'File is not valid JSON'}), 400
# Validate it looks like Google OAuth credentials. A bare scalar, a
# list, true/null — all valid JSON, none of them credentials. Reject
# rather than save: a file written as credentials.json but unusable
# as credentials only fails later, somewhere less obvious.
if not isinstance(creds_data, dict) or not (
'installed' in creds_data or 'web' in creds_data):
return jsonify({
'status': 'error',
'message': 'File does not appear to be a valid Google OAuth credentials file'
}), 400
plugin_dir = _plugin_directory('calendar')
if not plugin_dir:
return jsonify({'status': 'error', 'message': 'Plugin not found'}), 404
# Save file to plugin directory
credentials_path = Path(plugin_dir) / 'credentials.json'
# Backup existing file if it exists
if credentials_path.exists():
backup_path = Path(plugin_dir) / f'credentials.json.backup.{int(_pkg.time.time())}'
shutil.copy2(credentials_path, backup_path)
_prune_credential_backups(Path(plugin_dir))
# Save new file
file.save(str(credentials_path))
# Set proper permissions
os.chmod(credentials_path, 0o600) # Read/write for owner only
return jsonify({
'status': 'success',
'message': 'Credentials file uploaded successfully',
'path': str(credentials_path)
})
@api_v3.route('/plugins/calendar/authenticate', methods=['POST'])
def authenticate_calendar():
"""Google OAuth for the calendar plugin, in the two steps it requires.
Step 1 (no body) returns the consent URL to open. Step 2 posts back the
URL Google redirected to -- it fails to load, because the redirect points
at a loopback address nothing is listening on, but the address bar carries
the authorization code -- and the script exchanges it for a token.
Two calls rather than one because the user has to visit Google in between.
The script persists the PKCE verifier from step 1 for step 2 to reuse; the
exchange fails with "Missing code verifier" otherwise.
"""
plugin_dir = _pkg._calendar_plugin_dir()
if plugin_dir is None:
return jsonify({
'status': 'error',
'message': 'The calendar plugin is not installed'
}), 404
if not (plugin_dir / 'credentials.json').exists():
return jsonify({
'status': 'error',
'message': ('No credentials.json yet. Upload your Google OAuth '
'client file first (Step 1).')
}), 400
data = request.get_json(silent=True) or {}
redirect_url = (data.get('redirect_url') or data.get('code') or '').strip()
payload, error = _run_calendar_registration(plugin_dir, redirect_url)
if error:
return jsonify({'status': 'error', 'message': error}), 500
if payload.get('status') != 'success':
# The script's own diagnosis is more useful than anything that
# could be reconstructed here -- but it interpolates exceptions
# into its messages, so it reaches the client redacted and the
# original goes to the log.
logger.error('calendar authentication failed: %s', payload)
safe = dict(payload)
safe['message'] = redact_text(str(payload.get('message', '')
or 'Authentication failed'))
return jsonify(safe), 400
return jsonify(payload)
@api_v3.route('/plugins/calendar/list-calendars', methods=['GET'])
def list_calendar_calendars():
"""The calendars this account can see, for the config picker.
Reads the token the OAuth flow wrote rather than shelling out again: the
picker is used interactively and a subprocess per click is slower than the
API call it would be wrapping.
"""
plugin_dir = _pkg._calendar_plugin_dir()
if plugin_dir is None:
return jsonify({
'status': 'error',
'message': 'The calendar plugin is not installed'
}), 404
token_file = plugin_dir / 'token.pickle'
if not token_file.exists():
return jsonify({
'status': 'error',
'message': ('Not authenticated with Google yet. Complete Step 2 '
'first, then load your calendars.')
}), 400
try:
import pickle
from google.auth.transport.requests import Request as GoogleRequest
from googleapiclient.discovery import build as build_google_service
except ImportError as e:
return jsonify({
'status': 'error',
# The name of the missing module is the whole diagnosis, but it
# arrives as an exception, so it goes through the redactor like
# any other -- an ImportError can quote a path.
'message': ('The Google API libraries are not installed. Install '
"the calendar plugin's requirements.txt. (%s)"
% describe_exception(e))
}), 500
with open(token_file, 'rb') as handle:
# Written only by this plugin's own OAuth flow, into its own
# directory, and read here exactly as the plugin itself reads it.
creds = pickle.load(handle) # nosec B301 - locally generated token
if creds and creds.expired and creds.refresh_token:
creds.refresh(GoogleRequest())
with open(token_file, 'wb') as handle:
pickle.dump(creds, handle)
os.chmod(token_file, 0o600)
if not creds or not creds.valid:
return jsonify({
'status': 'error',
'message': ('Stored Google credentials are no longer valid. '
'Run Step 2 again to re-authenticate.')
}), 400
service = build_google_service('calendar', 'v3', credentials=creds)
# calendarList.list returns 100 entries per page by default and caps at
# 250, handing back a nextPageToken when there are more. Taking only
# the first page would silently hide calendars from the picker, and the
# user would have no way to tell the list was truncated.
entries = []
page_token = None
for _ in range(_CALENDAR_LIST_MAX_PAGES):
response = service.calendarList().list(
maxResults=250, pageToken=page_token).execute()
entries.extend(response.get('items', []))
page_token = response.get('nextPageToken')
if not page_token:
break
else:
# 2500 calendars in, something is wrong with the account or the
# token is looping; show what was collected rather than spin.
logger.warning(
'calendarList paging stopped at %d pages with more remaining',
_CALENDAR_LIST_MAX_PAGES)
calendars = [{
'id': entry.get('id'),
# The picker labels each row with summary and falls back to the id
# only in its own display, so send something either way.
'summary': entry.get('summary') or entry.get('id'),
'primary': bool(entry.get('primary', False)),
} for entry in entries if entry.get('id')]
# Primary first, then alphabetically: the list is usually short but the
# one the user wants is almost always their own calendar.
calendars.sort(key=lambda c: (not c['primary'], c['summary'].lower()))
return jsonify({'status': 'success', 'calendars': calendars})
@api_v3.route('/plugins/assets/delete', methods=['POST'])
def delete_plugin_asset():
"""Delete an asset file for a plugin"""
data = request.get_json()
plugin_id = data.get('plugin_id')
image_id = data.get('image_id')
if not plugin_id or not image_id:
return jsonify({'status': 'error', 'message': 'plugin_id and image_id are required'}), 400
# Get asset directory
assets_dir = _plugin_uploads_dir(plugin_id)
if assets_dir is None:
return jsonify({'status': 'error', 'message': 'Invalid plugin_id'}), 400
metadata_file = assets_dir / '.metadata.json'
if not metadata_file.exists():
return jsonify({'status': 'error', 'message': 'Metadata file not found'}), 404
# Load metadata
with open(metadata_file, 'r') as f:
metadata = json.load(f)
if image_id not in metadata:
return jsonify({'status': 'error', 'message': 'Image not found'}), 404
# Delete file. The stored path is data, not a trusted location: only
# unlink it when it resolves to a file directly inside this plugin's
# uploads. An entry pointing anywhere else is dropped from the
# metadata without touching the file it names.
entry = metadata[image_id] if isinstance(metadata[image_id], dict) else {}
parts = safe_relative_parts(entry.get('path'))
file_path = resolve_under(PROJECT_ROOT, *parts) if parts else None
if file_path is None or file_path.parent != assets_dir:
logger.warning('Asset %s has a path outside its uploads directory; '
'removing the entry without deleting a file', image_id)
elif file_path.exists():
file_path.unlink()
# Remove from metadata
del metadata[image_id]
# Save metadata
with open(metadata_file, 'w') as f:
json.dump(metadata, f, indent=2)
return jsonify({'status': 'success', 'message': 'Image deleted successfully'})
@api_v3.route('/plugins/assets/list', methods=['GET'])
def list_plugin_assets():
"""List asset files for a plugin"""
plugin_id = request.args.get('plugin_id')
if not plugin_id:
return jsonify({'status': 'error', 'message': 'plugin_id is required'}), 400
# Get asset directory
assets_dir = _plugin_uploads_dir(plugin_id)
if assets_dir is None:
return jsonify({'status': 'error', 'message': 'Invalid plugin_id'}), 400
metadata_file = assets_dir / '.metadata.json'
if not metadata_file.exists():
return jsonify({'status': 'success', 'data': {'assets': []}})
# Load metadata
with open(metadata_file, 'r') as f:
metadata = json.load(f)
# Convert to list
assets = list(metadata.values())
return jsonify({'status': 'success', 'data': {'assets': assets}})
Reference in New Issue View Git Blame Copy Permalink