From dba4ddc168cb7713683e5c532ed0482dc5195ce7 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Sat, 29 Aug 2026 15:23:33 +0800 Subject: [PATCH] =?UTF-8?q?fix(plugin):=20=E5=A2=9E=E5=8A=A0=E6=8F=92?= =?UTF-8?q?=E4=BB=B6=E5=B8=82=E5=9C=BA=E5=85=83=E6=95=B0=E6=8D=AE=E8=BE=B9?= =?UTF-8?q?=E7=95=8C=20(#6495)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 | 385 ++++++++++++++++-- app/application/plugin/identity.py | 4 +- app/domain/plugin.py | 11 + app/runtime/extensions/plugin/catalog.py | 6 +- docs/architecture-optimization-checklist.md | 4 +- docs/architecture-overview.md | 2 +- .../architecture/coverage-baseline.json | 6 +- .../architecture/dependency-baseline.json | 6 +- tests/test_plugin_catalog_runtime.py | 36 ++ tests/test_plugin_market_index_policy.py | 369 +++++++++++++++-- 10 files changed, 752 insertions(+), 77 deletions(-) diff --git a/app/adapters/external/plugin/client.py b/app/adapters/external/plugin/client.py index a4f5616d5..386f6b560 100644 --- a/app/adapters/external/plugin/client.py +++ b/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 diff --git a/app/application/plugin/identity.py b/app/application/plugin/identity.py index bdecc62f4..0b07e7691 100644 --- a/app/application/plugin/identity.py +++ b/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() diff --git a/app/domain/plugin.py b/app/domain/plugin.py index 6d8094403..a749c8070 100644 --- a/app/domain/plugin.py +++ b/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)) diff --git a/app/runtime/extensions/plugin/catalog.py b/app/runtime/extensions/plugin/catalog.py index 79e8e85f5..82c58d87a 100644 --- a/app/runtime/extensions/plugin/catalog.py +++ b/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") diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index 2d8aa152f..853dad9a4 100644 --- a/docs/architecture-optimization-checklist.md +++ b/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 热点文件 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index d53095784..a057d5010 100644 --- a/docs/architecture-overview.md +++ b/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） | diff --git a/tests/fixtures/architecture/coverage-baseline.json b/tests/fixtures/architecture/coverage-baseline.json index 7304b7971..61c8a8359 100644 --- a/tests/fixtures/architecture/coverage-baseline.json +++ b/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 } } diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index de39a7402..23716c3d4 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/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", diff --git a/tests/test_plugin_catalog_runtime.py b/tests/test_plugin_catalog_runtime.py index 329ea4f48..a7f4e3b54 100644 --- a/tests/test_plugin_catalog_runtime.py +++ b/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" + ] diff --git a/tests/test_plugin_market_index_policy.py b/tests/test_plugin_market_index_policy.py index d3f941850..b1e7c8fb1 100644 --- a/tests/test_plugin_market_index_policy.py +++ b/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"):