fix: reject invalid backup uploads

astrbot/dashboard/services/backup_service.py

@@ -221,12 +221,23 @@ def get_backup_manifest(self, zip_path: str) -> dict | None:
             with zipfile.ZipFile(zip_path, "r") as zf:
                 if "manifest.json" in zf.namelist():
                     manifest_data = zf.read("manifest.json")
-                    return json.loads(manifest_data.decode("utf-8"))
+                    manifest = json.loads(manifest_data.decode("utf-8"))
+                    if not isinstance(manifest, dict):
+                        logger.debug(
+                            "备份 manifest 格式无效: expected dict, got %s",
+                            type(manifest).__name__,
+                        )
+                        return None
+                    return manifest
                 return None
         except Exception as exc:
             logger.debug(f"读取备份 manifest 失败: {exc}")
         return None
 
+    def validate_uploaded_backup(self, zip_path: str) -> None:
+        if self.get_backup_manifest(zip_path) is None:
+            raise BackupServiceError("无效的备份文件：缺少或无法读取 manifest.json")
+
     def list_backups(self, *, page: int, page_size: int) -> dict:
         self.ensure_cleanup_task_started()
         Path(self.backup_dir).mkdir(parents=True, exist_ok=True)
@@ -315,7 +326,16 @@ async def upload_backup(self, file: Any | None) -> dict:
 
         Path(self.backup_dir).mkdir(parents=True, exist_ok=True)
         zip_path = os.path.join(self.backup_dir, unique_filename)
-        await self._save_upload(file, zip_path)
+        try:
+            await self._save_upload(file, zip_path)
+            self.validate_uploaded_backup(zip_path)
+        except Exception:
+            try:
+                if os.path.exists(zip_path):
+                    os.remove(zip_path)
+            except Exception as exc:
+                logger.warning(f"清理失败的备份文件失败: {exc}")
+            raise
 
         logger.info(
             f"上传的备份文件已保存: {unique_filename} (原始名称: {file.filename})"
@@ -463,6 +483,11 @@ async def upload_complete(self, data: object) -> dict:
                             outfile.write(data_block)
 
             file_size = os.path.getsize(output_path)
+            try:
+                self.validate_uploaded_backup(output_path)
+            except Exception:
+                await self.cleanup_upload_session(upload_id)
+                raise
             self.mark_backup_as_uploaded(output_path)
             logger.info(f"分片上传完成: {filename}, size={file_size}, chunks={total}")
             await self.cleanup_upload_session(upload_id)
@@ -473,8 +498,11 @@ async def upload_complete(self, data: object) -> dict:
                 "size": file_size,
             }
         except Exception:
-            if os.path.exists(output_path):
-                os.remove(output_path)
+            try:
+                if os.path.exists(output_path):
+                    os.remove(output_path)
+            except Exception as exc:
+                logger.warning(f"清理失败的备份文件失败: {exc}")
             raise
 
     async def upload_abort(self, data: object) -> tuple[dict | None, str | None]:

tests/test_backup.py

@@ -3,8 +3,10 @@
 import json
 import os
 import re
+import time
 import zipfile
 from datetime import datetime
+from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -29,6 +31,8 @@
 )
 from astrbot.core.utils.version_comparator import VersionComparator
 from astrbot.dashboard.services.backup_service import (
+    BackupService,
+    BackupServiceError,
     generate_unique_filename,
     secure_filename,
 )
@@ -751,6 +755,81 @@ def test_generate_unique_filename_with_complex_name(self):
         assert re.search(r"my_backup_file_\d{8}_\d{6}\.zip", result)
 
 
+class TestBackupUploadValidation:
+    """备份上传校验测试"""
+
+    @pytest.mark.asyncio
+    async def test_upload_backup_rejects_invalid_zip(
+        self, mock_main_db, tmp_path, monkeypatch
+    ):
+        """测试直接上传无效 ZIP 会失败且不留下文件"""
+        backup_dir = tmp_path / "backups"
+        data_dir = tmp_path / "data"
+        backup_dir.mkdir()
+        data_dir.mkdir()
+        monkeypatch.setattr(
+            "astrbot.dashboard.services.backup_service.get_astrbot_backups_path",
+            lambda: str(backup_dir),
+        )
+        monkeypatch.setattr(
+            "astrbot.dashboard.services.backup_service.get_astrbot_data_path",
+            lambda: str(data_dir),
+        )
+        core_lifecycle = MagicMock()
+        core_lifecycle.astrbot_config = {}
+        service = BackupService(mock_main_db, core_lifecycle)
+        upload = MagicMock()
+        upload.filename = "bad.zip"
+        upload.save = AsyncMock(
+            side_effect=lambda target: Path(target).write_bytes(b"not a zip")
+        )
+
+        with pytest.raises(BackupServiceError, match="无效的备份文件"):
+            await service.upload_backup(upload)
+
+        assert list(backup_dir.glob("*.zip")) == []
+
+    @pytest.mark.asyncio
+    async def test_upload_complete_rejects_invalid_zip(
+        self, mock_main_db, tmp_path, monkeypatch
+    ):
+        """测试分片上传无效 ZIP 会失败且清理合并文件"""
+        backup_dir = tmp_path / "backups"
+        data_dir = tmp_path / "data"
+        chunk_dir = backup_dir / ".chunks" / "upload-1"
+        chunk_dir.mkdir(parents=True)
+        data_dir.mkdir()
+        monkeypatch.setattr(
+            "astrbot.dashboard.services.backup_service.get_astrbot_backups_path",
+            lambda: str(backup_dir),
+        )
+        monkeypatch.setattr(
+            "astrbot.dashboard.services.backup_service.get_astrbot_data_path",
+            lambda: str(data_dir),
+        )
+        core_lifecycle = MagicMock()
+        core_lifecycle.astrbot_config = {}
+        service = BackupService(mock_main_db, core_lifecycle)
+        (chunk_dir / "0.part").write_bytes(b"not a zip")
+        service.upload_sessions["upload-1"] = {
+            "filename": "bad.zip",
+            "original_filename": "bad.zip",
+            "total_size": len(b"not a zip"),
+            "total_chunks": 1,
+            "received_chunks": {0},
+            "created_at": time.time(),
+            "last_activity": time.time(),
+            "chunk_dir": str(chunk_dir),
+        }
+
+        with pytest.raises(BackupServiceError, match="无效的备份文件"):
+            await service.upload_complete({"upload_id": "upload-1"})
+
+        assert not (backup_dir / "bad.zip").exists()
+        assert "upload-1" not in service.upload_sessions
+        assert not chunk_dir.exists()
+
+
 class TestVersionComparison:
     """版本比较函数测试 - 使用 VersionComparator"""