diff --git a/astrbot/core/star/updator.py b/astrbot/core/star/updator.py index cb462d8b2..c64777906 100644 --- a/astrbot/core/star/updator.py +++ b/astrbot/core/star/updator.py @@ -1,10 +1,9 @@ import os -import shutil import zipfile 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.utils.io import ensure_dir, remove_dir from ..star.star import StarMetadata from ..updator import RepoZipUpdator @@ -72,28 +71,9 @@ class PluginUpdator(RepoZipUpdator): def unzip_file(self, zip_path: str, target_dir: str) -> None: ensure_dir(target_dir) - update_dir = "" logger.info(f"Extracting archive: {zip_path}") with zipfile.ZipFile(zip_path, "r") as z: - update_dir = z.namelist()[0] + update_dir = self._resolve_archive_root_dir(z.namelist()) z.extractall(target_dir) - files = os.listdir(os.path.join(target_dir, update_dir)) - for f in files: - if os.path.isdir(os.path.join(target_dir, update_dir, f)): - if os.path.exists(os.path.join(target_dir, f)): - shutil.rmtree(os.path.join(target_dir, f), onerror=on_error) - elif os.path.exists(os.path.join(target_dir, f)): - os.remove(os.path.join(target_dir, f)) - shutil.move(os.path.join(target_dir, update_dir, f), target_dir) - - try: - logger.info( - f"Removing temporary files: {zip_path} and {os.path.join(target_dir, update_dir)}", - ) - shutil.rmtree(os.path.join(target_dir, update_dir), onerror=on_error) - os.remove(zip_path) - except BaseException: - logger.warning( - f"Failed to remove update files; you can manually delete {zip_path} and {os.path.join(target_dir, update_dir)}", - ) + self._finalize_extracted_archive(zip_path, target_dir, update_dir) diff --git a/astrbot/core/zip_updator.py b/astrbot/core/zip_updator.py index 1408a1c45..15096cc81 100644 --- a/astrbot/core/zip_updator.py +++ b/astrbot/core/zip_updator.py @@ -234,30 +234,87 @@ class RepoZipUpdator: 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 = z.namelist()[0] + update_dir = self._resolve_archive_root_dir(z.namelist()) z.extractall(target_dir) logger.debug(f"解压文件完成: {zip_path}") - files = os.listdir(os.path.join(target_dir, update_dir)) + self._finalize_extracted_archive(zip_path, target_dir, update_dir) + + @staticmethod + def _resolve_archive_root_dir(entries: list[str]) -> str: + normalized_entries = [os.path.normpath(entry) for entry in entries] + portable_entries = [entry.replace("\\", "/") for entry in normalized_entries] + root_candidates: list[str] = [] + + for raw_entry, normalized_entry, portable_entry in zip( + entries, normalized_entries, portable_entries + ): + if normalized_entry == ".": + continue + + has_children = any( + other_entry != portable_entry + and other_entry.startswith(f"{portable_entry}/") + for other_entry in portable_entries + ) + if raw_entry.endswith(("/", "\\")) or has_children: + root_candidates.append(normalized_entry) + continue + + parent_portable, _, _ = portable_entry.rpartition("/") + if not parent_portable: + return "" + root_candidates.append(parent_portable.replace("/", os.sep)) + + if not root_candidates: + return "" + return os.path.commonpath(root_candidates) + + def _finalize_extracted_archive( + self, + zip_path: str, + target_dir: str, + update_dir: str, + ) -> None: + target_root_path = os.path.normpath(target_dir) + + def _join_under_root(root: str, *parts: str) -> str: + path = os.path.normpath(os.path.join(root, *parts)) + try: + if os.path.commonpath([root, path]) != root: + raise ValueError("path escapes root directory") + except ValueError as exc: + raise ValueError("path escapes root directory") from exc + return path + + if not update_dir: + try: + os.remove(zip_path) + except Exception: + logger.warning(f"删除更新文件失败，可以手动删除 {zip_path}") + return + + update_root_path = _join_under_root(target_root_path, update_dir) + + files = os.listdir(update_root_path) for f in files: - if os.path.isdir(os.path.join(target_dir, update_dir, f)): - if os.path.exists(os.path.join(target_dir, f)): - shutil.rmtree(os.path.join(target_dir, f), onerror=on_error) - elif os.path.exists(os.path.join(target_dir, f)): - os.remove(os.path.join(target_dir, f)) - shutil.move(os.path.join(target_dir, update_dir, f), target_dir) + update_item_path = _join_under_root(update_root_path, f) + target_item_path = _join_under_root(target_root_path, f) + if os.path.isdir(update_item_path): + if os.path.exists(target_item_path): + shutil.rmtree(target_item_path, onerror=on_error) + elif os.path.exists(target_item_path): + os.remove(target_item_path) + shutil.move(update_item_path, target_root_path) try: - logger.debug( - f"删除临时更新文件: {zip_path} 和 {os.path.join(target_dir, update_dir)}", - ) - shutil.rmtree(os.path.join(target_dir, update_dir), onerror=on_error) + logger.debug(f"删除临时更新文件: {zip_path} 和 {update_root_path}") + shutil.rmtree(update_root_path, onerror=on_error) os.remove(zip_path) - except BaseException: + except Exception: logger.warning( - f"删除更新文件失败，可以手动删除 {zip_path} 和 {os.path.join(target_dir, update_dir)}", + f"删除更新文件失败，可以手动删除 {zip_path} 和 {update_root_path}" ) def format_name(self, name: str) -> str: diff --git a/tests/test_updator_socks.py b/tests/test_updator_socks.py index 11009ff85..dc4c6e250 100644 --- a/tests/test_updator_socks.py +++ b/tests/test_updator_socks.py @@ -1,3 +1,5 @@ +import ntpath +import posixpath from dataclasses import dataclass, field from pathlib import Path from types import SimpleNamespace @@ -108,6 +110,113 @@ class _FakeFailingStreamAsyncClient: return _FakeFailingStreamResponse() +class _FakeZipArchive: + def __init__(self, names: list[str]): + self._names = names + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb) -> None: + return None + + def namelist(self) -> list[str]: + return self._names + + def extractall(self, target_dir: str) -> None: # noqa: ARG002 + return None + + +def _build_fake_archive_entries(archive_root: str) -> list[str]: + return [archive_root, posixpath.join(archive_root, ".dockerignore")] + + +def _build_fake_archive_entries_with_first_file(root_dir: str) -> list[str]: + return [f"{root_dir}/README.md", f"{root_dir}/src/app.py"] + + +def _exercise_unzip_file_windows_path_normalization( + monkeypatch: pytest.MonkeyPatch, + *, + updater_module, + zip_updator_module, + updater, + target_dir: str, + archive_root: str, + logger_method: str, +) -> dict[str, object | None]: + captured: dict[str, object | None] = { + "listdir": None, + "move": None, + "cleanup": None, + "removed": None, + } + + def fake_listdir(path: str) -> list[str]: + captured["listdir"] = path + return [".dockerignore"] + + monkeypatch.setattr(updater_module.os, "makedirs", lambda path, exist_ok=True: None) + monkeypatch.setattr(updater_module.os.path, "join", ntpath.join) + monkeypatch.setattr(updater_module.os.path, "normpath", ntpath.normpath) + monkeypatch.setattr(updater_module.os.path, "commonpath", ntpath.commonpath) + monkeypatch.setattr(updater_module.os.path, "isdir", lambda path: False) + monkeypatch.setattr(updater_module.os.path, "exists", lambda path: False) + monkeypatch.setattr( + updater_module.zipfile, + "ZipFile", + lambda path, mode: _FakeZipArchive(_build_fake_archive_entries(archive_root)), + ) + monkeypatch.setattr(updater_module.logger, logger_method, lambda message: None) + monkeypatch.setattr(updater_module.logger, "warning", lambda message: None) + monkeypatch.setattr(updater_module.os, "listdir", fake_listdir) + monkeypatch.setattr( + zip_updator_module.shutil, + "move", + lambda src, dst: captured.__setitem__("move", (src, dst)), + ) + monkeypatch.setattr( + zip_updator_module.shutil, + "rmtree", + lambda path, onerror=None: captured.__setitem__("cleanup", path), + ) + monkeypatch.setattr( + updater_module.os, + "remove", + lambda path: captured.__setitem__("removed", path), + ) + + updater.unzip_file("temp.zip", target_dir) + + return captured + + +def _assert_unzip_file_windows_path_normalization( + captured: dict[str, object | None], + *, + target_dir: str, + archive_root: str, +) -> None: + normalized_root = ntpath.normpath(archive_root) + expected_root = ( + target_dir + if normalized_root == "." + else ntpath.join(target_dir, normalized_root) + ) + expected_file = ntpath.join(expected_root, ".dockerignore") + + 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 + + def _build_fake_httpx_module(state: _FakeAsyncClientState) -> SimpleNamespace: class _FakeAsyncClient: def __init__(self, **kwargs): @@ -400,3 +509,158 @@ async def test_download_file_logs_url_and_target_path_on_failure( assert any(url in message for message in log_messages) assert any(str(target_path) in message for message in log_messages) + + +@pytest.mark.parametrize( + "archive_root", + [ + "AstrBotDevs-AstrBot-39386ee/", + "AstrBotDevs-AstrBot-39386ee", + "owner-repo-branch/subdir/", + ".", + ], +) +def test_repo_unzip_file_normalizes_windows_extended_length_paths( + monkeypatch: pytest.MonkeyPatch, + archive_root: str, +) -> None: + import astrbot.core.zip_updator as zip_updator_module + + target_dir = r"\\?\C:\Users\admin\AppData\Local\AstrBot\backend\app" + captured = _exercise_unzip_file_windows_path_normalization( + monkeypatch, + updater_module=zip_updator_module, + zip_updator_module=zip_updator_module, + updater=RepoZipUpdator(), + target_dir=target_dir, + archive_root=archive_root, + logger_method="debug", + ) + + _assert_unzip_file_windows_path_normalization( + captured, target_dir=target_dir, archive_root=archive_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 + import astrbot.core.zip_updator as zip_updator_module + + target_dir = r"\\?\C:\Users\admin\AppData\Local\AstrBot\data\plugins\demo" + captured = _exercise_unzip_file_windows_path_normalization( + monkeypatch, + updater_module=plugin_updator_module, + zip_updator_module=zip_updator_module, + updater=PluginUpdator.__new__(PluginUpdator), + target_dir=target_dir, + archive_root=archive_root, + logger_method="info", + ) + + _assert_unzip_file_windows_path_normalization( + captured, target_dir=target_dir, archive_root=archive_root + ) + + +@pytest.mark.parametrize( + ("archive_root", "expected_error"), + [ + ("../escape/", "path escapes root directory"), + ("C:/escape", "path escapes root directory"), + ], +) +def test_repo_unzip_file_rejects_archive_roots_outside_target_dir( + monkeypatch: pytest.MonkeyPatch, + archive_root: str, + expected_error: str, +) -> None: + import astrbot.core.zip_updator as zip_updator_module + + monkeypatch.setattr( + zip_updator_module.os, "makedirs", lambda path, exist_ok=True: None + ) + monkeypatch.setattr(zip_updator_module.os.path, "join", ntpath.join) + monkeypatch.setattr(zip_updator_module.os.path, "normpath", ntpath.normpath) + monkeypatch.setattr(zip_updator_module.os.path, "commonpath", ntpath.commonpath) + monkeypatch.setattr( + zip_updator_module.zipfile, + "ZipFile", + lambda path, mode: _FakeZipArchive(_build_fake_archive_entries(archive_root)), + ) + + with pytest.raises(ValueError, match=expected_error): + RepoZipUpdator().unzip_file("temp.zip", r"\\?\C:\Users\admin\target") + + +def test_repo_unzip_file_handles_archives_without_explicit_root_dir_entry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import astrbot.core.zip_updator as zip_updator_module + + target_dir = r"\\?\C:\Users\admin\AppData\Local\AstrBot\backend\app" + archive_root = "repo-root" + expected_root = ntpath.join(target_dir, archive_root) + expected_file = ntpath.join(expected_root, "README.md") + captured: dict[str, object | None] = { + "listdir": None, + "move": None, + "cleanup": None, + "removed": None, + } + + def fake_listdir(path: str) -> list[str]: + captured["listdir"] = path + return ["README.md"] + + monkeypatch.setattr( + zip_updator_module.os, "makedirs", lambda path, exist_ok=True: None + ) + monkeypatch.setattr(zip_updator_module.os.path, "join", ntpath.join) + monkeypatch.setattr(zip_updator_module.os.path, "normpath", ntpath.normpath) + monkeypatch.setattr(zip_updator_module.os.path, "commonpath", ntpath.commonpath) + monkeypatch.setattr(zip_updator_module.os.path, "isdir", lambda path: False) + monkeypatch.setattr(zip_updator_module.os.path, "exists", lambda path: False) + monkeypatch.setattr( + zip_updator_module.zipfile, + "ZipFile", + lambda path, mode: _FakeZipArchive( + _build_fake_archive_entries_with_first_file(archive_root) + ), + ) + monkeypatch.setattr(zip_updator_module.logger, "debug", lambda message: None) + monkeypatch.setattr(zip_updator_module.logger, "warning", lambda message: None) + monkeypatch.setattr(zip_updator_module.os, "listdir", fake_listdir) + monkeypatch.setattr( + zip_updator_module.shutil, + "move", + lambda src, dst: captured.__setitem__("move", (src, dst)), + ) + monkeypatch.setattr( + zip_updator_module.shutil, + "rmtree", + lambda path, onerror=None: captured.__setitem__("cleanup", path), + ) + monkeypatch.setattr( + zip_updator_module.os, + "remove", + lambda path: captured.__setitem__("removed", path), + ) + + RepoZipUpdator().unzip_file("temp.zip", target_dir) + + assert captured["listdir"] == expected_root + assert captured["move"] == (expected_file, target_dir) + assert captured["cleanup"] == expected_root + assert captured["removed"] == "temp.zip"