fix: guard empty updater archive roots

Author: zouyonghe

astrbot/core/star/updator.py

@@ -5,6 +5,7 @@
 from astrbot.core import logger
 from astrbot.core.utils.astrbot_path import get_astrbot_plugin_path
 from astrbot.core.utils.io import ensure_dir, on_error, remove_dir
+from astrbot.core.zip_updator import normalize_archive_root_dir
 
 from ..star.star import StarMetadata
 from ..updator import RepoZipUpdator
@@ -75,9 +76,18 @@ def unzip_file(self, zip_path: str, target_dir: str) -> None:
         update_dir = ""
         logger.info(f"Extracting archive: {zip_path}")
         with zipfile.ZipFile(zip_path, "r") as z:
-            update_dir = self._normalize_archive_root_dir(z.namelist()[0])
+            update_dir = normalize_archive_root_dir(z.namelist()[0])
             z.extractall(target_dir)
 
+        if not update_dir:
+            try:
+                os.remove(zip_path)
+            except BaseException:
+                logger.warning(
+                    f"Failed to remove update files; you can manually delete {zip_path}",
+                )
+            return
+
         update_root_path = os.path.normpath(os.path.join(target_dir, update_dir))
         files = os.listdir(update_root_path)
         for f in files:

astrbot/core/zip_updator.py

@@ -13,6 +13,11 @@
 from astrbot.core.utils.version_comparator import VersionComparator
 
 
+def normalize_archive_root_dir(path: str) -> str:
+    normalized = os.path.normpath(path)
+    return "" if normalized == "." else normalized
+
+
 class ReleaseInfo:
     version: str
     published_at: str
@@ -231,20 +236,22 @@ def parse_github_url(self, url: str):
             return author, repo, branch
         raise ValueError("无效的 GitHub URL")
 
-    @staticmethod
-    def _normalize_archive_root_dir(path: str) -> str:
-        normalized = os.path.normpath(path)
-        return "" if normalized == "." else normalized
-
     def unzip_file(self, zip_path: str, target_dir: str) -> None:
         """解压缩文件, 并将压缩包内**第一个**文件夹内的文件移动到 target_dir"""
         ensure_dir(target_dir)
         update_dir = ""
         with zipfile.ZipFile(zip_path, "r") as z:
-            update_dir = self._normalize_archive_root_dir(z.namelist()[0])
+            update_dir = normalize_archive_root_dir(z.namelist()[0])
             z.extractall(target_dir)
         logger.debug(f"解压文件完成: {zip_path}")
 
+        if not update_dir:
+            try:
+                os.remove(zip_path)
+            except BaseException:
+                logger.warning(f"删除更新文件失败，可以手动删除 {zip_path}")
+            return
+
         update_root_path = os.path.normpath(os.path.join(target_dir, update_dir))
         files = os.listdir(update_root_path)
         for f in files:

tests/test_updator_socks.py

@@ -440,15 +440,23 @@ def test_repo_unzip_file_normalizes_windows_extended_length_paths(
 ) -> None:
     import astrbot.core.zip_updator as zip_updator_module
 
-    target_dir = r"\\?\\C:\\Users\\admin\\AppData\\Local\\AstrBot\\backend\\app"
-    expected_root = ntpath.normpath(ntpath.join(target_dir, archive_root))
-    expected_file = ntpath.normpath(
-        ntpath.join(target_dir, archive_root, ".dockerignore")
+    target_dir = r"\\?\C:\Users\admin\AppData\Local\AstrBot\backend\app"
+    normalized_root = ntpath.normpath(archive_root)
+    expected_root = (
+        target_dir
+        if normalized_root == "."
+        else ntpath.join(target_dir, normalized_root)
     )
-    captured: dict[str, object] = {}
+    expected_file = ntpath.join(expected_root, ".dockerignore")
+    captured: dict[str, object | None] = {
+        "listdir": None,
+        "move": None,
+        "cleanup": None,
+        "removed": None,
+    }
 
     def fake_listdir(path: str) -> list[str]:
-        captured.setdefault("listdir", path)
+        captured["listdir"] = path
         return [".dockerignore"]
 
     monkeypatch.setattr(
@@ -469,41 +477,65 @@ def fake_listdir(path: str) -> list[str]:
     monkeypatch.setattr(
         zip_updator_module.shutil,
         "move",
-        lambda src, dst: captured.setdefault("move", (src, dst)),
+        lambda src, dst: captured.__setitem__("move", (src, dst)),
     )
     monkeypatch.setattr(
         zip_updator_module.shutil,
         "rmtree",
-        lambda path, onerror=None: captured.setdefault("cleanup", path),
+        lambda path, onerror=None: captured.__setitem__("cleanup", path),
     )
     monkeypatch.setattr(
         zip_updator_module.os,
         "remove",
-        lambda path: captured.setdefault("removed", path),
+        lambda path: captured.__setitem__("removed", path),
     )
 
     RepoZipUpdator().unzip_file("temp.zip", target_dir)
 
+    assert captured["removed"] == "temp.zip"
+    if normalized_root == ".":
+        assert captured["listdir"] is None
+        assert captured["move"] is None
+        assert captured["cleanup"] is None
+        return
+
     assert captured["listdir"] == expected_root
     assert captured["move"] == (expected_file, target_dir)
     assert captured["cleanup"] == expected_root
 
 
+@pytest.mark.parametrize(
+    "archive_root",
+    [
+        "AstrBotDevs-demo-39386ee/",
+        "AstrBotDevs-demo-39386ee",
+        "owner-repo-branch/subdir/",
+        ".",
+    ],
+)
 def test_plugin_unzip_file_normalizes_windows_extended_length_paths(
     monkeypatch: pytest.MonkeyPatch,
+    archive_root: str,
 ) -> None:
     import astrbot.core.star.updator as plugin_updator_module
 
-    target_dir = r"\\?\\C:\\Users\\admin\\AppData\\Local\\AstrBot\\data\\plugins\\demo"
-    archive_root = "AstrBotDevs-demo-39386ee/"
-    expected_root = ntpath.normpath(ntpath.join(target_dir, archive_root))
-    expected_file = ntpath.normpath(
-        ntpath.join(target_dir, archive_root, ".dockerignore")
+    target_dir = r"\\?\C:\Users\admin\AppData\Local\AstrBot\data\plugins\demo"
+    normalized_root = ntpath.normpath(archive_root)
+    expected_root = (
+        target_dir
+        if normalized_root == "."
+        else ntpath.join(target_dir, normalized_root)
     )
-    captured: dict[str, object] = {}
+    expected_file = ntpath.join(expected_root, ".dockerignore")
+    captured: dict[str, object | None] = {
+        "listdir": None,
+        "move": None,
+        "cleanup": None,
+        "removed": None,
+    }
 
     def fake_listdir(path: str) -> list[str]:
-        captured.setdefault("listdir", path)
+        captured["listdir"] = path
         return [".dockerignore"]
 
     monkeypatch.setattr(
@@ -524,21 +556,28 @@ def fake_listdir(path: str) -> list[str]:
     monkeypatch.setattr(
         plugin_updator_module.shutil,
         "move",
-        lambda src, dst: captured.setdefault("move", (src, dst)),
+        lambda src, dst: captured.__setitem__("move", (src, dst)),
     )
     monkeypatch.setattr(
         plugin_updator_module.shutil,
         "rmtree",
-        lambda path, onerror=None: captured.setdefault("cleanup", path),
+        lambda path, onerror=None: captured.__setitem__("cleanup", path),
     )
     monkeypatch.setattr(
         plugin_updator_module.os,
         "remove",
-        lambda path: captured.setdefault("removed", path),
+        lambda path: captured.__setitem__("removed", path),
     )
 
     PluginUpdator.__new__(PluginUpdator).unzip_file("temp.zip", target_dir)
 
+    assert captured["removed"] == "temp.zip"
+    if normalized_root == ".":
+        assert captured["listdir"] is None
+        assert captured["move"] is None
+        assert captured["cleanup"] is None
+        return
+
     assert captured["listdir"] == expected_root
     assert captured["move"] == (expected_file, target_dir)
     assert captured["cleanup"] == expected_root