Skip to content

fix: remove unauthenticated hosted core updates #9081

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
LIghtJUNction wants to merge 1 commit into from
Closed

fix: remove unauthenticated hosted core updates #9081

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 0 additions & 50 deletions astrbot/core/updator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import os
import sys
import time
import zipfile
from pathlib import Path

import psutil
Expand All @@ -24,30 +23,6 @@ def __init__(self, repo_mirror: str = "", verify: str | bool | None = None) -> N
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:
"""终止当前进程的所有子进程
Expand Down Expand Up @@ -221,18 +196,15 @@ async def download_update_package(
"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} 的更新文件。")
Expand All @@ -248,28 +220,6 @@ async def download_update_package(

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),
Expand Down
103 changes: 3 additions & 100 deletions tests/test_updator_socks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,7 +291,7 @@ def fake_unzip_file(zip_path: str, target_dir: str):


@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:
Expand All @@ -307,59 +307,14 @@ async def fake_fetch_release_info(url: str, latest: bool = True): # noqa: ARG00
{
"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",
}
]

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)
Expand All @@ -373,59 +328,7 @@ async def fake_download_file(url: str, path: str, progress_callback=None): # no

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
Expand Down
Loading