LEDMatrix/src/backup_manager.py

"""
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")
# Sits in config/ next to the three above and is pure user state — a
# YouTube Music session that has to be re-authenticated by hand if lost.
# It was omitted from backups, so a restore silently signed the user out.
_YTM_REL = Path("config/ytm_auth.json")
_FONTS_REL = Path("assets/fonts")
_PLUGIN_UPLOADS_REL = Path("assets/plugins")
_STATE_REL = Path("data/plugin_state.json")

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(project_root: Path) -> str:
    """Best-effort version string for the current install."""
    version_file = project_root / "VERSION"
    if version_file.exists():
        try:
            return version_file.read_text(encoding="utf-8").strip() or "unknown"
        except OSError:
            pass
    head_file = project_root / ".git" / "HEAD"
    if head_file.exists():
        try:
            head = head_file.read_text(encoding="utf-8").strip()
            if head.startswith("ref: "):
                ref = head[5:]
                ref_path = project_root / ".git" / ref
                if ref_path.exists():
                    return ref_path.read_text(encoding="utf-8").strip()[:12] or "unknown"
            return head[:12] or "unknown"
        except OSError:
            pass
    return "unknown"


def _build_manifest(contents: List[str], project_root: Path) -> Dict[str, Any]:
    return {
        "schema_version": SCHEMA_VERSION,
        "created_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
        "ledmatrix_version": _ledmatrix_version(project_root),
        "hostname": socket.gethostname(),
        "contents": contents,
    }


# ---------------------------------------------------------------------------
# Installed-plugin enumeration
# ---------------------------------------------------------------------------


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`` and ``version``.

    Reads ``data/plugin_state.json`` if present; otherwise walks the plugin
    directory and reads each ``manifest.json``.
    """
    plugins: Dict[str, Dict[str, Any]] = {}

    state_file = project_root / _STATE_REL
    if state_file.exists():
        try:
            with state_file.open("r", encoding="utf-8") as f:
                state = json.load(f)
            raw_plugins = state.get("states", {}) if isinstance(state, dict) else {}
            if isinstance(raw_plugins, dict):
                for plugin_id, info in raw_plugins.items():
                    if not isinstance(info, dict):
                        continue
                    plugins[plugin_id] = {
                        "plugin_id": plugin_id,
                        "version": info.get("version") or "",
                        "enabled": bool(info.get("enabled", True)),
                    }
        except (OSError, json.JSONDecodeError) as e:
            logger.warning("Could not read plugin_state.json: %s", e)

    # Fall back to scanning plugin-repos/ for manifests.
    plugins_root = project_root / "plugin-repos"
    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
            plugin_id = data.get("id") or entry.name
            if plugin_id not in plugins:
                plugins[plugin_id] = {
                    "plugin_id": plugin_id,
                    "version": data.get("version", ""),
                    "enabled": True,
                }

    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.
    tmp_path = zip_path.with_suffix(".zip.tmp")
    try:
        with zipfile.ZipFile(tmp_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
            # Config files.
            if (project_root / _CONFIG_REL).exists():
                zf.write(project_root / _CONFIG_REL, _CONFIG_REL.as_posix())
                contents.append("config")
            if (project_root / _SECRETS_REL).exists():
                zf.write(project_root / _SECRETS_REL, _SECRETS_REL.as_posix())
                contents.append("secrets")
            if (project_root / _WIFI_REL).exists():
                zf.write(project_root / _WIFI_REL, _WIFI_REL.as_posix())
                contents.append("wifi")
            if (project_root / _YTM_REL).exists():
                zf.write(project_root / _YTM_REL, _YTM_REL.as_posix())
                contents.append("ytm_auth")

            # 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, project_root)
            zf.writestr(MANIFEST_NAME, json.dumps(manifest, indent=2))

        os.replace(tmp_path, zip_path)
    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()
    return {
        "has_config": (project_root / _CONFIG_REL).exists(),
        "has_secrets": (project_root / _SECRETS_REL).exists(),
        "has_wifi": (project_root / _WIFI_REL).exists(),
        "has_ytm_auth": (project_root / _YTM_REL).exists(),
        "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),
    }


# ---------------------------------------------------------------------------
# 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] = []
            if _CONFIG_REL.as_posix() in names:
                detected.append("config")
            if _SECRETS_REL.as_posix() in names:
                detected.append("secrets")
            if _WIFI_REL.as_posix() in names:
                detected.append("wifi")
            if _YTM_REL.as_posix() in names:
                detected.append("ytm_auth")
            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")

            plugins: List[Dict[str, 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) -> 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.
    """
    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)
        else:
            shutil.copymode(src, tmp_path)
        if existing_owner is not None:
            # 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.
            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

        # Main config.
        if options.restore_config and (tmp_dir / _CONFIG_REL).exists():
            try:
                _copy_file(tmp_dir / _CONFIG_REL, project_root / _CONFIG_REL)
                result.restored.append("config")
            except OSError as e:
                logger.error("[Backup] Failed to restore config.json: %s", e, exc_info=True)
                result.errors.append("Failed to restore config.json")
        elif (tmp_dir / _CONFIG_REL).exists():
            result.skipped.append("config")

        # Secrets.
        if options.restore_secrets and (tmp_dir / _SECRETS_REL).exists():
            try:
                _copy_file(tmp_dir / _SECRETS_REL, project_root / _SECRETS_REL)
                result.restored.append("secrets")
            except OSError as e:
                logger.error(
                    "[Backup] Failed to restore config_secrets.json: %s", e, exc_info=True
                )
                result.errors.append("Failed to restore config_secrets.json")
        elif (tmp_dir / _SECRETS_REL).exists():
            result.skipped.append("secrets")

        # WiFi.
        if options.restore_wifi and (tmp_dir / _WIFI_REL).exists():
            try:
                _copy_file(tmp_dir / _WIFI_REL, project_root / _WIFI_REL)
                result.restored.append("wifi")
            except OSError as e:
                logger.error(
                    "[Backup] Failed to restore wifi_config.json: %s", e, exc_info=True
                )
                result.errors.append("Failed to restore wifi_config.json")
        elif (tmp_dir / _WIFI_REL).exists():
            result.skipped.append("wifi")

        # YouTube Music session. Follows restore_wifi rather than getting its
        # own flag: it is device-local auth in the same sense, and a separate
        # toggle for one file would be noise in the restore dialog.
        if options.restore_wifi and (tmp_dir / _YTM_REL).exists():
            try:
                _copy_file(tmp_dir / _YTM_REL, project_root / _YTM_REL)
                result.restored.append("ytm_auth")
            except OSError as e:
                logger.error("[Backup] Failed to restore ytm_auth.json: %s", e, exc_info=True)
                result.errors.append("Failed to restore ytm_auth.json")
        elif (tmp_dir / _YTM_REL).exists():
            result.skipped.append("ytm_auth")

        # 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