Files
LEDMatrix/web_interface/blueprints/pages_v3.py
T
ChuckandClaude Opus 5.5 3a81f38f09 fix(web): uniqueItems saves, /health count, Vegas order wipe; one list-repair helper (#638)
* fix(web): drop repeats from uniqueItems lists before validating a plugin save

dedup_unique_arrays lost its only caller in #330, so submitting a value a
uniqueItems list already holds (a stock symbol saved once and posted again)
failed the whole save with a validation error. _prepare_plugin_config_for_save
runs it again just before validation, which covers both POST /plugins/config
and plugin sections posted to /config/main.

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

* fix(web): /health counts the discovered plugins and logs the checks it fails

The plugin check counted plugin_manager.get_available_plugins(), which
PluginManager does not have, behind a hasattr guard that made plugin_count 0
on every device. It now counts the discovered manifests, discovering first
when nothing has been scanned yet.

The config, plugin and hardware checks answered "see logs for details"
without logging anything. Each now logs a warning with the traceback.

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

* fix(web): store refresh no longer claims a commit-metadata refresh

POST /plugins/store/refresh read fetch_commit_info (or fetch_latest_versions)
only to append "(with refreshed commit metadata from GitHub)" to its message.
It never fetched any: the route re-downloads the registry and nothing else.
search_plugins takes the flag, but it reads commit info through its cache,
so passing it on would not refresh anything either. The flag is ignored now
and the message says what happened.

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

* fix(web): refuse a malformed Vegas plugin order instead of clearing it

A vegas_plugin_order or vegas_excluded_plugins value that was not JSON, or
not a list, was stored as [] and the save answered 200, so a bad value wiped
the saved order or exclusions. Both now answer 400 and save nothing, the way
plugin_rotation_order already did; the three share one parser. A list that
holds anything but plugin-id strings is refused as well.

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

* fix(web): per-plugin health and metrics read the display service's latest

GET /plugins/health/<id> and /plugins/metrics/<id> called get_health_summary
and get_metrics_summary without force_reload, so they answered with whatever
the web process read first and kept in memory, while the display service kept
writing newer state. They now pass force_reload=True, as the list routes do.

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

* fix(web): plugin config reset saves through the shared atomic save

POST /plugins/config/reset called config_manager.save_config directly, so it
took no backup, and a failed write escaped as an unhandled exception. It then
handed on_config_change the raw stored section, not the prepared config a
loaded plugin runs with. It now saves through _save_config_atomic with a
backup, answers CONFIG_SAVE_FAILED when that fails, and notifies with
_prepared_plugin_config, as POST /plugins/config does.

POST /plugins/toggle carried its own copy of _save_config_atomic's
save_config_atomic-or-save_config fallback; it calls the shared helper now.

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

* fix(web): one reading and one "unavailable" for each system metric

system_metrics.collect_system_metrics() promised None for a metric it could
not read, but returned cpu_temp as 0 off a Pi, and the whole no-psutil
fallback as zeros. GET /system/status measured the same numbers a second time
with its own code, and answered None there. Now both come from
collect_system_metrics(), and "unavailable" is None everywhere.

/system/status keeps its 0.1s CPU sample and its 10s cache, and gains
nothing it did not already send. Two differences: without psutil it answers
200 with null metrics instead of 503, and a disk it cannot stat is null
instead of a 500.

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

* fix(web): /display/current sends the snapshot as-is and logs a failed read

GET /display/current PIL-decoded the preview snapshot and re-encoded it before
base64-ing it, spending CPU on the Pi to send the same picture, and dropped
any failure with `except Exception: pass`. The /stream/display SSE stream
already passed the PNG's bytes straight through.

Both now read through web_interface/display_preview.py and answer with the
same payload. A missing snapshot is still a null image; any other read
failure is logged as a warning. /health reads the snapshot path from the same
module.

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

* refactor(web): one helper puts a submitted plugin config's lists back

The plugin-config save turned position-keyed dicts ({"0": ..., "1": ...})
back into lists in five copies: four in the form path's
fix_array_structures (whose prefix branches never ran, since no caller
passed one), and _fix_json_arrays on the JSON path. It then force-fixed
the news plugin's feeds.custom_feeds by name, in case the generic pass had
missed it. src/web_interface/config_arrays.coerce_array_shapes now does it
for both paths, custom_feeds included. ensure_array_defaults duplicated
_fix_none_arrays and is gone.

In the same function: the union-type re-checks that the null handling
above them made unreachable, the "(temporary)" random_seed debug log, and
a commented-out log line are removed. A failed validation is logged once
as a warning, not four ERROR lines and a WARNING.

Element types are left to normalize_config_values, which already converted
them for both paths. One difference: the form path no longer adds an empty
{} for a nested object the post left out that has no defaults.

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

* refactor(web): import at module top and log through the module logger

The web_interface.cache imports in config.py and fonts.py were wrapped in
`except ImportError` fallbacks. It is an in-repo module that imports nothing
from the project, so it cannot fail to import; it is imported once at module
top, as system.py now does. cache.py's docstring said blueprints import it
lazily "to avoid circular imports"; it now says why that is unnecessary.

Five logging.error calls in the dim-schedule GET and three logging.warning
calls in plugins.py went to the root logger; they use the module logger.
Function-local re-imports of json, os, shutil, logging and Path, all
already imported by the module, are gone. The `import os` inside two except
blocks of save_plugin_config also made os a local name for the whole function.

execute_plugin_action's step-1 handler gets a comment saying why it stays:
it looks like a copy of the blueprint handler, but without it a
TimeoutExpired from the plugin's script would reach the route's own
`except subprocess.TimeoutExpired` and be answered as a 408.

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

* refactor(web): app.py loses dead CSRF and reconciliation state, comments fixed

- csrf was always None, so `if csrf: csrf.exempt(...)` never ran, and its
  note that the api_v3 blueprint "is exempted above" named an exemption that
  does not exist. Both are gone; the reason there is no CSRF protection stays,
  shortened.
- The SSE rate-limit comment called the default "tight" at 20 per minute. The
  default is 1000 per minute and the streams' 200 is the tighter one; the
  comment now says so. The limits are unchanged.
- _reconciliation_done was written and never read. The docstring that
  explains why reconciliation runs once keeps its reason, in the present
  tense.
- Removed: a dangling "import cache functions" comment with no import under
  it, a "security check ... within project_root" label on an existence check,
  the "(simplified version)" narration, and the note that no redirect route is
  needed. The preview loop's sleep comment no longer mentions a PIL encode
  that the loop does not do.

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

* docs(web): api_v3 comments name the package __init__, not a _common module

Every route module's docstring said the shared blueprint comes "from
._common", a module the package split never created; they name the
package __init__. The PROJECT_ROOT comment described the path from
_common.py; it now describes this package and keeps the incident it
guards against. The "(corrected) in this commit" note in
resolve_pull_command and the /health comment the split's mechanical
time -> _pkg.time rewrite garbled ("Stamp the start _pkg.time") read
correctly again.

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

* refactor(web): drop hasattr checks for attributes PluginManager always has

PluginManager.__init__ sets health_tracker and resource_monitor (to None
until they are configured), so the seven
hasattr(api_v3.plugin_manager, ...) guards in the health, metrics and limits
routes were always true. The falsy checks that do the work stay.

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

* refactor(web): pages_v3 dispatches partials from a dict with one error handler

load_partial chose a loader through a fourteen-branch if/elif, and thirteen
of the loaders then wrapped themselves in the same try/except, logging
"Error loading partial" without saying which. The route now looks the name up
in _PARTIAL_LOADERS and has the one handler, which logs the partial's name.
The loaders just render. _load_tools_partial keeps its own messages. The
search index's _partial_html already catches a loader that raises.

serve_plugin_web_ui repeated _plugin_dir_for inline (containment plus the
ledmatrix- prefix fallback); it calls it now. Also removed: the unused
markupsafe.escape import, function-local json/Path re-imports, and unused
exception bindings.

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

* refactor(web): remove unused imports, locals and a try that cannot fail

- get_error_aggregator was imported by the api_v3 package and used by no
  one; seven names config.py imported, and Path in misc.py and logging in
  plugins.py, likewise.
- branch_info in install_plugin was built and never logged; test_config in
  /health was bound and never read (the load_config call is the check).
- An f-string with no placeholders in the asset upload route.
- _installed_plugin_ids wrapped list(manifests.keys()) in try/except;
  _discovered_plugin_manifests always returns a dict.

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

* refactor(web): start.py logs its startup lines and drops unreachable branches

The startup banner went to stdout with print(); it goes through a logger
now, which the app import has already configured, so it reaches the journal
with a level and timestamp like every other line. The "no addresses" branch
is gone: get_local_ips() always returns at least "localhost".

The except around app.run re-raised "only if it's not a client
disconnection error" from inside the branch that had just established it
was one, so that raise could not run. It is one check now, on a named
tuple of the errnos, which the werkzeug log filter uses too. The comment
on threaded=True counts three SSE endpoints, which is how many there are.
Trailing whitespace is stripped.

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

* refactor(web): save_main_config names its General fields once

The General tab's field names were listed twice, once to detect a General
form post and again, with four more, to keep the remaining-keys merge from
storing them as top-level keys. GENERAL_FIELDS and _MAPPED_TOP_LEVEL_FIELDS
hold them now, and the four per-section skip checks are one set.

The comment on that merge said plugin configs are handled "here too", and
"(including plugin keys)". Plugin sections are handled and removed from the
body before it runs; the comment says so.

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

* refactor(web): plugin directories come from the plugin manager only

Six lookups fell back to PROJECT_ROOT/plugins/<id> when there was no plugin
manager: GET /plugins/config's of-the-day data, POST /plugins/action, the
plugin static-file route, the calendar credentials upload and the calendar
OAuth routes. The loader never scans plugins/ (PluginManager.discover_plugins
reads only the configured directory, plugin-repos by default), so what they
found there was a plugin that never runs. _plugin_directory() asks the
manager and answers None without one, which each route already reports as
"not found".

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

* docs(changelog): web-backend

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-24 17:35:49 -04:00

970 lines
43 KiB
Python

from flask import Blueprint, Response, render_template, jsonify, url_for
from jinja2 import TemplateNotFound
from html.parser import HTMLParser
import json
import logging
import re
from pathlib import Path
# Strict allowlists for URL-derived values used in path and script operations.
_SAFE_PLUGIN_ID_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')
_SAFE_WEB_UI_FILE_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}\.html$')
_SAFE_WIDGET_NAME_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')
_SAFE_WIDGET_SCRIPT_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}\.js$')
from src.web_interface.secret_helpers import mask_secret_fields
from src.plugin_system.schema_manager import plugin_config_defaults, prepare_plugin_config
from src.common.path_safety import resolve_under, safe_path_component
from src.pi5_matrix_support import is_raspberry_pi_5
from web_interface import widget_bundle
logger = logging.getLogger(__name__)
# The managers live on the blueprint object: app.py sets
# pages_v3.config_manager, pages_v3.plugin_manager and the rest.
pages_v3 = Blueprint('pages_v3', __name__)
@pages_v3.route('/assets/widgets.js')
def widgets_bundle():
"""Every widget script in one request (see web_interface/widget_bundle.py).
The individual files stay served from /static for plugin-loader.js and
debugging; this saves the Pi ~30 round trips on each first page load.
Cached as immutable by app.py because the URL carries the version.
"""
body, version = widget_bundle.build_bundle()
response = Response(body, mimetype='application/javascript')
response.headers['X-Widget-Bundle-Version'] = str(version)
return response
@pages_v3.app_context_processor
def inject_widget_bundle_url():
"""`widgets_bundle_url()` for templates, versioned by file mtime."""
return {
'widgets_bundle_url': lambda: url_for(
'pages_v3.widgets_bundle', v=widget_bundle.bundle_version()
)
}
class _SettingsIndexParser(HTMLParser):
"""Extract searchable settings fields from a rendered partial's HTML.
Captures one entry per ``<div class="form-group" id="setting-…">``: the
anchor id, ``data-setting-key``, the field's ``<label>`` text, the
``.help-tip`` tooltip text (``data-tooltip``), and the nearest preceding
``<h3>``/``<h4>`` section heading. Parsing the *rendered* HTML (rather than
the schema) guarantees the anchor ids match the live DOM exactly, so the
search index cannot drift from what users actually see.
"""
def __init__(self, tab, tab_label):
super().__init__(convert_charrefs=True)
self.tab = tab
self.tab_label = tab_label
self.fields = []
self._section = ''
self._field = None
self._depth = 0 # open-div depth within the current field
self._in_label = False
self._label_parts = []
self._in_heading = False
self._heading_parts = []
def handle_starttag(self, tag, attrs):
a = {k: (v or '') for k, v in attrs}
classes = a.get('class', '').split()
# Section headings (only when not already inside a field)
if tag in ('h3', 'h4') and self._field is None:
self._in_heading = True
self._heading_parts = []
if tag == 'div':
fid = a.get('id', '')
if self._field is None and 'form-group' in classes and fid.startswith('setting-'):
self._field = {
'anchorId': fid,
'key': a.get('data-setting-key', '') or fid[len('setting-'):],
'label': '',
'help': '',
'section': self._section,
'tab': self.tab,
'tabLabel': self.tab_label,
}
self._depth = 1
return
if self._field is not None:
self._depth += 1
if self._field is not None:
if tag == 'label' and not self._field['label']:
self._in_label = True
self._label_parts = []
if tag == 'button' and 'help-tip' in classes and not self._field['help']:
self._field['help'] = a.get('data-tooltip', '')
def handle_data(self, data):
if self._in_label:
self._label_parts.append(data)
elif self._in_heading:
self._heading_parts.append(data)
def handle_endtag(self, tag):
if tag in ('h3', 'h4') and self._in_heading:
self._in_heading = False
self._section = ' '.join(''.join(self._heading_parts).split()).strip()
return
if self._field is None:
return
if tag == 'label' and self._in_label:
self._in_label = False
self._field['label'] = ' '.join(''.join(self._label_parts).split()).strip()
elif tag == 'div':
self._depth -= 1
if self._depth <= 0:
if self._field['label']:
self.fields.append(self._field)
self._field = None
self._depth = 0
def _partial_html(loader):
"""Run a partial loader and return its HTML string ('' on error)."""
try:
result = loader()
except Exception:
logger.warning("search-index: partial render failed", exc_info=True)
return ''
if isinstance(result, str):
return result
if isinstance(result, tuple): # loaders return (msg, status) on error
return ''
try:
return result.get_data(as_text=True)
except Exception:
return ''
def _extract_settings_fields(html, tab, tab_label):
parser = _SettingsIndexParser(tab, tab_label)
parser.feed(html)
return parser.fields
# Cache the built index keyed on the installed-plugin set. Core labels/tooltips
# are static template text, so only a change in installed plugins invalidates it.
_SEARCH_INDEX_CACHE = {'sig': None, 'fields': None}
@pages_v3.route('/')
def index():
"""Main v3 interface page: the app shell. Every tab loads as a partial."""
return render_template('v3/base.html')
@pages_v3.route('/partials/<partial_name>')
def load_partial(partial_name):
"""One tab's HTML for HTMX, by the names in _PARTIAL_LOADERS; 404 otherwise."""
loader = _PARTIAL_LOADERS.get(partial_name)
if loader is None:
return "Partial not found", 404
try:
return loader()
except Exception:
logger.error("Error loading partial %s", partial_name, exc_info=True)
return "Error loading partial", 500
@pages_v3.route('/partials/plugin-config/<plugin_id>')
def load_plugin_config_partial(plugin_id):
"""Load plugin configuration partial via HTMX - server-side rendered form"""
try:
return _load_plugin_config_partial(plugin_id)
except Exception:
logger.error("Error loading plugin config partial for %s", plugin_id, exc_info=True)
return '<div class="text-red-500 p-4">Error loading plugin config; see logs for details</div>', 500
@pages_v3.route('/settings/search-index')
def settings_search_index():
"""Return a flat JSON index of every searchable setting (core + plugin).
Powers the web UI's global settings search. Built by rendering the settings
partials server-side and extracting field metadata, then cached per
installed-plugin set so it is off the display's hot path.
"""
# Core settings tabs: (activeTab value, human label, loader).
core_tabs = [
('general', 'General', _load_general_partial),
('display', 'Display', _load_display_partial),
('durations', 'Durations', _load_durations_partial),
('schedule', 'Schedule', _load_schedule_partial),
('wifi', 'WiFi', _load_wifi_partial),
]
try:
plugin_ids = []
if pages_v3.plugin_manager:
try:
pages_v3.plugin_manager.discover_plugins()
plugin_ids = sorted(
pi.get('id') for pi in pages_v3.plugin_manager.get_all_plugin_info()
if pi.get('id')
)
except Exception:
logger.warning("search-index: could not enumerate plugins", exc_info=True)
sig = tuple(plugin_ids)
if _SEARCH_INDEX_CACHE['sig'] == sig and _SEARCH_INDEX_CACHE['fields'] is not None:
return jsonify({'fields': _SEARCH_INDEX_CACHE['fields']})
fields = []
for tab, label, loader in core_tabs:
fields.extend(_extract_settings_fields(_partial_html(loader), tab, label))
for pid in plugin_ids:
info = pages_v3.plugin_manager.get_plugin_info(pid) or {}
label = info.get('name', pid)
html = _partial_html(lambda pid=pid: _load_plugin_config_partial(pid))
fields.extend(_extract_settings_fields(html, pid, label))
_SEARCH_INDEX_CACHE['sig'] = sig
_SEARCH_INDEX_CACHE['fields'] = fields
return jsonify({'fields': fields})
except Exception:
logger.error("Error building settings search index", exc_info=True)
return jsonify({'fields': []}), 500
@pages_v3.route('/plugin-ui/<plugin_id>/web-ui/<path:filename>')
def serve_plugin_web_ui(plugin_id, filename):
"""Serve a plugin's web_ui/ HTML fragment as a standalone page.
Wraps the fragment with a minimal HTML page that injects window.PLUGIN_ID
and loads Tailwind CSS so the fragment runs correctly inside the iframe
that plugin_config.html embeds it in.
That iframe carries no ``sandbox`` attribute, so the fragment runs with
the interface's own origin. That is deliberate rather than an oversight:
the fragment is a file from an installed plugin, and an installed plugin
already runs Python on the device. The trust boundary is plugin install,
not this route. It is worth knowing when reading the code, which is why
it says so here instead of claiming a sandbox that is not there.
"""
# Validate URL-derived values against strict allowlists before any path or
# script operations.
if not _SAFE_PLUGIN_ID_RE.match(plugin_id):
return 'Invalid plugin ID', 400, {'Content-Type': 'text/plain'}
if not _SAFE_WEB_UI_FILE_RE.match(filename):
return 'Invalid filename', 400, {'Content-Type': 'text/plain'}
# The allowlists above already forbid a separator, but the value that gets
# joined has to be the checked one, not the argument -- see
# src/common/path_safety.py. safe_path_component rejects rather than
# truncates, so these cannot disagree.
safe_id = safe_path_component(plugin_id)
safe_fn = safe_path_component(filename)
if not safe_id or not safe_fn:
return 'Invalid path component', 400, {'Content-Type': 'text/plain'}
if not pages_v3.plugin_manager:
return 'Plugin manager not available', 503, {'Content-Type': 'text/plain'}
try:
web_ui_path = resolve_under(_plugin_dir_for(safe_id) / 'web_ui', safe_fn)
if web_ui_path is None:
return 'Forbidden', 403, {'Content-Type': 'text/plain'}
if not web_ui_path.exists():
return 'Not found', 404, {'Content-Type': 'text/plain'}
fragment = web_ui_path.read_text(encoding='utf-8')
# json.dumps wraps the value in quotes. Replace HTML meta-chars with
# their JS Unicode escape sequences so the value cannot close or escape
# the enclosing <script> tag.
# r'<' is the 6-char literal string <, which JavaScript
# interprets as <. This is the standard JSON-in-HTML hardening pattern.
safe_plugin_id_js = (
json.dumps(safe_id)
.replace('<', '\\u003c')
.replace('>', '\\u003e')
.replace('&', '\\u0026')
)
page = (
'<!DOCTYPE html>\n'
'<html lang="en">\n'
'<head>\n'
'<meta charset="UTF-8">\n'
'<meta name="viewport" content="width=device-width,initial-scale=1">\n'
'<script>\n'
# Inject plugin context before the fragment runs.
# plugin_id is validated to [a-zA-Z0-9_-] above, so this is safe,
# but we also Unicode-escape HTML meta-chars as defence in depth.
f' window.PLUGIN_ID = {safe_plugin_id_js};\n'
'</script>\n'
# Tailwind v2 CDN — same version used by the parent LEDMatrix UI
'<link rel="stylesheet" '
'href="https://cdnjs.cloudflare.com/ajax/libs/tailwindcss/2.2.19/tailwind.min.css" '
'crossorigin="anonymous">\n'
'<style>body{margin:0;padding:0;background:#fff;}</style>\n'
'</head>\n'
'<body>\n'
+ fragment +
'\n</body>\n</html>'
)
return page, 200, {'Content-Type': 'text/html; charset=utf-8'}
except ValueError:
return 'Forbidden', 403, {'Content-Type': 'text/plain'}
except Exception:
logger.error('Error serving plugin web_ui %s/%s', plugin_id, filename, exc_info=True)
return 'Error serving file', 500, {'Content-Type': 'text/plain'}
def _plugin_dir_for(safe_id):
"""A sanitised plugin id's directory, which may not exist.
Contained under the configured plugins directory, with PluginManager's
``ledmatrix-`` prefix fallback. Raises ValueError for an id that would
leave it; the routes answer that with a 403.
"""
plugins_base = Path(pages_v3.plugin_manager.plugins_dir).resolve()
plugin_dir = resolve_under(plugins_base, safe_id)
if plugin_dir is None:
raise ValueError('plugin id escapes the plugins directory')
if not plugin_dir.exists():
alt = resolve_under(plugins_base, f'ledmatrix-{safe_id}')
if alt is not None:
plugin_dir = alt
return plugin_dir
def _declared_widget_script(plugin_dir, widget_name):
"""The script filename a plugin's manifest declares for ``widget_name``.
The manifest is the allowlist: only a widget the plugin actually declares
can be served, so this route never exposes arbitrary files under the
plugin directory even though the directory itself is attacker-influenced
(plugins are user-installed). Returns None when the widget is not
declared, the manifest is unreadable, or the declared script name is not
a plain ``<name>.js`` basename.
"""
manifest_path = plugin_dir / 'manifest.json'
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
except (OSError, ValueError):
return None
if not isinstance(manifest, dict):
return None
for entry in manifest.get('widgets') or ():
if not isinstance(entry, dict):
continue
if entry.get('name') != widget_name:
continue
script = entry.get('script') or f'{widget_name}.js'
if not isinstance(script, str) or not _SAFE_WIDGET_SCRIPT_RE.match(script):
return None
return script
return None
@pages_v3.route('/static/plugin-widgets/<plugin_id>/<widget_name>.js')
def serve_plugin_widget(plugin_id, widget_name):
"""Serve a plugin-declared widget script from its ``widgets/`` directory.
This is the server half of ``LEDMatrixWidgets.loadPluginWidget`` (see
static/v3/js/widgets/plugin-loader.js), which fetches exactly this path.
The loader uses a dynamic ``import()``, so the response must carry a
JavaScript MIME type or the browser refuses the module.
The route is deliberately narrower than the plugin directory: a script is
served only when the plugin's own manifest declares a widget by that name,
so installing a plugin does not publish everything it ships.
"""
if not _SAFE_PLUGIN_ID_RE.match(plugin_id):
return 'Invalid plugin ID', 400, {'Content-Type': 'text/plain'}
if not _SAFE_WIDGET_NAME_RE.match(widget_name):
return 'Invalid widget name', 400, {'Content-Type': 'text/plain'}
# safe_path_component is this codebase's sanitiser (src/common/
# path_safety.py): it rejects rather than mangles, so a name that is not
# a plain path component never reaches the filesystem.
safe_id = safe_path_component(plugin_id)
safe_widget = safe_path_component(widget_name)
if not safe_id or not safe_widget:
return 'Invalid path component', 400, {'Content-Type': 'text/plain'}
if not pages_v3.plugin_manager:
return 'Plugin manager not available', 503, {'Content-Type': 'text/plain'}
try:
plugin_dir = _plugin_dir_for(safe_id)
if not plugin_dir.exists():
return 'Not found', 404, {'Content-Type': 'text/plain'}
script = _declared_widget_script(plugin_dir, safe_widget)
if script is None:
# Undeclared is a 404 rather than a 403: whether a plugin happens
# to ship an undeclared file is not something to confirm.
return 'Not found', 404, {'Content-Type': 'text/plain'}
# Contain widgets/ itself first: a symlinked widgets directory would
# otherwise become the base the script is checked against.
widgets_dir = resolve_under(plugin_dir, 'widgets')
if widgets_dir is None:
return 'Not found', 404, {'Content-Type': 'text/plain'}
# The script name comes from the plugin's manifest, not the request,
# so it gets the same containment treatment the URL parts got.
script_path = resolve_under(widgets_dir, script)
if script_path is None or not script_path.is_file():
return 'Not found', 404, {'Content-Type': 'text/plain'}
body = script_path.read_text(encoding='utf-8')
return body, 200, {
'Content-Type': 'text/javascript; charset=utf-8',
# Plugin updates replace this file in place; revalidate so a
# stale widget cannot outlive the plugin version that shipped it.
'Cache-Control': 'no-cache',
}
except ValueError:
return 'Forbidden', 403, {'Content-Type': 'text/plain'}
except Exception:
logger.error('Error serving plugin widget %s/%s', plugin_id, widget_name,
exc_info=True)
return 'Error serving file', 500, {'Content-Type': 'text/plain'}
def _load_overview_partial():
"""Load overview partial with system stats"""
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
# This would be populated with real system stats via SSE
return render_template('v3/partials/overview.html',
main_config=main_config)
def _load_general_partial():
"""Load general settings partial"""
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
try:
from web_interface.auto_update import describe_status
auto_update_status = describe_status(main_config)
except Exception:
logger.debug("Could not read auto-update status", exc_info=True)
auto_update_status = None
return render_template('v3/partials/general.html',
main_config=main_config,
auto_update_status=auto_update_status)
def _load_display_partial():
"""Load display settings partial"""
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
return render_template('v3/partials/display.html',
main_config=main_config,
is_pi5=is_raspberry_pi_5())
def _plugin_default_duration(plugin_id, plugin_config):
"""Seconds a plugin shows each screen when the Rotation page sets none.
Mirrors BasePlugin.get_display_duration's config fallback: the plugin's
display_duration, else its schema default, else 15.
"""
def _valid(v):
return isinstance(v, (int, float)) and not isinstance(v, bool) and v > 0
value = plugin_config.get('display_duration')
if not _valid(value):
schema = None
schema_mgr = getattr(pages_v3, 'schema_manager', None)
if schema_mgr is not None:
try:
schema = schema_mgr.load_schema(plugin_id)
except Exception:
logger.debug("durations: no schema for %s", plugin_id, exc_info=True)
value = plugin_config_defaults(schema if isinstance(schema, dict) else None).get(
'display_duration')
if not _valid(value):
value = 15
return int(value) if float(value).is_integer() else value
def _load_durations_partial():
"""Load rotation & durations partial.
Builds one duration entry per display mode of every enabled plugin. A mode
with a value saved in display.display_durations shows it; the rest are
blank, with the plugin's own duration as the placeholder, because a saved
value overrides the plugin (see DisplayController._get_display_duration).
Pre-filling every mode would pin them all on the first save.
"""
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
duration_groups = []
covered_keys = set()
if pages_v3.plugin_manager:
try:
pages_v3.plugin_manager.discover_plugins()
saved = (main_config.get('display', {}) or {}).get('display_durations', {}) or {}
infos = sorted(pages_v3.plugin_manager.get_all_plugin_info(),
key=lambda i: (i.get('name') or i.get('id') or '').lower())
for info in infos:
pid = info.get('id')
if not pid or not (main_config.get(pid, {}) or {}).get('enabled', False):
continue
modes = pages_v3.plugin_manager.get_plugin_display_modes(pid) or [pid]
covered_keys.update(modes)
default = _plugin_default_duration(pid, main_config.get(pid, {}) or {})
duration_groups.append({
'plugin_id': pid,
'plugin_name': info.get('name') or pid,
'modes': [{'key': m, 'value': saved.get(m, ''), 'default': default}
for m in modes],
})
# Saved keys not owned by any enabled plugin (disabled or
# uninstalled plugins) stay visible rather than vanishing.
leftovers = [{'key': k, 'value': v} for k, v in saved.items()
if k not in covered_keys]
if leftovers:
duration_groups.append({
'plugin_id': '',
'plugin_name': 'Other saved entries',
'modes': leftovers,
})
except Exception:
logger.warning("durations: could not enumerate plugin modes", exc_info=True)
return render_template('v3/partials/durations.html',
main_config=main_config,
duration_groups=duration_groups)
def _load_schedule_partial():
"""Load schedule settings partial"""
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
schedule_config = main_config.get('schedule', {})
dim_schedule_config = main_config.get('dim_schedule', {})
# Get normal brightness for display in dim schedule UI
normal_brightness = main_config.get('display', {}).get('hardware', {}).get('brightness', 90)
return render_template('v3/partials/schedule.html',
schedule_config=schedule_config,
dim_schedule_config=dim_schedule_config,
normal_brightness=normal_brightness)
def _load_plugins_partial():
"""Load plugins management partial"""
# Load plugin data from the plugin system
plugins_data = []
# Get installed plugins if managers are available
if pages_v3.plugin_manager and pages_v3.plugin_store_manager:
try:
# Get all installed plugin info
all_plugin_info = pages_v3.plugin_manager.get_all_plugin_info()
# Load config once before the loop (not per-plugin)
full_config = pages_v3.config_manager.load_config() if pages_v3.config_manager else {}
# Format for the web interface
for plugin_info in all_plugin_info:
plugin_id = plugin_info.get('id')
# Re-read manifest from disk to ensure we have the latest metadata
manifest_path = Path(pages_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)
# Update plugin_info with fresh manifest data
plugin_info.update(fresh_manifest)
except Exception:
# If we can't read the fresh manifest, use the cached one
logger.warning("Could not read fresh manifest for plugin: %s", plugin_id)
# Get enabled status from config (source of truth)
# Read from config file first, fall back to plugin instance if config doesn't have the key
enabled = None
if pages_v3.config_manager:
plugin_config = full_config.get(plugin_id, {})
# Check if 'enabled' key exists in config (even if False)
if 'enabled' in plugin_config:
enabled = bool(plugin_config['enabled'])
# Fallback to plugin instance if config doesn't have enabled key
if enabled is None:
plugin_instance = pages_v3.plugin_manager.get_plugin(plugin_id)
if plugin_instance:
enabled = plugin_instance.enabled
else:
# Default to True if no config key and plugin not loaded (matches BasePlugin default)
enabled = True
# Get verified status from store registry (no GitHub API calls needed)
store_info = pages_v3.plugin_store_manager.get_registry_info(plugin_id)
verified = store_info.get('verified', False) if store_info else False
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')
plugins_data.append({
'id': plugin_id,
'name': plugin_info.get('name', plugin_id),
'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', []),
'enabled': enabled,
'verified': verified,
'loaded': plugin_info.get('loaded', False),
'last_updated': last_updated,
'last_commit': last_commit,
'branch': branch
})
except Exception:
logger.error("Error loading plugin data", exc_info=True)
return render_template('v3/partials/plugins.html',
plugins=plugins_data)
def _load_fonts_partial():
"""Load fonts management partial"""
# This would load font data from the font system
fonts_data = {} # Placeholder for font data
return render_template('v3/partials/fonts.html',
fonts=fonts_data)
def _load_logs_partial():
"""Load logs viewer partial"""
return render_template('v3/partials/logs.html')
def _load_raw_json_partial():
"""Load raw JSON editor partial"""
if pages_v3.config_manager:
main_config_data = pages_v3.config_manager.get_raw_file_content('main')
secrets_config_data = pages_v3.config_manager.get_raw_file_content('secrets')
main_config_json = json.dumps(main_config_data, indent=4)
secrets_config_json = json.dumps(secrets_config_data, indent=4)
return render_template('v3/partials/raw_json.html',
main_config_json=main_config_json,
secrets_config_json=secrets_config_json,
main_config_path=pages_v3.config_manager.get_config_path(),
secrets_config_path=pages_v3.config_manager.get_secrets_path())
def _load_backup_restore_partial():
"""Load backup & restore partial."""
return render_template('v3/partials/backup_restore.html')
@pages_v3.route('/setup')
def captive_setup():
"""Lightweight captive portal setup page — self-contained, no frameworks."""
return render_template('v3/captive_setup.html')
def _load_wifi_partial():
"""Load WiFi setup partial"""
return render_template('v3/partials/wifi.html')
def _load_cache_partial():
"""Load cache management partial"""
return render_template('v3/partials/cache.html')
def _load_operation_history_partial():
"""Load operation history partial"""
return render_template('v3/partials/operation_history.html')
def _load_tools_partial():
"""Load tools/utilities partial."""
try:
return render_template('v3/partials/tools.html')
except TemplateNotFound:
logger.error("[Pages V3][Tools] Template not found: v3/partials/tools.html", exc_info=True)
return "[Pages V3][Tools] Template is missing.", 500
except OSError as exc:
logger.error("[Pages V3][Tools] I/O error loading tools partial: %s", exc, exc_info=True)
return "[Pages V3][Tools] Failed to load due to a file system error. Check logs.", 500
_PARTIAL_LOADERS = {
'overview': _load_overview_partial,
'general': _load_general_partial,
'display': _load_display_partial,
'durations': _load_durations_partial,
'schedule': _load_schedule_partial,
'plugins': _load_plugins_partial,
'fonts': _load_fonts_partial,
'logs': _load_logs_partial,
'raw-json': _load_raw_json_partial,
'backup-restore': _load_backup_restore_partial,
'wifi': _load_wifi_partial,
'cache': _load_cache_partial,
'operation-history': _load_operation_history_partial,
'tools': _load_tools_partial,
}
def _load_plugin_config_partial(plugin_id):
"""
Load plugin configuration partial - server-side rendered form.
This replaces the client-side generateConfigForm() JavaScript.
"""
# Refuse an id that is not a plain directory name, rather than quietly
# basename-ing it down to one: "../weather" used to become "weather" and
# render a partial the caller never asked for.
plugin_id = safe_path_component(plugin_id)
if not plugin_id or not re.match(r'^[a-zA-Z0-9][a-zA-Z0-9._\-:]*$', plugin_id):
return '<div class="text-red-500 p-4">Invalid plugin ID</div>', 400
try:
if not pages_v3.plugin_manager:
return '<div class="text-red-500 p-4">Plugin manager not available</div>', 500
# Handle starlark app config (starlark:<app_id>)
if plugin_id.startswith('starlark:'):
return _load_starlark_config_partial(plugin_id[len('starlark:'):])
# Resolve and validate all plugin paths against the plugins base directory
_plugins_base = Path(pages_v3.plugin_manager.plugins_dir).resolve()
_plugin_dir = resolve_under(_plugins_base, plugin_id)
if _plugin_dir is None:
return '<div class="text-red-500 p-4">Invalid plugin ID</div>', 400
# Try to get plugin info first
plugin_info = pages_v3.plugin_manager.get_plugin_info(plugin_id)
# If not found, re-discover plugins (handles plugins added after startup)
if not plugin_info:
pages_v3.plugin_manager.discover_plugins()
plugin_info = pages_v3.plugin_manager.get_plugin_info(plugin_id)
if not plugin_info:
return '<div class="text-red-500 p-4">Plugin not found</div>', 404
# Get plugin instance (may be None if not loaded)
plugin_instance = pages_v3.plugin_manager.get_plugin(plugin_id)
# Get plugin configuration from config file
config = {}
if pages_v3.config_manager:
full_config = pages_v3.config_manager.load_config()
config = full_config.get(plugin_id, {})
# Load uploaded images from metadata file if images field exists in schema
schema_path_temp = resolve_under(_plugin_dir, "config_schema.json")
if schema_path_temp is not None and schema_path_temp.exists():
try:
with open(schema_path_temp, 'r', encoding='utf-8') as f:
temp_schema = json.load(f)
if (temp_schema.get('properties', {}).get('images', {}).get('x-widget') == 'file-upload' or
temp_schema.get('properties', {}).get('images', {}).get('x_widget') == 'file-upload'):
_assets_base = (Path(__file__).parent.parent.parent / 'assets' / 'plugins').resolve()
metadata_file = resolve_under(
_assets_base, plugin_id, 'uploads', '.metadata.json'
)
if metadata_file and metadata_file.exists():
try:
with open(metadata_file, 'r', encoding='utf-8') as mf:
metadata = json.load(mf)
images_from_metadata = list(metadata.values())
if not config.get('images') or len(config.get('images', [])) == 0:
config['images'] = images_from_metadata
else:
config_image_ids = {img.get('id') for img in config.get('images', []) if img.get('id')}
new_images = [img for img in images_from_metadata if img.get('id') not in config_image_ids]
if new_images:
config['images'] = config.get('images', []) + new_images
except Exception as e:
logger.warning("Could not load plugin upload metadata: %s", e)
except Exception as e: # nosec B110 - metadata pre-load is optional; schema loads fully below
logger.debug("Metadata pre-load skipped for plugin %s: %s", plugin_id, e)
# Get plugin schema.
#
# Through SchemaManager, not a raw json.load, because that is what
# the save route uses (api_v3.save_plugin_config) -- and the two
# disagreeing is not academic. SchemaManager applies
# expand_style_elements, which turns a compact
# customization.x-style-elements declaration into the per-element
# blocks this form renders. Reading the file directly meant a plugin
# using that form (of-the-day ships one) had a customization section
# that rendered nothing at all, while saving still validated against
# the expanded shape.
#
# use_cache=False matches the save route: a plugin's schema changes
# on disk during development, and a cached copy would keep serving
# the old form.
#
# The raw read stays as a fallback for callers that never set a
# schema_manager (several tests, and any embedder of this blueprint).
schema = {}
schema_mgr = getattr(pages_v3, 'schema_manager', None)
if schema_mgr is not None:
try:
schema = schema_mgr.load_schema(plugin_id, use_cache=False) or {}
except Exception as e:
logger.warning("SchemaManager could not load schema for %s: %s",
plugin_id, e)
if not schema:
# resolve_under keeps the containment guard main added here; the
# SchemaManager path above does its own.
schema_path = resolve_under(_plugin_dir, "config_schema.json")
if schema_path is not None and schema_path.exists():
try:
with open(schema_path, 'r', encoding='utf-8') as f:
schema = json.load(f)
except Exception as e:
logger.warning("Could not load schema for plugin: %s", e)
# Get web UI actions from plugin manifest
web_ui_actions = []
manifest_path = resolve_under(_plugin_dir, "manifest.json")
if manifest_path is not None and manifest_path.exists():
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
web_ui_actions = manifest.get('web_ui_actions', [])
except Exception as e:
logger.warning("Could not load manifest for plugin: %s", e)
schema_properties = schema.get('properties') if isinstance(schema, dict) else None
if not isinstance(schema_properties, dict):
return '<div class="text-red-500 p-4">Error loading plugin config securely: schema unavailable.</div>', 500
# Fill in schema defaults for keys the saved config doesn't have yet,
# as GET /api/v3/plugins/config does. Without this, an option added in
# a plugin update (geochron 1.2.0's show_date, default true) renders
# as an unchecked box, and because the save treats every drawn but
# unposted checkbox as false, the first save turns it off for good.
try:
defaults = plugin_config_defaults(schema)
if schema_mgr is not None:
defaults = schema_mgr.apply_device_location(defaults)
config = prepare_plugin_config(config, schema, defaults)
except Exception as e:
logger.warning("Could not merge schema defaults for %s: %s", plugin_id, e)
# Mask secret fields before rendering template (fail closed — never
# leak secrets). After the merge, so a secret's default is masked too.
config = mask_secret_fields(config, schema_properties)
# Determine enabled status
enabled = config.get('enabled', True)
if plugin_instance:
enabled = plugin_instance.enabled
# Build plugin data for template
plugin_data = {
'id': plugin_id,
'name': plugin_info.get('name', plugin_id),
'author': plugin_info.get('author', 'Unknown'),
'version': plugin_info.get('version', ''),
'description': plugin_info.get('description', ''),
'category': plugin_info.get('category', 'General'),
'tags': plugin_info.get('tags', []),
'enabled': enabled,
'last_commit': plugin_info.get('last_commit') or plugin_info.get('last_commit_sha', ''),
'branch': plugin_info.get('branch', ''),
}
return render_template(
'v3/partials/plugin_config.html',
plugin=plugin_data,
config=config,
schema=schema,
web_ui_actions=web_ui_actions
)
except Exception:
logger.error("Error loading plugin config partial for %s", plugin_id, exc_info=True)
return '<div class="text-red-500 p-4">Error loading plugin config; see logs for details</div>', 500
def _load_starlark_config_partial(app_id):
"""Load configuration partial for a Starlark app."""
# Refuse an id that is not a plain directory name rather than basename-ing
# it down to one -- see _load_plugin_config_partial for why.
app_id = safe_path_component(app_id)
if not app_id or not re.match(r'^[a-zA-Z0-9][a-zA-Z0-9_\-]*$', app_id):
return '<div class="text-red-500 p-4">Invalid app ID</div>', 400
try:
starlark_plugin = pages_v3.plugin_manager.get_plugin('starlark-apps') if pages_v3.plugin_manager else None
if starlark_plugin and hasattr(starlark_plugin, 'apps'):
app = starlark_plugin.apps.get(app_id)
if not app:
return '<div class="text-red-500 p-4">Starlark app not found</div>', 404
return render_template(
'v3/partials/starlark_config.html',
app_id=app_id,
app_name=app.manifest.get('name', app_id),
app_enabled=app.is_enabled(),
render_interval=app.get_render_interval(),
display_duration=app.get_display_duration(),
config=app.config,
schema=app.schema,
has_frames=app.frames is not None,
frame_count=len(app.frames) if app.frames else 0,
last_render_time=app.last_render_time,
)
# Standalone: read from manifest file
starlark_base = (Path(__file__).resolve().parent.parent.parent / 'starlark-apps').resolve()
manifest_file = starlark_base / 'manifest.json'
if not manifest_file.exists():
return '<div class="text-red-500 p-4">Starlark app not found</div>', 404
with open(manifest_file, 'r') as f:
manifest = json.load(f)
app_data = manifest.get('apps', {}).get(app_id)
if not app_data:
return '<div class="text-red-500 p-4">Starlark app not found</div>', 404
# Load schema from schema.json if it exists — validate path stays within starlark_base
schema = None
schema_file = resolve_under(starlark_base, app_id, 'schema.json')
if schema_file and schema_file.exists():
try:
with open(schema_file, 'r') as f:
schema = json.load(f)
except (OSError, json.JSONDecodeError) as e:
logger.warning("Could not load starlark schema for app: %s", e)
# Load config from config.json if it exists — validate path stays within starlark_base
config = {}
config_file = resolve_under(starlark_base, app_id, 'config.json')
if config_file and config_file.exists():
try:
with open(config_file, 'r') as f:
config = json.load(f)
except (OSError, json.JSONDecodeError) as e:
logger.warning("Could not load starlark config for app: %s", e)
return render_template(
'v3/partials/starlark_config.html',
app_id=app_id,
app_name=app_data.get('name', app_id),
app_enabled=app_data.get('enabled', True),
render_interval=app_data.get('render_interval', 300),
display_duration=app_data.get('display_duration', 15),
config=config,
schema=schema,
has_frames=False,
frame_count=0,
last_render_time=None,
)
except Exception:
logger.error("[Pages V3] Error loading starlark config for app", exc_info=True)
return '<div class="text-red-500 p-4">Error loading starlark config; see logs for details</div>', 500