Files
appfactory-portal/app/routes/backups.py
T
JiriUhlir 7df5dc77b4 stzles
2026-06-16 10:43:00 +02:00

220 lines
7.6 KiB
Python

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'<a class="icon-action" href="/portal/backups/download/{backup_url_name}" title="Stáhnout" aria-label="Stáhnout">'
f'<i class="fa-solid fa-download" aria-hidden="true"></i></a>'
if can_download
else ""
)
# Restore overwrites live data, so it is admin-only and guarded by an explicit confirm.
restore_action = (
f"""
<form method="post" action="/portal/backups/restore"
onsubmit="return confirm('Obnovit zálohu {backup_name}?\\n\\nTato akce přepíše aktuální data AppFactory a může restartovat běžící služby. Akce je nevratná.');">
<input type="hidden" name="backup_path" value="{backup_path}">
<button type="submit" class="icon-action" title="Obnovit" aria-label="Obnovit">
<i class="fa-solid fa-clock-rotate-left" aria-hidden="true"></i>
</button>
</form>
"""
if can_download
else ""
)
rows += f"""
<tr class="backup-row">
<td>
<strong>{backup_name}</strong><br>
<span class="muted">{size_mb:.2f} MB</span>
</td>
<td>
<div class="copy-row">
<input readonly value="{backup_path}">
<button type="button" class="copy-button" data-copy-value="{backup_path}" onclick="copyText(this)">
<i class="fa-solid fa-copy" aria-hidden="true"></i> Kopírovat
</button>
</div>
</td>
<td class="actions-cell service-actions">
{download_action}
{restore_action}
<form method="post" action="/portal/backups/delete" onsubmit="return confirm('Smazat zálohu {backup_name}?');">
<input type="hidden" name="backup_path" value="{backup_path}">
<button type="submit" class="icon-action icon-action-danger" title="Smazat" aria-label="Smazat">
<i class="fa-solid fa-trash" aria-hidden="true"></i>
</button>
</form>
</td>
</tr>
"""
if not rows:
rows = '<tr><td colspan="3">Nebyly nalezeny žádné zálohy.</td></tr>'
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"""
<div class="card">
<h2><i class="fa-solid fa-box-archive" aria-hidden="true"></i> Správa záloh</h2>
<p class="muted">
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á.
</p>
<form method="post" action="/portal/backups/create">
<button type="submit"><i class="fa-solid fa-plus" aria-hidden="true"></i> Vytvořit zálohu</button>
</form>
</div>
<div class="card">
<h2>Dostupné zálohy</h2>
<table>
<tr>
<th>Záloha</th>
<th>Cesta</th>
<th>Akce</th>
</tr>
{rows}
</table>
</div>
""",
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,
)