mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-01 01:10:21 +08:00
fix: remove unauthenticated hosted core updates
This commit is contained in:
@@ -1,7 +1,6 @@
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import psutil
|
||||
@@ -24,30 +23,6 @@ class AstrBotUpdator(RepoZipUpdator):
|
||||
super().__init__(repo_mirror, verify=verify)
|
||||
self.MAIN_PATH = get_astrbot_path()
|
||||
self.ASTRBOT_RELEASE_API = "https://api.soulter.top/releases"
|
||||
self.CORE_PACKAGE_BASE_URL = (
|
||||
"https://astrbot-registry.soulter.top/download/astrbot-core"
|
||||
)
|
||||
|
||||
def _build_core_package_url(self, version: str | None) -> str | None:
|
||||
"""Build the hosted core package URL for a release tag.
|
||||
|
||||
Args:
|
||||
version: Release tag, such as ``v4.26.0``.
|
||||
|
||||
Returns:
|
||||
Public package URL, or None when hosted package download is disabled.
|
||||
"""
|
||||
|
||||
if not version or not str(version).startswith("v"):
|
||||
return None
|
||||
|
||||
base_url = os.environ.get(
|
||||
"ASTRBOT_CORE_PACKAGE_BASE_URL",
|
||||
self.CORE_PACKAGE_BASE_URL,
|
||||
).strip()
|
||||
if not base_url:
|
||||
return None
|
||||
return f"{base_url.rstrip('/')}/{version}/source.zip"
|
||||
|
||||
def terminate_child_processes(self) -> None:
|
||||
"""终止当前进程的所有子进程
|
||||
@@ -221,18 +196,15 @@ class AstrBotUpdator(RepoZipUpdator):
|
||||
"Error: You are running AstrBot via CLI, please use `pip` or `uv tool upgrade` to update AstrBot."
|
||||
) # 避免版本管理混乱
|
||||
|
||||
target_version = None
|
||||
if latest:
|
||||
latest_version = update_data[0]["tag_name"]
|
||||
if self.compare_version(VERSION, latest_version) >= 0:
|
||||
raise Exception("当前已经是最新版本。")
|
||||
target_version = latest_version
|
||||
file_url = update_data[0]["zipball_url"]
|
||||
elif str(version).startswith("v"):
|
||||
# 更新到指定版本
|
||||
for data in update_data:
|
||||
if data["tag_name"] == version:
|
||||
target_version = data["tag_name"]
|
||||
file_url = data["zipball_url"]
|
||||
if not file_url:
|
||||
raise Exception(f"未找到版本号为 {version} 的更新文件。")
|
||||
@@ -248,28 +220,6 @@ class AstrBotUpdator(RepoZipUpdator):
|
||||
|
||||
zip_path = Path(path)
|
||||
ensure_dir(zip_path.parent)
|
||||
hosted_package_url = self._build_core_package_url(target_version)
|
||||
if hosted_package_url:
|
||||
try:
|
||||
logger.info(
|
||||
f"优先从托管存储下载 AstrBot Core 更新包: {hosted_package_url}"
|
||||
)
|
||||
await self._download_file(
|
||||
hosted_package_url,
|
||||
str(zip_path),
|
||||
progress_callback=progress_callback,
|
||||
)
|
||||
if not zipfile.is_zipfile(zip_path):
|
||||
raise RuntimeError(
|
||||
"Downloaded hosted package is not a valid ZIP file"
|
||||
)
|
||||
return zip_path
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
f"从托管存储下载 AstrBot Core 更新包失败: {exc},"
|
||||
"将回退到当前更新源。"
|
||||
)
|
||||
|
||||
await self._download_file(
|
||||
file_url,
|
||||
str(zip_path),
|
||||
|
||||
@@ -291,7 +291,7 @@ async def test_plugin_updator_install_prefers_download_url(
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_astrbot_updator_prefers_hosted_core_package(
|
||||
async def test_astrbot_updator_uses_release_zipball_for_core_package(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
@@ -307,7 +307,7 @@ async def test_astrbot_updator_prefers_hosted_core_package(
|
||||
{
|
||||
"version": "AstrBot v99.0.0",
|
||||
"published_at": "2026-06-19T00:00:00Z",
|
||||
"body": "hosted core package",
|
||||
"body": "core package",
|
||||
"tag_name": "v99.0.0",
|
||||
"zipball_url": "https://github.example/archive.zip",
|
||||
}
|
||||
@@ -315,51 +315,6 @@ async def test_astrbot_updator_prefers_hosted_core_package(
|
||||
|
||||
async def fake_download_file(url: str, path: str, progress_callback=None): # noqa: ARG001
|
||||
calls.append(url)
|
||||
with zipfile.ZipFile(path, "w") as archive:
|
||||
archive.writestr("AstrBot-v99.0.0/README.md", "hosted-core")
|
||||
|
||||
monkeypatch.setattr(updator, "fetch_release_info", fake_fetch_release_info)
|
||||
monkeypatch.setattr(updator, "_download_file", fake_download_file)
|
||||
|
||||
zip_path = await updator.download_update_package(
|
||||
latest=False,
|
||||
version="v99.0.0",
|
||||
path=tmp_path / "core.zip",
|
||||
)
|
||||
|
||||
assert zip_path == tmp_path / "core.zip"
|
||||
assert zipfile.is_zipfile(zip_path)
|
||||
assert calls == ["https://cdn.example/core/v99.0.0/source.zip"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_astrbot_updator_falls_back_when_hosted_core_package_fails(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
monkeypatch.delenv("ASTRBOT_CLI", raising=False)
|
||||
monkeypatch.delenv("ASTRBOT_LAUNCHER", raising=False)
|
||||
monkeypatch.setenv("ASTRBOT_CORE_PACKAGE_BASE_URL", "https://cdn.example/core")
|
||||
|
||||
updator = AstrBotUpdator()
|
||||
calls: list[str] = []
|
||||
|
||||
async def fake_fetch_release_info(url: str, latest: bool = True): # noqa: ARG001
|
||||
return [
|
||||
{
|
||||
"version": "AstrBot v99.0.0",
|
||||
"published_at": "2026-06-19T00:00:00Z",
|
||||
"body": "hosted core package",
|
||||
"tag_name": "v99.0.0",
|
||||
"zipball_url": "https://github.example/archive.zip",
|
||||
}
|
||||
]
|
||||
|
||||
async def fake_download_file(url: str, path: str, progress_callback=None): # noqa: ARG001
|
||||
calls.append(url)
|
||||
parsed = urlparse(url)
|
||||
if parsed.scheme == "https" and parsed.hostname == "cdn.example":
|
||||
raise RuntimeError("404")
|
||||
Path(path).write_bytes(b"github-core")
|
||||
|
||||
monkeypatch.setattr(updator, "fetch_release_info", fake_fetch_release_info)
|
||||
@@ -373,59 +328,7 @@ async def test_astrbot_updator_falls_back_when_hosted_core_package_fails(
|
||||
|
||||
assert zip_path == tmp_path / "core.zip"
|
||||
assert zip_path.read_bytes() == b"github-core"
|
||||
assert calls == [
|
||||
"https://cdn.example/core/v99.0.0/source.zip",
|
||||
"https://github.example/archive.zip",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_astrbot_updator_falls_back_when_hosted_core_package_is_not_zip(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
monkeypatch.delenv("ASTRBOT_CLI", raising=False)
|
||||
monkeypatch.delenv("ASTRBOT_LAUNCHER", raising=False)
|
||||
monkeypatch.setenv("ASTRBOT_CORE_PACKAGE_BASE_URL", "https://cdn.example/core")
|
||||
|
||||
updator = AstrBotUpdator()
|
||||
calls: list[str] = []
|
||||
|
||||
async def fake_fetch_release_info(url: str, latest: bool = True): # noqa: ARG001
|
||||
return [
|
||||
{
|
||||
"version": "AstrBot v99.0.0",
|
||||
"published_at": "2026-06-19T00:00:00Z",
|
||||
"body": "hosted core package",
|
||||
"tag_name": "v99.0.0",
|
||||
"zipball_url": "https://github.example/archive.zip",
|
||||
}
|
||||
]
|
||||
|
||||
async def fake_download_file(url: str, path: str, progress_callback=None): # noqa: ARG001
|
||||
calls.append(url)
|
||||
parsed = urlparse(url)
|
||||
if parsed.scheme == "https" and parsed.hostname == "cdn.example":
|
||||
Path(path).write_bytes(b"not a zip")
|
||||
return
|
||||
with zipfile.ZipFile(path, "w") as archive:
|
||||
archive.writestr("AstrBot-v99.0.0/README.md", "github-core")
|
||||
|
||||
monkeypatch.setattr(updator, "fetch_release_info", fake_fetch_release_info)
|
||||
monkeypatch.setattr(updator, "_download_file", fake_download_file)
|
||||
|
||||
zip_path = await updator.download_update_package(
|
||||
latest=False,
|
||||
version="v99.0.0",
|
||||
path=tmp_path / "core.zip",
|
||||
)
|
||||
|
||||
assert zip_path == tmp_path / "core.zip"
|
||||
assert zipfile.is_zipfile(zip_path)
|
||||
assert calls == [
|
||||
"https://cdn.example/core/v99.0.0/source.zip",
|
||||
"https://github.example/archive.zip",
|
||||
]
|
||||
assert calls == ["https://github.example/archive.zip"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
Reference in New Issue
Block a user