import asyncio import tempfile import zipfile from pathlib import Path import pytest from bot.services.backup_archive import ( attach_archive_integrity, build_file_records, write_manifest, write_zip_from_directory, ) from bot.services.backup_restore_service import ( BackupArchiveError, BackupRestoreService, ) from bot.services.backup_worker import BACKUP_FILENAME_PREFIX from config.settings import Settings def _settings(tmp_path: Path, compose_dir: Path, **overrides) -> Settings: values = { "BOT_TOKEN": "token", "POSTGRES_USER": "app_user", "POSTGRES_PASSWORD": "app_password", "POSTGRES_DB": "shop", "BACKUP_DIR": str(tmp_path / "backups"), "BACKUP_COMPOSE_SOURCE_DIR": str(compose_dir), "_env_file": None, } values.update(overrides) return Settings(**values) def _write_backup_archive( path: Path, *, include_db=True, include_compose=True, unsafe=False, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(dir=path.parent) as tmp: staging_dir = Path(tmp) if include_db: dump_dir = staging_dir / "database" dump_dir.mkdir(parents=True) (dump_dir / "shop.dump").write_bytes(b"fake dump") if include_compose: compose_dir = staging_dir / "compose" compose_dir.mkdir(parents=True) (compose_dir / "docker-compose.yml").write_text("services: {}\n", encoding="utf-8") (compose_dir / ".env").write_text("POSTGRES_PASSWORD=secret\n", encoding="utf-8") manifest = { "app": "remnawave-minishop", "format_version": 1, "type": "test", "created_at": "2026-05-27T09:00:00+00:00", "postgres": {"database": "shop", "included": include_db}, "compose": {"included": include_compose, "files_count": 2 if include_compose else 0}, "warnings": [], } attach_archive_integrity( manifest, file_records=build_file_records(staging_dir), ) write_manifest(staging_dir, manifest) write_zip_from_directory(staging_dir, path) if unsafe: # Add a malicious member after signing; validation must reject before restore. with zipfile.ZipFile(path, "a") as archive: archive.writestr("compose/../evil.txt", "nope") def test_backup_restore_service_lists_archives_with_contents(tmp_path): compose_dir = tmp_path / "compose" compose_dir.mkdir() settings = _settings(tmp_path, compose_dir) archive_path = Path(settings.BACKUP_DIR) / f"{BACKUP_FILENAME_PREFIX}20260527-120000+0300.zip" _write_backup_archive(archive_path) archives = BackupRestoreService(settings).list_archives() assert [item.name for item in archives] == [archive_path.name] assert archives[0].has_database is True assert archives[0].has_compose is True assert archives[0].database_name == "shop" assert archives[0].compose_files_count == 2 def test_backup_restore_service_rejects_path_traversal_archive_name(tmp_path): settings = _settings(tmp_path, tmp_path / "compose") service = BackupRestoreService(settings) with pytest.raises(BackupArchiveError): service.archive_path_for_name("../backup.zip") def test_backup_restore_service_restores_compose_and_snapshots_current(tmp_path): compose_dir = tmp_path / "compose" compose_dir.mkdir() (compose_dir / "docker-compose.yml").write_text("old: true\n", encoding="utf-8") settings = _settings(tmp_path, compose_dir) archive_path = Path(settings.BACKUP_DIR) / f"{BACKUP_FILENAME_PREFIX}20260527-120000+0300.zip" _write_backup_archive(archive_path, include_db=False) service = BackupRestoreService(settings) result = service.restore_archive_sync( archive_path.name, restore_database=False, restore_compose=True, ) assert result.database_restored is False assert result.compose_files_restored == 2 assert (compose_dir / "docker-compose.yml").read_text(encoding="utf-8") == "services: {}\n" assert (compose_dir / ".env").read_text(encoding="utf-8") == "POSTGRES_PASSWORD=secret\n" assert result.compose_pre_restore_archive assert Path(result.compose_pre_restore_archive).is_file() snapshot = service.inspect_archive(Path(result.compose_pre_restore_archive)) assert snapshot.has_compose is True assert snapshot.compose_files_count == 1 def test_backup_restore_service_prevents_zip_slip_in_compose_restore(tmp_path): compose_dir = tmp_path / "compose" compose_dir.mkdir() settings = _settings(tmp_path, compose_dir) archive_path = Path(settings.BACKUP_DIR) / f"{BACKUP_FILENAME_PREFIX}20260527-120000+0300.zip" _write_backup_archive(archive_path, include_db=False, unsafe=True) with pytest.raises(BackupArchiveError): BackupRestoreService(settings).restore_archive_sync( archive_path.name, restore_database=False, restore_compose=True, ) assert not (tmp_path / "evil.txt").exists() def test_backup_restore_service_runs_pg_restore_for_dump(tmp_path): compose_dir = tmp_path / "compose" compose_dir.mkdir() settings = _settings(tmp_path, compose_dir) archive_path = Path(settings.BACKUP_DIR) / f"{BACKUP_FILENAME_PREFIX}20260527-120000+0300.zip" _write_backup_archive(archive_path, include_compose=False) service = BackupRestoreService(settings) restored_payloads = [] def fake_pg_restore(dump_path: Path) -> None: restored_payloads.append(dump_path.read_bytes()) service._run_pg_restore = fake_pg_restore result = asyncio.run( service.restore_archive( archive_path.name, restore_database=True, restore_compose=False, ) ) assert result.database_restored is True assert restored_payloads == [b"fake dump"] def test_backup_restore_service_accepts_archive_from_another_instance(tmp_path): compose_dir = tmp_path / "compose" compose_dir.mkdir() target_settings = _settings(tmp_path, compose_dir, BOT_TOKEN="target-token") archive_path = Path(target_settings.BACKUP_DIR) / ( f"{BACKUP_FILENAME_PREFIX}20260527-120000+0300.zip" ) _write_backup_archive(archive_path, include_compose=False) service = BackupRestoreService(target_settings) restored_payloads = [] def fake_pg_restore(dump_path: Path) -> None: restored_payloads.append(dump_path.read_bytes()) service._run_pg_restore = fake_pg_restore result = service.restore_archive_sync( archive_path.name, restore_database=True, restore_compose=False, ) assert result.database_restored is True assert restored_payloads == [b"fake dump"] def test_backup_restore_service_validates_uploaded_zip(tmp_path): compose_dir = tmp_path / "compose" compose_dir.mkdir() settings = _settings(tmp_path, compose_dir) temp_path = tmp_path / "not-a-backup.zip" temp_path.write_text("not zip", encoding="utf-8") with pytest.raises(BackupArchiveError): BackupRestoreService(settings).import_uploaded_archive(temp_path, "backup.zip") def test_backup_restore_service_rejects_tampered_archive(tmp_path): compose_dir = tmp_path / "compose" compose_dir.mkdir() settings = _settings(tmp_path, compose_dir) archive_path = Path(settings.BACKUP_DIR) / f"{BACKUP_FILENAME_PREFIX}20260527-120000+0300.zip" _write_backup_archive(archive_path, include_compose=False) tampered_path = archive_path.with_name("tampered.zip") with zipfile.ZipFile(archive_path) as source, zipfile.ZipFile(tampered_path, "w") as target: for member in source.infolist(): payload = source.read(member.filename) if member.filename == "database/shop.dump": payload = b"not the signed dump" target.writestr(member, payload) with pytest.raises(BackupArchiveError): BackupRestoreService(settings).import_uploaded_archive(tampered_path, "tampered.zip")