mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
ChuckandClaude Opus 5.5
9a1f94f793
* fix(starlark): blank app locations use the device location, not San Francisco A Starlark (Tidbyt) app whose Location field is blank rendered at its author's hard-coded DEFAULT_LOCATION -- usually San Francisco -- even with the device city set under General settings. A user in Charlotte, NC got San Francisco weather and radar with nothing in config.json to explain it. src/device_location.py fills unset location fields at render time (display plugin and the web standalone render): the device city is geocoded once via Open-Meteo, preferring a match in the configured state/country, and cached permanently. A saved location always wins; if the lookup fails the field is dropped so the app uses its own default, and the failure is not retried for 30 minutes. Also fixes the config form: clearing a location omitted the key, and the save merges, so the old value could never be removed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(starlark): say what happens when the device location can't be used A blank app Location only renders at the device's city when one is set and the Open-Meteo lookup finds it. With no city, no match, or the geocoder unreachable (retried after 30 minutes), the app gets no location and keeps its author's default. The guide, the config page hint, CONFIG_REFERENCE and the CHANGELOG entry now say so. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2247 lines
97 KiB
Python
2247 lines
97 KiB
Python
"""The /api/v3 blueprint.
|
|
|
|
This was one 10,469-line module. It is now a package, but deliberately not a
|
|
set of *separate* blueprints: every route module beside this one decorates the
|
|
single `api_v3` object defined here, so endpoint names stay `api_v3.<function>`
|
|
and the URL map is unchanged. web_interface/app.py registers it as before.
|
|
|
|
The shared imports, constants and helpers stay in this file rather than moving
|
|
to a _common submodule, because tests monkeypatch some of them by module
|
|
attribute -- `monkeypatch.setattr(api_v3_module, "_BACKUP_EXPORT_DIR", ...)`.
|
|
Keeping them here means that keeps working exactly as it did. Route modules
|
|
read the mutable ones back through this module (see _pkg below) rather than
|
|
binding them by value, for the same reason.
|
|
|
|
The route modules are imported at the *bottom*: they import names from here, so
|
|
everything they need has to exist first.
|
|
"""
|
|
from flask import Blueprint, request, jsonify, Response
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import sys
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import hashlib
|
|
import uuid
|
|
import logging
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional, Tuple, Type
|
|
from urllib.parse import urlparse, urlunparse
|
|
logger = logging.getLogger(__name__)
|
|
# Import new infrastructure
|
|
from src.web_interface.api_helpers import success_response, error_response, validate_request_json
|
|
from src.web_interface.errors import ErrorCode
|
|
from src.web_interface.secret_helpers import (find_secret_fields, mask_all_secret_values,
|
|
merge_secrets, remove_empty_secrets,
|
|
separate_secrets,
|
|
strip_masked_values)
|
|
from src.web_interface.error_handler import describe_exception, redact_text
|
|
from src.plugin_system.operation_types import OperationType
|
|
from src.web_interface.validators import (
|
|
validate_file_upload
|
|
)
|
|
from src.error_aggregator import get_error_aggregator
|
|
from src.common.permission_utils import install_requirements_file
|
|
from src.common.path_safety import resolve_under
|
|
from src.device_location import DeviceLocationResolver, apply_device_location
|
|
_SUDO = shutil.which('sudo')
|
|
_JOURNALCTL = shutil.which('journalctl')
|
|
_GIT = shutil.which('git')
|
|
# Cap subprocess output returned to the browser — pip can produce MBs on build failures.
|
|
_MAX_OUTPUT_BYTES = 51_200 # 50 KB
|
|
def _truncate_output(stdout: str, stderr: str) -> str:
|
|
"""Combine stdout+stderr and truncate to _MAX_OUTPUT_BYTES (keeping the tail)."""
|
|
combined = (stdout + stderr).strip()
|
|
if len(combined) > _MAX_OUTPUT_BYTES:
|
|
combined = '[...output truncated...]\n' + combined[-_MAX_OUTPUT_BYTES:]
|
|
return combined
|
|
def _pip_install_requirements(req_file: Path, timeout: int) -> subprocess.CompletedProcess:
|
|
"""Install a requirements.txt file, preferring the vetted sudo wrapper so
|
|
the packages are visible to root-run ledmatrix.service — not just to
|
|
whichever non-root user runs this web process. Falls back to installing
|
|
for the current process only if the wrapper isn't set up yet (i.e. the
|
|
admin hasn't run scripts/install/configure_web_sudo.sh since upgrading),
|
|
so the button still does *something* useful rather than hard-failing.
|
|
|
|
Thin wrapper around the shared implementation in permission_utils so the
|
|
Plugin Store's own dependency installation (store_manager.py) follows the
|
|
exact same root-visible install path instead of a divergent one.
|
|
"""
|
|
return install_requirements_file(req_file, timeout=timeout)
|
|
def _scrub_git_remote_url(url: str) -> str:
|
|
"""Strip embedded username/password from an HTTPS remote URL before returning it to the UI."""
|
|
try:
|
|
p = urlparse(url)
|
|
if p.scheme in ('http', 'https') and (p.username or p.password):
|
|
netloc = p.hostname or ''
|
|
if p.port:
|
|
netloc += f':{p.port}'
|
|
return urlunparse(p._replace(netloc=netloc))
|
|
except Exception:
|
|
pass
|
|
return url
|
|
# NOTE: the managers live on the blueprint object (app.py sets
|
|
# api_v3.config_manager, api_v3.plugin_manager, api_v3.cache_manager and
|
|
# the rest). Deliberately not mirrored as module globals: a bare
|
|
# `config_manager` used to resolve to a None that was never assigned, which
|
|
# silently disabled the /health checks and made /display/current fall back
|
|
# to a hardcoded 128x64.
|
|
# Get project root directory (web_interface/../..)
|
|
# web_interface/blueprints/api_v3/_common.py -> up four to the project root.
|
|
# This was three levels when everything lived in web_interface/blueprints/api_v3.py;
|
|
# the split moved the file one directory deeper and silently pointed PROJECT_ROOT
|
|
# at web_interface/ instead. Nothing failed at import -- it surfaced as routes
|
|
# 404ing and "installation script not found", because every path built from it
|
|
# was wrong. Asserted in test_api_v3_url_map.py so the next move cannot repeat it.
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
|
# System fonts that cannot be deleted (used by catalog API and delete endpoint)
|
|
SYSTEM_FONTS = frozenset([
|
|
'pressstart2p-regular', 'pressstart2p',
|
|
'4x6-font', '4x6',
|
|
'5by7.regular', '5by7', '5x7',
|
|
'5x8', '6x9', '6x10', '6x12', '6x13', '6x13b', '6x13o',
|
|
'7x13', '7x13b', '7x13o', '7x14', '7x14b',
|
|
'8x13', '8x13b', '8x13o',
|
|
'9x15', '9x15b', '9x18', '9x18b',
|
|
'10x20',
|
|
'matrixchunky8', 'matrixlight6', 'tom-thumb',
|
|
'clr6x12', 'helvr12', 'texgyre-27'
|
|
])
|
|
api_v3 = Blueprint('api_v3', __name__)
|
|
def _get_plugin_version(plugin_id: str) -> str:
|
|
"""Read the installed version from a plugin's manifest.json.
|
|
|
|
Returns the version string on success, or '' if the manifest
|
|
cannot be read (missing, corrupt, permission denied, etc.) or if
|
|
``plugin_id`` is not a plain directory name. Several callers pass an id
|
|
that arrived in a request body, so the name is validated here rather
|
|
than relying on each of them to have done it.
|
|
"""
|
|
manifest_path = resolve_under(
|
|
api_v3.plugin_store_manager.plugins_dir, plugin_id, "manifest.json"
|
|
)
|
|
if manifest_path is None:
|
|
logger.warning("[PluginVersion] Rejected unsafe plugin id %r", plugin_id)
|
|
return ''
|
|
try:
|
|
with open(manifest_path, 'r', encoding='utf-8') as f:
|
|
manifest = json.load(f)
|
|
return manifest.get('version', '')
|
|
except (FileNotFoundError, PermissionError, OSError) as e:
|
|
logger.warning("[PluginVersion] Could not read manifest for %s at %s: %s", plugin_id, manifest_path, e)
|
|
except json.JSONDecodeError as e:
|
|
logger.warning("[PluginVersion] Invalid JSON in manifest for %s at %s: %s", plugin_id, manifest_path, e)
|
|
return ''
|
|
def _is_plugin_update_available(installed_version: str, latest_version: str) -> bool:
|
|
"""Return True when the registry's ``latest_version`` is strictly newer
|
|
than the installed version.
|
|
|
|
Thin alias for the shared comparator in
|
|
`src.plugin_system.compatibility.is_update_available` — the store's
|
|
`update_plugin` uses the same function, so the UI badge and the actual
|
|
reinstall decision can never disagree.
|
|
"""
|
|
from src.plugin_system.compatibility import is_update_available
|
|
return is_update_available(installed_version, latest_version)
|
|
def _save_config_atomic(config_manager, config_data, create_backup=True):
|
|
"""
|
|
Save configuration using atomic save if available, fallback to regular save.
|
|
|
|
Returns:
|
|
tuple: (success: bool, error_message: str or None)
|
|
"""
|
|
if hasattr(config_manager, 'save_config_atomic'):
|
|
result = config_manager.save_config_atomic(config_data, create_backup=create_backup)
|
|
if result.status.value != 'success':
|
|
return False, result.message
|
|
return True, None
|
|
else:
|
|
try:
|
|
config_manager.save_config(config_data)
|
|
return True, None
|
|
except Exception as e:
|
|
return False, str(e)
|
|
def _coerce_to_bool(value):
|
|
"""
|
|
Coerce a form value to a proper Python boolean.
|
|
|
|
HTML checkboxes send string values like "true", "on", "1" when checked.
|
|
This ensures we store actual booleans in config JSON, not strings.
|
|
|
|
Args:
|
|
value: The form value (string, bool, int, or None)
|
|
|
|
Returns:
|
|
bool: True if value represents a truthy checkbox state, False otherwise
|
|
"""
|
|
if value is None:
|
|
return False
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, int):
|
|
return value == 1
|
|
if isinstance(value, str):
|
|
return value.lower() in ('true', 'on', '1', 'yes')
|
|
return False
|
|
def _get_display_service_status():
|
|
"""Return status information about the ledmatrix service."""
|
|
try:
|
|
result = subprocess.run(
|
|
['systemctl', 'is-active', 'ledmatrix'],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=3
|
|
)
|
|
return {
|
|
'active': result.stdout.strip() == 'active',
|
|
'returncode': result.returncode,
|
|
'stdout': result.stdout.strip(),
|
|
'stderr': result.stderr.strip()
|
|
}
|
|
except subprocess.TimeoutExpired:
|
|
return {
|
|
'active': False,
|
|
'returncode': -1,
|
|
'stdout': '',
|
|
'stderr': 'timeout'
|
|
}
|
|
except Exception as err:
|
|
return {
|
|
'active': False,
|
|
'returncode': -1,
|
|
'stdout': '',
|
|
'stderr': str(err)
|
|
}
|
|
def _run_systemctl_command(args):
|
|
"""Run a systemctl command safely."""
|
|
try:
|
|
result = subprocess.run(
|
|
args,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15
|
|
)
|
|
return {
|
|
'returncode': result.returncode,
|
|
'stdout': result.stdout,
|
|
'stderr': result.stderr
|
|
}
|
|
except subprocess.TimeoutExpired:
|
|
return {
|
|
'returncode': -1,
|
|
'stdout': '',
|
|
'stderr': 'timeout'
|
|
}
|
|
except Exception as err:
|
|
return {
|
|
'returncode': -1,
|
|
'stdout': '',
|
|
'stderr': str(err)
|
|
}
|
|
def _ensure_display_service_running():
|
|
"""Ensure the ledmatrix display service is running."""
|
|
status = _get_display_service_status()
|
|
if status.get('active'):
|
|
status['started'] = False
|
|
return status
|
|
result = _run_systemctl_command(['sudo', 'systemctl', 'start', 'ledmatrix.service'])
|
|
service_status = _get_display_service_status()
|
|
result['started'] = result.get('returncode') == 0
|
|
result['active'] = service_status.get('active')
|
|
result['status'] = service_status
|
|
return result
|
|
def _stop_display_service():
|
|
"""Stop the ledmatrix display service."""
|
|
result = _run_systemctl_command(['sudo', 'systemctl', 'stop', 'ledmatrix.service'])
|
|
status = _get_display_service_status()
|
|
result['active'] = status.get('active')
|
|
result['status'] = status
|
|
return result
|
|
#: Field names whose value is a credential. Matched by name because this
|
|
#: endpoint returns the whole config, core keys included, and core config has
|
|
#: no schema to carry x-secret markers.
|
|
_CREDENTIAL_NAME_PARTS = ("password", "passwd", "secret", "token", "api_key",
|
|
"apikey", "access_key", "private_key", "client_secret")
|
|
def _looks_like_a_credential(name: str) -> bool:
|
|
lowered = name.lower()
|
|
return any(part in lowered for part in _CREDENTIAL_NAME_PARTS)
|
|
def _redact_credentials(value):
|
|
"""A copy of `value` with credential-named fields blanked.
|
|
|
|
/config/main returned the raw config to anyone who could reach the port,
|
|
and this interface has no authentication. On one rig that meant a 40-char
|
|
GitHub token, a 183-char Home Assistant token and five API keys were
|
|
readable by anything on the LAN.
|
|
|
|
The x-secret masking used by the plugin config endpoints does not help
|
|
here: this endpoint never consults a schema, and core keys such as
|
|
github.api_token have no schema to mark. Matching on the field name is
|
|
blunt, but for a whole-config dump the right default is that anything
|
|
named like a credential does not leave the process.
|
|
|
|
Blanked rather than removed, and safe to blank: POST /config/main merges
|
|
into the loaded config and only writes the keys it was given, so a client
|
|
that round-trips this response cannot erase a secret it never saw.
|
|
"""
|
|
if isinstance(value, dict):
|
|
return {k: (_blank_credential_value(v) if _looks_like_a_credential(k)
|
|
else _redact_credentials(v))
|
|
for k, v in value.items()}
|
|
if isinstance(value, list):
|
|
return [_redact_credentials(item) for item in value]
|
|
return value
|
|
def _blank_all_scalars(value):
|
|
"""Blank every scalar reached from `value`, at any depth.
|
|
|
|
Unlike `_blank_credential_value`, this never delegates back to
|
|
`_redact_credentials`'s name-based walk: an object reached through a
|
|
credential-owned list (e.g. `tokens: [{"value": "secret"}]`) has no
|
|
field name of its own to test, so every scalar inside it is blanked
|
|
regardless of what its keys are called.
|
|
"""
|
|
if isinstance(value, dict):
|
|
return {k: _blank_all_scalars(v) for k, v in value.items()}
|
|
if isinstance(value, list):
|
|
return [_blank_all_scalars(item) for item in value]
|
|
return ""
|
|
def _blank_credential_value(value):
|
|
"""Blank a value that sits under a credential-shaped key.
|
|
|
|
A dict is still walked -- `secrets: {api_key: ..., note: ...}` is a
|
|
section name, not a value to blank in one go, so sub-fields that are not
|
|
themselves credential-named survive (see
|
|
test_a_credential_shaped_container_is_still_walked). A list has no field
|
|
names to test for its items, though, so every scalar reached through it
|
|
is blanked outright: a bare list of secrets (`tokens: ["a", "b"]`) and a
|
|
list of credential-shaped objects (`tokens: [{"value": "secret"}]`) are
|
|
both blanked at any depth via `_blank_all_scalars`.
|
|
"""
|
|
if isinstance(value, dict):
|
|
return _redact_credentials(value)
|
|
if isinstance(value, list):
|
|
return [_blank_all_scalars(item) for item in value]
|
|
return ""
|
|
def _validate_time_format(time_str):
|
|
"""Validate time format is HH:MM"""
|
|
try:
|
|
datetime.strptime(time_str, '%H:%M')
|
|
return True, None
|
|
except (ValueError, TypeError):
|
|
return False, f"Invalid time format: {time_str}. Expected HH:MM format."
|
|
def _git_current_branch(project_dir):
|
|
"""Current branch name, or '' when detached or git fails."""
|
|
try:
|
|
r = subprocess.run(['git', 'branch', '--show-current'],
|
|
capture_output=True, text=True, timeout=10, cwd=str(project_dir))
|
|
return r.stdout.strip() if r.returncode == 0 else ''
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
return ''
|
|
def _git_upstream(project_dir):
|
|
"""Configured upstream for the current branch (e.g. 'origin/main'), or ''."""
|
|
try:
|
|
r = subprocess.run(['git', 'rev-parse', '--abbrev-ref', '--symbolic-full-name', '@{u}'],
|
|
capture_output=True, text=True, timeout=10, cwd=str(project_dir))
|
|
return r.stdout.strip() if r.returncode == 0 else ''
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
return ''
|
|
def _git_remote_branch_exists(project_dir, branch):
|
|
"""True when origin/<branch> exists locally as a remote-tracking ref."""
|
|
if not branch:
|
|
return False
|
|
try:
|
|
r = subprocess.run(
|
|
['git', 'show-ref', '--verify', '--quiet', f'refs/remotes/origin/{branch}'],
|
|
capture_output=True, text=True, timeout=10, cwd=str(project_dir))
|
|
return r.returncode == 0
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
return False
|
|
def resolve_pull_command(project_dir):
|
|
"""Work out how to pull, for branches with and without an upstream.
|
|
|
|
A plain ``git pull --rebase`` fails outright on a branch that has no
|
|
upstream ("There is no tracking information for the current branch"),
|
|
which is easy to end up on: checking out a branch by name, restoring a
|
|
backup, or following an install guide that names one. The update button
|
|
then reports a failure the user cannot act on.
|
|
|
|
``--autostash`` is passed for the same reason. Rebase refuses to start
|
|
when any tracked file is modified, and on these installs something always
|
|
is: first_time_install.sh chmods five scripts that git tracked as 644, so
|
|
every machine that ran the installer carries five permanent mode changes
|
|
and the update button reports "cannot pull with rebase: You have unstaged
|
|
changes". Those modes are corrected in this commit, but a user cannot pull
|
|
the correction while the pull is what is blocked, and any other local edit
|
|
would reproduce it anyway. Autostash reapplies the changes afterwards.
|
|
|
|
Returns ``(args, note, error)``. When ``origin/<branch>`` exists the pull
|
|
is made explicit against it, so the update proceeds and the branch is
|
|
given tracking information afterwards.
|
|
"""
|
|
upstream = _git_upstream(project_dir)
|
|
if upstream:
|
|
return ['git', 'pull', '--rebase', '--autostash'], '', None
|
|
|
|
branch = _git_current_branch(project_dir)
|
|
if not branch:
|
|
return None, '', (
|
|
"This checkout is in a detached HEAD state, so there is no branch "
|
|
"to update. Switch to a branch first (Tools -> Switch branch)."
|
|
)
|
|
if _git_remote_branch_exists(project_dir, branch):
|
|
return (
|
|
['git', 'pull', '--rebase', '--autostash', 'origin', branch],
|
|
f"Branch '{branch}' had no upstream; pulled from origin/{branch} and set it as the upstream.",
|
|
None,
|
|
)
|
|
return None, '', (
|
|
f"Branch '{branch}' has no upstream and there is no origin/{branch} to "
|
|
f"pull from. Use Switch branch to move to a branch that exists on the "
|
|
f"remote, or push this one first."
|
|
)
|
|
_BRANCH_NAME_RE = re.compile(r'[A-Za-z0-9._/-]{1,200}')
|
|
def is_valid_branch_name(name):
|
|
"""Accept only plain branch names.
|
|
|
|
This value becomes a subprocess argument, so anything exotic is refused
|
|
rather than escaped. '..' is excluded because it is range syntax to git.
|
|
"""
|
|
if not name or not _BRANCH_NAME_RE.fullmatch(name):
|
|
return False
|
|
return '..' not in name and not name.startswith('-')
|
|
def checkout_branch(project_dir, target, stash=False):
|
|
"""Switch the checkout to `target`, returning (payload, http_status).
|
|
|
|
Split out of the route so it can be tested against real repositories.
|
|
Attaches tracking when the branch exists on origin, so the next
|
|
Pull Latest is a plain `git pull` rather than the no-upstream fallback.
|
|
"""
|
|
target = (target or '').strip()
|
|
if not target:
|
|
return {'status': 'error', 'message': 'Branch name required'}, 400
|
|
if not is_valid_branch_name(target):
|
|
return {'status': 'error', 'message': 'Invalid branch name'}, 400
|
|
|
|
try:
|
|
subprocess.run(['git', 'fetch', 'origin', '--prune'],
|
|
capture_output=True, text=True, timeout=60, cwd=project_dir)
|
|
|
|
local_exists = subprocess.run(
|
|
['git', 'show-ref', '--verify', '--quiet', f'refs/heads/{target}'],
|
|
capture_output=True, text=True, timeout=10, cwd=project_dir).returncode == 0
|
|
remote_exists = _git_remote_branch_exists(project_dir, target)
|
|
if not local_exists and not remote_exists:
|
|
return {'status': 'error',
|
|
'message': f"No branch '{target}' locally or on origin"}, 404
|
|
|
|
# Local edits block a checkout. Pull Latest already stashes for the
|
|
# same reason, so offer it here too -- but only when asked, never
|
|
# silently: putting someone's edits away unasked is worse than
|
|
# refusing the switch.
|
|
stash_note = ''
|
|
if stash:
|
|
stashed = subprocess.run(['git', 'stash', 'push', '-m', f'switch to {target}'],
|
|
capture_output=True, text=True, timeout=60, cwd=project_dir)
|
|
if stashed.returncode == 0 and 'No local changes' not in stashed.stdout:
|
|
stash_note = ' Local changes were stashed (recover them with git stash list).'
|
|
|
|
if local_exists:
|
|
co = subprocess.run(['git', 'checkout', target],
|
|
capture_output=True, text=True, timeout=60, cwd=project_dir)
|
|
else:
|
|
# -B so a stale local ref does not block the checkout.
|
|
co = subprocess.run(['git', 'checkout', '-B', target, f'origin/{target}'],
|
|
capture_output=True, text=True, timeout=60, cwd=project_dir)
|
|
|
|
if co.returncode != 0:
|
|
logger.warning("git checkout %s failed: %s", target, co.stderr)
|
|
return {
|
|
'status': 'error',
|
|
'message': f"Could not switch to '{target}'.",
|
|
# Keep git's full list of blocking files: naming them is the
|
|
# difference between an error the user can act on and one they
|
|
# cannot.
|
|
'detail': (co.stderr or '').strip(),
|
|
'can_retry_with_stash': 'would be overwritten by checkout' in (co.stderr or ''),
|
|
}, 200
|
|
|
|
if remote_exists:
|
|
subprocess.run(['git', 'branch', f'--set-upstream-to=origin/{target}', target],
|
|
capture_output=True, text=True, timeout=10, cwd=project_dir)
|
|
|
|
logger.info("Switched checkout to branch %s", target)
|
|
return {
|
|
'status': 'success',
|
|
'message': f"Now on '{target}'.{stash_note} Use Pull Latest to fetch its newest code.",
|
|
}, 200
|
|
except subprocess.TimeoutExpired:
|
|
return {'status': 'error', 'message': 'Timed out talking to git'}, 504
|
|
except OSError as exc:
|
|
logger.error("checkout_branch failed: %s", exc, exc_info=True)
|
|
return {'status': 'error', 'message': 'Could not switch branch'}, 500
|
|
def get_git_version(project_dir=None):
|
|
"""Get git version information from the repository"""
|
|
if project_dir is None:
|
|
project_dir = PROJECT_ROOT
|
|
|
|
try:
|
|
# Try to get tag description (e.g., v2.4-10-g123456)
|
|
result = subprocess.run(
|
|
['git', 'describe', '--tags', '--dirty'],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
cwd=str(project_dir)
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
version_str = result.stdout.strip()
|
|
if re.match(r'^[a-zA-Z0-9._\-]+$', version_str):
|
|
return version_str
|
|
|
|
# Fallback to short commit hash
|
|
result = subprocess.run(
|
|
['git', 'rev-parse', '--short', 'HEAD'],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
cwd=str(project_dir)
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
version_str = result.stdout.strip()
|
|
if re.match(r'^[a-zA-Z0-9._\-]+$', version_str):
|
|
return version_str
|
|
|
|
return 'Unknown'
|
|
except Exception:
|
|
return 'Unknown'
|
|
_update_check_cache: Dict[str, Any] = {'result': None, 'ts': 0.0}
|
|
_UPDATE_CHECK_TTL = 300 # 5 minutes — avoids a git fetch on every page load
|
|
def _update_check_failed(detail: str) -> Dict[str, Any]:
|
|
"""A check that could not run is not the same as being up to date.
|
|
|
|
Reporting update_available=False on a git failure hides the banner, and
|
|
the banner is the only route to the update button -- so a checkout git
|
|
refuses to touch looks exactly like a current one, permanently. The most
|
|
common cause is an install performed as root: git then reports "dubious
|
|
ownership" and every command fails, including the fetch here.
|
|
"""
|
|
return {'update_available': False, 'remote_sha': 'unknown',
|
|
'commits_behind': 0, 'check_failed': True, 'error': detail}
|
|
def _describe_git_failure(stderr: str) -> str:
|
|
"""Turn git's stderr into something the user can act on."""
|
|
text = (stderr or '').strip()
|
|
if 'dubious ownership' in text or 'detected dubious ownership' in text:
|
|
return ("This checkout is owned by a different user than the one "
|
|
"running the web interface, so git refuses to use it. It is "
|
|
"usually the result of installing as root. Fix the ownership "
|
|
"and the update will work: sudo chown -R $USER:$USER "
|
|
+ str(PROJECT_ROOT))
|
|
if 'could not resolve host' in text.lower() or 'network is unreachable' in text.lower():
|
|
return "Could not reach GitHub to check for updates."
|
|
return "Could not check for updates: " + (text.splitlines()[0] if text else "git failed")
|
|
def _installed_plugin_ids():
|
|
"""Best-effort list of installed plugin IDs for the web process.
|
|
|
|
Health/metrics state is written by the separate display service to the
|
|
shared on-disk cache, so the tracker's in-memory set is empty here. We
|
|
enumerate the installed plugins and read each one's persisted summary by ID
|
|
instead of relying on the tracker's in-memory `get_all_*` view.
|
|
"""
|
|
manifests = _discovered_plugin_manifests()
|
|
try:
|
|
return list(manifests.keys()) if manifests else []
|
|
except Exception:
|
|
logger.debug('listing plugin_manifests failed while building plugin ids', exc_info=True)
|
|
return []
|
|
def _discovered_plugin_manifests(plugin_id=None, rescan=False):
|
|
"""The plugin manager's manifests, discovering plugins first if needed.
|
|
|
|
The web process discovers plugins lazily (see app.py): nothing scans at
|
|
startup, so plugin_manifests is empty until some endpoint calls
|
|
discover_plugins(). A route that looks a plugin up without coming through
|
|
here answers "not found" for every installed plugin until something else
|
|
has run -- after a web restart, POST /display/on-demand/start returned 404
|
|
for minutes on a real rig, and only API-only callers ever noticed.
|
|
|
|
Scans when nothing is discovered yet, when ``plugin_id`` is given and not
|
|
among the manifests (it may have been installed since the last scan), or
|
|
when ``rescan`` is set (for lookups that are not by id, such as a mode).
|
|
Otherwise the existing map is reused, so a steady stream of requests
|
|
for known plugins costs nothing.
|
|
|
|
Returns the manifest map, or {} when there is no plugin manager.
|
|
"""
|
|
pm = api_v3.plugin_manager
|
|
if pm is None:
|
|
return {}
|
|
manifests = getattr(pm, 'plugin_manifests', None)
|
|
if not manifests or rescan or (plugin_id is not None and plugin_id not in manifests):
|
|
try:
|
|
pm.discover_plugins()
|
|
except Exception:
|
|
logger.warning('Plugin discovery failed', exc_info=True)
|
|
manifests = getattr(pm, 'plugin_manifests', None)
|
|
return manifests or {}
|
|
def _do_transactional_uninstall(plugin_id, preserve_config):
|
|
"""Execute an uninstall with snapshot-based rollback.
|
|
|
|
Order of operations:
|
|
1. Snapshot main config + secrets (abort on unexpected errors, proceed on expected I/O errors).
|
|
2. Clean up plugin config (abort with 500 if this raises — avoids orphaned files).
|
|
3. Unload plugin from runtime if loaded (rollback + 500 if this raises).
|
|
4. Remove plugin files (rollback + 500 if this returns False or raises).
|
|
5. Finish (remove state, invalidate caches).
|
|
|
|
Rollback restores the config snapshot and, if the plugin had been
|
|
loaded before unload, calls load_plugin to restore runtime state.
|
|
|
|
Returns (True, None) on success or (False, error_message) on failure.
|
|
"""
|
|
from src.exceptions import ConfigError
|
|
|
|
# --- Step 1: snapshot main + secrets ---
|
|
main_snapshot = None
|
|
secrets_snapshot = None
|
|
try:
|
|
main_snapshot = api_v3.config_manager.get_raw_file_content('main')
|
|
except (OSError, ConfigError):
|
|
pass # Proceed without snapshot; narrow catch preserves TypeError/AttributeError
|
|
try:
|
|
secrets_snapshot = api_v3.config_manager.get_raw_file_content('secrets')
|
|
except (OSError, ConfigError):
|
|
pass
|
|
|
|
# --- Step 2: cleanup config first (abort before touching filesystem) ---
|
|
if not preserve_config:
|
|
api_v3.config_manager.cleanup_plugin_config(plugin_id, remove_secrets=True)
|
|
|
|
# Record whether the plugin was running before we touch anything.
|
|
was_loaded = (
|
|
api_v3.plugin_manager is not None
|
|
and plugin_id in api_v3.plugin_manager.plugins
|
|
)
|
|
|
|
def _rollback(reload_plugin):
|
|
if main_snapshot is not None:
|
|
try:
|
|
api_v3.config_manager.save_raw_file_content('main', main_snapshot)
|
|
except Exception as restore_err:
|
|
logger.error("Failed to restore main config snapshot for %s: %s", plugin_id, restore_err)
|
|
if secrets_snapshot is not None:
|
|
try:
|
|
api_v3.config_manager.save_raw_file_content('secrets', secrets_snapshot)
|
|
except Exception as restore_err:
|
|
logger.error("Failed to restore secrets snapshot for %s: %s", plugin_id, restore_err)
|
|
if reload_plugin and api_v3.plugin_manager is not None:
|
|
try:
|
|
api_v3.plugin_manager.load_plugin(plugin_id)
|
|
except Exception as reload_err:
|
|
logger.error("Failed to reload plugin %s during rollback: %s", plugin_id, reload_err)
|
|
|
|
# --- Step 3: unload ---
|
|
if was_loaded:
|
|
try:
|
|
api_v3.plugin_manager.unload_plugin(plugin_id)
|
|
except Exception as unload_err:
|
|
_rollback(reload_plugin=False) # unload failed — runtime state unchanged
|
|
return False, f"Failed to unload plugin {plugin_id}: {unload_err}"
|
|
|
|
# --- Step 4: remove files ---
|
|
try:
|
|
success = api_v3.plugin_store_manager.uninstall_plugin(plugin_id)
|
|
except Exception as remove_err:
|
|
_rollback(reload_plugin=was_loaded)
|
|
return False, f"Failed to remove plugin {plugin_id}: {remove_err}"
|
|
|
|
if not success:
|
|
_rollback(reload_plugin=was_loaded)
|
|
return False, f"Failed to uninstall plugin {plugin_id}"
|
|
|
|
# --- Step 5: finish ---
|
|
if api_v3.schema_manager:
|
|
api_v3.schema_manager.invalidate_cache(plugin_id)
|
|
if api_v3.plugin_state_manager:
|
|
api_v3.plugin_state_manager.remove_plugin_state(plugin_id)
|
|
# Persistently record the uninstall so a later core `git pull` update
|
|
# cannot resurrect a built-in plugin (committed under plugin-repos/) that
|
|
# the user removed. Best-effort: never fail the uninstall over this.
|
|
try:
|
|
api_v3.plugin_store_manager.record_uninstalled_plugin(plugin_id)
|
|
except Exception as record_err:
|
|
logger.warning("Could not record uninstall for %s: %s", plugin_id, record_err)
|
|
return True, None
|
|
def deep_merge(base_dict, update_dict):
|
|
"""
|
|
Deep merge update_dict into base_dict.
|
|
For nested dicts, recursively merge. For other types, update_dict takes precedence.
|
|
|
|
Lists are intentionally REPLACED wholesale, never index-merged: form posts
|
|
carry complete arrays, and index-merging would resurrect items the user
|
|
deleted. This also applies to the parallel secrets lists produced by
|
|
separate_secrets — a newly saved secrets list is authoritative.
|
|
"""
|
|
result = base_dict.copy()
|
|
for key, value in update_dict.items():
|
|
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
|
|
# Recursively merge nested dicts
|
|
result[key] = deep_merge(result[key], value)
|
|
else:
|
|
# For non-dict values or new keys, use the update value
|
|
result[key] = value
|
|
return result
|
|
def _parse_form_value(value):
|
|
"""
|
|
Parse a form value into the appropriate Python type.
|
|
Handles booleans, numbers, JSON arrays/objects, and strings.
|
|
"""
|
|
import json
|
|
|
|
if value is None:
|
|
return None
|
|
|
|
# Handle string values
|
|
if isinstance(value, str):
|
|
stripped = value.strip()
|
|
|
|
# Check for boolean strings
|
|
if stripped.lower() == 'true':
|
|
return True
|
|
if stripped.lower() == 'false':
|
|
return False
|
|
if stripped.lower() in ('null', 'none') or stripped == '':
|
|
return None
|
|
|
|
# Try parsing as JSON (for arrays and objects) - do this BEFORE number parsing
|
|
# This handles RGB arrays like "[255, 0, 0]" correctly
|
|
if stripped.startswith('[') or stripped.startswith('{'):
|
|
try:
|
|
return json.loads(stripped)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Try parsing as number
|
|
try:
|
|
if '.' in stripped:
|
|
return float(stripped)
|
|
return int(stripped)
|
|
except ValueError:
|
|
pass
|
|
|
|
# Return as string (original value, not stripped)
|
|
return value
|
|
|
|
return value
|
|
def _get_schema_property(schema, key_path):
|
|
"""
|
|
Get the schema property for a given key path (supports dot notation).
|
|
|
|
Args:
|
|
schema: The JSON schema dict
|
|
key_path: Dot-separated path like "customization.time_text.font"
|
|
|
|
Returns:
|
|
The property schema dict or None if not found
|
|
"""
|
|
if not schema or 'properties' not in schema:
|
|
return None
|
|
|
|
parts = key_path.split('.')
|
|
current = schema['properties']
|
|
i = 0
|
|
|
|
while i < len(parts):
|
|
# Try progressively longer candidates, longest first, so schema keys that
|
|
# themselves contain dots (e.g. league keys like "fifa.world") are matched
|
|
# instead of being mistaken for nested "fifa" -> "world" objects.
|
|
matched = False
|
|
for j in range(len(parts), i, -1):
|
|
candidate = '.'.join(parts[i:j])
|
|
if isinstance(current, dict) and candidate in current:
|
|
prop = current[candidate]
|
|
# Consumed all remaining parts — this is the target property.
|
|
if j == len(parts):
|
|
return prop
|
|
# Navigate deeper through an object with properties.
|
|
if isinstance(prop, dict) and 'properties' in prop:
|
|
current = prop['properties']
|
|
i = j
|
|
matched = True
|
|
break
|
|
# Matched a non-object before consuming the path — can't go deeper.
|
|
return None
|
|
if not matched:
|
|
return None
|
|
|
|
return None
|
|
def _is_field_required(key_path, schema):
|
|
"""
|
|
Check if a field is required according to the schema.
|
|
|
|
Args:
|
|
key_path: Dot-separated path like "mqtt.username"
|
|
schema: The JSON schema dict
|
|
|
|
Returns:
|
|
True if field is required, False otherwise
|
|
"""
|
|
if not schema or 'properties' not in schema:
|
|
return False
|
|
|
|
parts = key_path.split('.')
|
|
if len(parts) == 1:
|
|
# Top-level field
|
|
required = schema.get('required', [])
|
|
return parts[0] in required
|
|
else:
|
|
# Nested field - navigate to parent object
|
|
parent_path = '.'.join(parts[:-1])
|
|
field_name = parts[-1]
|
|
|
|
# Get parent property
|
|
parent_prop = _get_schema_property(schema, parent_path)
|
|
if not parent_prop or 'properties' not in parent_prop:
|
|
return False
|
|
|
|
# Check if field is required in parent
|
|
required = parent_prop.get('required', [])
|
|
return field_name in required
|
|
# Sentinel object to indicate a field should be skipped (not set in config)
|
|
_SKIP_FIELD = object()
|
|
|
|
|
|
def _schema_type_is(prop, wanted):
|
|
"""Whether a schema property is of ``wanted`` type.
|
|
|
|
JSON Schema allows a union (``["array", "null"]``), which the per-element
|
|
style system uses for its per-mode override fields: null there means
|
|
"inherit the base", so the type genuinely is "an array or nothing". A
|
|
bare ``prop.get('type') == 'array'`` reads False for those, which meant
|
|
the indexed colour inputs a form posts as ``...text_color.0/.1/.2`` were
|
|
never recombined into a list.
|
|
"""
|
|
if not isinstance(prop, dict):
|
|
return False
|
|
declared = prop.get('type')
|
|
if isinstance(declared, list):
|
|
return wanted in declared
|
|
return declared == wanted
|
|
|
|
|
|
def _schema_allows_null(prop):
|
|
"""Whether a schema property's declared type includes null."""
|
|
return _schema_type_is(prop, 'null')
|
|
|
|
|
|
def _parse_form_value_with_schema(value, key_path, schema):
|
|
"""
|
|
Parse a form value using schema information to determine correct type.
|
|
Handles arrays (comma-separated strings), objects, and other types.
|
|
|
|
Args:
|
|
value: The form value (usually a string)
|
|
key_path: Dot-separated path like "category_order" or "customization.time_text.font"
|
|
schema: The plugin's JSON schema
|
|
|
|
Returns:
|
|
Parsed value with correct type, or _SKIP_FIELD to indicate the field should not be set
|
|
"""
|
|
import json
|
|
|
|
# Get the schema property for this field
|
|
prop = _get_schema_property(schema, key_path)
|
|
|
|
# Handle None/empty values
|
|
if value is None or (isinstance(value, str) and value.strip() == ''):
|
|
# A nullable field left blank means null, not an empty container.
|
|
# This is the inherit sentinel for per-mode style overrides: an
|
|
# empty list there would read as "the user chose no colour" rather
|
|
# than "follow the base element".
|
|
if _schema_allows_null(prop):
|
|
return None
|
|
# If schema says it's an array, return empty array instead of None
|
|
if prop and _schema_type_is(prop, 'array'):
|
|
return []
|
|
# If schema says it's an object, return empty dict instead of None
|
|
if prop and _schema_type_is(prop, 'object'):
|
|
return {}
|
|
# If it's an optional string field, preserve empty string instead of None
|
|
if prop and prop.get('type') == 'string':
|
|
if not _is_field_required(key_path, schema):
|
|
return "" # Return empty string for optional string fields
|
|
# For number/integer fields, check if they have defaults or are required
|
|
if prop:
|
|
prop_type = prop.get('type')
|
|
if prop_type in ('number', 'integer'):
|
|
# If field has a default, use it
|
|
if 'default' in prop:
|
|
return prop['default']
|
|
# If field is not required and has no default, skip setting it
|
|
if not _is_field_required(key_path, schema):
|
|
return _SKIP_FIELD
|
|
# If field is required but empty, return None (validation will fail, which is correct)
|
|
return None
|
|
return None
|
|
|
|
# Handle string values
|
|
if isinstance(value, str):
|
|
stripped = value.strip()
|
|
|
|
# Check for boolean strings
|
|
if stripped.lower() == 'true':
|
|
return True
|
|
if stripped.lower() == 'false':
|
|
return False
|
|
# "on"/"off" come from HTML checkboxes — only coerce when schema says boolean
|
|
if prop and prop.get('type') == 'boolean':
|
|
if stripped.lower() == 'on':
|
|
return True
|
|
if stripped.lower() == 'off':
|
|
return False
|
|
|
|
# Handle arrays based on schema
|
|
if prop and _schema_type_is(prop, 'array'):
|
|
# Try parsing as JSON first (handles "[1,2,3]" format)
|
|
if stripped.startswith('['):
|
|
try:
|
|
return json.loads(stripped)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Otherwise, treat as comma-separated string
|
|
if stripped:
|
|
# Split by comma and strip each item
|
|
items = [item.strip() for item in stripped.split(',') if item.strip()]
|
|
# Try to convert items to numbers if schema items are numbers
|
|
items_schema = prop.get('items', {})
|
|
if items_schema.get('type') in ('number', 'integer'):
|
|
try:
|
|
return [int(item) if '.' not in item else float(item) for item in items]
|
|
except ValueError:
|
|
pass
|
|
return items
|
|
return []
|
|
|
|
# Handle objects based on schema
|
|
if prop and _schema_type_is(prop, 'object'):
|
|
# Try parsing as JSON
|
|
if stripped.startswith('{'):
|
|
try:
|
|
return json.loads(stripped)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
# If it's not JSON, return empty dict (form shouldn't send objects as strings)
|
|
return {}
|
|
|
|
# Try parsing as JSON (for arrays and objects) - do this BEFORE number parsing
|
|
if stripped.startswith('[') or stripped.startswith('{'):
|
|
try:
|
|
return json.loads(stripped)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Handle numbers based on schema
|
|
if prop:
|
|
prop_type = prop.get('type')
|
|
if prop_type == 'integer':
|
|
try:
|
|
return int(stripped)
|
|
except ValueError:
|
|
return prop.get('default', 0)
|
|
elif prop_type == 'number':
|
|
try:
|
|
return float(stripped)
|
|
except ValueError:
|
|
return prop.get('default', 0.0)
|
|
|
|
# Try parsing as number (fallback) — skip when schema explicitly says string
|
|
if not (prop and prop.get('type') == 'string'):
|
|
try:
|
|
if '.' in stripped:
|
|
return float(stripped)
|
|
return int(stripped)
|
|
except ValueError:
|
|
pass
|
|
|
|
# Return as string
|
|
return value
|
|
|
|
return value
|
|
def _resolve_key_segments(key_path, config):
|
|
"""Split a dot-notation path into segments, greedily preserving keys that
|
|
themselves contain dots (e.g. league keys like "fifa.world").
|
|
|
|
At each level the longest candidate that matches a key already present in the
|
|
config wins; otherwise the path splits on the next dot (the normal
|
|
nested-create case). Because dotted keys such as ``leagues."fifa.world"``
|
|
always exist in the saved config being updated, this routes the value to the
|
|
real league object instead of fabricating a ``leagues.fifa.world`` tree.
|
|
"""
|
|
parts = key_path.split('.')
|
|
segments = []
|
|
node = config
|
|
i = 0
|
|
while i < len(parts):
|
|
matched = False
|
|
if isinstance(node, dict):
|
|
for j in range(len(parts), i, -1):
|
|
candidate = '.'.join(parts[i:j])
|
|
if candidate in node:
|
|
segments.append(candidate)
|
|
node = node[candidate]
|
|
i = j
|
|
matched = True
|
|
break
|
|
if not matched:
|
|
part = parts[i]
|
|
segments.append(part)
|
|
node = node.get(part) if isinstance(node, dict) else None
|
|
i += 1
|
|
return segments
|
|
def _set_nested_value(config, key_path, value):
|
|
"""
|
|
Set a value in a nested dict using dot notation path.
|
|
Handles existing nested dicts correctly by merging instead of replacing.
|
|
Keys containing dots (e.g. league keys like "fifa.world") are preserved when
|
|
they already exist in the config rather than being split into nested objects.
|
|
|
|
Args:
|
|
config: The config dict to modify
|
|
key_path: Dot-separated path (e.g., "customization.period_text.font")
|
|
value: The value to set (or _SKIP_FIELD to skip setting)
|
|
"""
|
|
# Skip setting if value is the sentinel
|
|
if value is _SKIP_FIELD:
|
|
return
|
|
|
|
segments = _resolve_key_segments(key_path, config)
|
|
current = config
|
|
|
|
# Navigate/create intermediate dicts
|
|
for seg in segments[:-1]:
|
|
if seg not in current:
|
|
current[seg] = {}
|
|
elif not isinstance(current[seg], dict):
|
|
# If the existing value is not a dict, replace it with a dict
|
|
current[seg] = {}
|
|
current = current[seg]
|
|
|
|
# Set the final value. _SKIP_FIELD (checked above) is the only sentinel
|
|
# for "leave the existing value alone" -- an explicit None here is a real
|
|
# value (e.g. the per-mode "inherit the base" override) and must overwrite
|
|
# whatever was already stored.
|
|
current[segments[-1]] = value
|
|
|
|
|
|
#: Hidden field the rendered plugin form repeats once per top-level section it
|
|
#: drew. Named with a leading underscore pair so the save path can drop it (and
|
|
#: anything else meta) before treating form keys as config paths.
|
|
_RENDERED_SECTION_FIELD = '__rendered_section'
|
|
|
|
|
|
def _is_hidden_prop(prop_schema):
|
|
"""Whether a schema property is declared ``"x-display": "hidden"``.
|
|
|
|
Mirrors ``prop_is_hidden`` in ``plugin_config.html``: an object whose every
|
|
child is hidden counts as hidden too. A hidden property stays declared (so
|
|
stored configs carrying it keep validating) but the form draws no control
|
|
for it, which puts it outside everything the save infers from the form.
|
|
"""
|
|
if not isinstance(prop_schema, dict):
|
|
return False
|
|
if prop_schema.get('x-display') == 'hidden':
|
|
return True
|
|
children = prop_schema.get('properties')
|
|
if isinstance(children, dict) and children:
|
|
return all(_is_hidden_prop(child) for child in children.values())
|
|
return False
|
|
|
|
|
|
def _hidden_array_item_property(schema, key_path):
|
|
"""The hidden schema property a posted array-row key names, else None.
|
|
|
|
Only paths through an array index (``countdowns.0.id``) qualify: that is
|
|
the one place the rendered form posts a hidden value, because a posted row
|
|
replaces the stored item wholesale. The value is JSON-encoded there so its
|
|
exact type survives the round trip.
|
|
"""
|
|
parts = key_path.split('.')
|
|
digit_at = next((i for i, p in enumerate(parts) if p.isdigit()), None)
|
|
if digit_at is None or digit_at == 0 or digit_at == len(parts) - 1:
|
|
return None
|
|
array_prop = _get_schema_property(schema, '.'.join(parts[:digit_at]))
|
|
if not _schema_type_is(array_prop, 'array'):
|
|
return None
|
|
items = array_prop.get('items')
|
|
if not isinstance(items, dict) or not isinstance(items.get('properties'), dict):
|
|
return None
|
|
rest = '.'.join(parts[digit_at + 1:])
|
|
item_schema = {'properties': items['properties']}
|
|
prop = _get_schema_property(item_schema, rest)
|
|
if prop is None:
|
|
# Deeper array inside the item (rare): keep walking.
|
|
return _hidden_array_item_property(item_schema, rest)
|
|
return prop if _is_hidden_prop(prop) else None
|
|
|
|
|
|
def _boolean_is_in_scope(full_path, prefix, sections, submitted_parents):
|
|
"""Whether a missing checkbox at ``full_path`` may be forced to False.
|
|
|
|
"Missing" only means "unchecked" for a form that actually rendered the
|
|
control. A caller that posts a handful of fields -- a script, the MQTT
|
|
bridge, a curl against the documented endpoint -- never rendered anything,
|
|
and reading its silence as "every other checkbox is off" turns a one-field
|
|
save into a mass disable. That is not hypothetical: a partial post of four
|
|
``customization.*`` keys switched off ``nfl.enabled``, ``ncaa_fb.enabled``
|
|
and every display-mode toggle on a live device.
|
|
|
|
Two ways to be in scope:
|
|
|
|
* ``sections`` -- the rendered form lists the top-level sections it drew
|
|
(``__rendered_section``). Anything it drew is fair game, including a
|
|
section whose only fields are checkboxes that are all unchecked, which
|
|
is the case no heuristic can recover.
|
|
* ``submitted_parents`` -- no marker, so fall back to evidence: the
|
|
containing object must have had at least one field posted.
|
|
"""
|
|
if sections is not None:
|
|
return full_path.split('.', 1)[0] in sections
|
|
if submitted_parents is None:
|
|
return True
|
|
return prefix in submitted_parents
|
|
|
|
|
|
def _submitted_parents(form_keys):
|
|
"""The object paths a form actually posted a field from ('' = top level)."""
|
|
parents = set()
|
|
for key in form_keys:
|
|
parents.add(key.rsplit('.', 1)[0] if '.' in key else '')
|
|
return parents
|
|
|
|
|
|
def _set_missing_booleans_to_false(plugin_config, schema_props, form_keys, prefix='', config_node=None,
|
|
sections=None, submitted_parents=None):
|
|
"""Walk schema and set missing boolean form fields to False.
|
|
|
|
HTML checkboxes don't submit values when unchecked. When saving plugin plugin_config,
|
|
the backend starts from existing plugin_config (to support partial form updates), which
|
|
means an unchecked checkbox's old ``True`` value persists. This function detects
|
|
boolean schema properties not present in the form submission and explicitly sets
|
|
them to ``False`` -- but only where that silence is evidence, see
|
|
:func:`_boolean_is_in_scope`.
|
|
|
|
The top-level ``enabled`` field is excluded because it has its own preservation
|
|
logic in the save endpoint.
|
|
|
|
Handles boolean fields inside nested objects and inside arrays of objects
|
|
(e.g. ``feeds.custom_feeds.0.enabled``).
|
|
|
|
Args:
|
|
plugin_config: The root plugin plugin_config dict (used for pure-dict paths)
|
|
schema_props: Schema ``properties`` dict at the current nesting level
|
|
form_keys: Set of form field names that were submitted
|
|
prefix: Dot-notation prefix for the current nesting level
|
|
config_node: The current plugin_config subtree when inside an array item (avoids
|
|
using _set_nested_value which corrupts lists)
|
|
sections: Top-level sections the form reported rendering, or None when it
|
|
reported none (then submitted_parents decides)
|
|
submitted_parents: Object paths with at least one posted field; computed
|
|
on the first call when there are no section markers
|
|
"""
|
|
if sections is None and submitted_parents is None:
|
|
submitted_parents = _submitted_parents(form_keys)
|
|
# Determine which plugin_config node to operate on
|
|
node = config_node if config_node is not None else plugin_config
|
|
|
|
for prop_name, prop_schema in schema_props.items():
|
|
if not isinstance(prop_schema, dict):
|
|
continue
|
|
|
|
# A hidden property (or an object of nothing but hidden ones) was never
|
|
# drawn, so its absence from the form is not an unchecked box. Leave it
|
|
# -- and everything under it -- exactly as stored.
|
|
if _is_hidden_prop(prop_schema):
|
|
continue
|
|
|
|
full_path = f"{prefix}.{prop_name}" if prefix else prop_name
|
|
prop_type = prop_schema.get('type')
|
|
|
|
if prop_type == 'boolean' and full_path != 'enabled':
|
|
# If this boolean wasn't submitted in the form, it's an unchecked
|
|
# checkbox -- provided the form drew it at all.
|
|
if (full_path not in form_keys
|
|
and _boolean_is_in_scope(full_path, prefix, sections,
|
|
submitted_parents)):
|
|
if config_node is not None:
|
|
# Inside an array item — set directly on the item dict
|
|
node[prop_name] = False
|
|
else:
|
|
# Pure dict path — use helper
|
|
_set_nested_value(plugin_config, full_path, False)
|
|
|
|
elif prop_type == 'object' and 'properties' in prop_schema:
|
|
# Recurse into nested objects
|
|
if config_node is not None:
|
|
# Inside an array item. Walk into the existing dict, or a scratch
|
|
# one that is attached only if a boolean actually landed in it:
|
|
# an optional object the item never had (a custom feed with no
|
|
# logo) must stay absent, because an empty stub fails the
|
|
# object's own `required` list and 400s every save.
|
|
existing = node.get(prop_name)
|
|
child = existing if isinstance(existing, dict) else {}
|
|
_set_missing_booleans_to_false(
|
|
plugin_config, prop_schema['properties'], form_keys, full_path,
|
|
config_node=child,
|
|
sections=sections, submitted_parents=submitted_parents
|
|
)
|
|
if child is not existing and child:
|
|
node[prop_name] = child
|
|
else:
|
|
_set_missing_booleans_to_false(
|
|
plugin_config, prop_schema['properties'], form_keys, full_path,
|
|
sections=sections, submitted_parents=submitted_parents
|
|
)
|
|
|
|
elif prop_type == 'array':
|
|
# Handle arrays of objects that may contain boolean fields
|
|
# Form keys use indexed notation: "path.0.field", "path.1.field"
|
|
items_schema = prop_schema.get('items', {})
|
|
if isinstance(items_schema, dict) and items_schema.get('type') == 'object' and 'properties' in items_schema:
|
|
array_prefix = f"{full_path}."
|
|
# Collect unique item indices from submitted form keys
|
|
indices = set()
|
|
for k in form_keys:
|
|
if k.startswith(array_prefix):
|
|
# Extract index: "path.0.field" -> "0"
|
|
rest = k[len(array_prefix):]
|
|
idx = rest.split('.', 1)[0]
|
|
if idx.isdigit():
|
|
indices.add(int(idx))
|
|
|
|
if not indices:
|
|
continue
|
|
|
|
# Navigate to the array in the plugin_config (create if missing)
|
|
if config_node is not None:
|
|
if prop_name not in node or not isinstance(node[prop_name], list):
|
|
node[prop_name] = []
|
|
array_list = node[prop_name]
|
|
else:
|
|
# Navigate from root plugin_config through dict keys to get the list
|
|
parts = full_path.split('.')
|
|
current = plugin_config
|
|
for part in parts[:-1]:
|
|
if part not in current or not isinstance(current[part], dict):
|
|
current[part] = {}
|
|
current = current[part]
|
|
arr_key = parts[-1]
|
|
if arr_key not in current or not isinstance(current[arr_key], list):
|
|
current[arr_key] = []
|
|
array_list = current[arr_key]
|
|
|
|
# Recurse into each array item so its missing booleans get set to False
|
|
for idx in indices:
|
|
# Ensure list is long enough and item is a dict
|
|
while len(array_list) <= idx:
|
|
array_list.append({})
|
|
if not isinstance(array_list[idx], dict):
|
|
array_list[idx] = {}
|
|
item_prefix = f"{full_path}.{idx}"
|
|
_set_missing_booleans_to_false(
|
|
plugin_config, items_schema['properties'], form_keys, item_prefix,
|
|
config_node=array_list[idx],
|
|
sections=sections, submitted_parents=submitted_parents
|
|
)
|
|
def _enhance_schema_with_core_properties(schema):
|
|
"""
|
|
Enhance schema with the core-owned per-plugin properties.
|
|
|
|
``enabled``, ``display_duration``, ``live_priority`` and the ``vegas_*``
|
|
tuning keys are system-managed and always allowed, even when the plugin's schema doesn't declare them. The
|
|
list is ``schema_manager.CORE_PLUGIN_PROPERTIES``, the one validation uses,
|
|
so the save filter keeps exactly what validation accepts.
|
|
|
|
Args:
|
|
schema: The original JSON schema dict
|
|
|
|
Returns:
|
|
Enhanced schema dict with core properties injected
|
|
"""
|
|
from src.plugin_system.schema_manager import with_core_plugin_properties
|
|
|
|
if not schema:
|
|
return schema
|
|
return with_core_plugin_properties(schema)
|
|
|
|
|
|
def _prepared_plugin_config(plugin_id, raw_config):
|
|
"""A plugin's config section as the plugin runs with it, for on_config_change.
|
|
|
|
Loading a plugin reads legacy booleans as objects and fills in schema
|
|
defaults (PluginManager.prepare_plugin_config); a save's notification must
|
|
hand over the same shape. Falls back to the raw section.
|
|
"""
|
|
prepare = getattr(api_v3.plugin_manager, 'prepare_plugin_config', None)
|
|
if callable(prepare):
|
|
try:
|
|
prepared = prepare(plugin_id, raw_config)
|
|
if isinstance(prepared, dict):
|
|
return prepared
|
|
except Exception:
|
|
logger.debug("Could not prepare config for %s", plugin_id, exc_info=True)
|
|
return raw_config
|
|
|
|
|
|
def _filter_config_by_schema(config, schema, prefix=''):
|
|
"""
|
|
Filter config to only include fields defined in the schema.
|
|
Removes fields not in schema, especially important when additionalProperties is false.
|
|
|
|
Args:
|
|
config: The config dict to filter
|
|
schema: The JSON schema dict
|
|
prefix: Prefix for nested paths (used recursively)
|
|
|
|
Returns:
|
|
Filtered config dict containing only schema-defined fields
|
|
"""
|
|
if not schema or 'properties' not in schema:
|
|
return config
|
|
|
|
filtered = {}
|
|
schema_props = schema.get('properties', {})
|
|
|
|
for key, value in config.items():
|
|
if key not in schema_props:
|
|
# Field not in schema, skip it
|
|
continue
|
|
|
|
prop_schema = schema_props[key]
|
|
|
|
# Handle nested objects recursively
|
|
if isinstance(value, dict) and prop_schema.get('type') == 'object' and 'properties' in prop_schema:
|
|
filtered[key] = _filter_config_by_schema(value, prop_schema, f"{prefix}.{key}" if prefix else key)
|
|
else:
|
|
# Keep the value as-is for non-object types
|
|
filtered[key] = value
|
|
|
|
return filtered
|
|
_MAX_CREDENTIAL_BACKUPS = 5
|
|
def _prune_credential_backups(plugin_dir: Path) -> None:
|
|
"""Keep only the newest _MAX_CREDENTIAL_BACKUPS credential backups.
|
|
|
|
Every re-upload copies the previous credentials.json aside. Without
|
|
pruning those accumulate for the life of the install — each one a
|
|
complete set of OAuth client credentials sitting in the plugin
|
|
directory.
|
|
"""
|
|
backups = sorted(
|
|
plugin_dir.glob('credentials.json.backup.*'),
|
|
key=lambda p: p.stat().st_mtime,
|
|
reverse=True,
|
|
)
|
|
for stale in backups[_MAX_CREDENTIAL_BACKUPS:]:
|
|
try:
|
|
stale.unlink()
|
|
except OSError:
|
|
logger.warning("Could not remove old credential backup %s", stale.name)
|
|
# calendarList.list pages at 250 entries maximum. Ten pages is far past any
|
|
# real account and exists only so a malformed nextPageToken cannot spin here.
|
|
_CALENDAR_LIST_MAX_PAGES = 10
|
|
def _calendar_plugin_dir() -> Optional[Path]:
|
|
"""Where the calendar plugin is installed, or None if it is not."""
|
|
if api_v3.plugin_manager:
|
|
plugin_dir = api_v3.plugin_manager.get_plugin_directory('calendar')
|
|
else:
|
|
plugin_dir = PROJECT_ROOT / 'plugins' / 'calendar'
|
|
if not plugin_dir:
|
|
return None
|
|
plugin_dir = Path(plugin_dir)
|
|
return plugin_dir if plugin_dir.exists() else None
|
|
def _run_calendar_registration(plugin_dir: Path, stdin_payload: str):
|
|
"""Run the plugin's OAuth script and return the JSON object it prints.
|
|
|
|
The script decides between web and terminal mode by whether stdin is a
|
|
tty, so it must be given a pipe. It emits one JSON object on stdout; the
|
|
last parsable line is taken, because an import warning or a library's
|
|
stderr redirection can land in front of it.
|
|
|
|
Returns (payload, error_message). Exactly one is None.
|
|
"""
|
|
script = plugin_dir / 'calendar_registration.py'
|
|
if not script.exists():
|
|
return None, 'Authentication script not found in the calendar plugin'
|
|
|
|
try:
|
|
result = subprocess.run( # nosec B603 - fixed script path inside the plugin dir
|
|
[sys.executable, str(script)],
|
|
input=stdin_payload,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
cwd=str(plugin_dir),
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return None, 'Authentication timed out after 120s'
|
|
except OSError as e:
|
|
logger.error('Could not run calendar_registration.py', exc_info=True)
|
|
return None, 'Could not run the authentication script: %s' % describe_exception(e)
|
|
|
|
for line in reversed((result.stdout or '').splitlines()):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
payload = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(payload, dict):
|
|
return payload, None
|
|
|
|
raw = (result.stderr or result.stdout or '').strip()
|
|
# Redacted in the log too: this is a script that handles OAuth client
|
|
# secrets, and its stderr can quote them verbatim (CWE-532).
|
|
if raw:
|
|
logger.error('calendar_registration.py failed (exit %s): %s',
|
|
result.returncode, redact_text(raw))
|
|
return None, 'Authentication script produced no result%s' % (
|
|
': %s' % redact_text(raw) if raw else '')
|
|
def _resolve_backup_export_dir() -> Path:
|
|
"""Where exported backups live: beside the install, not inside it.
|
|
|
|
They used to be written to ``<project>/config/backups/exports``. That is
|
|
inside the directory a reinstall deletes, so the documented recovery path
|
|
-- export a backup, then reinstall -- destroyed the backup it had just
|
|
told the user to make. Anyone who downloaded the ZIP was fine; anyone
|
|
relying on the on-device copy was not.
|
|
|
|
Falls back to the old location when the parent directory is not writable,
|
|
so an unusual layout degrades to previous behaviour instead of failing to
|
|
export at all.
|
|
"""
|
|
preferred = PROJECT_ROOT.parent / "ledmatrix-backups"
|
|
fallback = PROJECT_ROOT / "config" / "backups" / "exports"
|
|
try:
|
|
preferred.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile(dir=preferred, prefix=".writetest-"):
|
|
pass
|
|
return preferred
|
|
except OSError as e:
|
|
logger.warning(
|
|
f"[Backup] Export dir {preferred} is not writable ({e}); "
|
|
f"falling back to {fallback}, which a reinstall will delete"
|
|
)
|
|
return fallback
|
|
_BACKUP_EXPORT_DIR = _resolve_backup_export_dir()
|
|
def _safe_backup_path(filename: str) -> Path:
|
|
"""Resolve a filename to an absolute path inside the export dir,
|
|
rejecting any traversal attempts. Returns None if unsafe."""
|
|
# Use basename first (CodeQL-recognized sanitizer) then validate format
|
|
filename = os.path.basename(filename or '')
|
|
if not filename or not re.match(r'^[a-zA-Z0-9][a-zA-Z0-9._-]{0,200}\.zip$', filename):
|
|
return None
|
|
path = (_BACKUP_EXPORT_DIR / filename).resolve()
|
|
try:
|
|
path.relative_to(_BACKUP_EXPORT_DIR.resolve())
|
|
except ValueError:
|
|
return None
|
|
return path
|
|
_STARLARK_APPS_DIR = PROJECT_ROOT / 'starlark-apps'
|
|
_STARLARK_MANIFEST_FILE = _STARLARK_APPS_DIR / 'manifest.json'
|
|
# A dedicated, never-replaced file to flock -- see _starlark_manifest_lock.
|
|
_STARLARK_MANIFEST_LOCK_FILE = _STARLARK_APPS_DIR / 'manifest.json.lock'
|
|
def _get_starlark_plugin() -> Optional[Any]:
|
|
"""Get the starlark-apps plugin instance, or None."""
|
|
if not api_v3.plugin_manager:
|
|
return None
|
|
return api_v3.plugin_manager.get_plugin('starlark-apps')
|
|
def _find_pixlet_binary(explicit_path: Optional[str] = None) -> Optional[str]:
|
|
"""Find pixlet binary: explicit path → bundled binary → system PATH."""
|
|
import platform
|
|
if explicit_path and os.path.isfile(explicit_path) and os.access(explicit_path, os.X_OK):
|
|
return explicit_path
|
|
bin_dir = PROJECT_ROOT / "bin" / "pixlet"
|
|
system = platform.system().lower()
|
|
machine = platform.machine().lower()
|
|
if system == "linux":
|
|
if "aarch64" in machine or "arm64" in machine:
|
|
name = "pixlet-linux-arm64"
|
|
elif "x86_64" in machine or "amd64" in machine:
|
|
name = "pixlet-linux-amd64"
|
|
else:
|
|
name = None
|
|
elif system == "darwin":
|
|
name = "pixlet-darwin-arm64" if "arm64" in machine else "pixlet-darwin-amd64"
|
|
else:
|
|
name = None
|
|
if name:
|
|
bundled = bin_dir / name
|
|
if bundled.is_file():
|
|
if os.access(str(bundled), os.X_OK):
|
|
return str(bundled)
|
|
try:
|
|
bundled.chmod(0o755)
|
|
except OSError:
|
|
logger.warning("Could not make pixlet bundled binary executable (%s); falling back to PATH", bundled)
|
|
else:
|
|
if os.access(str(bundled), os.X_OK):
|
|
return str(bundled)
|
|
logger.warning("Pixlet bundled binary still not executable after chmod (%s); falling back to PATH", bundled)
|
|
return shutil.which("pixlet")
|
|
@contextlib.contextmanager
|
|
def _starlark_manifest_lock():
|
|
"""Hold an exclusive lock across a standalone starlark-manifest read-modify-write.
|
|
|
|
StarlarkAppsPlugin._update_manifest_safe (plugin-repos/starlark-apps/manager.py)
|
|
already does this -- fcntl.flock held for the whole read-modify-write cycle --
|
|
when the plugin instance is loaded. These routes fall back to reading and
|
|
writing manifest.json directly when it is not, and did so with no lock: each
|
|
_write_starlark_manifest() call is atomic on its own (temp file + rename), but
|
|
two concurrent requests can each read the manifest, mutate their own copy, and
|
|
write it back, and the second write silently discards the first's change.
|
|
|
|
Locks _STARLARK_MANIFEST_LOCK_FILE, a sidecar that is never written to or
|
|
renamed over -- not manifest.json itself. manifest.json is replaced by an
|
|
atomic rename on every write (here and in the plugin), which swaps in a
|
|
fresh inode; a lock held on the old inode does not exclude a second locker
|
|
that opens the path afresh right after the rename and gets the new inode,
|
|
so two writers could still race each other despite both "holding a lock".
|
|
A stable sidecar path always resolves to the same inode, so every locker
|
|
-- standalone or plugin-owned -- contends for the same lock. The plugin
|
|
must lock this same sidecar file for that guarantee to hold across both.
|
|
|
|
Callers should do their read, mutation and _write_starlark_manifest() call
|
|
entirely inside the `with` block, mirroring the plugin's lock scope.
|
|
"""
|
|
# Imported here, not at module scope: fcntl is POSIX-only, and a top-level
|
|
# import made the whole api_v3 package unimportable on Windows -- which the
|
|
# monolithic blueprint never was, so the split would have broken local dev
|
|
# and the test suite there. Deliberately NOT degraded to a no-op lock off
|
|
# POSIX: the docstring above describes a real lost-update race, and silently
|
|
# not locking would be worse than failing loudly on a platform that cannot
|
|
# run the display anyway.
|
|
import fcntl
|
|
|
|
_STARLARK_APPS_DIR.mkdir(parents=True, exist_ok=True)
|
|
lock_fd = os.open(str(_STARLARK_MANIFEST_LOCK_FILE), os.O_CREAT | os.O_RDWR, 0o644)
|
|
try:
|
|
fcntl.flock(lock_fd, fcntl.LOCK_EX)
|
|
try:
|
|
yield
|
|
finally:
|
|
fcntl.flock(lock_fd, fcntl.LOCK_UN)
|
|
finally:
|
|
os.close(lock_fd)
|
|
def _read_starlark_manifest() -> Dict[str, Any]:
|
|
"""Read the starlark-apps manifest.json directly from disk."""
|
|
try:
|
|
if _STARLARK_MANIFEST_FILE.exists():
|
|
with open(_STARLARK_MANIFEST_FILE, 'r') as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
logger.error(f"Error reading starlark manifest: {e}")
|
|
return {'apps': {}}
|
|
def _starlark_github_token() -> Optional[str]:
|
|
"""The GitHub token the Starlark store should authenticate with.
|
|
|
|
These routes used to read `github_token` off config.json, a key that is
|
|
written nowhere and offered by no setting -- so the store always ran
|
|
unauthenticated at 60 requests/hour, on the same per-IP budget every
|
|
plugin update check spends, while the token the user had actually
|
|
configured sat in config_secrets.json raising the same budget to 5000.
|
|
The store going blank was that budget running out.
|
|
|
|
Prefer the store manager's token, which is the one the settings UI
|
|
writes and validates; keep the config.json key as a fallback so a
|
|
hand-edited config still works.
|
|
"""
|
|
token = getattr(api_v3.plugin_store_manager, 'github_token', None)
|
|
if token:
|
|
return token
|
|
|
|
try:
|
|
config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
|
|
return config.get('github_token')
|
|
except Exception:
|
|
logger.warning("[Starlark] Could not read config for a GitHub token", exc_info=True)
|
|
return None
|
|
def _get_tronbyte_repository_class() -> Type[Any]:
|
|
"""Import TronbyteRepository from plugin-repos directory."""
|
|
import importlib.util
|
|
import importlib
|
|
|
|
module_path = PROJECT_ROOT / 'plugin-repos' / 'starlark-apps' / 'tronbyte_repository.py'
|
|
if not module_path.exists():
|
|
raise ImportError(f"TronbyteRepository module not found at {module_path}")
|
|
|
|
# If already imported, return cached class
|
|
if "tronbyte_repository" in sys.modules:
|
|
return sys.modules["tronbyte_repository"].TronbyteRepository
|
|
|
|
spec = importlib.util.spec_from_file_location("tronbyte_repository", str(module_path))
|
|
if spec is None:
|
|
raise ImportError(f"Failed to create module spec for tronbyte_repository at {module_path}")
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
if module is None:
|
|
raise ImportError("Failed to create module from spec for tronbyte_repository")
|
|
|
|
sys.modules["tronbyte_repository"] = module
|
|
try:
|
|
spec.loader.exec_module(module)
|
|
except BaseException:
|
|
# A module that failed to execute must not stay in sys.modules: the
|
|
# cache branch above would hand back the half-initialised object for
|
|
# the rest of the process, so one transient failure would disable
|
|
# this path permanently and surface as AttributeError, not ImportError.
|
|
sys.modules.pop("tronbyte_repository", None)
|
|
raise
|
|
return module.TronbyteRepository
|
|
def _get_pixlet_renderer_class() -> Type[Any]:
|
|
"""Import PixletRenderer from plugin-repos directory."""
|
|
import importlib.util
|
|
import importlib
|
|
|
|
module_path = PROJECT_ROOT / 'plugin-repos' / 'starlark-apps' / 'pixlet_renderer.py'
|
|
if not module_path.exists():
|
|
raise ImportError(f"PixletRenderer module not found at {module_path}")
|
|
|
|
# If already imported, return cached class
|
|
if "pixlet_renderer" in sys.modules:
|
|
return sys.modules["pixlet_renderer"].PixletRenderer
|
|
|
|
spec = importlib.util.spec_from_file_location("pixlet_renderer", str(module_path))
|
|
if spec is None:
|
|
raise ImportError(f"Failed to create module spec for pixlet_renderer at {module_path}")
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
if module is None:
|
|
raise ImportError("Failed to create module from spec for pixlet_renderer")
|
|
|
|
sys.modules["pixlet_renderer"] = module
|
|
try:
|
|
spec.loader.exec_module(module)
|
|
except BaseException:
|
|
# A module that failed to execute must not stay in sys.modules: the
|
|
# cache branch above would hand back the half-initialised object for
|
|
# the rest of the process, so one transient failure would disable
|
|
# this path permanently and surface as AttributeError, not ImportError.
|
|
sys.modules.pop("pixlet_renderer", None)
|
|
raise
|
|
return module.PixletRenderer
|
|
def _validate_and_sanitize_app_id(app_id: Optional[str], fallback_source: Optional[str] = None) -> Tuple[Optional[str], Optional[str]]:
|
|
"""Validate and sanitize app_id to a safe slug."""
|
|
if not app_id and fallback_source:
|
|
app_id = fallback_source
|
|
if not app_id:
|
|
return None, "app_id is required"
|
|
if '..' in app_id or '/' in app_id or '\\' in app_id:
|
|
return None, "app_id contains invalid characters"
|
|
|
|
sanitized = re.sub(r'[^a-z0-9_]', '_', app_id.lower()).strip('_')
|
|
if not sanitized:
|
|
sanitized = f"app_{hashlib.sha256(app_id.encode()).hexdigest()[:12]}"
|
|
if sanitized[0].isdigit():
|
|
sanitized = f"app_{sanitized}"
|
|
return sanitized, None
|
|
def _validate_timing_value(value: Any, field_name: str, min_val: int = 1, max_val: int = 86400) -> Tuple[Optional[int], Optional[str]]:
|
|
"""Validate and coerce timing values."""
|
|
if value is None:
|
|
return None, None
|
|
try:
|
|
int_value = int(value)
|
|
except (ValueError, TypeError):
|
|
return None, f"{field_name} must be an integer"
|
|
if int_value < min_val:
|
|
return None, f"{field_name} must be at least {min_val}"
|
|
if int_value > max_val:
|
|
return None, f"{field_name} must be at most {max_val}"
|
|
return int_value, None
|
|
def _validate_starlark_app_path(app_id: str) -> Tuple[Optional[Path], Optional[str]]:
|
|
"""The app's directory, or an error if app_id could escape the base dir.
|
|
|
|
Returns the *resolved* path rather than a boolean, and every caller uses
|
|
what it returns instead of re-joining ``_STARLARK_APPS_DIR / app_id``
|
|
afterwards. The old shape validated in one place and rebuilt the path in
|
|
another, which is two things that have to stay in step -- and is why
|
|
CodeQL reported twenty-four path-injection alerts across these handlers
|
|
even though the guard was effective: a boolean is not a sanitiser it can
|
|
follow, and the value reaching the filesystem was the raw one.
|
|
|
|
The name is unchanged so the call sites read the same.
|
|
"""
|
|
if not isinstance(app_id, str) or not app_id:
|
|
return None, "Invalid app_id"
|
|
|
|
# Reject the traversal characters outright before touching the filesystem.
|
|
if '..' in app_id or '/' in app_id or '\\' in app_id:
|
|
return None, "Invalid app_id: contains path traversal characters"
|
|
|
|
# os.path.basename strips any directory component, so what is joined below
|
|
# cannot carry one. The equality check means this rejects rather than
|
|
# silently truncates -- behaviour is identical to the character test above,
|
|
# and it is the sanitiser CodeQL's path-injection query actually follows.
|
|
# relative_to() alone is a check it cannot trace, which is why twenty-four
|
|
# of these stayed flagged after the value was threaded through properly.
|
|
safe_name = os.path.basename(app_id)
|
|
if safe_name != app_id or safe_name in ('', '.', '..'):
|
|
return None, "Invalid app_id: contains path traversal characters"
|
|
|
|
try:
|
|
base_path = _STARLARK_APPS_DIR.resolve()
|
|
app_path = (base_path / safe_name).resolve()
|
|
try:
|
|
app_path.relative_to(base_path)
|
|
except ValueError:
|
|
return None, "Invalid app_id: path traversal attempt"
|
|
return app_path, None
|
|
except OSError as e:
|
|
logger.warning("Path validation error for app_id %r: %s", app_id, e)
|
|
return None, "Invalid app_id"
|
|
_starlark_device_location: Optional[DeviceLocationResolver] = None
|
|
|
|
|
|
def _get_starlark_device_location() -> DeviceLocationResolver:
|
|
"""One resolver per process, so a geocode failure's backoff is shared."""
|
|
global _starlark_device_location
|
|
if _starlark_device_location is None:
|
|
_starlark_device_location = DeviceLocationResolver(
|
|
getattr(api_v3, 'cache_manager', None) or _ensure_cache_manager(), logger)
|
|
return _starlark_device_location
|
|
|
|
|
|
def _read_starlark_schema(app_dir: Path) -> Optional[Dict[str, Any]]:
|
|
"""An installed app's schema.json, or None if it has none or it's unreadable."""
|
|
schema_file = app_dir / 'schema.json'
|
|
if not schema_file.exists():
|
|
return None
|
|
try:
|
|
with open(schema_file) as f:
|
|
schema = json.load(f)
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
logger.warning("Could not read schema.json at %s: %s", schema_file, e)
|
|
return None
|
|
return schema if isinstance(schema, dict) else None
|
|
|
|
|
|
def _standalone_render_starlark_app(app_id: str) -> Tuple[bool, int, Optional[str]]:
|
|
"""Render a Starlark app via pixlet directly (no plugin required).
|
|
|
|
Reads the .star file and config from starlark-apps/{app_id}/, runs pixlet,
|
|
and saves the output to cached_render.webp in the same directory.
|
|
This is the web-service fallback when starlark-apps plugin is not loaded.
|
|
|
|
Returns (success, http_status_code, error_message).
|
|
"""
|
|
manifest = _read_starlark_manifest()
|
|
if not isinstance(manifest, dict):
|
|
return False, 400, "Invalid manifest shape: expected object with 'apps' mapping"
|
|
apps = manifest.get('apps', {})
|
|
if not isinstance(apps, dict):
|
|
return False, 400, "Invalid manifest shape: expected object with 'apps' mapping"
|
|
app_data = apps.get(app_id)
|
|
if not app_data:
|
|
return False, 404, f"App not found: {app_id}"
|
|
|
|
# Validated here as well as at the handler: this is reachable on its own,
|
|
# and a path built from app_id should never be assembled without it.
|
|
app_dir, path_error = _validate_starlark_app_path(app_id)
|
|
if path_error:
|
|
return False, 400, path_error
|
|
star_file = app_dir / app_data.get('star_file', f'{app_id}.star')
|
|
if not star_file.exists():
|
|
return False, 404, f"Star file not found: {star_file}"
|
|
|
|
full_config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
|
|
plugin_config = full_config.get('starlark-apps', {})
|
|
|
|
pixlet_path = _find_pixlet_binary(plugin_config.get('pixlet_path'))
|
|
if not pixlet_path:
|
|
return False, 503, "Pixlet binary not found — install pixlet first"
|
|
|
|
magnify = plugin_config.get('magnify')
|
|
if magnify is None:
|
|
# The size DisplayManager renders at (shared defaults, double-sided
|
|
# applied), so the Pixlet render matches the screen it lands on
|
|
from src.display_geometry import logical_size
|
|
try:
|
|
width, height = logical_size(full_config)
|
|
except (TypeError, ValueError):
|
|
width, height = 64, 32
|
|
magnify = max(1, min(8, int(min(width / 64, height / 32))))
|
|
else:
|
|
try:
|
|
magnify = max(1, min(8, int(magnify)))
|
|
except (ValueError, TypeError):
|
|
magnify = 1
|
|
|
|
config_file = app_dir / 'config.json'
|
|
app_config: Dict[str, Any] = {}
|
|
if config_file.exists():
|
|
try:
|
|
with open(config_file) as f:
|
|
app_config = json.load(f)
|
|
except json.JSONDecodeError as e:
|
|
logger.warning("Invalid config.json for %r at %s: %s", app_id, config_file, e)
|
|
return False, 400, f"Invalid config.json for {app_id}"
|
|
except OSError as e:
|
|
logger.warning("Cannot read config.json for %r at %s: %s", app_id, config_file, e)
|
|
return False, 400, f"Cannot read config.json for {app_id}"
|
|
if not isinstance(app_config, dict):
|
|
return False, 400, (
|
|
f"config.json for {app_id} must be a JSON object, "
|
|
f"got {type(app_config).__name__}"
|
|
)
|
|
|
|
INTERNAL_KEYS = {'render_interval', 'display_duration'}
|
|
pixlet_config = {k: v for k, v in app_config.items() if k not in INTERNAL_KEYS}
|
|
pixlet_config = apply_device_location(
|
|
pixlet_config, _read_starlark_schema(app_dir),
|
|
_get_starlark_device_location(), full_config)
|
|
|
|
output_path = str(app_dir / 'cached_render.webp')
|
|
cmd = [pixlet_path, 'render', str(star_file)]
|
|
for key, value in pixlet_config.items():
|
|
if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', key):
|
|
continue
|
|
value_str = 'true' if value is True else 'false' if value is False else str(value)
|
|
if re.search(r'[`$|<>&;\x00]|\$\(', value_str):
|
|
continue
|
|
cmd.append(f'{key}={value_str}')
|
|
cmd.extend(['-o', output_path, '-m', str(magnify)])
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30, cwd=str(app_dir))
|
|
if result.returncode == 0 and os.path.isfile(output_path):
|
|
return True, 200, None
|
|
return False, 502, f"Pixlet failed (exit {result.returncode}): {result.stderr.strip()}"
|
|
except subprocess.TimeoutExpired:
|
|
return False, 504, "Render timed out after 30s"
|
|
except Exception as e:
|
|
logger.exception("Starlark render failed for %r", app_id)
|
|
return False, 500, "Render error"
|
|
def _write_starlark_manifest(manifest: Dict[str, Any]) -> bool:
|
|
"""Write the starlark-apps manifest.json to disk with atomic write."""
|
|
temp_file = None
|
|
try:
|
|
_STARLARK_APPS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Atomic write: unique temp file in the target directory, then rename.
|
|
# with_suffix('.tmp') gave every caller the same manifest.tmp, and
|
|
# Flask serves concurrently -- upload, uninstall, config, toggle and
|
|
# the plugin toggle all reach here. Two writers shared one file,
|
|
# interleaved their json.dump output, and both renamed it, so the
|
|
# rename was atomic over content that was a mix of two manifests.
|
|
fd, temp_name = tempfile.mkstemp(
|
|
dir=str(_STARLARK_APPS_DIR), prefix='.manifest.', suffix='.tmp')
|
|
temp_file = Path(temp_name)
|
|
with os.fdopen(fd, 'w') as f:
|
|
json.dump(manifest, f, indent=2)
|
|
f.flush()
|
|
os.fsync(f.fileno()) # Ensure data is written to disk
|
|
os.chmod(temp_name, 0o644) # mkstemp creates 0600; match a normal write
|
|
|
|
# Atomic rename (overwrites destination)
|
|
temp_file.replace(_STARLARK_MANIFEST_FILE)
|
|
return True
|
|
except OSError as e:
|
|
logger.error(f"Error writing starlark manifest: {e}")
|
|
# Clean up temp file if it exists
|
|
if temp_file and temp_file.exists():
|
|
try:
|
|
temp_file.unlink()
|
|
except Exception:
|
|
pass
|
|
return False
|
|
def _install_star_file(app_id: str, star_file_path: str, metadata: Dict[str, Any], assets_dir: Optional[str] = None) -> bool:
|
|
"""Install a .star file and update the manifest (standalone, no plugin needed)."""
|
|
import shutil
|
|
import json
|
|
app_dir, path_error = _validate_starlark_app_path(app_id)
|
|
if path_error:
|
|
logger.warning("Refusing to install %r: %s", app_id, path_error)
|
|
return False
|
|
app_dir.mkdir(parents=True, exist_ok=True)
|
|
dest = app_dir / f"{app_id}.star"
|
|
shutil.copy2(star_file_path, str(dest))
|
|
|
|
# Copy asset directories if provided (images/, sources/, etc.)
|
|
if assets_dir and Path(assets_dir).exists():
|
|
assets_path = Path(assets_dir)
|
|
for item in assets_path.iterdir():
|
|
if item.is_dir():
|
|
# Copy entire directory (e.g., images/, sources/)
|
|
dest_dir = app_dir / item.name
|
|
if dest_dir.exists():
|
|
shutil.rmtree(dest_dir)
|
|
shutil.copytree(item, dest_dir)
|
|
logger.debug(f"Copied assets directory: {item.name}")
|
|
logger.info(f"Installed assets for {app_id}")
|
|
|
|
# Try to extract schema using PixletRenderer
|
|
schema = None
|
|
try:
|
|
PixletRenderer = _get_pixlet_renderer_class()
|
|
pixlet = PixletRenderer()
|
|
if pixlet.is_available():
|
|
_, schema, _ = pixlet.extract_schema(str(dest))
|
|
if schema:
|
|
schema_path = app_dir / "schema.json"
|
|
with open(schema_path, 'w') as f:
|
|
json.dump(schema, f, indent=2)
|
|
logger.info(f"Extracted schema for {app_id}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to extract schema for {app_id}: {e}")
|
|
|
|
# Create default config — pre-populate with schema defaults
|
|
default_config = {}
|
|
if schema:
|
|
fields = schema.get('fields') or schema.get('schema') or []
|
|
for field in fields:
|
|
if isinstance(field, dict) and 'id' in field and 'default' in field:
|
|
default_config[field['id']] = field['default']
|
|
|
|
# Create config.json file
|
|
config_path = app_dir / "config.json"
|
|
with open(config_path, 'w') as f:
|
|
json.dump(default_config, f, indent=2)
|
|
|
|
with _starlark_manifest_lock():
|
|
manifest = _read_starlark_manifest()
|
|
manifest.setdefault('apps', {})[app_id] = {
|
|
'name': metadata.get('name', app_id),
|
|
'enabled': True,
|
|
'render_interval': metadata.get('render_interval', 300),
|
|
'display_duration': metadata.get('display_duration', 15),
|
|
'config': metadata.get('config', {}),
|
|
# The filename, not the full path. Readers join this to the app's own
|
|
# directory and fall back to a bare '<app_id>.star', so an absolute
|
|
# value gave the key two meanings -- and Path.__truediv__ discards the
|
|
# left side when the right is absolute, which pinned the manifest to
|
|
# whatever PROJECT_ROOT installed it. Moving or redeploying the install
|
|
# then left the app unable to find its own file.
|
|
'star_file': dest.name,
|
|
}
|
|
return _write_starlark_manifest(manifest)
|
|
def _starlark_virtual_plugins() -> list:
|
|
"""Installed Starlark apps, shaped like plugin entries.
|
|
|
|
#253 surfaced these alongside real plugins so an installed .star app can
|
|
be seen, enabled and disabled from the same list as everything else; #330
|
|
dropped it with the rest of the Starlark code, which is why an app
|
|
installs successfully and then appears nowhere.
|
|
|
|
Reads the loaded plugin when there is one and the on-disk manifest
|
|
otherwise, so the list is right before starlark-apps has been loaded too.
|
|
"""
|
|
entries = []
|
|
base = {
|
|
'version': 'starlark', 'category': 'Starlark App', 'tags': ['starlark'],
|
|
'verified': False, 'last_updated': None, 'last_commit': None,
|
|
'last_commit_message': None, 'branch': None, 'web_ui_actions': [],
|
|
'vegas_mode': 'fixed', 'vegas_content_type': 'multi',
|
|
'is_starlark_app': True,
|
|
}
|
|
try:
|
|
plugin = _get_starlark_plugin()
|
|
if plugin is not None and hasattr(plugin, 'apps'):
|
|
for app_id, app in plugin.apps.items():
|
|
m = getattr(app, 'manifest', {}) or {}
|
|
entries.append({**base,
|
|
'id': f'starlark:{app_id}',
|
|
'name': m.get('name', app_id),
|
|
'author': m.get('author', 'Tronbyte Community'),
|
|
'description': m.get('summary', 'Starlark app'),
|
|
'enabled': app.is_enabled(),
|
|
'loaded': True})
|
|
return entries
|
|
|
|
for app_id, data in (_read_starlark_manifest().get('apps', {}) or {}).items():
|
|
entries.append({**base,
|
|
'id': f'starlark:{app_id}',
|
|
'name': data.get('name', app_id),
|
|
'author': data.get('author', 'Tronbyte Community'),
|
|
'description': data.get('summary', 'Starlark app'),
|
|
'enabled': data.get('enabled', True),
|
|
'loaded': False})
|
|
except Exception:
|
|
# Never let a Starlark problem empty the whole plugins list.
|
|
logger.exception('Could not build Starlark virtual plugin entries')
|
|
return entries
|
|
def _toggle_starlark_app(app_id: str, enabled: bool):
|
|
"""Enable or disable one Starlark app, loaded or not."""
|
|
# Check for traversal, but toggle the key that was listed.
|
|
# _starlark_virtual_plugins publishes the raw manifest key, while
|
|
# _validate_and_sanitize_app_id lowercases it and rewrites every character
|
|
# outside [a-z0-9_]: an app stored as 'My-App' was offered to the UI as
|
|
# 'starlark:My-App' and looked up here as 'my_app', so toggling an app the
|
|
# page had just drawn answered 404. Keys written by _install_star_file are
|
|
# already sanitised; ones written by the plugin, or edited by hand, are
|
|
# not. _validate_starlark_app_path rejects traversal without rewriting.
|
|
_, err = _validate_starlark_app_path(app_id)
|
|
if err:
|
|
# err already names app_id; do not prefix it a second time.
|
|
return jsonify({'status': 'error', 'message': err}), 400
|
|
safe_id = app_id
|
|
|
|
plugin = _get_starlark_plugin()
|
|
if plugin is not None and safe_id in getattr(plugin, 'apps', {}):
|
|
def _update(manifest):
|
|
# setdefault rather than indexing: the app is loaded, but its
|
|
# on-disk entry need not exist, and _update_manifest_safe does not
|
|
# catch KeyError -- it would escape as a 500 rather than the error
|
|
# this returns.
|
|
manifest.setdefault('apps', {}).setdefault(safe_id, {})['enabled'] = enabled
|
|
|
|
if plugin._update_manifest_safe(_update) is False:
|
|
return jsonify({'status': 'error',
|
|
'message': 'Failed to save app state'}), 500
|
|
# Only now is the in-memory copy allowed to disagree with disk.
|
|
plugin.apps[safe_id].manifest['enabled'] = enabled
|
|
else:
|
|
with _starlark_manifest_lock():
|
|
manifest = _read_starlark_manifest()
|
|
app_data = manifest.get('apps', {}).get(safe_id)
|
|
if not app_data:
|
|
return jsonify({'status': 'error',
|
|
'message': f'Starlark app not found: {safe_id}'}), 404
|
|
app_data['enabled'] = enabled
|
|
if not _write_starlark_manifest(manifest):
|
|
return jsonify({'status': 'error', 'message': 'Failed to save manifest'}), 500
|
|
|
|
return jsonify({'status': 'success',
|
|
'message': f"Starlark app {'enabled' if enabled else 'disabled'}",
|
|
'enabled': enabled})
|
|
|
|
|
|
_PIXLET_EDITOR_SCRIPT = PROJECT_ROOT / 'scripts' / 'utils' / 'pixlet_config_editor.sh'
|
|
|
|
# Deliberately under /tmp: a session cannot survive a reboot, so neither should
|
|
# the record of one.
|
|
_PIXLET_EDITOR_STATE = Path(tempfile.gettempdir()) / 'ledmatrix_pixlet_editor.json'
|
|
|
|
_PIXLET_EDITOR_DEFAULT_PORT = 8080
|
|
|
|
_PIXLET_EDITOR_DEFAULT_TIMEOUT = 1800
|
|
|
|
_PIXLET_EDITOR_MAX_TIMEOUT = 14400
|
|
|
|
def _read_pixlet_editor_state() -> Optional[Dict[str, Any]]:
|
|
try:
|
|
if _PIXLET_EDITOR_STATE.is_file():
|
|
with open(_PIXLET_EDITOR_STATE, encoding='utf-8') as handle:
|
|
state = json.load(handle)
|
|
return state if isinstance(state, dict) else None
|
|
except (OSError, json.JSONDecodeError):
|
|
logger.debug('Unreadable pixlet editor state; treating as no session', exc_info=True)
|
|
return None
|
|
|
|
def _clear_pixlet_editor_state() -> None:
|
|
with contextlib.suppress(OSError):
|
|
_PIXLET_EDITOR_STATE.unlink()
|
|
|
|
def _pixlet_editor_alive(pid: Optional[int]) -> bool:
|
|
"""Is the recorded session still running?
|
|
|
|
os.kill(pid, 0) is not enough on its own: the script is a child of this
|
|
process, so once it exits it stays a zombie until reaped, and signal 0
|
|
succeeds against a zombie. Left at that, a finished session would read as
|
|
running forever and the UI would keep offering a Stop button for it.
|
|
"""
|
|
if not pid:
|
|
return False
|
|
|
|
# Reap it if it is ours and already finished; harmless if it is not.
|
|
with contextlib.suppress(ChildProcessError, OSError):
|
|
reaped, _ = os.waitpid(pid, os.WNOHANG)
|
|
if reaped == pid:
|
|
return False
|
|
|
|
try:
|
|
os.kill(pid, 0)
|
|
except ProcessLookupError:
|
|
return False
|
|
except PermissionError:
|
|
# Exists but is not ours to signal, which still counts as running.
|
|
return True
|
|
|
|
# Not our child (e.g. the web service restarted under a live session), so
|
|
# waitpid told us nothing -- ask /proc whether it is merely a zombie.
|
|
with contextlib.suppress(OSError, IndexError, ValueError):
|
|
with open(f'/proc/{pid}/stat', encoding='utf-8') as handle:
|
|
# The comm field can contain spaces and parens; everything after
|
|
# the final ')' is positional, and state is the first of those.
|
|
fields = handle.read().rsplit(')', 1)[1].split()
|
|
if fields and fields[0] == 'Z':
|
|
return False
|
|
return True
|
|
|
|
def _pixlet_editor_status() -> Dict[str, Any]:
|
|
"""Current session, reconciled against reality.
|
|
|
|
The state file records what we started; the process may have ended on its
|
|
own (its timeout, a crash, a manual kill). Anything stale is cleared here so
|
|
the UI never offers a Stop button for a session that is already over.
|
|
"""
|
|
state = _read_pixlet_editor_state()
|
|
if not state:
|
|
return {'running': False}
|
|
if not _pixlet_editor_alive(state.get('pid')):
|
|
_clear_pixlet_editor_state()
|
|
return {'running': False}
|
|
|
|
remaining = None
|
|
deadline = state.get('deadline')
|
|
if isinstance(deadline, (int, float)):
|
|
remaining = max(0, int(deadline - time.time()))
|
|
return {
|
|
'running': True,
|
|
'app_id': state.get('app_id'),
|
|
'port': state.get('port', _PIXLET_EDITOR_DEFAULT_PORT),
|
|
'pid': state.get('pid'),
|
|
'started_at': state.get('started_at'),
|
|
'timeout': state.get('timeout'),
|
|
'seconds_remaining': remaining,
|
|
'host_bound': state.get('host', '0.0.0.0'),
|
|
}
|
|
|
|
_MQTT_BRIDGE_DIR = PROJECT_ROOT / 'integrations' / 'mqtt_bridge'
|
|
|
|
_MQTT_BRIDGE_CONFIG = _MQTT_BRIDGE_DIR / 'bridge_config.json'
|
|
|
|
_MQTT_BRIDGE_EXAMPLE = _MQTT_BRIDGE_DIR / 'bridge_config.example.json'
|
|
|
|
_MQTT_BRIDGE_SERVICE = 'ledmatrix-mqtt-bridge.service'
|
|
|
|
# Mirrors DEFAULTS in ledmatrix_mqtt_bridge.py. Duplicated rather than imported
|
|
# because that module pulls in paho-mqtt, which the web process does not need
|
|
# installed just to render a settings form.
|
|
_MQTT_BRIDGE_DEFAULTS = {
|
|
'mqtt_host': 'localhost',
|
|
'mqtt_port': 1883,
|
|
'mqtt_username': None,
|
|
'mqtt_client_id': 'ledmatrix-mqtt-bridge',
|
|
'mqtt_topic': 'ledmatrix/command',
|
|
'mqtt_tls': False,
|
|
'mqtt_tls_insecure': False,
|
|
# Opt-in acknowledgement that credentials may cross an untrusted
|
|
# network in cleartext. Off by default: the save is refused instead.
|
|
'allow_insecure_mqtt': False,
|
|
'ledmatrix_api_base': 'http://localhost:5000',
|
|
'request_timeout': 15,
|
|
'on_demand_duration': None,
|
|
'log_level': 'INFO',
|
|
}
|
|
|
|
_MQTT_BRIDGE_LOG_LEVELS = ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL')
|
|
|
|
def _mqtt_bridge_service_state() -> Dict[str, Any]:
|
|
"""installed / active / enabled for the bridge unit."""
|
|
state = {'installed': False, 'active': False, 'enabled': False}
|
|
try:
|
|
listed = _run_systemctl_command(
|
|
['systemctl', 'list-unit-files', _MQTT_BRIDGE_SERVICE, '--no-legend'])
|
|
state['installed'] = bool((listed.get('stdout') or '').strip())
|
|
state['active'] = _run_systemctl_command(
|
|
['systemctl', 'is-active', _MQTT_BRIDGE_SERVICE]).get('stdout', '').strip() == 'active'
|
|
state['enabled'] = _run_systemctl_command(
|
|
['systemctl', 'is-enabled', _MQTT_BRIDGE_SERVICE]).get('stdout', '').strip() == 'enabled'
|
|
except Exception:
|
|
logger.debug('Could not read %s state', _MQTT_BRIDGE_SERVICE, exc_info=True)
|
|
return state
|
|
|
|
def _read_mqtt_bridge_config() -> Dict[str, Any]:
|
|
"""Stored config overlaid on the defaults. Missing file is not an error.
|
|
|
|
Named `settings`, not `config`: this module imports a submodule called
|
|
`config` at the bottom for its route side effects, and a local of the same
|
|
name shadows it.
|
|
"""
|
|
settings = dict(_MQTT_BRIDGE_DEFAULTS)
|
|
settings['mqtt_password'] = None
|
|
try:
|
|
if _MQTT_BRIDGE_CONFIG.is_file():
|
|
with open(_MQTT_BRIDGE_CONFIG, encoding='utf-8') as handle:
|
|
stored = json.load(handle)
|
|
if isinstance(stored, dict):
|
|
settings.update(stored)
|
|
except (OSError, json.JSONDecodeError) as err:
|
|
logger.warning('Could not read %s: %s', _MQTT_BRIDGE_CONFIG, err)
|
|
return settings
|
|
|
|
def _coerce_mqtt_bridge_value(key: str, raw: Any) -> Tuple[Any, Optional[str]]:
|
|
"""Validate one submitted field. Returns (value, error)."""
|
|
if key in ('mqtt_port',):
|
|
try:
|
|
port = int(raw)
|
|
except (TypeError, ValueError):
|
|
return None, 'Port must be a whole number'
|
|
if not 1 <= port <= 65535:
|
|
return None, 'Port must be between 1 and 65535'
|
|
return port, None
|
|
if key in ('request_timeout',):
|
|
try:
|
|
timeout = int(raw)
|
|
except (TypeError, ValueError):
|
|
return None, 'Request timeout must be a whole number'
|
|
if not 1 <= timeout <= 300:
|
|
return None, 'Request timeout must be between 1 and 300 seconds'
|
|
return timeout, None
|
|
if key == 'on_demand_duration':
|
|
if raw in (None, ''):
|
|
return None, None
|
|
try:
|
|
duration = int(raw)
|
|
except (TypeError, ValueError):
|
|
return None, 'On-demand duration must be a whole number of seconds'
|
|
if not 1 <= duration <= 86400:
|
|
return None, 'On-demand duration must be between 1 and 86400 seconds'
|
|
return duration, None
|
|
if key in ('mqtt_tls', 'mqtt_tls_insecure', 'allow_insecure_mqtt'):
|
|
return bool(raw) if isinstance(raw, bool) else str(raw).lower() in ('1', 'true', 'yes', 'on'), None
|
|
if key == 'log_level':
|
|
level = str(raw or '').upper()
|
|
if level not in _MQTT_BRIDGE_LOG_LEVELS:
|
|
return None, f"Log level must be one of {', '.join(_MQTT_BRIDGE_LOG_LEVELS)}"
|
|
return level, None
|
|
if key == 'ledmatrix_api_base':
|
|
base = str(raw or '').strip().rstrip('/')
|
|
if not base.startswith(('http://', 'https://')):
|
|
return None, 'API base must start with http:// or https://'
|
|
if len(base) > 300:
|
|
return None, 'API base is too long'
|
|
return base, None
|
|
# Remaining keys are free text; empty means "unset" for the optional ones.
|
|
text = '' if raw is None else str(raw).strip()
|
|
if len(text) > 300:
|
|
return None, f'{key} is too long'
|
|
if key == 'mqtt_username' and not text:
|
|
return None, None
|
|
if key in ('mqtt_host', 'mqtt_client_id', 'mqtt_topic') and not text:
|
|
return None, f'{key.replace("_", " ")} cannot be empty'
|
|
return text, None
|
|
|
|
|
|
# Imported last, and for their side effect: each registers its routes on
|
|
# api_v3. They import names from this module, so this module must be fully
|
|
# executed before they run.
|
|
from web_interface.blueprints.api_v3 import ( # noqa: E402,F401
|
|
backup,
|
|
config,
|
|
display,
|
|
fonts,
|
|
misc,
|
|
plugins,
|
|
starlark,
|
|
system,
|
|
wifi,
|
|
)
|
|
|
|
__all__ = ["api_v3"]
|