fix(plugin): 增加插件市场元数据边界 (#6495)

* fix(plugin): bound market metadata inputs

* test(architecture): sync plugin metadata baseline

* fix(plugin): narrow validated metadata identifiers

* test(architecture): sync coverage baseline

app/adapters/external/plugin/client.py

@@ -8,6 +8,7 @@ import re
 import threading
 import time
 import traceback
+from collections import Counter
 from pathlib import Path
 from typing import Any, Dict, Iterator, List, Optional, cast
 from urllib.parse import urlparse, urlsplit
@@ -22,6 +23,7 @@ from app.domain.plugin import (
     build_local_plugin_source,
     check_plugin_system_version,
     is_local_plugin_source,
+    is_physical_plugin_id,
     is_plugin_generation_compatible,
     parse_local_plugin_generation,
     parse_local_plugin_path,
@@ -42,6 +44,28 @@ PluginReleaseList = list[PluginPayload]
 PluginRequestOptions = dict[str, Any]
 PluginReleaseTask = asyncio.Task[Optional[PluginReleaseList]]
 
+PLUGIN_INDEX_MAX_BYTES = 1024 * 1024
+PLUGIN_INDEX_MAX_ENTRIES = 4096
+PLUGIN_INDEX_MAX_HISTORY_ENTRIES = 512
+PLUGIN_INDEX_MAX_NESTING = 64
+PLUGIN_INDEX_READ_CHUNK_SIZE = 64 * 1024
+PLUGIN_INDEX_COMPATIBILITY_FLAG_PATTERN = re.compile(r"^v\d+t?$")
+PLUGIN_INDEX_TEXT_LIMITS = {
+    "name": 256,
+    "author": 256,
+    "description": 4096,
+    "icon": 2048,
+    "author_url": 2048,
+    "authorUrl": 2048,
+    "project_url": 2048,
+    "homepage": 2048,
+    "repo_url": 2048,
+}
+
+
+class _PluginIndexTooLargeError(RuntimeError):
+    """插件索引的声明长度或实际读取字节超过资源边界。"""
+
 
 def build_local_repo_url(
     plugin_id: str,
@@ -418,9 +442,9 @@ class PluginMarketTransport:
             paths.append(path.resolve())
         return paths
 
-    @staticmethod
+    @classmethod
     def __get_local_package(
-        repo_path: Path, package_version: Optional[str] = None
+        cls, repo_path: Path, package_version: Optional[str] = None
     ) -> Optional[PluginIndex]:
         """
         从本地插件仓库读取 package.json 或 package.{version}.json
@@ -431,14 +455,22 @@ class PluginMarketTransport:
         if not package_file.exists():
             return {}
         try:
-            content = package_file.read_text(encoding="utf-8", errors="replace")
-            payload = json.loads(content)
-        except Exception as e:
-            logger.warn(f"读取本地插件包 {package_file} 失败：{e}")
-            return None
-        if not isinstance(payload, dict):
-            logger.warn(f"本地插件包 {package_file} 格式不正确")
-            return None
+            if package_file.stat().st_size > PLUGIN_INDEX_MAX_BYTES:
+                raise _PluginIndexTooLargeError(
+                    f"插件索引超过 {PLUGIN_INDEX_MAX_BYTES} 字节"
+                )
+            with package_file.open("rb") as file_handle:
+                raw_content = file_handle.read(PLUGIN_INDEX_MAX_BYTES + 1)
+            if len(raw_content) > PLUGIN_INDEX_MAX_BYTES:
+                raise _PluginIndexTooLargeError(
+                    f"插件索引超过 {PLUGIN_INDEX_MAX_BYTES} 字节"
+                )
+            content = raw_content.decode("utf-8", errors="replace")
+        except OSError as error:
+            raise RuntimeError(f"读取本地插件包 {package_file} 失败：{error}") from error
+        payload = cls.__parse_plugin_index_response(content)
+        if payload is None:
+            raise RuntimeError(f"本地插件包 {package_file} 格式无效")
         return payload
 
     @staticmethod
@@ -610,22 +642,236 @@ class PluginMarketTransport:
         return parts._replace(query=query).geturl()
 
     @staticmethod
-    def __parse_plugin_index_response(content: str) -> Optional[PluginIndex]:
-        """
-        解析插件索引响应，仅缓存成功解析出的字典结果。
-        """
+    def __safe_plugin_id(plugin_id: object) -> str:
+        """生成适合日志展示的有界插件 ID，不回显完整外部载荷。"""
+        if not isinstance(plugin_id, str):
+            return f"<{type(plugin_id).__name__}>"
+        return json.dumps(plugin_id[:64], ensure_ascii=True)
+
+    @staticmethod
+    def __plugin_index_nesting_is_valid(payload: object) -> bool:
+        """限制外部 JSON 的容器深度，避免后续序列化递归耗尽调用栈。"""
+        pending = [(payload, 0)]
+        while pending:
+            value, depth = pending.pop()
+            if depth > PLUGIN_INDEX_MAX_NESTING:
+                return False
+            if isinstance(value, dict):
+                pending.extend((item, depth + 1) for item in value.values())
+            elif isinstance(value, list):
+                pending.extend((item, depth + 1) for item in value)
+        return True
+
+    @classmethod
+    def __normalize_plugin_index_entry(
+        cls,
+        plugin_id: object,
+        plugin_info: object,
+    ) -> tuple[Optional[PluginPayload], Optional[str], list[str]]:
+        """校验影响安装决策的字段，并宽容丢弃异常展示字段。"""
+        if not isinstance(plugin_id, str) or not is_physical_plugin_id(plugin_id):
+            return None, "插件 ID 非法", []
+        if not isinstance(plugin_info, dict):
+            return None, "条目不是对象", []
+
+        declared_id = plugin_info.get("id")
+        if "id" in plugin_info and (
+            not isinstance(declared_id, str)
+            or not is_physical_plugin_id(declared_id)
+            or declared_id.lower() != plugin_id.lower()
+        ):
+            return None, "id 与索引键不一致", []
+
+        version = plugin_info.get("version")
+        if not isinstance(version, str) or not version or len(version) > 64:
+            return None, "version 非法", []
+
+        system_version = plugin_info.get("system_version")
+        if "system_version" in plugin_info and (
+            not isinstance(system_version, str) or len(system_version) > 256
+        ):
+            return None, "system_version 非法", []
+
+        level = plugin_info.get("level")
+        if "level" in plugin_info and (
+            isinstance(level, bool)
+            or not isinstance(level, int)
+            or not 0 <= level <= 99
+        ):
+            return None, "level 非法", []
+
+        public_key = plugin_info.get("key")
+        if "key" in plugin_info and (
+            not isinstance(public_key, str) or len(public_key) > 16 * 1024
+        ):
+            return None, "key 非法", []
+
+        for field, value in plugin_info.items():
+            if field == "release" and not isinstance(value, bool):
+                return None, "release 非法", []
+            if (
+                PLUGIN_INDEX_COMPATIBILITY_FLAG_PATTERN.fullmatch(field)
+                and not isinstance(value, bool)
+            ):
+                return None, "兼容标志非法", []
+
+        normalized = plugin_info.copy()
+        if declared_id is not None:
+            normalized["id"] = plugin_id
+        dropped_fields: list[str] = []
+        for field, max_length in PLUGIN_INDEX_TEXT_LIMITS.items():
+            if field not in normalized:
+                continue
+            value = normalized[field]
+            if not isinstance(value, str) or len(value) > max_length:
+                normalized.pop(field)
+                dropped_fields.append(field)
+
+        labels = normalized.get("labels")
+        if labels is not None:
+            valid_labels = (
+                isinstance(labels, str)
+                and len(labels) <= 1024
+            ) or (
+                isinstance(labels, list)
+                and len(labels) <= 64
+                and all(
+                    isinstance(item, str) and len(item) <= 128
+                    for item in labels
+                )
+            )
+            if not valid_labels:
+                normalized.pop("labels")
+                dropped_fields.append("labels")
+
+        history = normalized.get("history")
+        if history is not None and (
+            not isinstance(history, dict)
+            or len(history) > PLUGIN_INDEX_MAX_HISTORY_ENTRIES
+        ):
+            normalized.pop("history")
+            dropped_fields.append("history")
+
+        return normalized, None, dropped_fields
+
+    @classmethod
+    def __parse_plugin_index_response(cls, content: str) -> Optional[PluginIndex]:
+        """解析并规范化插件索引，仅缓存满足索引级边界的结果。"""
         try:
             payload = json.loads(content)
-        except json.JSONDecodeError:
-            if "404: Not Found" not in content:
-                logger.warn(f"插件包数据解析失败：{content}")
+        except (ValueError, RecursionError):
+            logger.warning("插件包数据解析失败：响应不是有效 JSON")
             return None
 
         if not isinstance(payload, dict):
-            logger.warn(f"插件包数据格式不正确，期望 dict，实际为 {type(payload).__name__}")
+            logger.warning(
+                f"插件包数据格式不正确，期望 dict，实际为 {type(payload).__name__}"
+            )
+            return None
+        if not cls.__plugin_index_nesting_is_valid(payload):
+            logger.warning(
+                f"插件包 JSON 嵌套超过上限：{PLUGIN_INDEX_MAX_NESTING} 层"
+            )
+            return None
+        if len(payload) > PLUGIN_INDEX_MAX_ENTRIES:
+            logger.warning(
+                f"插件包条目超过上限：{len(payload)} > {PLUGIN_INDEX_MAX_ENTRIES}"
+            )
             return None
 
-        return payload
+        normalized: PluginIndex = {}
+        skipped_reasons: Counter[str] = Counter()
+        dropped_fields: Counter[str] = Counter()
+        skipped_examples: list[str] = []
+        for plugin_id, plugin_info in payload.items():
+            item, reason, dropped = cls.__normalize_plugin_index_entry(
+                plugin_id,
+                plugin_info,
+            )
+            if item is None:
+                skipped_reasons[reason or "未知原因"] += 1
+                if len(skipped_examples) < 5:
+                    skipped_examples.append(cls.__safe_plugin_id(plugin_id))
+                continue
+            normalized[plugin_id] = item
+            dropped_fields.update(dropped)
+
+        summaries = []
+        if skipped_reasons:
+            reasons = "、".join(
+                f"{reason} {count} 条"
+                for reason, count in sorted(skipped_reasons.items())
+            )
+            examples = "、".join(skipped_examples)
+            summaries.append(
+                f"跳过 {sum(skipped_reasons.values())} 个条目（{reasons}；示例 {examples}）"
+            )
+        if dropped_fields:
+            fields = "、".join(
+                f"{field} {count} 次"
+                for field, count in sorted(dropped_fields.items())
+            )
+            summaries.append(f"丢弃异常展示字段（{fields}）")
+        if summaries:
+            logger.warning(f"插件包数据已隔离异常内容：{'；'.join(summaries)}")
+
+        return normalized
+
+    @staticmethod
+    def __declared_plugin_index_too_large(headers: Any) -> bool:
+        """仅把合法的 Content-Length 用作流式读取前的快速拒绝依据。"""
+        try:
+            value = headers.get("Content-Length")
+            return value is not None and int(value) > PLUGIN_INDEX_MAX_BYTES
+        except (AttributeError, TypeError, ValueError):
+            return False
+
+    @classmethod
+    def __read_plugin_index_response(cls, response: Response) -> tuple[int, str]:
+        """从同步流式响应读取有界的解压后索引文本。"""
+        status_code = response.status_code
+        if status_code != 200:
+            return status_code, ""
+        if cls.__declared_plugin_index_too_large(response.headers):
+            raise _PluginIndexTooLargeError(
+                f"插件索引响应超过 {PLUGIN_INDEX_MAX_BYTES} 字节"
+            )
+        content = bytearray()
+        for chunk in response.iter_content(chunk_size=PLUGIN_INDEX_READ_CHUNK_SIZE):
+            if not chunk:
+                continue
+            content.extend(chunk)
+            if len(content) > PLUGIN_INDEX_MAX_BYTES:
+                raise _PluginIndexTooLargeError(
+                    f"插件索引响应超过 {PLUGIN_INDEX_MAX_BYTES} 字节"
+                )
+        return status_code, content.decode("utf-8", errors="replace")
+
+    @classmethod
+    async def __async_read_plugin_index_response(
+        cls,
+        response: httpx2.Response,
+    ) -> tuple[int, str]:
+        """从异步流式响应读取有界的解压后索引文本。"""
+        status_code = response.status_code
+        if status_code != 200:
+            return status_code, ""
+        if cls.__declared_plugin_index_too_large(response.headers):
+            raise _PluginIndexTooLargeError(
+                f"插件索引响应超过 {PLUGIN_INDEX_MAX_BYTES} 字节"
+            )
+        content = bytearray()
+        async for chunk in response.aiter_bytes(
+            chunk_size=PLUGIN_INDEX_READ_CHUNK_SIZE
+        ):
+            if not chunk:
+                continue
+            content.extend(chunk)
+            if len(content) > PLUGIN_INDEX_MAX_BYTES:
+                raise _PluginIndexTooLargeError(
+                    f"插件索引响应超过 {PLUGIN_INDEX_MAX_BYTES} 字节"
+                )
+        return status_code, content.decode("utf-8", errors="replace")
 
     @classmethod
     def _build_plugin_index_request(
@@ -651,6 +897,78 @@ class PluginMarketTransport:
         headers = get_runtime_setting('REPO_GITHUB_HEADERS')(repo=f"{user}/{repo}")
         return package_url, headers
 
+    @classmethod
+    def __request_plugin_index_with_fallback(
+        cls,
+        url: str,
+        headers: Optional[dict[str, str]] = None,
+        timeout: Optional[int] = 60,
+    ) -> Optional[tuple[int, str]]:
+        """按 GitHub 降级顺序流式读取同步插件索引，并限制解压后字节数。"""
+        strategies = cls._build_github_request_strategies(
+            url=url,
+            headers=headers,
+            timeout=timeout,
+        )
+        for strategy_name, target_url, request_params in strategies:
+            logger.debug(
+                f"[GitHub] 尝试使用策略：{strategy_name} 请求插件索引：{target_url}"
+            )
+            try:
+                request = RequestUtils(**request_params)
+                with request.get_stream(
+                    url=target_url,
+                    raise_exception=True,
+                ) as response:
+                    if response is None:
+                        continue
+                    return cls.__read_plugin_index_response(response)
+            except _PluginIndexTooLargeError:
+                raise
+            except Exception as error:  # noqa: BLE001 - 失败后尝试下一传输策略
+                logger.error(
+                    f"[GitHub] 插件索引请求失败，策略：{strategy_name}，"
+                    f"URL：{target_url}，错误：{error}"
+                )
+        logger.error(f"[GitHub] 所有策略均无法读取插件索引，URL：{url}")
+        return None
+
+    @classmethod
+    async def __async_request_plugin_index_with_fallback(
+        cls,
+        url: str,
+        headers: Optional[dict[str, str]] = None,
+        timeout: Optional[int] = 60,
+    ) -> Optional[tuple[int, str]]:
+        """按 GitHub 降级顺序流式读取异步插件索引，并限制解压后字节数。"""
+        strategies = cls._build_github_request_strategies(
+            url=url,
+            headers=headers,
+            timeout=timeout,
+        )
+        for strategy_name, target_url, request_params in strategies:
+            logger.debug(
+                f"[GitHub] 尝试使用策略：{strategy_name} 请求插件索引：{target_url}"
+            )
+            try:
+                request = AsyncRequestUtils(**request_params)
+                async with request.get_stream(
+                    url=target_url,
+                    raise_exception=True,
+                ) as response:
+                    if response is None:
+                        continue
+                    return await cls.__async_read_plugin_index_response(response)
+            except _PluginIndexTooLargeError:
+                raise
+            except Exception as error:  # noqa: BLE001 - 失败后尝试下一传输策略
+                logger.error(
+                    f"[GitHub] 插件索引请求失败，策略：{strategy_name}，"
+                    f"URL：{target_url}，错误：{error}"
+                )
+        logger.error(f"[GitHub] 所有策略均无法读取插件索引，URL：{url}")
+        return None
+
     @classmethod
     def _resolve_plugin_index_response(
             cls,
@@ -798,14 +1116,18 @@ class PluginMarketTransport:
         if request is None:
             raise ValueError("插件仓库地址无效")
         package_url, headers = request
-        res = self.__request_with_fallback(package_url, headers=headers)
-        if res is None:
+        response = self.__request_plugin_index_with_fallback(
+            package_url,
+            headers=headers,
+        )
+        if response is None:
             raise RuntimeError("插件索引请求失败：连接失败")
-        if res.status_code == 404:
+        status_code, content = response
+        if status_code == 404:
             return None
-        if res.status_code != 200:
-            raise RuntimeError(f"插件索引请求失败：HTTP {res.status_code}")
-        payload = self.__parse_plugin_index_response(res.text)
+        if status_code != 200:
+            raise RuntimeError(f"插件索引请求失败：HTTP {status_code}")
+        payload = self.__parse_plugin_index_response(content)
         if payload is None:
             raise RuntimeError("插件索引响应格式无效")
         return payload
@@ -1133,17 +1455,18 @@ class PluginMarketTransport:
         if request is None:
             raise ValueError("插件仓库地址无效")
         package_url, headers = request
-        res = await self.__async_request_with_fallback(
+        response = await self.__async_request_plugin_index_with_fallback(
             package_url,
             headers=headers,
         )
-        if res is None:
+        if response is None:
             raise RuntimeError("插件索引请求失败：连接失败")
-        if res.status_code == 404:
+        status_code, content = response
+        if status_code == 404:
             return None
-        if res.status_code != 200:
-            raise RuntimeError(f"插件索引请求失败：HTTP {res.status_code}")
-        payload = self.__parse_plugin_index_response(res.text)
+        if status_code != 200:
+            raise RuntimeError(f"插件索引请求失败：HTTP {status_code}")
+        payload = self.__parse_plugin_index_response(content)
         if payload is None:
             raise RuntimeError("插件索引响应格式无效")
         return payload

app/application/plugin/identity.py

@@ -10,8 +10,8 @@ from enum import StrEnum
 from typing import Protocol
 
 from app.application.plugin.declaration import PluginDeclaredMetadata
+from app.domain.plugin import is_physical_plugin_id
 
-_PLUGIN_ID_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,127}$")
 _ONLINE_SOURCE_KEY_PATTERN = re.compile(
     r"^github:[a-z0-9](?:[a-z0-9-]{0,38})/"
     r"[a-z0-9._-]{1,100}$"
@@ -60,7 +60,7 @@ class PluginIdentityConflictError(RuntimeError):
 
 def normalize_physical_plugin_id(plugin_id: str) -> str:
     """校验可安全持久化的物理插件 ID，并返回大小写无关身份键。"""
-    if plugin_id != plugin_id.strip() or not _PLUGIN_ID_PATTERN.fullmatch(plugin_id):
+    if not is_physical_plugin_id(plugin_id):
         raise ValueError("插件 ID 必须以字母开头且只能包含 ASCII 字母、数字或下划线")
     return plugin_id.lower()
 

app/domain/plugin.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from typing import Any
@@ -12,6 +13,7 @@ from packaging.version import InvalidVersion, Version
 
 LOCAL_PLUGIN_SOURCE_PREFIX = "local://"
 PLUGIN_SYSTEM_VERSION_FIELD = "system_version"
+_PHYSICAL_PLUGIN_ID_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,127}$")
 PLUGIN_GENERATION_COMPATIBILITY: dict[str, tuple[str, ...]] = {
     "v3": ("v2",),
 }
@@ -25,6 +27,15 @@ class PluginReleaseInstallPlan:
     fallback_to_filelist: bool
 
 
+def is_physical_plugin_id(plugin_id: object) -> bool:
+    """判断值是否符合物理插件在市场、目录和来源身份中共用的 ID 规则。"""
+    return bool(
+        isinstance(plugin_id, str)
+        and plugin_id == plugin_id.strip()
+        and _PHYSICAL_PLUGIN_ID_PATTERN.fullmatch(plugin_id)
+    )
+
+
 def is_local_plugin_source(repo_url: str | None) -> bool:
     """判断插件来源是否为宿主定义的不透明本地仓库标识。"""
     return bool(repo_url and repo_url.startswith(LOCAL_PLUGIN_SOURCE_PREFIX))

app/runtime/extensions/plugin/catalog.py

@@ -145,7 +145,11 @@ class PluginCatalogFacade:
     def local_repository(self) -> list[Plugin]:
         """读取本地插件仓候选并映射为目录 DTO。"""
         installed = self._storage().read(SystemConfigKey.UserInstalledPlugins) or []
-        candidates = self._system().local_candidates()
+        try:
+            candidates = self._system().local_candidates()
+        except Exception as error:  # noqa: BLE001 - 展示失败不能阻断整个插件目录
+            self._logger.warning(f"读取本地插件仓候选失败，已跳过本地目录展示：{error}")
+            return []
         plugins: list[Plugin] = []
         for plugin_id, info in candidates.items():
             package_version = info.get("package_version")

docs/architecture-optimization-checklist.md

@@ -70,7 +70,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 959 / 8,149 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 959 / 8,151 | `dependency-baseline.json` 当前快照 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
@@ -80,7 +80,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 | 长方法 | 299 个超过 80 行 | 70 个超过 150 行，23 个超过 250 行；大量是私有方法 |
 | 全量 mypy 历史债务 | 10,415 / 593 文件 | strict frontier 当前覆盖 41 个文件，Plugin、LLM Provider 与 Agent Manager owner 拆分及受益调用方低水位已固化 |
 | Ruff 历史诊断 | 651 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
-| 覆盖率低水位 | Application 80.92%，Domain 79.87% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
+| 覆盖率低水位 | Application 80.92%，Domain 79.89% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件
 

docs/architecture-overview.md

@@ -742,7 +742,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 959 |
-| 内部导入边 | 8,149 |
+| 内部导入边 | 8,151 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 53（债务已清零，53 条精确 containment） |

tests/fixtures/architecture/coverage-baseline.json

@@ -5,8 +5,8 @@
     "statements": 15397
   },
   "domain": {
-    "covered_lines": 3511,
-    "percent": 79.87,
-    "statements": 4396
+    "covered_lines": 3515,
+    "percent": 79.89,
+    "statements": 4400
   }
 }

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8149,
-  "edge_sha256": "750a3668e8c7e795f7d7f48958a6e539ea2184980b813010a08ebb8103153d3a",
+  "edge_count": 8151,
+  "edge_sha256": "8dde7ef068410659e55be5ebb0548b32d9fa1cef3cb888182b5aab6386e8b343",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -3984,6 +3984,8 @@
     "app.application.plugin.identity -> app.application",
     "app.application.plugin.identity -> app.application.plugin",
     "app.application.plugin.identity -> app.application.plugin.declaration",
+    "app.application.plugin.identity -> app.domain",
+    "app.application.plugin.identity -> app.domain.plugin",
     "app.application.plugin.install -> app.application",
     "app.application.plugin.install -> app.application.plugin",
     "app.application.plugin.install -> app.application.plugin.admission",

tests/test_plugin_catalog_runtime.py

@@ -57,3 +57,39 @@ def test_installed_catalog_keeps_plugins_that_are_not_loaded():
     ]
     assert plugins[1].plugin_name == "DependencyPending"
     assert plugins[2].installed is True
+
+
+def test_local_repository_failure_does_not_break_catalog_projection():
+    """本地仓索引读取失败只影响本地候选展示，不能拖垮整个插件目录。"""
+    warnings: list[str] = []
+
+    def local_candidates():
+        raise RuntimeError("invalid local package")
+
+    facade = PluginCatalogFacade(
+        classes=lambda: {},
+        running=lambda: {},
+        storage=lambda: SimpleNamespace(read=lambda _key: []),
+        system=lambda: SimpleNamespace(
+            local_candidates=local_candidates,
+        ),
+        market_catalog=lambda: None,
+        market_loader=lambda *_args, **_kwargs: [],
+        async_market_loader=lambda *_args, **_kwargs: [],
+        map_plugin=lambda **_kwargs: None,
+        auth_checker=lambda **_kwargs: True,
+        plugin_attr=lambda _plugin_id, _attr: None,
+        plugin_instance=lambda _plugin_id: None,
+        plugin_instances=lambda: {},
+        runtime_status=lambda _plugin_id: None,
+        log=SimpleNamespace(
+            error=lambda *_args: None,
+            info=lambda *_args: None,
+            warning=warnings.append,
+        ),
+    )
+
+    assert facade.local_repository() == []
+    assert warnings == [
+        "读取本地插件仓候选失败，已跳过本地目录展示：invalid local package"
+    ]

tests/test_plugin_market_index_policy.py

@@ -1,10 +1,106 @@
 """插件市场索引同步与异步策略一致性测试。"""
 
-from types import SimpleNamespace
+import json
+from contextlib import asynccontextmanager, contextmanager
 
 import pytest
 
-from app.adapters.external.plugin.client import PluginMarketClient, PluginMarketTransport
+from app.adapters.external.plugin.client import (
+    PLUGIN_INDEX_MAX_BYTES,
+    PLUGIN_INDEX_MAX_ENTRIES,
+    PluginMarketClient,
+    PluginMarketTransport,
+)
+
+SYNC_INDEX_REQUEST = "_PluginMarketTransport__request_plugin_index_with_fallback"
+ASYNC_INDEX_REQUEST = "_PluginMarketTransport__async_request_plugin_index_with_fallback"
+
+
+class _SyncStreamResponse:
+    """提供 requests 流式响应所需的最小测试契约。"""
+
+    def __init__(self, content: bytes, content_length: str | None = None) -> None:
+        self.status_code = 200
+        self.headers = {}
+        if content_length is not None:
+            self.headers["Content-Length"] = content_length
+        self._content = content
+
+    def iter_content(self, chunk_size: int):
+        for offset in range(0, len(self._content), chunk_size):
+            yield self._content[offset:offset + chunk_size]
+
+
+class _AsyncStreamResponse:
+    """提供 httpx 流式响应所需的最小测试契约。"""
+
+    def __init__(self, content: bytes, content_length: str | None = None) -> None:
+        self.status_code = 200
+        self.headers = {}
+        if content_length is not None:
+            self.headers["Content-Length"] = content_length
+        self._content = content
+
+    async def aiter_bytes(self, chunk_size: int):
+        for offset in range(0, len(self._content), chunk_size):
+            yield self._content[offset:offset + chunk_size]
+
+
+def test_plugin_index_request_uses_sync_stream_transport(monkeypatch) -> None:
+    """同步索引专用请求必须进入流式传输，不能预先加载完整响应体。"""
+    calls: list[tuple[dict, str, bool]] = []
+
+    class FakeRequestUtils:
+        def __init__(self, **kwargs) -> None:
+            self._kwargs = kwargs
+
+        @contextmanager
+        def get_stream(self, url: str, raise_exception: bool):
+            calls.append((self._kwargs, url, raise_exception))
+            yield _SyncStreamResponse(b'{"DemoPlugin":{"version":"1.0.0"}}')
+
+    monkeypatch.setattr(
+        "app.adapters.external.plugin.client.RequestUtils",
+        FakeRequestUtils,
+    )
+
+    result = PluginMarketTransport._PluginMarketTransport__request_plugin_index_with_fallback(
+        "https://raw.githubusercontent.com/example/repo/main/package.v3.json",
+        headers={"Accept": "application/json"},
+    )
+
+    assert result == (200, '{"DemoPlugin":{"version":"1.0.0"}}')
+    assert calls[0][0]["headers"] == {"Accept": "application/json"}
+    assert calls[0][2] is True
+
+
+@pytest.mark.asyncio
+async def test_plugin_index_request_uses_async_stream_transport(monkeypatch) -> None:
+    """异步索引专用请求与同步入口使用相同的流式资源边界。"""
+    calls: list[tuple[dict, str, bool]] = []
+
+    class FakeAsyncRequestUtils:
+        def __init__(self, **kwargs) -> None:
+            self._kwargs = kwargs
+
+        @asynccontextmanager
+        async def get_stream(self, url: str, raise_exception: bool):
+            calls.append((self._kwargs, url, raise_exception))
+            yield _AsyncStreamResponse(b'{"DemoPlugin":{"version":"1.0.0"}}')
+
+    monkeypatch.setattr(
+        "app.adapters.external.plugin.client.AsyncRequestUtils",
+        FakeAsyncRequestUtils,
+    )
+
+    result = await PluginMarketTransport._PluginMarketTransport__async_request_plugin_index_with_fallback(
+        "https://raw.githubusercontent.com/example/repo/main/package.v3.json",
+        headers={"Accept": "application/json"},
+    )
+
+    assert result == (200, '{"DemoPlugin":{"version":"1.0.0"}}')
+    assert calls[0][0]["headers"] == {"Accept": "application/json"}
+    assert calls[0][2] is True
 
 
 @pytest.mark.asyncio
@@ -14,10 +110,7 @@ async def test_sync_and_async_plugin_indexes_share_request_and_result(
     """同步与异步索引读取必须使用同一请求计划并返回相同三态结果。"""
     helper = PluginMarketTransport()
     repo_url = "https://github.com/policy-owner/policy-repository"
-    response = SimpleNamespace(
-        status_code=200,
-        text='{"DemoPlugin": {"version": "1.2.3"}}',
-    )
+    response = (200, '{"DemoPlugin": {"version": "1.2.3"}}')
     sync_requests: list[tuple[str, dict]] = []
     async_requests: list[tuple[str, dict]] = []
 
@@ -33,12 +126,12 @@ async def test_sync_and_async_plugin_indexes_share_request_and_result(
 
     monkeypatch.setattr(
         helper,
-        "_PluginMarketTransport__request_with_fallback",
+        SYNC_INDEX_REQUEST,
         sync_request,
     )
     monkeypatch.setattr(
         helper,
-        "_PluginMarketTransport__async_request_with_fallback",
+        ASYNC_INDEX_REQUEST,
         async_request,
     )
 
@@ -69,14 +162,11 @@ async def test_market_read_warms_source_inventory_cache(monkeypatch) -> None:
     async def request(_url: str, *, headers: dict):
         nonlocal requests
         requests += 1
-        return SimpleNamespace(
-            status_code=200,
-            text='{"DemoPlugin": {"version": "1.2.3"}}',
-        )
+        return 200, '{"DemoPlugin": {"version": "1.2.3"}}'
 
     monkeypatch.setattr(
         helper,
-        "_PluginMarketTransport__async_request_with_fallback",
+        ASYNC_INDEX_REQUEST,
         request,
     )
     await helper.async_get_plugin_index_result.cache_clear()
@@ -101,14 +191,11 @@ def test_sync_force_refresh_bypasses_index_cache(
     def request(_url: str, *, headers: dict):
         nonlocal requests
         requests += 1
-        return SimpleNamespace(
-            status_code=200,
-            text=f'{{"DemoPlugin": {{"version": "{version}"}}}}',
-        )
+        return 200, f'{{"DemoPlugin": {{"version": "{version}"}}}}'
 
     monkeypatch.setattr(
         helper,
-        "_PluginMarketTransport__request_with_fallback",
+        SYNC_INDEX_REQUEST,
         request,
     )
     helper.get_plugin_index_result.cache_clear()
@@ -142,14 +229,11 @@ async def test_async_force_refresh_bypasses_index_cache(
     async def request(_url: str, *, headers: dict):
         nonlocal requests
         requests += 1
-        return SimpleNamespace(
-            status_code=200,
-            text=f'{{"DemoPlugin": {{"version": "{version}"}}}}',
-        )
+        return 200, f'{{"DemoPlugin": {{"version": "{version}"}}}}'
 
     monkeypatch.setattr(
         helper,
-        "_PluginMarketTransport__async_request_with_fallback",
+        ASYNC_INDEX_REQUEST,
         request,
     )
     await helper.async_get_plugin_index_result.cache_clear()
@@ -179,11 +263,11 @@ async def test_absent_plugin_generation_is_cached(monkeypatch) -> None:
     async def request(_url: str, *, headers: dict):
         nonlocal requests
         requests += 1
-        return SimpleNamespace(status_code=404, text="404: Not Found")
+        return 404, ""
 
     monkeypatch.setattr(
         helper,
-        "_PluginMarketTransport__async_request_with_fallback",
+        ASYNC_INDEX_REQUEST,
         request,
     )
     await helper.async_get_plugin_index_result.cache_clear()
@@ -206,11 +290,11 @@ async def test_index_cache_retains_multi_market_generation_working_set(
     async def request(_url: str, *, headers: dict):
         nonlocal requests
         requests += 1
-        return SimpleNamespace(status_code=200, text='{"DemoPlugin": {}}')
+        return 200, '{"DemoPlugin": {"version": "1.0.0"}}'
 
     monkeypatch.setattr(
         helper,
-        "_PluginMarketTransport__async_request_with_fallback",
+        ASYNC_INDEX_REQUEST,
         request,
     )
     await helper.async_get_plugin_index_result.cache_clear()
@@ -227,12 +311,204 @@ async def test_index_cache_retains_multi_market_generation_working_set(
     assert requests == len(targets)
 
 
+@pytest.mark.parametrize("content_length", [None, str(PLUGIN_INDEX_MAX_BYTES)])
+def test_sync_plugin_index_stream_accepts_exact_byte_limit(content_length) -> None:
+    """同步索引以解压后实际字节计数，恰好达到上限仍可读取。"""
+    content = b"x" * PLUGIN_INDEX_MAX_BYTES
+    response = _SyncStreamResponse(content, content_length)
+
+    status_code, text = PluginMarketTransport._PluginMarketTransport__read_plugin_index_response(
+        response
+    )
+
+    assert status_code == 200
+    assert len(text) == PLUGIN_INDEX_MAX_BYTES
+
+
+@pytest.mark.parametrize("content_length", [None, "invalid", str(PLUGIN_INDEX_MAX_BYTES)])
+def test_sync_plugin_index_stream_rejects_actual_byte_overflow(content_length) -> None:
+    """缺失、伪造或偏小的声明长度都不能绕过同步实际读取上限。"""
+    response = _SyncStreamResponse(
+        b"x" * (PLUGIN_INDEX_MAX_BYTES + 1),
+        content_length,
+    )
+
+    with pytest.raises(RuntimeError, match="插件索引响应超过"):
+        PluginMarketTransport._PluginMarketTransport__read_plugin_index_response(
+            response
+        )
+
+
+def test_sync_plugin_index_stream_rejects_declared_byte_overflow() -> None:
+    """可信声明已超限时无需读取响应体。"""
+    response = _SyncStreamResponse(
+        b"",
+        str(PLUGIN_INDEX_MAX_BYTES + 1),
+    )
+
+    with pytest.raises(RuntimeError, match="插件索引响应超过"):
+        PluginMarketTransport._PluginMarketTransport__read_plugin_index_response(
+            response
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("content_length", [None, str(PLUGIN_INDEX_MAX_BYTES)])
+async def test_async_plugin_index_stream_accepts_exact_byte_limit(content_length) -> None:
+    """异步索引与同步入口共享恰好达到上限时仍成功的合同。"""
+    response = _AsyncStreamResponse(
+        b"x" * PLUGIN_INDEX_MAX_BYTES,
+        content_length,
+    )
+
+    status_code, text = await PluginMarketTransport._PluginMarketTransport__async_read_plugin_index_response(
+        response
+    )
+
+    assert status_code == 200
+    assert len(text) == PLUGIN_INDEX_MAX_BYTES
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("content_length", [None, "invalid", str(PLUGIN_INDEX_MAX_BYTES)])
+async def test_async_plugin_index_stream_rejects_actual_byte_overflow(content_length) -> None:
+    """异步流式读取同样不能被缺失或不可信的声明长度绕过。"""
+    response = _AsyncStreamResponse(
+        b"x" * (PLUGIN_INDEX_MAX_BYTES + 1),
+        content_length,
+    )
+
+    with pytest.raises(RuntimeError, match="插件索引响应超过"):
+        await PluginMarketTransport._PluginMarketTransport__async_read_plugin_index_response(
+            response
+        )
+
+
+@pytest.mark.parametrize(
+    ("entry_count", "accepted"),
+    [
+        (PLUGIN_INDEX_MAX_ENTRIES, True),
+        (PLUGIN_INDEX_MAX_ENTRIES + 1, False),
+    ],
+)
+def test_plugin_index_entry_count_boundary(entry_count: int, accepted: bool) -> None:
+    """索引条目总数采用含上限边界，超出后整仓读取失败。"""
+    content = json.dumps(
+        {
+            f"Plugin{index}": {"version": "1.0.0"}
+            for index in range(entry_count)
+        }
+    )
+
+    result = PluginMarketTransport()._resolve_plugin_index_response(200, content)
+
+    assert (result is not None) is accepted
+    if result is not None:
+        assert len(result) == PLUGIN_INDEX_MAX_ENTRIES
+
+
+def test_plugin_index_isolates_bad_entries_and_preserves_unknown_fields() -> None:
+    """决策字段非法只隔离该条目，展示字段降级且未知字段保持前向兼容。"""
+    history = {str(index): "change" for index in range(513)}
+    content = json.dumps(
+        {
+            "GoodPlugin": {
+                "version": "1.0.0",
+                "release": True,
+                "future_contract": {"enabled": True},
+            },
+            "BadRelease": {"version": "1.0.0", "release": "false"},
+            "NullRelease": {"version": "1.0.0", "release": None},
+            "BadLevel": {"version": "1.0.0", "level": "99"},
+            "Bad-ID": {"version": "1.0.0"},
+            "MismatchedInnerId": {
+                "id": "OtherPlugin",
+                "version": "1.0.0",
+            },
+            "InvalidInnerId": {"id": "Bad-ID", "version": "1.0.0"},
+            "CanonicalInnerId": {
+                "id": "canonicalinnerid",
+                "version": "1.0.0",
+            },
+            "NotObject": [],
+            "DisplayFallback": {
+                "version": "1.0.0",
+                "description": "x" * 4097,
+                "labels": ["x"] * 65,
+                "history": history,
+            },
+        }
+    )
+
+    result = PluginMarketTransport()._resolve_plugin_index_response(200, content)
+
+    assert set(result or {}) == {
+        "GoodPlugin",
+        "CanonicalInnerId",
+        "DisplayFallback",
+    }
+    assert result["GoodPlugin"]["future_contract"] == {"enabled": True}
+    assert result["CanonicalInnerId"]["id"] == "CanonicalInnerId"
+    assert set(result["DisplayFallback"]) == {"version"}
+
+
+def test_plugin_index_parse_error_does_not_log_response_payload(monkeypatch) -> None:
+    """解析失败日志不得回显可能包含凭据或超长内容的原始响应。"""
+    messages: list[str] = []
+    monkeypatch.setattr(
+        "app.adapters.external.plugin.client.logger.warning",
+        lambda message: messages.append(message),
+    )
+
+    result = PluginMarketTransport()._resolve_plugin_index_response(
+        200,
+        'not-json-PRIVATE-PAYLOAD',
+    )
+
+    assert result is None
+    assert messages
+    assert all("PRIVATE-PAYLOAD" not in message for message in messages)
+
+
+def test_plugin_index_rejects_excessive_json_nesting() -> None:
+    """深度异常的 JSON 必须形成读取失败，不能把解析器异常泄漏到市场调用方。"""
+    content = '{"DemoPlugin":{"version":"1.0.0","future":' + "[" * 2000
+    content += "0" + "]" * 2000 + "}}"
+
+    result = PluginMarketTransport()._resolve_plugin_index_response(200, content)
+
+    assert result is None
+
+
+def test_local_plugin_index_rejects_oversize_and_invalid_files(tmp_path) -> None:
+    """已存在的本地索引超限或非法时必须失败，不能形成部分成功库存。"""
+    package_file = tmp_path / "package.v3.json"
+    package_file.write_bytes(b"x" * (PLUGIN_INDEX_MAX_BYTES + 1))
+
+    with pytest.raises(RuntimeError, match="插件索引超过"):
+        PluginMarketTransport._PluginMarketTransport__get_local_package(
+            tmp_path,
+            "v3",
+        )
+
+    package_file.write_text("not-json", encoding="utf-8")
+    with pytest.raises(RuntimeError, match="格式无效"):
+        PluginMarketTransport._PluginMarketTransport__get_local_package(
+            tmp_path,
+            "v3",
+        )
+
+
 @pytest.mark.parametrize(
     ("status_code", "content", "expected"),
     [
         (404, "404: Not Found", {}),
         (500, "upstream failed", None),
-        (200, '{"DemoPlugin": {}}', {"DemoPlugin": {}}),
+        (
+            200,
+            '{"DemoPlugin": {"version": "1.0.0"}}',
+            {"DemoPlugin": {"version": "1.0.0"}},
+        ),
         (200, "not-json", None),
         (200, "[]", None),
     ],
@@ -266,9 +542,9 @@ def test_plugin_index_result_preserves_read_state(
     repo_url = f"https://github.com/policy-owner/policy-repository-{status_code}"
 
     def request(_url: str, *, headers: dict):
-        return SimpleNamespace(status_code=status_code, text=content)
+        return status_code, content
 
-    monkeypatch.setattr(helper, "_PluginMarketTransport__request_with_fallback", request)
+    monkeypatch.setattr(helper, SYNC_INDEX_REQUEST, request)
     helper.get_plugin_index_result.cache_clear()
 
     result = helper.get_plugin_index_result(repo_url, "v3")
@@ -293,9 +569,9 @@ def test_plugin_index_result_raises_for_unusable_reads(
     helper = PluginMarketTransport()
 
     def request(_url: str, *, headers: dict):
-        return SimpleNamespace(status_code=status_code, text=content)
+        return status_code, content
 
-    monkeypatch.setattr(helper, "_PluginMarketTransport__request_with_fallback", request)
+    monkeypatch.setattr(helper, SYNC_INDEX_REQUEST, request)
     helper.get_plugin_index_result.cache_clear()
 
     with pytest.raises(RuntimeError, match=message):
@@ -305,17 +581,40 @@ def test_plugin_index_result_raises_for_unusable_reads(
         )
 
 
+def test_invalid_plugin_index_is_not_cached(monkeypatch) -> None:
+    """非法索引每次都应重新读取，避免失败载荷占用正常缓存周期。"""
+    helper = PluginMarketTransport()
+    requests = 0
+
+    def request(_url: str, *, headers: dict):
+        nonlocal requests
+        requests += 1
+        return 200, "not-json"
+
+    monkeypatch.setattr(helper, SYNC_INDEX_REQUEST, request)
+    helper.get_plugin_index_result.cache_clear()
+
+    for _ in range(2):
+        with pytest.raises(RuntimeError, match="插件索引响应格式无效"):
+            helper.get_plugin_index_result(
+                "https://github.com/policy-owner/invalid-index-cache",
+                "v3",
+            )
+
+    assert requests == 2
+
+
 @pytest.mark.asyncio
 async def test_async_plugin_index_result_preserves_absent_state(monkeypatch) -> None:
     """异步只读入口也必须保留 404 不存在事实。"""
     helper = PluginMarketTransport()
 
     async def request(_url: str, *, headers: dict):
-        return SimpleNamespace(status_code=404, text="404: Not Found")
+        return 404, ""
 
     monkeypatch.setattr(
         helper,
-        "_PluginMarketTransport__async_request_with_fallback",
+        ASYNC_INDEX_REQUEST,
         request,
     )
     await helper.async_get_plugin_index_result.cache_clear()
@@ -335,7 +634,7 @@ def test_plugin_index_result_propagates_adapter_exception(monkeypatch) -> None:
     def request(_url: str, *, headers: dict):
         raise OSError("socket closed")
 
-    monkeypatch.setattr(helper, "_PluginMarketTransport__request_with_fallback", request)
+    monkeypatch.setattr(helper, SYNC_INDEX_REQUEST, request)
     helper.get_plugin_index_result.cache_clear()
 
     with pytest.raises(OSError, match="socket closed"):