fix: validate plugin install sources (#9061)

Author: Soulter

astrbot/core/star/star_manager.py

@@ -52,7 +52,7 @@
 from .filter.permission import PermissionType, PermissionTypeFilter
 from .star import star_map, star_registry
 from .star_handler import EventType, star_handlers_registry
-from .updator import PluginUpdator
+from .updator import PLUGIN_METADATA_FILENAMES, PluginUpdator
 
 try:
     from watchfiles import PythonFilter, awatch
@@ -465,38 +465,44 @@ async def _import_plugin_with_dependency_recovery(
 
     @staticmethod
     def _load_plugin_metadata(plugin_path: str, plugin_obj=None) -> StarMetadata | None:
-        """先寻找 metadata.yaml 文件，如果不存在，则使用插件对象的 info() 函数获取元数据。
+        """Load plugin metadata from metadata.yaml or metadata.yml.
 
-        Notes: 旧版本 AstrBot 插件可能使用的是 info() 函数来获取元数据。
+        Args:
+            plugin_path: Plugin directory path.
+            plugin_obj: Deprecated compatibility argument; ignored.
+
+        Returns:
+            Loaded plugin metadata, or None if no metadata file exists.
         """
+        del plugin_obj
         metadata = None
+        metadata_label = "metadata.yaml"
+        plugin_root = Path(plugin_path)
 
-        if not os.path.exists(plugin_path):
+        if not plugin_root.exists():
             raise Exception("插件不存在。")
 
-        if os.path.exists(os.path.join(plugin_path, "metadata.yaml")):
-            with open(
-                os.path.join(plugin_path, "metadata.yaml"),
-                encoding="utf-8",
-            ) as f:
+        metadata_path = next(
+            (
+                plugin_root / filename
+                for filename in PLUGIN_METADATA_FILENAMES
+                if (plugin_root / filename).exists()
+            ),
+            None,
+        )
+        if metadata_path:
+            metadata_label = metadata_path.name
+            with metadata_path.open(encoding="utf-8") as f:
                 metadata = yaml.safe_load(f)
-        elif plugin_obj and hasattr(plugin_obj, "info"):
-            # 使用 info() 函数
-            metadata = plugin_obj.info()
 
         if isinstance(metadata, dict):
             if "desc" not in metadata and "description" in metadata:
                 metadata["desc"] = metadata["description"]
 
-            if (
-                "name" not in metadata
-                or "desc" not in metadata
-                or "version" not in metadata
-                or "author" not in metadata
-            ):
-                raise Exception(
-                    "插件元数据信息不完整。name, desc, version, author 是必须的字段。",
-                )
+            try:
+                PluginUpdator.validate_plugin_metadata(metadata, metadata_label)
+            except ValueError as exc:
+                raise Exception(f"插件元数据校验失败：{exc!s}") from exc
             metadata = StarMetadata(
                 name=metadata["name"],
                 author=metadata["author"],
@@ -577,32 +583,42 @@ def _normalize_plugin_dir_name(plugin_name: str) -> str:
     def _validate_importable_name(plugin_name: str) -> None:
         if "/" in plugin_name or "\\" in plugin_name:
             raise ValueError(
-                "metadata.yaml 中 name 含有路径分隔符，不可用于 importlib 加载。"
+                "metadata 文件中 name 含有路径分隔符，不可用于 importlib 加载。"
             )
         if not plugin_name.isidentifier() or keyword.iskeyword(plugin_name):
             raise Exception(
-                "metadata.yaml 中 name 不是合法的模块名称（应为合法 Python 标识符且非关键字）。"
+                "metadata 文件中 name 不是合法的模块名称（应为合法 Python 标识符且非关键字）。"
             )
 
     @staticmethod
     def _get_plugin_dir_name_from_metadata(plugin_path: str) -> str:
-        metadata_path = os.path.join(plugin_path, "metadata.yaml")
-        if not os.path.exists(metadata_path):
-            raise Exception("未找到 metadata.yaml，无法获取插件目录名。")
+        plugin_root = Path(plugin_path)
+        metadata_path = next(
+            (
+                plugin_root / filename
+                for filename in PLUGIN_METADATA_FILENAMES
+                if (plugin_root / filename).exists()
+            ),
+            None,
+        )
+        if metadata_path is None:
+            raise Exception(
+                "未找到 metadata.yaml 或 metadata.yml，无法获取插件目录名。"
+            )
 
-        with open(metadata_path, encoding="utf-8") as f:
+        with metadata_path.open(encoding="utf-8") as f:
             metadata = yaml.safe_load(f)
 
         if not isinstance(metadata, dict):
-            raise Exception("metadata.yaml 格式错误。")
+            raise Exception(f"{metadata_path.name} 格式错误。")
 
         plugin_name = metadata.get("name")
         if not isinstance(plugin_name, str) or not plugin_name.strip():
-            raise Exception("metadata.yaml 中缺少 name 字段。")
+            raise Exception(f"{metadata_path.name} 中缺少 name 字段。")
 
         plugin_dir_name = PluginManager._normalize_plugin_dir_name(plugin_name)
         if not plugin_dir_name:
-            raise Exception("metadata.yaml 中 name 字段内容非法。")
+            raise Exception(f"{metadata_path.name} 中 name 字段内容非法。")
         PluginManager._validate_importable_name(plugin_dir_name)
         return plugin_dir_name
 

astrbot/core/star/updator.py

@@ -1,13 +1,18 @@
 import os
 import zipfile
 
+import yaml
+
 from astrbot.core import logger
 from astrbot.core.utils.astrbot_path import get_astrbot_plugin_path
 from astrbot.core.utils.io import ensure_dir, remove_dir
 
 from ..star.star import StarMetadata
 from ..updator import RepoZipUpdator
 
+PLUGIN_METADATA_FILENAMES = ("metadata.yaml", "metadata.yml")
+PLUGIN_METADATA_REQUIRED_FIELDS = ("name", "desc", "version", "author")
+
 
 class PluginUpdator(RepoZipUpdator):
     def __init__(self, repo_mirror: str = "", verify: str | bool | None = None) -> None:
@@ -58,6 +63,7 @@ async def update(
         elif repo_url:
             await self.download_from_repo_url(plugin_path, repo_url, proxy=proxy)
 
+        self.validate_plugin_archive(plugin_path + ".zip")
         try:
             remove_dir(plugin_path)
         except BaseException as e:
@@ -69,7 +75,135 @@ async def update(
 
         return plugin_path
 
+    @classmethod
+    def find_plugin_metadata_entry(cls, entries: list[str]) -> str | None:
+        """Find AstrBot plugin metadata in archive entries.
+
+        Args:
+            entries: Zip archive member names.
+
+        Returns:
+            The original archive entry name for plugin metadata, or None.
+        """
+        update_dir = cls._resolve_archive_root_dir(entries)
+        portable_update_dir = os.path.normpath(update_dir).replace("\\", "/")
+        if portable_update_dir == ".":
+            portable_update_dir = ""
+
+        entries_by_portable_path = {}
+        for entry in entries:
+            portable_entry = os.path.normpath(entry).replace("\\", "/")
+            if portable_entry in ("", "."):
+                continue
+            entries_by_portable_path[portable_entry] = entry
+
+        metadata_candidates = (
+            [
+                f"{portable_update_dir}/{filename}"
+                for filename in PLUGIN_METADATA_FILENAMES
+            ]
+            if portable_update_dir
+            else list(PLUGIN_METADATA_FILENAMES)
+        )
+        for candidate in metadata_candidates:
+            if candidate in entries_by_portable_path:
+                return entries_by_portable_path[candidate]
+        return None
+
+    @staticmethod
+    def validate_plugin_metadata(metadata: object, metadata_label: str) -> None:
+        """Validate AstrBot plugin metadata content.
+
+        Args:
+            metadata: Parsed metadata YAML content.
+            metadata_label: Metadata filename or archive entry for error messages.
+
+        Raises:
+            ValueError: If metadata is malformed or misses required fields.
+        """
+        if not isinstance(metadata, dict):
+            raise ValueError(f"{metadata_label} 格式错误。")
+
+        normalized_metadata = dict(metadata)
+        if "desc" not in normalized_metadata and "description" in normalized_metadata:
+            normalized_metadata["desc"] = normalized_metadata["description"]
+
+        missing_fields = [
+            field
+            for field in PLUGIN_METADATA_REQUIRED_FIELDS
+            if field not in normalized_metadata
+        ]
+        if missing_fields:
+            raise ValueError(
+                f"{metadata_label} 中缺少必需字段: {', '.join(missing_fields)}。"
+            )
+
+        invalid_fields = [
+            field
+            for field in PLUGIN_METADATA_REQUIRED_FIELDS
+            if not isinstance(normalized_metadata[field], str)
+            or not normalized_metadata[field].strip()
+        ]
+        if invalid_fields:
+            raise ValueError(
+                f"{metadata_label} 中字段 {', '.join(invalid_fields)} 必须是非空字符串。"
+            )
+
+    @classmethod
+    def inspect_plugin_archive(cls, zip_path: str) -> dict[str, object]:
+        """Inspect plugin metadata in an AstrBot plugin archive.
+
+        Args:
+            zip_path: Path to the plugin archive.
+
+        Returns:
+            A dict containing the metadata entry and parsed metadata.
+
+        Raises:
+            ValueError: If the archive is not a valid AstrBot plugin.
+        """
+        try:
+            with zipfile.ZipFile(zip_path, "r") as z:
+                metadata_entry = cls.find_plugin_metadata_entry(z.namelist())
+                if metadata_entry is None:
+                    raise ValueError(
+                        "压缩包不是合法的 AstrBot 插件：未找到 metadata.yaml 或 metadata.yml。"
+                    )
+
+                try:
+                    metadata_text = z.read(metadata_entry).decode("utf-8")
+                    metadata = yaml.safe_load(metadata_text)
+                except UnicodeDecodeError as exc:
+                    raise ValueError(f"{metadata_entry} 必须使用 UTF-8 编码。") from exc
+                except yaml.YAMLError as exc:
+                    raise ValueError(f"{metadata_entry} 格式错误。") from exc
+
+                cls.validate_plugin_metadata(metadata, metadata_entry)
+                return {
+                    "metadata_entry": metadata_entry,
+                    "metadata": metadata,
+                }
+        except zipfile.BadZipFile as exc:
+            raise ValueError("插件压缩包格式错误。") from exc
+
+    @classmethod
+    def validate_plugin_archive(cls, zip_path: str) -> str:
+        """Validate that an archive contains a valid AstrBot plugin.
+
+        Args:
+            zip_path: Path to the plugin archive.
+
+        Returns:
+            The archive entry name of the plugin metadata file.
+
+        Raises:
+            ValueError: If the archive is not a valid AstrBot plugin.
+        """
+        inspection = cls.inspect_plugin_archive(zip_path)
+        return str(inspection["metadata_entry"])
+
     def unzip_file(self, zip_path: str, target_dir: str) -> None:
+        self.validate_plugin_archive(zip_path)
         ensure_dir(target_dir)
         logger.info(f"Extracting archive: {zip_path}")
         with zipfile.ZipFile(zip_path, "r") as z:

astrbot/core/zip_updator.py

@@ -54,6 +54,55 @@ def _truncate_response_body(body: str, max_len: int = 1000) -> str:
             return body
         return body[:max_len] + "...[truncated]"
 
+    async def fetch_github_default_branch(self, author: str, repo: str) -> str | None:
+        """Fetch the default branch for a GitHub repository.
+
+        Args:
+            author: GitHub repository owner.
+            repo: GitHub repository name.
+
+        Returns:
+            The default branch name, or None if it cannot be resolved.
+        """
+        url = f"https://api.github.com/repos/{author}/{repo}"
+        try:
+            async with self._create_httpx_client(timeout=10.0) as client:
+                response = await client.get(url)
+                response.raise_for_status()
+                repo_info = response.json()
+        except Exception as exc:
+            logger.debug("获取 GitHub 默认分支失败 %s/%s: %s", author, repo, exc)
+            return None
+
+        default_branch = str(repo_info.get("default_branch") or "").strip()
+        return default_branch or None
+
+    async def resolve_github_source_branch(
+        self,
+        repo_url: str,
+    ) -> tuple[str, str, str]:
+        """Resolve the GitHub branch used for repository source downloads.
+
+        Args:
+            repo_url: GitHub repository URL, optionally with a tree branch.
+
+        Returns:
+            Repository owner, name, and resolved source branch.
+
+        Raises:
+            ValueError: If the repository URL is invalid.
+        """
+        author, repo, branch = self.parse_github_url(repo_url)
+        if branch:
+            return author, repo, branch
+
+        default_branch = await self.fetch_github_default_branch(author, repo)
+        if default_branch:
+            return author, repo, default_branch
+
+        logger.info("未能获取 %s/%s 的默认分支，将尝试 main 分支", author, repo)
+        return author, repo, "main"
+
     async def _download_file(
         self,
         url: str,
@@ -225,32 +274,13 @@ async def check_update(
     async def download_from_repo_url(
         self, target_path: str, repo_url: str, proxy=""
     ) -> None:
-        author, repo, branch = self.parse_github_url(repo_url)
+        author, repo, branch = await self.resolve_github_source_branch(repo_url)
 
         logger.info(f"正在下载更新 {repo} ...")
-
-        if branch:
-            logger.info(f"正在从指定分支 {branch} 下载 {author}/{repo}")
-            release_url = (
-                f"https://github.com/{author}/{repo}/archive/refs/heads/{branch}.zip"
-            )
-        else:
-            try:
-                release_url = f"https://api.github.com/repos/{author}/{repo}/releases"
-                releases = await self.fetch_release_info(url=release_url)
-            except Exception as e:
-                logger.warning(
-                    f"获取 {author}/{repo} 的 GitHub Releases 失败: {e}，将尝试下载默认分支",
-                )
-                releases = []
-            if not releases:
-                # 如果没有最新版本，下载默认分支
-                logger.info(f"正在从默认分支下载 {author}/{repo}")
-                release_url = (
-                    f"https://github.com/{author}/{repo}/archive/refs/heads/master.zip"
-                )
-            else:
-                release_url = releases[0]["zipball_url"]
+        logger.info(f"正在从分支 {branch} 下载 {author}/{repo}")
+        release_url = (
+            f"https://github.com/{author}/{repo}/archive/refs/heads/{branch}.zip"
+        )
 
         if proxy:
             proxy = proxy.rstrip("/")