mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-16 01:40:15 +08:00
fix windows updater zip root path normalization (#8019)
* fix: normalize updater zip root paths on windows * refactor: share updater archive path normalization * test: expand updater archive root coverage * fix: guard empty updater archive roots * refactor: share updater extraction safeguards * refactor: simplify updater extraction cleanup * refactor: inline updater root normalization * fix: infer archive root from zip entries
This commit is contained in:
エイカク
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user