LEDMatrix/web_interface/blueprints/api_v3/backup.py

"""Backup creation, listing and restore.

Routes decorate the shared `api_v3` Blueprint from the package `__init__`,
so their endpoint names are unchanged by living here.

No route here catches Exception: the blueprint's handler in `__init__`
logs it and answers 500 with the same `status`/`message` fields the
backup page reads (backup_restore.html), plus `details`.
"""
from web_interface.blueprints.api_v3 import (
    PROJECT_ROOT, Path, _coerce_to_bool, _safe_backup_path, api_v3,
    datetime, json, jsonify, logger, os, request, tempfile,
)
import web_interface.blueprints.api_v3 as _pkg
# Read through the module rather than bound by value: tests patch these
# as module attributes, and a value binding would not see the patch.
# Several are also called from helpers that live in __init__, so the
# package is the only patch point that covers every caller.
from web_interface.cache import delete_cached


@api_v3.route('/backup/preview', methods=['GET'])
def backup_preview():
    """Return a summary of what a new backup would include."""
    from src.backup_manager import preview_backup_contents
    data = preview_backup_contents(PROJECT_ROOT)
    return jsonify({'status': 'success', 'data': data})
@api_v3.route('/backup/list', methods=['GET'])
def backup_list():
    """List backup ZIPs stored in the export directory."""
    _pkg._BACKUP_EXPORT_DIR.mkdir(parents=True, exist_ok=True)
    entries = []
    for p in sorted(_pkg._BACKUP_EXPORT_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True):
        if not p.is_file() or p.suffix != '.zip':
            continue
        st = p.stat()
        entries.append({
            'filename': p.name,
            'size': st.st_size,
            'created_at': datetime.fromtimestamp(st.st_mtime).strftime('%Y-%m-%d %H:%M:%S'),
        })
    return jsonify({'status': 'success', 'data': entries})
@api_v3.route('/backup/export', methods=['POST'])
def backup_export():
    """Create a new backup ZIP and return its filename."""
    from src.backup_manager import create_backup
    zip_path = create_backup(PROJECT_ROOT, output_dir=_pkg._BACKUP_EXPORT_DIR)
    return jsonify({'status': 'success', 'filename': zip_path.name})
@api_v3.route('/backup/validate', methods=['POST'])
def backup_validate():
    """Validate an uploaded backup ZIP and return its manifest."""
    from src.backup_manager import validate_backup
    if 'backup_file' not in request.files:
        return jsonify({'status': 'error', 'message': 'No backup_file in request'}), 400
    f = request.files['backup_file']
    with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp:
        tmp_path = tmp.name
        f.save(tmp_path)
    try:
        ok, err_msg, manifest = validate_backup(Path(tmp_path))
    finally:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
    if not ok:
        logger.warning("Backup validation failed: %s", err_msg)
        return jsonify({'status': 'error', 'message': 'Invalid or corrupted backup file'}), 400
    safe_manifest = {
        'schema_version': manifest.get('schema_version'),
        'created_at': manifest.get('created_at'),
        'ledmatrix_version': manifest.get('ledmatrix_version'),
        'hostname': manifest.get('hostname'),
        'contents': manifest.get('contents', []),
        'detected_contents': manifest.get('detected_contents', []),
        'plugins': manifest.get('plugins', []),
        'total_uncompressed': manifest.get('total_uncompressed'),
        'file_count': manifest.get('file_count'),
    }
    return jsonify({'status': 'success', 'data': safe_manifest})
#: The only keys RestoreOptions recognizes. A typo'd or renamed key (e.g.
#: "restoreSecrets") would otherwise be silently ignored by opts_dict.get(),
#: leaving that flag at its True default -- restoring secrets a caller's
#: request clearly meant to exclude, with no indication anything was wrong.
_RESTORE_OPTION_KEYS = frozenset((
    'restore_config', 'restore_secrets', 'restore_wifi', 'restore_fonts',
    'restore_plugin_uploads', 'reinstall_plugins',
))
def _installed_path(psm, plugin_id):
    """Where the store finds ``plugin_id`` installed, or None.

    The same lookup install_plugin makes to decide that a copy exists: the
    id, or an id the registry proves is the same plugin (``aliases``, the
    ``plugin_path`` name), never a bare ``ledmatrix-<id>`` folder.
    """
    found = psm._existing_install(plugin_id)
    return found if isinstance(found, Path) and found.exists() else None


@api_v3.route('/backup/restore', methods=['POST'])
def backup_restore():
    """Restore a backup ZIP with optional RestoreOptions."""
    from src.backup_manager import restore_backup, RestoreOptions
    if 'backup_file' not in request.files:
        return jsonify({'status': 'error', 'message': 'No backup_file in request'}), 400
    f = request.files['backup_file']
    options_raw = request.form.get('options', '{}')
    try:
        opts_dict = json.loads(options_raw)
    except json.JSONDecodeError:
        opts_dict = None
    if not isinstance(opts_dict, dict):
        # Every option defaults to True, so falling back to {} on a
        # parse failure would silently perform a FULL restore —
        # secrets and all — for a caller who asked for a narrow one
        # and mis-serialized it. Refuse instead of guessing.
        return jsonify({
            'status': 'error',
            'message': 'Invalid options: expected a JSON object',
        }), 400
    unknown_keys = set(opts_dict) - _RESTORE_OPTION_KEYS
    if unknown_keys:
        return jsonify({
            'status': 'error',
            'message': f'Unknown restore option(s): {", ".join(sorted(unknown_keys))}',
        }), 400
    # _coerce_to_bool (not bare bool()) because a request can send these
    # as JSON strings: bool("false") is True in Python, so a caller who
    # explicitly asked to skip secrets would have had them restored
    # anyway.
    options = RestoreOptions(
        restore_config=_coerce_to_bool(opts_dict.get('restore_config', True)),
        restore_secrets=_coerce_to_bool(opts_dict.get('restore_secrets', True)),
        restore_wifi=_coerce_to_bool(opts_dict.get('restore_wifi', True)),
        restore_fonts=_coerce_to_bool(opts_dict.get('restore_fonts', True)),
        restore_plugin_uploads=_coerce_to_bool(opts_dict.get('restore_plugin_uploads', True)),
        reinstall_plugins=_coerce_to_bool(opts_dict.get('reinstall_plugins', True)),
    )
    with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp:
        tmp_path = tmp.name
        f.save(tmp_path)
    try:
        result = restore_backup(Path(tmp_path), PROJECT_ROOT, options)
    finally:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
    # Restored fonts reach the Fonts tab through a catalog cached for five
    # minutes (fonts.py); upload and delete clear it, and so must this.
    if any(str(item).startswith('fonts') for item in result.restored):
        delete_cached('fonts_catalog')

    # Reinstall plugins if requested and store manager available
    if options.reinstall_plugins and result.plugins_to_install:
        psm = getattr(api_v3, 'plugin_store_manager', None)
        for plug in result.plugins_to_install:
            pid = plug.get('plugin_id')
            if not pid:
                continue
            try:
                # Only what is missing. install_plugin replaces an installed
                # copy with a fresh download, so restoring onto the same
                # device re-downloaded every plugin, and one installed from
                # its own URL (not in the registry) "failed" and failed the
                # whole restore while it sat there installed. The store's
                # own lookup, so registry aliases count as installed too.
                if psm and _installed_path(psm, pid) is not None:
                    result.skipped.append(f'plugin:{pid} (installed)')
                    continue
                if psm and hasattr(psm, 'install_plugin'):
                    ok = psm.install_plugin(pid)
                    if ok:
                        result.plugins_installed.append(pid)
                    else:
                        result.plugins_failed.append({'plugin_id': pid, 'error': 'install_plugin returned False'})
                else:
                    result.plugins_failed.append({'plugin_id': pid, 'error': 'Store manager unavailable'})
            except Exception as pe:
                logger.error(
                    "[Backup] Failed to reinstall plugin %r: %s", pid, pe, exc_info=True
                )
                result.plugins_failed.append({'plugin_id': pid, 'error': 'Installation failed; see server logs'})

    # A restore that dropped files can still report success if the only
    # failures were plugin reinstalls, since those don't touch result.errors.
    if result.plugins_failed:
        result.success = False

    data = result.to_dict()
    if not result.success:
        # Name what failed, and what nonetheless landed. A restore is
        # partial far more often than it is total -- a fresh install can
        # leave config_secrets.json unwritable by the web service, so
        # config restores and secrets do not. "Restore had errors" alone
        # left the user unable to tell a wholly failed restore from one
        # that quietly dropped their API keys.
        failed_plugins = [
            str(p.get('plugin_id')) for p in (result.plugins_failed or []) if p.get('plugin_id')
        ]
        parts = []
        if result.restored:
            parts.append(f"restored: {', '.join(result.restored)}")
        if result.errors:
            parts.append(f"failed: {'; '.join(result.errors)}")
        if failed_plugins:
            parts.append(f"plugins not reinstalled: {', '.join(failed_plugins)}")
        message = 'Restore incomplete — ' + ('. '.join(parts) if parts else 'see logs')
        return jsonify({'status': 'error', 'message': message, 'data': data}), 500
    return jsonify({'status': 'success', 'data': data})
@api_v3.route('/backup/download/<path:filename>', methods=['GET'])
def backup_download(filename):
    """Stream a backup ZIP to the browser."""
    from flask import send_from_directory
    if _safe_backup_path(filename) is None:
        return jsonify({'status': 'error', 'message': 'Backup not found'}), 404
    try:
        # send_from_directory uses werkzeug safe_join internally — CodeQL-recognized sanitizer.
        return send_from_directory(_pkg._BACKUP_EXPORT_DIR, filename, as_attachment=True)
    except FileNotFoundError:
        return jsonify({'status': 'error', 'message': 'Backup not found'}), 404
@api_v3.route('/backup/<path:filename>', methods=['DELETE'])
def backup_delete(filename):
    """Delete a stored backup ZIP."""
    safe = _safe_backup_path(filename)
    if safe is None:
        return jsonify({'status': 'error', 'message': 'Backup not found'}), 404
    # Enumerate the export directory and match by name so the unlink target is
    # a filesystem-derived path rather than one constructed from user input.
    try:
        for entry in _pkg._BACKUP_EXPORT_DIR.iterdir():
            if entry.is_file() and entry.name == safe.name:
                entry.unlink()
                return jsonify({'status': 'success'})
    except OSError as e:
        logger.error("backup_delete failed: %s", e, exc_info=True)
        return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500
    return jsonify({'status': 'error', 'message': 'Backup not found'}), 404