fix: validate release payload shape before parsing

Author: zouyonghe

astrbot/core/zip_updator.py

@@ -48,6 +48,56 @@ def __init__(self, repo_mirror: str = "") -> None:
         self.repo_mirror = repo_mirror
         self.rm_on_error = on_error
 
+    @staticmethod
+    def _normalize_release_payload(result: object, url: str) -> list:
+        if isinstance(result, dict):
+            releases = [result]
+        elif isinstance(result, list):
+            releases = result
+        else:
+            logger.error(
+                f"版本信息格式异常，期望列表或字典，实际为: {type(result).__name__}, url: {url}",
+            )
+            raise FetchReleaseError("版本信息格式异常")
+
+        if not releases:
+            return []
+
+        required_fields = (
+            "name",
+            "published_at",
+            "body",
+            "tag_name",
+            "zipball_url",
+        )
+        normalized = []
+        for idx, release in enumerate(releases):
+            if not isinstance(release, dict):
+                logger.error(
+                    f"版本信息第 {idx} 项格式异常，期望字典，实际为: {type(release).__name__}, url: {url}",
+                )
+                raise FetchReleaseError("版本信息格式异常")
+
+            missing_fields = [
+                field for field in required_fields if field not in release
+            ]
+            if missing_fields:
+                logger.error(
+                    f"版本信息第 {idx} 项缺少字段: {missing_fields}, url: {url}",
+                )
+                raise FetchReleaseError("版本信息字段缺失")
+
+            normalized.append(
+                {
+                    "version": release["name"] or release["tag_name"],
+                    "published_at": release["published_at"],
+                    "body": release["body"],
+                    "tag_name": release["tag_name"],
+                    "zipball_url": release["zipball_url"],
+                },
+            )
+        return normalized
+
     async def fetch_release_info(self, url: str) -> list:
         """请求版本信息。
         返回一个列表，每个元素是一个字典，包含版本号、发布时间、更新内容、commit hash等信息。
@@ -80,22 +130,7 @@ async def fetch_release_info(self, url: str) -> list:
             logger.error(f"解析版本信息时发生异常: {e}")
             raise FetchReleaseError("解析版本信息失败") from e
 
-        if isinstance(result, dict):
-            result = [result]
-        if not result:
-            return []
-        ret = []
-        for release in result:
-            ret.append(
-                {
-                    "version": release["name"],
-                    "published_at": release["published_at"],
-                    "body": release["body"],
-                    "tag_name": release["tag_name"],
-                    "zipball_url": release["zipball_url"],
-                },
-            )
-        return ret
+        return self._normalize_release_payload(result, url)
 
     def unzip(self) -> NoReturn:
         raise NotImplementedError

tests/unit/test_updator.py

@@ -4,6 +4,54 @@
 from astrbot.core.zip_updator import FetchReleaseError, RepoZipUpdator
 
 
+def test_normalize_release_payload_raises_on_missing_fields():
+    updator = RepoZipUpdator()
+    malformed_payload = [
+        {
+            "name": "v1.0.0",
+            "published_at": "2026-03-01T00:00:00Z",
+            "tag_name": "v1.0.0",
+            "zipball_url": "https://example.com/v1.0.0.zip",
+        }
+    ]
+
+    with pytest.raises(FetchReleaseError, match="版本信息字段缺失"):
+        updator._normalize_release_payload(
+            malformed_payload,
+            "https://example.invalid/releases",
+        )
+
+
+def test_normalize_release_payload_raises_on_invalid_item_type():
+    updator = RepoZipUpdator()
+
+    with pytest.raises(FetchReleaseError, match="版本信息格式异常"):
+        updator._normalize_release_payload(
+            ["invalid-release-item"],
+            "https://example.invalid/releases",
+        )
+
+
+def test_normalize_release_payload_accepts_valid_payload():
+    updator = RepoZipUpdator()
+    payload = {
+        "name": "v1.0.0",
+        "published_at": "2026-03-01T00:00:00Z",
+        "body": "release body",
+        "tag_name": "v1.0.0",
+        "zipball_url": "https://example.com/v1.0.0.zip",
+    }
+
+    releases = updator._normalize_release_payload(
+        payload,
+        "https://example.invalid/releases",
+    )
+
+    assert len(releases) == 1
+    assert releases[0]["tag_name"] == "v1.0.0"
+    assert releases[0]["version"] == "v1.0.0"
+
+
 @pytest.mark.asyncio
 async def test_update_supports_nightly_tag(monkeypatch, tmp_path):
     updator = AstrBotUpdator()