mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
* fix(web): mask the Config Editor's secrets like GET /config/secrets The Config Editor tab (/partials/raw-json) filled its config_secrets.json editor with the file as it is on disk. GET /api/v3/config/secrets masks every value because the interface is reachable without a login by default, but this page handed the same credentials (GitHub token, Home Assistant token, plugin API keys) to anyone who loaded it. The masked-save path in save_raw_secrets_config was written for a masked editor and never got one. _load_raw_json_partial now masks the section with mask_all_secret_values after strip_auth_section, exactly as the GET does. Saving it back is safe: save_raw_secrets_config drops the masks (strip_masked_values) and merges the rest onto the stored file (deep_merge), so an untouched secret stays as it is and a replaced mask is the only value that changes. The config.json editor is left as it is. Its save (save_raw_main_config) writes the posted object verbatim, with no mask stripping or merge, so a masked main editor would write the bullets over any credential it holds. Masking it needs a merge-on-save of its own first. Tests: TestConfigEditorRoundTrip renders the partial over a real ConfigManager, checks no real value is in the editor, and posts the editor back unchanged (the file is identical) and with one mask replaced (only that value changes). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(web): keep disabled plugins in the saved rotation order and Vegas exclusions PluginOrderList draws one row per enabled plugin and, once drawn, rewrites its hidden inputs (plugin_rotation_order, vegas_plugin_order, vegas_excluded_plugins) from those rows. A disabled plugin has no row, so merely opening the Display or Rotation & Durations tab took it out of the inputs, and the next save of that form stored the lists without it. Exclude Clock from Vegas, disable it, change the brightness, re-enable it: Clock was scrolling in Vegas again and had moved to the end of the rotation. syncInputs now keeps the saved ids that have no row. In the order, each one keeps its saved slot and the rows fill the other slots in their current order, with rows not in the saved order last, as before. In the exclusions they follow the unchecked rows. Only string ids are carried over, once each: /config/main refuses a list holding anything else, which would block every later save of the tab. Tests: test/js/unit/test_plugin_order_list.js runs the shipped widget in a vm with a fake DOM (draw, reorder, include/exclude, the rotation list, junk ids) and is in run_all.js and the README. The durations DOM suite now reads only its own rows' ids from the input, since a rig's saved order can hold others. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(web): a restore reinstalls only the plugins that are missing POST /backup/restore with reinstall_plugins (the "Reinstall missing plugins" box) passed every plugin in the backup's plugins.json to install_plugin(). That replaces an installed copy with a fresh download, so a restore onto the same device re-downloaded every plugin inside the request. A plugin installed from its own URL is not in the registry, so its install returned False, plugins_failed set success to False, and the restore answered 500 "Restore incomplete ... plugins not reinstalled: <id>" (shown as "Restore failed") with the plugin still installed and the config restored. Each plugin is now looked up first with the store's _existing_install, the same lookup install_plugin makes to decide a copy exists: the id, or an id the registry proves is the same plugin (aliases, the plugin_path name), and never a bare ledmatrix-<id> folder (#686). One that is installed is recorded in result.skipped as "plugin:<id> (installed)", which the page lists under Skipped; a missing one is installed as before. The list_installed_plugins docstring said every listed plugin is reinstalled and now says otherwise. Tests: TestInstalledPluginsAreNotReinstalled, with a mocked store (installed skipped, missing installed; an installed plugin the store can't install is not a failure) and with a real PluginStoreManager (a registry alias and a third-party install are skipped, a missing plugin installed). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(web): /config/main answers malformed JSON with a 400 save_main_config read a JSON body with request.get_json(), which raises Werkzeug's BadRequest for a body that does not parse (or an empty one sent as application/json). That happened inside the handler's try, so the catch-all answered 500 CONFIG_SAVE_FAILED with "Check file permissions on config directory" among its suggested fixes and logged a traceback at ERROR, for what was the caller's mistake. It now reads with get_json(silent=True), as save_raw_main_config does, and answers a sent-but-unparseable body with the same 400 {"status": "error", "message": "Invalid JSON in request body"}. An empty JSON body falls through to the existing 400 "No data provided". The change is limited to the lines that read the body. Tests: TestMalformedBody in test_api_v3_partial_main_save.py (the 400 and its shape, identical to /config/raw/main's, and nothing saved; the empty body). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(web): a restore that brings back fonts clears the font catalog cache GET /api/v3/fonts/catalog caches its answer as fonts_catalog for five minutes. Font upload and delete clear that entry (fonts.py), but POST /backup/restore copies user fonts into assets/fonts without touching it, so restored fonts were missing from the Fonts tab and every font picker until the cache expired. backup_restore now clears fonts_catalog when the result lists restored fonts (restore_backup records them as "fonts (<count>)"). A restore that restored no fonts leaves the cache alone. Tests: TestFontsCatalogCache in test_api_v3_backup_restore.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(web): drop uninstalled plugins from the carried-over order and exclusions2b34f254made the plugin order list keep every saved id that has no row, so a disabled plugin keeps its rotation slot and Vegas exclusion. That also kept the ids of plugins that have since been uninstalled: they stayed in plugin_rotation_order and vegas_excluded_plugins for good, where before the next save of the tab dropped them. The widget already fetches /api/v3/plugins/installed, every installed plugin with its enabled flag, and draws only the enabled ones. It now keeps that response's full id set and carries over only saved ids that are installed but have no row (disabled). An id outside the set is dropped, as before. With no list, nothing is dropped: a failed request draws no rows and leaves the inputs as saved, and the carry-over keeps everything if the set was never filled. Tests: test/js/unit/test_plugin_order_list.js adds a disabled plugin kept while an uninstalled one is dropped (order and exclusions; fails on2b34f254), and a failed plugin list leaving both inputs as saved. The CHANGELOG bullet and the README row say so. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(js): register the order-list suite apart from other branches' suites Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
698 lines
27 KiB
Python
698 lines
27 KiB
Python
"""
|
|
User configuration backup and restore.
|
|
|
|
Packages the user's LEDMatrix configuration, secrets, WiFi settings,
|
|
user-uploaded fonts, plugin image uploads, and installed-plugin manifest
|
|
into a single ``.zip`` that can be exported from one installation and
|
|
imported on a fresh install.
|
|
|
|
This module is intentionally Flask-free so it can be unit-tested and
|
|
used from scripts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import socket
|
|
import tempfile
|
|
import zipfile
|
|
from dataclasses import dataclass, field, asdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SCHEMA_VERSION = 1
|
|
|
|
# Filenames shipped with the LEDMatrix repository under ``assets/fonts/``.
|
|
# Anything present on disk but NOT in this set is treated as a user upload
|
|
# and included in backups. Keep this snapshot in sync with the repo — regenerate
|
|
# with::
|
|
#
|
|
# ls assets/fonts/
|
|
#
|
|
# Tests assert the set matches the checked-in fonts.
|
|
BUNDLED_FONTS: frozenset[str] = frozenset({
|
|
"10x20.bdf",
|
|
"4x6.bdf",
|
|
"4x6-font.ttf",
|
|
"5by7.regular.ttf",
|
|
"5x7.bdf",
|
|
"5x8.bdf",
|
|
"6x9.bdf",
|
|
"6x10.bdf",
|
|
"6x12.bdf",
|
|
"6x13.bdf",
|
|
"6x13B.bdf",
|
|
"6x13O.bdf",
|
|
"7x13.bdf",
|
|
"7x13B.bdf",
|
|
"7x13O.bdf",
|
|
"7x14.bdf",
|
|
"7x14B.bdf",
|
|
"8x13.bdf",
|
|
"8x13B.bdf",
|
|
"8x13O.bdf",
|
|
"9x15.bdf",
|
|
"9x15B.bdf",
|
|
"9x18.bdf",
|
|
"9x18B.bdf",
|
|
"AUTHORS",
|
|
"bdf_font_guide",
|
|
"clR6x12.bdf",
|
|
"helvR12.bdf",
|
|
"ic8x8u.bdf",
|
|
"MatrixChunky8.bdf",
|
|
"MatrixChunky8X.bdf",
|
|
"MatrixLight6.bdf",
|
|
"MatrixLight6X.bdf",
|
|
"MatrixLight8X.bdf",
|
|
"PressStart2P-Regular.ttf",
|
|
"README",
|
|
"README.md",
|
|
"texgyre-27.bdf",
|
|
"tom-thumb.bdf",
|
|
})
|
|
|
|
# Relative paths inside the project that the backup knows how to round-trip.
|
|
_CONFIG_REL = Path("config/config.json")
|
|
_SECRETS_REL = Path("config/config_secrets.json")
|
|
_WIFI_REL = Path("config/wifi_config.json")
|
|
# A YouTube Music session: pure user state that has to be re-authenticated by
|
|
# hand if lost, so a restore must bring it back.
|
|
_YTM_REL = Path("config/ytm_auth.json")
|
|
_FONTS_REL = Path("assets/fonts")
|
|
_PLUGIN_UPLOADS_REL = Path("assets/plugins")
|
|
|
|
#: The sections that are one file each: (section name, path, the
|
|
#: RestoreOptions flag that restores it). create, preview, validate and
|
|
#: restore all walk this table. ytm_auth follows restore_wifi: it is
|
|
#: device-local auth like the Wi-Fi settings, and a toggle of its own for one
|
|
#: file would be noise in the restore dialog.
|
|
_SINGLE_FILE_SECTIONS: Tuple[Tuple[str, Path, str], ...] = (
|
|
("config", _CONFIG_REL, "restore_config"),
|
|
("secrets", _SECRETS_REL, "restore_secrets"),
|
|
("wifi", _WIFI_REL, "restore_wifi"),
|
|
("ytm_auth", _YTM_REL, "restore_wifi"),
|
|
)
|
|
|
|
#: Sections holding credentials. Restored onto a device that has no copy yet,
|
|
#: they would otherwise take the extracted temp file's umask mode (0o644,
|
|
#: world-readable); 0o640 matches what config_manager_atomic gives secrets.
|
|
_PRIVATE_SECTION_RELS = frozenset({_SECRETS_REL, _WIFI_REL, _YTM_REL})
|
|
_PRIVATE_FILE_MODE = 0o640
|
|
|
|
MANIFEST_NAME = "manifest.json"
|
|
PLUGINS_MANIFEST_NAME = "plugins.json"
|
|
|
|
# Hard cap on the size of a single file we'll accept inside an uploaded ZIP
|
|
# to limit zip-bomb risk. 50 MB matches the existing plugin-image upload cap.
|
|
_MAX_MEMBER_BYTES = 50 * 1024 * 1024
|
|
# Hard cap on the total uncompressed size of an uploaded ZIP.
|
|
_MAX_TOTAL_BYTES = 200 * 1024 * 1024
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data classes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class RestoreOptions:
|
|
"""Which sections of a backup should be restored."""
|
|
|
|
restore_config: bool = True
|
|
restore_secrets: bool = True
|
|
restore_wifi: bool = True
|
|
restore_fonts: bool = True
|
|
restore_plugin_uploads: bool = True
|
|
reinstall_plugins: bool = True
|
|
|
|
|
|
@dataclass
|
|
class RestoreResult:
|
|
"""Outcome of a restore operation."""
|
|
|
|
success: bool = False
|
|
restored: List[str] = field(default_factory=list)
|
|
skipped: List[str] = field(default_factory=list)
|
|
plugins_to_install: List[Dict[str, Any]] = field(default_factory=list)
|
|
plugins_installed: List[str] = field(default_factory=list)
|
|
plugins_failed: List[Dict[str, str]] = field(default_factory=list)
|
|
errors: List[str] = field(default_factory=list)
|
|
manifest: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Manifest helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _ledmatrix_version() -> str:
|
|
"""The release of the running core (``src.__version__``), recorded in the
|
|
manifest so a restore can tell which release wrote the backup."""
|
|
from src import __version__
|
|
return __version__
|
|
|
|
|
|
def _build_manifest(contents: List[str]) -> Dict[str, Any]:
|
|
return {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"created_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
|
|
"ledmatrix_version": _ledmatrix_version(),
|
|
"hostname": socket.gethostname(),
|
|
"contents": contents,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Installed-plugin enumeration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _read_config(project_root: Path) -> Dict[str, Any]:
|
|
"""config/config.json as a dict; empty when missing or unreadable."""
|
|
try:
|
|
with (project_root / _CONFIG_REL).open("r", encoding="utf-8") as f:
|
|
config = json.load(f)
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
return config if isinstance(config, dict) else {}
|
|
|
|
|
|
def _plugins_directory(project_root: Path,
|
|
config: Optional[Dict[str, Any]] = None) -> Path:
|
|
"""The plugin install directory: ``plugin_system.plugins_directory`` from
|
|
config/config.json (relative to ``project_root`` unless absolute), or
|
|
``plugin-repos`` when the config does not say or cannot be read."""
|
|
if config is None:
|
|
config = _read_config(project_root)
|
|
configured: Any = None
|
|
plugin_system = config.get("plugin_system")
|
|
if isinstance(plugin_system, dict):
|
|
configured = plugin_system.get("plugins_directory")
|
|
if not isinstance(configured, str) or not configured.strip():
|
|
configured = "plugin-repos"
|
|
path = Path(configured)
|
|
return path if path.is_absolute() else project_root / path
|
|
|
|
|
|
def list_installed_plugins(project_root: Path) -> List[Dict[str, Any]]:
|
|
"""
|
|
Return a list of currently-installed plugins suitable for the backup
|
|
manifest. Each entry has ``plugin_id``, ``version`` and ``enabled``.
|
|
|
|
The plugins are the ``manifest.json`` files in the configured plugin
|
|
directory (see :func:`_plugins_directory`), with the manifest's version;
|
|
``enabled`` is config.json's flag by the display's rule (a missing flag
|
|
is disabled). A restore installs each listed plugin that is missing and
|
|
takes enabled state from the restored config.json, so ``enabled`` is
|
|
informational.
|
|
|
|
``data/plugin_state.json`` is not read: it only ever repeated config's
|
|
enabled flags and the manifests' versions, and is retired (nothing
|
|
writes it any more). An old backup that listed a plugin only from that
|
|
file still restores it, since restore reads ``plugins.json`` as written.
|
|
"""
|
|
plugins: Dict[str, Dict[str, Any]] = {}
|
|
config = _read_config(project_root)
|
|
|
|
plugins_root = _plugins_directory(project_root, config)
|
|
if plugins_root.exists():
|
|
for entry in sorted(plugins_root.iterdir()):
|
|
if not entry.is_dir():
|
|
continue
|
|
manifest = entry / "manifest.json"
|
|
if not manifest.exists():
|
|
continue
|
|
try:
|
|
with manifest.open("r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
# Valid JSON that is not an object (a list, a bare string) would
|
|
# raise AttributeError on .get() and abort the whole export.
|
|
if not isinstance(data, dict):
|
|
continue
|
|
plugin_id = data.get("id") or entry.name
|
|
if plugin_id not in plugins:
|
|
section = config.get(plugin_id)
|
|
plugins[plugin_id] = {
|
|
"plugin_id": plugin_id,
|
|
"version": data.get("version", ""),
|
|
"enabled": isinstance(section, dict) and bool(section.get("enabled", False)),
|
|
}
|
|
|
|
return sorted(plugins.values(), key=lambda p: p["plugin_id"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Font filtering
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def iter_user_fonts(project_root: Path) -> List[Path]:
|
|
"""Return absolute paths to user-uploaded fonts (anything in
|
|
``assets/fonts/`` not listed in :data:`BUNDLED_FONTS`)."""
|
|
fonts_dir = project_root / _FONTS_REL
|
|
if not fonts_dir.exists():
|
|
return []
|
|
user_fonts: List[Path] = []
|
|
for entry in sorted(fonts_dir.iterdir()):
|
|
if entry.is_file() and entry.name not in BUNDLED_FONTS:
|
|
user_fonts.append(entry)
|
|
return user_fonts
|
|
|
|
|
|
def iter_plugin_uploads(project_root: Path) -> List[Path]:
|
|
"""Return every file under ``assets/plugins/*/uploads/`` (recursive)."""
|
|
plugin_root = project_root / _PLUGIN_UPLOADS_REL
|
|
if not plugin_root.exists():
|
|
return []
|
|
out: List[Path] = []
|
|
for plugin_dir in sorted(plugin_root.iterdir()):
|
|
if not plugin_dir.is_dir():
|
|
continue
|
|
uploads = plugin_dir / "uploads"
|
|
if not uploads.exists():
|
|
continue
|
|
for root, _dirs, files in os.walk(uploads):
|
|
for name in sorted(files):
|
|
out.append(Path(root) / name)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Export
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def create_backup(
|
|
project_root: Path,
|
|
output_dir: Optional[Path] = None,
|
|
) -> Path:
|
|
"""
|
|
Build a backup ZIP and write it into ``output_dir`` (defaults to
|
|
``<project_root>/config/backups/exports/``). Returns the path to the
|
|
created file.
|
|
"""
|
|
project_root = Path(project_root).resolve()
|
|
if output_dir is None:
|
|
output_dir = project_root / "config" / "backups" / "exports"
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
hostname = socket.gethostname() or "ledmatrix"
|
|
safe_host = "".join(c for c in hostname if c.isalnum() or c in "-_") or "ledmatrix"
|
|
zip_name = f"ledmatrix-backup-{safe_host}-{timestamp}.zip"
|
|
zip_path = output_dir / zip_name
|
|
|
|
contents: List[str] = []
|
|
|
|
# Stream directly to a temp file so we never hold the whole ZIP in memory.
|
|
# The name is unique per call: a fixed "<zip>.tmp" was shared by two
|
|
# exports started in the same second, which then wrote the same file.
|
|
fd, tmp_name = tempfile.mkstemp(dir=str(output_dir), prefix=f".{zip_name}.", suffix=".tmp")
|
|
os.close(fd)
|
|
tmp_path = Path(tmp_name)
|
|
try:
|
|
with zipfile.ZipFile(tmp_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
for section, rel, _flag in _SINGLE_FILE_SECTIONS:
|
|
if (project_root / rel).exists():
|
|
zf.write(project_root / rel, rel.as_posix())
|
|
contents.append(section)
|
|
|
|
# User-uploaded fonts.
|
|
user_fonts = iter_user_fonts(project_root)
|
|
if user_fonts:
|
|
for font in user_fonts:
|
|
arcname = font.relative_to(project_root).as_posix()
|
|
zf.write(font, arcname)
|
|
contents.append("fonts")
|
|
|
|
# Plugin uploads.
|
|
plugin_uploads = iter_plugin_uploads(project_root)
|
|
if plugin_uploads:
|
|
for upload in plugin_uploads:
|
|
arcname = upload.relative_to(project_root).as_posix()
|
|
zf.write(upload, arcname)
|
|
contents.append("plugin_uploads")
|
|
|
|
# Installed plugins manifest.
|
|
plugins = list_installed_plugins(project_root)
|
|
if plugins:
|
|
zf.writestr(
|
|
PLUGINS_MANIFEST_NAME,
|
|
json.dumps(plugins, indent=2),
|
|
)
|
|
contents.append("plugins")
|
|
|
|
# Manifest goes last so that `contents` reflects what we actually wrote.
|
|
manifest = _build_manifest(contents)
|
|
zf.writestr(MANIFEST_NAME, json.dumps(manifest, indent=2))
|
|
|
|
# Same-second exports share a timestamp; number the later one rather
|
|
# than replacing the backup the first one just returned. The name is
|
|
# claimed with an exclusive create (O_EXCL fails if it exists), so two
|
|
# exports finishing together can't both pick the same free name; the
|
|
# replace then swaps the finished archive in over our own placeholder.
|
|
suffix = 2
|
|
while True:
|
|
try:
|
|
os.close(os.open(zip_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600))
|
|
break
|
|
except FileExistsError:
|
|
zip_path = output_dir / f"{Path(zip_name).stem}-{suffix}.zip"
|
|
suffix += 1
|
|
try:
|
|
os.replace(tmp_path, zip_path)
|
|
except BaseException:
|
|
zip_path.unlink(missing_ok=True)
|
|
raise
|
|
except Exception:
|
|
tmp_path.unlink(missing_ok=True)
|
|
raise
|
|
logger.info("Created backup %s (%d bytes)", zip_path, zip_path.stat().st_size)
|
|
return zip_path
|
|
|
|
|
|
def preview_backup_contents(project_root: Path) -> Dict[str, Any]:
|
|
"""Return a summary of what ``create_backup`` would include."""
|
|
project_root = Path(project_root).resolve()
|
|
preview: Dict[str, Any] = {
|
|
f"has_{section}": (project_root / rel).exists()
|
|
for section, rel, _flag in _SINGLE_FILE_SECTIONS
|
|
}
|
|
preview.update({
|
|
"user_fonts": [p.name for p in iter_user_fonts(project_root)],
|
|
"plugin_uploads": len(iter_plugin_uploads(project_root)),
|
|
"plugins": list_installed_plugins(project_root),
|
|
})
|
|
return preview
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _safe_extract_path(base_dir: Path, member_name: str) -> Optional[Path]:
|
|
"""Resolve a ZIP member name against ``base_dir`` and reject anything
|
|
that escapes it. Returns the resolved absolute path, or ``None`` if the
|
|
name is unsafe."""
|
|
# Reject absolute paths and Windows-style drives outright.
|
|
if member_name.startswith(("/", "\\")) or (len(member_name) >= 2 and member_name[1] == ":"):
|
|
return None
|
|
target = (base_dir / member_name).resolve()
|
|
try:
|
|
target.relative_to(base_dir.resolve())
|
|
except ValueError:
|
|
return None
|
|
return target
|
|
|
|
|
|
def validate_backup(zip_path: Path) -> Tuple[bool, str, Dict[str, Any]]:
|
|
"""
|
|
Inspect a backup ZIP without extracting to disk.
|
|
|
|
Returns ``(ok, error_message, manifest_dict)``. ``manifest_dict`` contains
|
|
the parsed manifest plus diagnostic fields:
|
|
- ``detected_contents``: list of section names present in the archive
|
|
- ``plugins``: parsed plugins.json if present
|
|
- ``total_uncompressed``: sum of uncompressed sizes
|
|
"""
|
|
zip_path = Path(zip_path)
|
|
if not zip_path.exists():
|
|
return False, f"Backup file not found: {zip_path}", {}
|
|
|
|
try:
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
names = zf.namelist()
|
|
if MANIFEST_NAME not in names:
|
|
return False, "Backup is missing manifest.json", {}
|
|
|
|
total = 0
|
|
with tempfile.TemporaryDirectory() as _sandbox:
|
|
sandbox = Path(_sandbox)
|
|
for info in zf.infolist():
|
|
if info.file_size > _MAX_MEMBER_BYTES:
|
|
return False, f"Member {info.filename} is too large", {}
|
|
total += info.file_size
|
|
if total > _MAX_TOTAL_BYTES:
|
|
return False, "Backup exceeds maximum allowed size", {}
|
|
# Safety: reject members with unsafe paths up front.
|
|
if _safe_extract_path(sandbox, info.filename) is None:
|
|
return False, f"Unsafe path in backup: {info.filename}", {}
|
|
|
|
try:
|
|
manifest_raw = zf.read(MANIFEST_NAME).decode("utf-8")
|
|
manifest = json.loads(manifest_raw)
|
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError):
|
|
return False, "Invalid manifest.json", {}
|
|
|
|
if not isinstance(manifest, dict) or "schema_version" not in manifest:
|
|
return False, "Invalid manifest structure", {}
|
|
if manifest.get("schema_version") != SCHEMA_VERSION:
|
|
return (
|
|
False,
|
|
f"Unsupported backup schema version: {manifest.get('schema_version')}",
|
|
{},
|
|
)
|
|
|
|
detected: List[str] = [
|
|
section for section, rel, _flag in _SINGLE_FILE_SECTIONS
|
|
if rel.as_posix() in names
|
|
]
|
|
if any(n.startswith(_FONTS_REL.as_posix() + "/") for n in names):
|
|
detected.append("fonts")
|
|
if any(
|
|
n.startswith(_PLUGIN_UPLOADS_REL.as_posix() + "/") and "/uploads/" in n
|
|
for n in names
|
|
):
|
|
detected.append("plugin_uploads")
|
|
|
|
# Whatever the archive's manifest holds; checked below.
|
|
plugins: Any = []
|
|
if PLUGINS_MANIFEST_NAME in names:
|
|
try:
|
|
plugins = json.loads(zf.read(PLUGINS_MANIFEST_NAME).decode("utf-8"))
|
|
if not isinstance(plugins, list):
|
|
plugins = []
|
|
else:
|
|
detected.append("plugins")
|
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError):
|
|
plugins = []
|
|
|
|
result_manifest = dict(manifest)
|
|
result_manifest["detected_contents"] = detected
|
|
result_manifest["plugins"] = plugins
|
|
result_manifest["total_uncompressed"] = total
|
|
result_manifest["file_count"] = len(names)
|
|
return True, "", result_manifest
|
|
except zipfile.BadZipFile:
|
|
return False, "File is not a valid ZIP archive", {}
|
|
except OSError:
|
|
return False, "Could not read backup", {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Restore
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _extract_zip_safe(zip_path: Path, dest_dir: Path) -> None:
|
|
"""Extract ``zip_path`` into ``dest_dir`` rejecting any unsafe members."""
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
for info in zf.infolist():
|
|
target = _safe_extract_path(dest_dir, info.filename)
|
|
if target is None:
|
|
raise ValueError(f"Unsafe path in backup: {info.filename}")
|
|
if info.is_dir():
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
with zf.open(info, "r") as src, open(target, "wb") as dst:
|
|
shutil.copyfileobj(src, dst, length=64 * 1024)
|
|
|
|
|
|
def _copy_file(src: Path, dst: Path, new_mode: Optional[int] = None) -> None:
|
|
"""Replace ``dst`` with ``src``, atomically, without needing to own ``dst``.
|
|
|
|
``shutil.copy2`` opens the destination for writing, so it needs write
|
|
permission on the *existing file*. Several config files are installed
|
|
root-owned and group-readable while the web interface — which is what runs
|
|
a restore — deliberately runs as a non-root user. Restoring those failed
|
|
with EACCES even though the account could create files in the same
|
|
directory perfectly well.
|
|
|
|
Writing a temporary file alongside and renaming over the target needs only
|
|
directory permission, which the web user has. It is also atomic: a crash
|
|
mid-restore can no longer leave a half-written config behind.
|
|
|
|
The destination's existing mode is preserved when there is one, so
|
|
restoring secrets does not silently widen them to the umask default.
|
|
When there is none, ``new_mode`` (if given) is used instead of ``src``'s.
|
|
"""
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
existing_mode: Optional[int] = None
|
|
existing_owner: Optional[Tuple[int, int]] = None
|
|
if dst.exists():
|
|
try:
|
|
info = dst.stat()
|
|
existing_mode = stat.S_IMODE(info.st_mode)
|
|
existing_owner = (info.st_uid, info.st_gid)
|
|
except OSError:
|
|
existing_mode = None
|
|
existing_owner = None
|
|
|
|
fd, tmp_name = tempfile.mkstemp(dir=str(dst.parent), prefix=f".{dst.name}.", suffix=".tmp")
|
|
os.close(fd)
|
|
tmp_path = Path(tmp_name)
|
|
try:
|
|
shutil.copyfile(src, tmp_path)
|
|
if existing_mode is not None:
|
|
os.chmod(tmp_path, existing_mode)
|
|
elif new_mode is not None:
|
|
os.chmod(tmp_path, new_mode)
|
|
else:
|
|
shutil.copymode(src, tmp_path)
|
|
if existing_owner is not None and hasattr(os, 'chown'):
|
|
# Replacing a file creates a new inode owned by whoever is running,
|
|
# which would silently move a root-owned config to the web user.
|
|
# Carry the previous owner across when the OS permits it — only
|
|
# root can hand a file to another user, so this is best-effort and
|
|
# a plain restore as the web user simply keeps its own ownership.
|
|
# os.chown does not exist on Windows (where st_uid/st_gid are just
|
|
# 0); looking it up there raises AttributeError, which no caller
|
|
# catches, so every restore over an existing file aborted.
|
|
try:
|
|
os.chown(tmp_path, existing_owner[0], existing_owner[1])
|
|
except (OSError, PermissionError):
|
|
pass
|
|
os.replace(tmp_path, dst)
|
|
except BaseException:
|
|
try:
|
|
tmp_path.unlink()
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
def restore_backup(
|
|
zip_path: Path,
|
|
project_root: Path,
|
|
options: Optional[RestoreOptions] = None,
|
|
) -> RestoreResult:
|
|
"""
|
|
Restore ``zip_path`` into ``project_root`` according to ``options``.
|
|
|
|
Plugin reinstalls are NOT performed here — the caller is responsible for
|
|
walking ``result.plugins_to_install`` and calling the store manager. This
|
|
keeps this module Flask-free and side-effect free beyond the filesystem.
|
|
"""
|
|
if options is None:
|
|
options = RestoreOptions()
|
|
project_root = Path(project_root).resolve()
|
|
result = RestoreResult()
|
|
|
|
ok, err, manifest = validate_backup(zip_path)
|
|
if not ok:
|
|
result.errors.append(err)
|
|
return result
|
|
result.manifest = manifest
|
|
|
|
with tempfile.TemporaryDirectory(prefix="ledmatrix_restore_") as tmp:
|
|
tmp_dir = Path(tmp)
|
|
try:
|
|
_extract_zip_safe(Path(zip_path), tmp_dir)
|
|
except (ValueError, zipfile.BadZipFile, OSError) as e:
|
|
logger.error("[Backup] Failed to extract backup: %s", e, exc_info=True)
|
|
result.errors.append("Failed to extract backup")
|
|
return result
|
|
|
|
for section, rel, flag in _SINGLE_FILE_SECTIONS:
|
|
if not (tmp_dir / rel).exists():
|
|
continue
|
|
if not getattr(options, flag):
|
|
result.skipped.append(section)
|
|
continue
|
|
try:
|
|
_copy_file(tmp_dir / rel, project_root / rel,
|
|
new_mode=_PRIVATE_FILE_MODE if rel in _PRIVATE_SECTION_RELS else None)
|
|
result.restored.append(section)
|
|
except OSError as e:
|
|
logger.error("[Backup] Failed to restore %s: %s", rel.name, e, exc_info=True)
|
|
result.errors.append(f"Failed to restore {rel.name}")
|
|
|
|
# User fonts — skip anything that collides with a bundled font.
|
|
tmp_fonts = tmp_dir / _FONTS_REL
|
|
if options.restore_fonts and tmp_fonts.exists():
|
|
restored_count = 0
|
|
for font in sorted(tmp_fonts.iterdir()):
|
|
if not font.is_file():
|
|
continue
|
|
if font.name in BUNDLED_FONTS:
|
|
result.skipped.append(f"font:{font.name} (bundled)")
|
|
continue
|
|
try:
|
|
_copy_file(font, project_root / _FONTS_REL / font.name)
|
|
restored_count += 1
|
|
except OSError as e:
|
|
logger.error(
|
|
"[Backup] Failed to restore font %s: %s", font.name, e, exc_info=True
|
|
)
|
|
result.errors.append(f"Failed to restore font {font.name}")
|
|
if restored_count:
|
|
result.restored.append(f"fonts ({restored_count})")
|
|
elif tmp_fonts.exists():
|
|
result.skipped.append("fonts")
|
|
|
|
# Plugin uploads.
|
|
tmp_uploads = tmp_dir / _PLUGIN_UPLOADS_REL
|
|
if options.restore_plugin_uploads and tmp_uploads.exists():
|
|
count = 0
|
|
for root, _dirs, files in os.walk(tmp_uploads):
|
|
for name in files:
|
|
src = Path(root) / name
|
|
rel = src.relative_to(tmp_dir)
|
|
if "/uploads/" not in rel.as_posix():
|
|
result.errors.append(f"Rejected unexpected plugin path: {rel}")
|
|
continue
|
|
try:
|
|
_copy_file(src, project_root / rel)
|
|
count += 1
|
|
except OSError as e:
|
|
logger.error("[Backup] Failed to restore %s: %s", rel, e, exc_info=True)
|
|
result.errors.append(f"Failed to restore {rel}")
|
|
if count:
|
|
result.restored.append(f"plugin_uploads ({count})")
|
|
elif tmp_uploads.exists():
|
|
result.skipped.append("plugin_uploads")
|
|
|
|
# Plugins list (for caller to reinstall).
|
|
if options.reinstall_plugins and (tmp_dir / PLUGINS_MANIFEST_NAME).exists():
|
|
try:
|
|
with (tmp_dir / PLUGINS_MANIFEST_NAME).open("r", encoding="utf-8") as f:
|
|
plugins = json.load(f)
|
|
if isinstance(plugins, list):
|
|
result.plugins_to_install = [
|
|
{"plugin_id": p.get("plugin_id"), "version": p.get("version", "")}
|
|
for p in plugins
|
|
if isinstance(p, dict) and p.get("plugin_id")
|
|
]
|
|
except (OSError, json.JSONDecodeError) as e:
|
|
logger.error("[Backup] Could not read plugins.json: %s", e, exc_info=True)
|
|
result.errors.append("Could not read plugins.json")
|
|
|
|
result.success = not result.errors
|
|
return result
|