from __future__ import annotations from dataclasses import dataclass from datetime import UTC, datetime import hashlib import json import os from pathlib import Path import shutil import subprocess import tempfile import zipfile from fastapi import HTTPException, status from sqlalchemy.engine import make_url from app.core.config import settings from app.models.user import User from app.schemas.backup import ( BackupListResponse, BackupManifest, BackupSummary, BackupValidationResponse, ) BACKUP_FILENAME_PREFIX = "olympus-backup-" BACKUP_FILENAME_SUFFIX = ".zip" DATABASE_DUMP_NAME = "database.dump" MANIFEST_NAME = "manifest.json" STORAGE_DIR_NAME = "storage" BACKUP_CONFIRM_TEXT = "ICH VERSTEHE DAS RISIKO" RESTORE_DISABLED_MESSAGE = ( "Automatischer Restore ist vorbereitet, aber in v0.9.1 deaktiviert. " "Bitte Restore ueber CLI-Script ausfuehren." ) @dataclass(frozen=True) class BackupStats: total_count: int total_size_bytes: int latest_backup_at: datetime | None class BackupService: @staticmethod def get_backup_dir() -> Path: backup_dir = (Path(settings.storage_base_path) / "backups").resolve() backup_dir.mkdir(parents=True, exist_ok=True) return backup_dir @staticmethod def list_backups() -> BackupListResponse: items = [ BackupService._read_summary(path) for path in sorted( BackupService.get_backup_dir().glob(f"{BACKUP_FILENAME_PREFIX}*{BACKUP_FILENAME_SUFFIX}"), key=lambda item: item.stat().st_mtime, reverse=True, ) ] latest_backup_at = next((item.created_at for item in items if item.created_at is not None), None) return BackupListResponse( items=items, total_count=len(items), total_size_bytes=sum(item.size_bytes for item in items), latest_backup_at=latest_backup_at, ) @staticmethod def get_backup_stats() -> BackupStats: backups = BackupService.list_backups() return BackupStats( total_count=backups.total_count, total_size_bytes=backups.total_size_bytes, latest_backup_at=backups.latest_backup_at, ) @staticmethod def create_backup(*, actor: User) -> BackupSummary: backup_dir = BackupService.get_backup_dir() timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S") filename = f"{BACKUP_FILENAME_PREFIX}{timestamp}{BACKUP_FILENAME_SUFFIX}" target_path = backup_dir / filename with tempfile.TemporaryDirectory(prefix="backup-", dir=backup_dir) as temp_dir_name: temp_dir = Path(temp_dir_name) dump_path = temp_dir / DATABASE_DUMP_NAME storage_temp_dir = temp_dir / STORAGE_DIR_NAME manifest_path = temp_dir / MANIFEST_NAME archive_path = temp_dir / filename BackupService._run_pg_dump(dump_path) file_count, total_size_bytes = BackupService._copy_storage_snapshot(storage_temp_dir) dump_size = dump_path.stat().st_size checksum_sha256 = BackupService._calculate_archive_checksum( dump_path=dump_path, storage_dir=storage_temp_dir, ) manifest = BackupManifest( backup_id=hashlib.sha256(f"{filename}:{actor.id}:{timestamp}".encode("utf-8")).hexdigest()[:24], created_at=datetime.now(UTC), app_version=settings.app_version, backup_type="full", database_url_host_anonymized=BackupService._anonymized_database_host(), database_name=BackupService._database_name(), storage_base_path=settings.storage_base_path, included_sections=["database", "storage"], file_count=file_count + 1, total_size_bytes=total_size_bytes + dump_size, checksum_sha256=checksum_sha256, created_by_user_id=actor.id, created_by_username=actor.username, ) manifest_path.write_text( json.dumps(manifest.model_dump(mode="json"), indent=2, ensure_ascii=True), encoding="utf-8", ) BackupService._write_archive( archive_path=archive_path, manifest_path=manifest_path, dump_path=dump_path, storage_dir=storage_temp_dir, ) shutil.move(str(archive_path), target_path) return BackupService._read_summary(target_path) @staticmethod def validate_backup(filename: str) -> BackupValidationResponse: path = BackupService.resolve_backup_path(filename) issues: list[str] = [] manifest: BackupManifest | None = None checksum_valid = False try: with zipfile.ZipFile(path) as archive: names = set(archive.namelist()) if MANIFEST_NAME not in names: issues.append("manifest.json fehlt") if DATABASE_DUMP_NAME not in names: issues.append("database.dump fehlt") if not any(name == f"{STORAGE_DIR_NAME}/" or name.startswith(f"{STORAGE_DIR_NAME}/") for name in names): issues.append("storage/ fehlt") if MANIFEST_NAME in names: try: with archive.open(MANIFEST_NAME) as manifest_file: manifest = BackupManifest.model_validate_json(manifest_file.read().decode("utf-8")) except Exception: issues.append("manifest.json ist ungueltig") if manifest is not None: checksum_valid = BackupService._validate_archive_checksum(archive, manifest.checksum_sha256) if not checksum_valid: issues.append("Checksumme ist ungueltig") except zipfile.BadZipFile: issues.append("ZIP-Datei ist ungueltig") valid = len(issues) == 0 return BackupValidationResponse( filename=path.name, valid=valid, message="Backup ist gueltig" if valid else "Backup-Pruefung fehlgeschlagen", issues=issues, checksum_valid=checksum_valid, restore_supported=False, requires_cli_restore=True, manifest=manifest, ) @staticmethod def restore_backup(filename: str, *, confirm_text: str) -> BackupValidationResponse: if confirm_text != BACKUP_CONFIRM_TEXT: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Bestaetigungstext stimmt nicht ueberein", ) validation = BackupService.validate_backup(filename) if not validation.valid: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Backup ist ungueltig und kann nicht wiederhergestellt werden", ) raise HTTPException( status_code=status.HTTP_501_NOT_IMPLEMENTED, detail=RESTORE_DISABLED_MESSAGE, ) @staticmethod def delete_backup(filename: str) -> None: path = BackupService.resolve_backup_path(filename) path.unlink(missing_ok=False) @staticmethod def resolve_backup_path(filename: str) -> Path: if not filename.endswith(BACKUP_FILENAME_SUFFIX): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Backup nicht gefunden") if Path(filename).name != filename or ".." in Path(filename).parts: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Backup nicht gefunden") path = (BackupService.get_backup_dir() / filename).resolve() backup_dir = BackupService.get_backup_dir() if backup_dir != path.parent or not path.exists() or not path.is_file(): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Backup nicht gefunden") return path @staticmethod def _run_pg_dump(dump_path: Path) -> None: pg_dump_url = BackupService._pg_dump_database_url() command = [ "pg_dump", "--format=custom", "--no-owner", "--no-privileges", f"--file={dump_path}", f"--dbname={pg_dump_url}", ] try: subprocess.run( command, check=True, capture_output=True, text=True, env=os.environ.copy(), ) except FileNotFoundError as exc: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="pg_dump ist im Hermes-Container nicht verfuegbar", ) from exc except subprocess.CalledProcessError as exc: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="PostgreSQL-Dump konnte nicht erstellt werden", ) from exc @staticmethod def _copy_storage_snapshot(target_dir: Path) -> tuple[int, int]: source_dir = Path(settings.storage_base_path).resolve() backup_dir = BackupService.get_backup_dir() source_dir.mkdir(parents=True, exist_ok=True) target_dir.mkdir(parents=True, exist_ok=True) file_count = 0 total_size_bytes = 0 for source_path in sorted(source_dir.rglob("*")): if source_path == backup_dir or backup_dir in source_path.parents: continue relative_path = source_path.relative_to(source_dir) destination_path = target_dir / relative_path if source_path.is_dir(): destination_path.mkdir(parents=True, exist_ok=True) continue if not source_path.is_file(): continue destination_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_path, destination_path) file_count += 1 total_size_bytes += source_path.stat().st_size return file_count, total_size_bytes @staticmethod def _write_archive( *, archive_path: Path, manifest_path: Path, dump_path: Path, storage_dir: Path, ) -> None: with zipfile.ZipFile(archive_path, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: archive.writestr(f"{STORAGE_DIR_NAME}/", "") archive.write(manifest_path, MANIFEST_NAME) archive.write(dump_path, DATABASE_DUMP_NAME) for file_path in sorted(storage_dir.rglob("*")): if file_path.is_dir(): continue archive.write(file_path, file_path.relative_to(storage_dir.parent).as_posix()) @staticmethod def _read_summary(path: Path) -> BackupSummary: default_summary = BackupSummary( filename=path.name, size_bytes=path.stat().st_size, validation_status="warning", validation_message="Manifest konnte nicht gelesen werden", ) try: with zipfile.ZipFile(path) as archive: with archive.open(MANIFEST_NAME) as manifest_file: manifest = BackupManifest.model_validate_json(manifest_file.read().decode("utf-8")) return BackupSummary( filename=path.name, size_bytes=path.stat().st_size, created_at=manifest.created_at, app_version=manifest.app_version, backup_type=manifest.backup_type, database_name=manifest.database_name, storage_base_path=manifest.storage_base_path, file_count=manifest.file_count, total_size_bytes=manifest.total_size_bytes, created_by_user_id=manifest.created_by_user_id, created_by_username=manifest.created_by_username, validation_status="valid", validation_message="Backup ist lesbar", ) except Exception: return default_summary @staticmethod def _calculate_archive_checksum(*, dump_path: Path, storage_dir: Path) -> str: digest = hashlib.sha256() digest.update(DATABASE_DUMP_NAME.encode("utf-8")) BackupService._update_digest_from_file(digest, dump_path) for file_path in sorted(storage_dir.rglob("*")): if file_path.is_dir(): continue digest.update(file_path.relative_to(storage_dir.parent).as_posix().encode("utf-8")) BackupService._update_digest_from_file(digest, file_path) return digest.hexdigest() @staticmethod def _validate_archive_checksum(archive: zipfile.ZipFile, expected_checksum: str) -> bool: digest = hashlib.sha256() if DATABASE_DUMP_NAME not in archive.namelist(): return False digest.update(DATABASE_DUMP_NAME.encode("utf-8")) with archive.open(DATABASE_DUMP_NAME) as dump_file: BackupService._update_digest_from_stream(digest, dump_file) for name in sorted(item for item in archive.namelist() if item.startswith(f"{STORAGE_DIR_NAME}/") and not item.endswith("/")): digest.update(name.encode("utf-8")) with archive.open(name) as storage_file: BackupService._update_digest_from_stream(digest, storage_file) return digest.hexdigest() == expected_checksum @staticmethod def _anonymized_database_host() -> str: parsed = make_url(settings.database_url) host = parsed.host or "unknown" digest = hashlib.sha256(host.encode("utf-8")).hexdigest()[:12] return f"sha256:{digest}" @staticmethod def _database_name() -> str: parsed = make_url(settings.database_url) return parsed.database or "unknown" @staticmethod def _pg_dump_database_url() -> str: parsed = make_url(settings.database_url) normalized = parsed.set(drivername="postgresql") return normalized.render_as_string(hide_password=False) @staticmethod def _update_digest_from_file(digest, file_path: Path) -> None: with file_path.open("rb") as file_handle: BackupService._update_digest_from_stream(digest, file_handle) @staticmethod def _update_digest_from_stream(digest, stream) -> None: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk)