import html from pathlib import Path from urllib.parse import quote from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse from ..auth import require_user from ..backups import backup_dir, is_backup_path, list_backups from ..config import BACKUP_SCRIPT, RESTORE_SCRIPT from ..db.audit import log_audit_event from ..shell import run_command from ..templates.layout import page, render_result router = APIRouter() def is_admin(user) -> bool: return (user.get("role") or "").lower() == "admin" @router.get("/backups", response_class=HTMLResponse) def backups_page(request: Request, user=Depends(require_user)): rows = "" can_download = is_admin(user) for path in list_backups(): size_mb = path.stat().st_size / 1024 / 1024 backup_name = html.escape(path.name) backup_path = html.escape(str(path), quote=True) backup_url_name = quote(path.name, safe="") download_action = ( f'' f' Stáhnout' if can_download else "" ) # Restore overwrites live data, so it is admin-only and guarded by an explicit confirm. restore_action = ( f"""
""" if can_download else "" ) rows += f""" {backup_name}
{size_mb:.2f} MB
{download_action} {restore_action}
""" if not rows: rows = 'Nebyly nalezeny žádné zálohy.' restore_hint = ( "Obnovu lze spustit přímo z portálu tlačítkem Obnovit (pouze administrátor)." if can_download else "Obnovu zálohy může spustit pouze administrátor." ) return page( "Zálohy", f"""

Správa záloh

Zálohy neobsahují samotný adresář se zálohami. Obnova starší zálohy by neměla smazat novější soubory .tar.gz. {restore_hint} Obnova přepíše aktuální data a je nevratná.

Dostupné zálohy

{rows}
Záloha Cesta Akce
""", user=user, ) @router.get("/backups/download/{backup_name}") def download_backup(backup_name: str, user=Depends(require_user)): if not is_admin(user): raise HTTPException(status_code=403, detail="Only admins can download backups") target = backup_dir() / backup_name if not is_backup_path(target): raise HTTPException(status_code=400, detail="Invalid backup path") target = target.resolve() if not target.exists() or not target.is_file(): raise HTTPException(status_code=404, detail="Backup not found") log_audit_event( user, action="backup_download", target_type="backup", target_id=target.name, metadata={"backup_path": str(target)}, ) return FileResponse( path=target, filename=target.name, media_type="application/gzip", ) @router.post("/backups/create", response_class=HTMLResponse) def create_backup(user=Depends(require_user)): result = run_command([BACKUP_SCRIPT]) status = "OK" if result.returncode == 0 else "FAILED" log_audit_event( user, action="backup_create", target_type="backup", metadata={ "status": status, "returncode": result.returncode, }, ) return render_result( title=f"Vytvoření zálohy: {status}", back_url="/portal/backups", sections=[("Výstup", result.stdout), ("Chyba", result.stderr)], user=user, ) @router.post("/backups/delete") def delete_backup(backup_path: str = Form(...), user=Depends(require_user)): target = Path(backup_path) if not is_backup_path(target): return HTMLResponse("Neplatná cesta k záloze", status_code=400) target = target.resolve() if target.exists(): target.unlink() return RedirectResponse(url="/portal/backups", status_code=303) @router.post("/backups/restore", response_class=HTMLResponse) def restore_backup(backup_path: str = Form(...), user=Depends(require_user)): # Restoring overwrites live data and may restart services, so it is restricted to admins. if not is_admin(user): raise HTTPException(status_code=403, detail="Only admins can restore backups") target = Path(backup_path) if not is_backup_path(target): return HTMLResponse("Neplatná cesta k záloze", status_code=400) target = target.resolve() if not target.exists() or not target.is_file(): raise HTTPException(status_code=404, detail="Backup not found") result = run_command([RESTORE_SCRIPT, "--force", str(target)]) status = "OK" if result.returncode == 0 else "FAILED" log_audit_event( user, action="backup_restore", target_type="backup", target_id=target.name, metadata={ "backup_path": str(target), "status": status, "returncode": result.returncode, }, ) return render_result( title=f"Obnova zálohy: {status}", back_url="/portal/backups", sections=[("Výstup", result.stdout), ("Chyba", result.stderr)], user=user, )