diff --git a/app/routes/backups.py b/app/routes/backups.py index 3bfb84c..e0db1d3 100644 --- a/app/routes/backups.py +++ b/app/routes/backups.py @@ -1,11 +1,12 @@ import html from pathlib import Path +from urllib.parse import quote -from fastapi import APIRouter, Depends, Form, Request -from fastapi.responses import HTMLResponse, RedirectResponse +from fastapi import APIRouter, Depends, Form, HTTPException, Request +from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse from ..auth import require_user -from ..backups import is_backup_path, list_backups +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 @@ -14,14 +15,25 @@ 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'

Stáhnout

' + if can_download + else "" + ) restore_cmd = html.escape( f"sudo /home/jiri/workspace/appfactory-tools/scripts/restore-appfactory.sh --force {path}", @@ -48,6 +60,7 @@ def backups_page(request: Request, user=Depends(require_user)):
Obnova je nebezpečná a musí se spustit ručně přes SSH.
+ {download_action}
@@ -92,6 +105,34 @@ def backups_page(request: Request, user=Depends(require_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])