download backup admin possibility
This commit is contained in:
+44
-3
@@ -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'<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}",
|
||||
@@ -48,6 +60,7 @@ def backups_page(request: Request, user=Depends(require_user)):
|
||||
<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>
|
||||
@@ -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])
|
||||
|
||||
Reference in New Issue
Block a user