""" 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 ``/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 ".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