Files
appfactory-portal/app/routes/backups.py
T
2026-05-29 10:56:20 +02:00

190 lines
6.0 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
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'<p><a class="btn btn-secondary" href="/portal/backups/download/{backup_url_name}">Stáhnout</a></p>'
if can_download
else ""
)
restore_cmd = html.escape(
f"sudo /home/jiri/workspace/appfactory-tools/scripts/restore-appfactory.sh --force {path}",
quote=True,
)
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)">Kopírovat</button>
</div>
</td>
<td>
<div class="copy-row">
<input readonly value="{restore_cmd}">
<button type="button" class="copy-button" data-copy-value="{restore_cmd}" onclick="copyText(this)">Kopírovat</button>
</div>
<div class="muted">Obnova je nebezpečná a musí se spustit ručně přes SSH.</div>
</td>
<td class="actions-cell">
{download_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="danger">Smazat</button>
</form>
</td>
</tr>
"""
if not rows:
rows = '<tr><td colspan="4">Nebyly nalezeny žádné zálohy.</td></tr>'
return page(
"Zálohy",
f"""
<div class="card">
<h2>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.
Obnova zůstává ruční, aby nedošlo k nechtěnému přepsání dat.
</p>
<form method="post" action="/portal/backups/create">
<button type="submit">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>Příkaz pro obnovu</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-trigger")
def restore_trigger(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)
log_audit_event(
user,
action="restore_trigger",
target_type="backup",
target_id=target.name,
metadata={
"backup_path": str(target),
},
)
return RedirectResponse(url="/portal/backups", status_code=303)