diff --git a/.gitignore b/.gitignore index 44cc045..3b4f2e9 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,10 @@ backend/.env # 运行期生成的 SQLite 索引（vault 下的 Markdown 测试数据需提交） backend/data/*.db* backend/data/credentials/ +# 本机 MCP 配置、授权状态及服务器工作目录不得提交。 +backend/data/mcp/ +server.json +servers.json # Editors and operating systems .idea/ diff --git a/backend/app/extensions/mcp.py b/backend/app/extensions/mcp.py index c3852fc..8247edd 100644 --- a/backend/app/extensions/mcp.py +++ b/backend/app/extensions/mcp.py @@ -10,6 +10,7 @@ import asyncio import json import os import queue +import re import signal import subprocess import threading @@ -49,6 +50,7 @@ MAX_MCP_MESSAGE_BYTES = 2 * 1024 * 1024 MAX_MCP_TOOL_RESULT_BYTES = 256 * 1024 MAX_MCP_TOOLS = 500 MAX_MCP_LIST_PAGES = 100 +_SSE_NEWLINE = re.compile(rb"\r\n?|\n") class McpBridgeError(RuntimeError): @@ -1019,7 +1021,8 @@ class McpBridge: if status.status == PluginHostState.unhealthy: raise McpBridgeError( "PLUGIN_HOST_UNAVAILABLE", - status.error or "MCP event stream became unavailable during startup.", + status.error + or "MCP event stream became unavailable during startup.", status_code=503, ) status.status = PluginHostState.ready @@ -1383,12 +1386,58 @@ def _bounded_json_response(response: httpx.Response) -> dict[str, Any]: return payload +def _bounded_sse_lines(response: httpx.Response): + """Split UTF-8 lines without httpx.iter_lines()'s unbounded line buffer. + + Check each segment before appending it, including partial/no-newline input. + SSE allows LF, CR and CRLF; a CRLF pair can span network chunks. + """ + + pending = bytearray() + event_size = 0 + skip_lf = False + first_line = True + for chunk in response.iter_bytes(): + offset = 0 + if skip_lf and chunk: + offset = int(chunk.startswith(b"\n")) + skip_lf = False + for match in _SSE_NEWLINE.finditer(chunk, offset): + start, end = match.span() + segment = memoryview(chunk)[offset:start] + if event_size + len(pending) + len(segment) + 1 > MAX_MCP_MESSAGE_BYTES: + raise McpBridgeError( + "MCP_HTTP_RESPONSE_INVALID", "MCP SSE event is too large." + ) + pending.extend(segment) + line = pending.decode("utf-8", errors="replace") + event_size += len(pending) + 1 + pending.clear() + if first_line: + line = line.removeprefix("\ufeff") + first_line = False + if not line: + event_size = 0 + yield line + skip_lf = chunk[end - 1 : end] == b"\r" and end == len(chunk) + offset = end + tail = memoryview(chunk)[offset:] + if event_size + len(pending) + len(tail) > MAX_MCP_MESSAGE_BYTES: + raise McpBridgeError( + "MCP_HTTP_RESPONSE_INVALID", "MCP SSE event is too large." + ) + pending.extend(tail) + if pending: + line = pending.decode("utf-8", errors="replace") + yield line.removeprefix("\ufeff") if first_line else line + + def _iter_sse(response: httpx.Response): event = "message" event_id: str | None = None data_lines: list[str] = [] size = 0 - for line in response.iter_lines(): + for line in _bounded_sse_lines(response): size += len(line.encode("utf-8")) + 1 if size > MAX_MCP_MESSAGE_BYTES: raise McpBridgeError( diff --git a/backend/app/extensions/mcp_registry.py b/backend/app/extensions/mcp_registry.py index 7307ede..a81aced 100644 --- a/backend/app/extensions/mcp_registry.py +++ b/backend/app/extensions/mcp_registry.py @@ -9,7 +9,7 @@ import threading from datetime import UTC, datetime from functools import wraps from pathlib import Path -from typing import Any +from typing import Any, Literal from urllib.parse import urlsplit from uuid import uuid4 @@ -45,10 +45,22 @@ _SERVER_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$") _MAX_MCP_SERVERS = 256 +class _McpConnectionBackend(PluginBackend): + """Bridge adapter for the independent server's float timeout contract. + + Plugin manifests retain their integer/60-second startup restrictions. + Reusing that validation here used to reject valid 120-second server configs. + """ + + startup_timeout_seconds: float = Field(default=15, ge=1, le=120) + tool_timeout_seconds: float = Field(default=30, ge=1, le=300) + + class _McpServerRecord(McpServerConfig): """Validated on-disk representation with defaults for older C.1 records.""" version: int = Field(default=1, ge=1) + secret_environment_version: Literal[1, 2] = 1 enabled: bool = False approved_digest: str | None = Field( default=None, min_length=64, max_length=64, pattern=r"^[0-9a-f]{64}$" @@ -102,6 +114,8 @@ class McpServerRegistry: self._registered: dict[str, list[str]] = {} self._summaries: dict[str, list[McpToolSummary]] = {} self._last_status: dict[str, dict[str, Any]] = {} + self._generations: dict[str, object] = {} + self._migrate_environment_secrets() def list(self) -> list[McpServer]: with self._lock: @@ -137,6 +151,7 @@ class McpServerRegistry: record["url"] = request.url.strip() if request.url else None record.update( version=1, + secret_environment_version=2, enabled=False, approved_digest=None, tested_digest=None, @@ -163,7 +178,7 @@ class McpServerRegistry: with self._lock: previous = self._record(server_id) removed_secret_ids = [ - self._secret_id(server_id, key, kind) + secret_id for kind, old_keys, new_keys in ( ( "environment", @@ -176,7 +191,8 @@ class McpServerRegistry: request.secret_header_keys, ), ) - for key in set(old_keys) - set(new_keys) + for secret_id in self._secret_ids(server_id, old_keys, kind) + - self._secret_ids(server_id, new_keys, kind) ] try: self.credentials.delete_many(removed_secret_ids) @@ -191,6 +207,7 @@ class McpServerRegistry: record["url"] = request.url.strip() if request.url else None record.update( version=request.version + 1, + secret_environment_version=2, enabled=False, approved_digest=None, tested_digest=None, @@ -210,12 +227,12 @@ class McpServerRegistry: with self._lock: record = self._record(server_id) secret_ids = [ - self._secret_id(server_id, key, kind) + secret_id for kind, keys in ( ("environment", record.get("secret_environment_keys", [])), ("header", record.get("secret_header_keys", [])), ) - for key in keys + for secret_id in self._secret_ids(server_id, keys, kind) ] try: self.credentials.delete_many(secret_ids) @@ -307,6 +324,8 @@ class McpServerRegistry: try: discovered = self._start(server_id, record) except Exception as exc: + self._generations.pop(server_id, None) + self.bridge.remove(self._host_id(server_id)) tested_at = datetime.now(UTC) failure = { "status": PluginHostState.error, @@ -339,6 +358,7 @@ class McpServerRegistry: "last_test_succeeded": True, } self._summaries[server_id] = self._tool_summaries(discovered) + self._generations.pop(server_id, None) self.bridge.stop(self._host_id(server_id)) with self._lock: tested_record = { @@ -368,6 +388,7 @@ class McpServerRegistry: except Exception: for name in registered: self.tools.unregister(name) + self._generations.pop(server_id, None) self.bridge.stop(self._host_id(server_id)) raise try: @@ -380,6 +401,7 @@ class McpServerRegistry: except McpRegistryError: for name in registered: self.tools.unregister(name) + self._generations.pop(server_id, None) self.bridge.stop(self._host_id(server_id)) raise return self.get(server_id) @@ -394,6 +416,7 @@ class McpServerRegistry: self._records = updated for name in self._registered.pop(server_id, []): self.tools.unregister(name) + self._generations.pop(server_id, None) self.bridge.stop(self._host_id(server_id)) return self.get(server_id) @@ -416,6 +439,7 @@ class McpServerRegistry: @_serialized_lifecycle def shutdown(self) -> None: for server_id in list(self._records): + self._generations.pop(server_id, None) for name in self._registered.pop(server_id, []): self.tools.unregister(name) self.bridge.stop(self._host_id(server_id)) @@ -456,6 +480,9 @@ class McpServerRegistry: ) headers[key] = value host_id = self._host_id(server_id) + # A queued callback from the previous process must not affect its replacement. + generation = object() + self._generations[server_id] = generation self.bridge.remove(host_id) try: return self.bridge.start( @@ -463,7 +490,9 @@ class McpServerRegistry: self._backend(record), self._server_dir(server_id), list(record.get("permissions", [])), - lambda _host, message: self._unavailable(server_id, message), + lambda _host, message: self._unavailable( + server_id, generation, message + ), command_override=( [record["command"], *record.get("args", [])] if record.get("command") @@ -476,6 +505,8 @@ class McpServerRegistry: headers=headers, ) except McpBridgeError as exc: + self._generations.pop(server_id, None) + self.bridge.remove(host_id) raise McpRegistryError( exc.code, exc.message, status_code=exc.status_code ) from exc @@ -496,10 +527,13 @@ class McpServerRegistry: self.tools.register(definition, arguments_model, executor) - def _unavailable(self, server_id: str, message: str) -> None: + def _unavailable(self, server_id: str, generation: object, message: str) -> None: # A failure may race with enable(). Waiting for the lifecycle mutation makes # sure tools registered immediately before the callback are also removed. with self._lifecycle_lock: + if self._generations.get(server_id) is not generation: + return + self._generations.pop(server_id, None) try: with self._lock: record = self._records.get(server_id) @@ -694,7 +728,7 @@ class McpServerRegistry: @staticmethod def _backend(record: dict[str, Any]) -> PluginBackend: - return PluginBackend( + return _McpConnectionBackend( type="mcp", transport="stdio", command=record.get("command") or "http", @@ -755,9 +789,89 @@ class McpServerRegistry: @staticmethod def _secret_id(server_id: str, key: str, kind: str = "environment") -> str: - suffix = hashlib.sha256(f"{kind}\0{key.casefold()}".encode()).hexdigest()[:20] + identity = ( + f"environment-v2\0{key}" + if kind == "environment" + else f"{kind}\0{key.casefold()}" + ) + suffix = hashlib.sha256(identity.encode()).hexdigest()[:20] return f"mcp.{server_id}.{suffix}" + @staticmethod + def _legacy_environment_secret_id(server_id: str, key: str) -> str: + suffix = hashlib.sha256(f"environment\0{key.casefold()}".encode()).hexdigest()[ + :20 + ] + return f"mcp.{server_id}.{suffix}" + + def _secret_ids(self, server_id: str, keys: list[str], kind: str) -> set[str]: + ids = {self._secret_id(server_id, key, kind) for key in keys} + if kind == "environment": + # Include retained ambiguous legacy ciphertext when its last declaration is removed. + ids.update( + self._legacy_environment_secret_id(server_id, key) for key in keys + ) + return ids + + def _migrate_environment_secrets(self) -> None: + """迁移旧的大小写折叠 ID；已碰撞的值无法恢复，保留原密文并要求重新录入。""" + + replacements: dict[str, str] = {} + ambiguous: dict[str, list[str]] = {} + legacy_records = { + server_id: record + for server_id, record in self._records.items() + if record.get("secret_environment_version", 1) == 1 + } + if not legacy_records: + return + for server_id, record in legacy_records.items(): + groups: dict[str, set[str]] = {} + for key in record.get("secret_environment_keys", []): + groups.setdefault(key.casefold(), set()).add(key) + for keys in groups.values(): + key = next(iter(keys)) + legacy_id = self._legacy_environment_secret_id(server_id, key) + if len(keys) == 1: + replacements[legacy_id] = self._secret_id(server_id, key) + else: + ambiguous.setdefault(server_id, []).extend(keys) + try: + self.credentials.move_many(replacements) + for server_id, keys in ambiguous.items(): + if not any( + self.credentials.has( + self._legacy_environment_secret_id(server_id, key) + ) + for key in keys + ): + continue + if all( + self.credentials.has(self._secret_id(server_id, key)) + for key in keys + ): + continue + self._records[server_id].update( + enabled=False, + tested_digest=None, + last_test_succeeded=None, + last_tested_at=None, + ) + self._last_status[server_id] = { + "status": PluginHostState.error, + "error": "环境变量密钥名称曾发生大小写冲突，请分别重新录入密钥并测试连接。", + } + # Persist a migration marker even when legacy values were ambiguous. + # Otherwise a later key removal could make that old shared value look + # unambiguous and resurrect a deleted credential on the next restart. + for server_id in legacy_records: + self._records[server_id]["secret_environment_version"] = 2 + self._write() + except CredentialStoreError as exc: + raise McpRegistryError( + "MCP_SECRET_STORE_ERROR", str(exc), status_code=500 + ) from exc + def _secret_configured( self, server_id: str, key: str, kind: str = "environment" ) -> bool: @@ -857,7 +971,9 @@ class McpServerRegistry: config_fields = set(McpServerConfig.model_fields) try: for server_id, raw in value.items(): - if not isinstance(server_id, str) or not _SERVER_ID.fullmatch(server_id): + if not isinstance(server_id, str) or not _SERVER_ID.fullmatch( + server_id + ): raise ValueError("invalid server id") record = _McpServerRecord.model_validate(raw) config = record.model_dump(mode="json", include=config_fields) diff --git a/backend/app/providers/credentials.py b/backend/app/providers/credentials.py index accdacb..8eec9a2 100644 --- a/backend/app/providers/credentials.py +++ b/backend/app/providers/credentials.py @@ -5,13 +5,12 @@ import os import re import threading from pathlib import Path -from typing import Protocol +from typing import ClassVar, Protocol from cryptography.fernet import Fernet, InvalidToken from app.config import get_settings - _CREDENTIAL_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") _PLUGIN_CREDENTIAL_PREFIX = "plugin." _MCP_CREDENTIAL_PREFIX = "mcp." @@ -29,7 +28,9 @@ def validate_provider_credential_id(credential_id: str | None) -> None: """阻止 Provider 和通用凭据 API 跨入 Plugin 私有命名空间。""" if credential_id and credential_id.casefold().startswith(_PLUGIN_CREDENTIAL_PREFIX): - raise CredentialStoreError("Credential namespace is reserved for Plugin settings.") + raise CredentialStoreError( + "Credential namespace is reserved for Plugin settings." + ) if credential_id and credential_id.casefold().startswith(_MCP_CREDENTIAL_PREFIX): raise CredentialStoreError("Credential namespace is reserved for MCP settings.") @@ -37,7 +38,7 @@ def validate_provider_credential_id(credential_id: str | None) -> None: class EnvironmentCredentialResolver: """解析由桌面 Host 注入 Sidecar 进程的临时凭证上下文。""" - _development_aliases = { + _development_aliases: ClassVar[dict[str, str]] = { "openai": "OPENAI_API_KEY", "deepseek": "DEEPSEEK_API_KEY", } @@ -85,7 +86,9 @@ class EncryptedCredentialStore: try: return Fernet(environment_key.encode("ascii")) except (ValueError, UnicodeEncodeError) as exc: - raise CredentialStoreError("APP_CREDENTIAL_MASTER_KEY is invalid.") from exc + raise CredentialStoreError( + "APP_CREDENTIAL_MASTER_KEY is invalid." + ) from exc key_path.parent.mkdir(parents=True, exist_ok=True) self._restrict(key_path.parent, 0o700) @@ -102,7 +105,9 @@ class EncryptedCredentialStore: try: return Fernet(key_path.read_bytes().strip()) except (OSError, ValueError) as exc: - raise CredentialStoreError("Credential master key cannot be loaded.") from exc + raise CredentialStoreError( + "Credential master key cannot be loaded." + ) from exc def _read_tokens(self) -> dict[str, str]: _, store_path = self._paths() @@ -111,11 +116,16 @@ class EncryptedCredentialStore: try: data = json.loads(store_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: - raise CredentialStoreError("Encrypted credential store cannot be loaded.") from exc + raise CredentialStoreError( + "Encrypted credential store cannot be loaded." + ) from exc if not isinstance(data, dict) or not all( - isinstance(key, str) and isinstance(value, str) for key, value in data.items() + isinstance(key, str) and isinstance(value, str) + for key, value in data.items() ): - raise CredentialStoreError("Encrypted credential store has an invalid format.") + raise CredentialStoreError( + "Encrypted credential store has an invalid format." + ) return data def _write_tokens(self, tokens: dict[str, str]) -> None: @@ -196,6 +206,22 @@ class EncryptedCredentialStore: self._write_tokens(tokens) return removed + def move_many(self, replacements: dict[str, str]) -> None: + """原子迁移凭据 ID，直接移动密文且不覆盖已经写入的新凭据。""" + + for old_id, new_id in replacements.items(): + self._validate_id(old_id) + self._validate_id(new_id) + with self._lock: + tokens = self._read_tokens() + changed = False + for old_id, new_id in replacements.items(): + if old_id != new_id and old_id in tokens: + tokens.setdefault(new_id, tokens.pop(old_id)) + changed = True + if changed: + self._write_tokens(tokens) + class ChainedCredentialResolver: def __init__(self, *resolvers: CredentialResolver) -> None: diff --git a/backend/app/routes.py b/backend/app/routes.py index 27c3586..ec8908f 100644 --- a/backend/app/routes.py +++ b/backend/app/routes.py @@ -100,15 +100,8 @@ from app.services import ( router = APIRouter(prefix="/api") -def mcp_call(operation): - try: - return operation() - except McpRegistryError as exc: - raise ApiError(exc.status_code, exc.code, exc.message) from exc - - async def mcp_call_async(operation): - """MCP process operations wait on stdio and must not block the API event loop.""" + """Even registry reads can wait on lifecycle locks; keep all MCP work off the event loop.""" try: return await asyncio.to_thread(operation) except McpRegistryError as exc: @@ -521,19 +514,19 @@ async def uninstall_skill(skill_id: str) -> OperationResponse: # Independent MCP Server Registry @router.get("/mcp/servers", response_model=McpServerListResponse, tags=["MCP Servers"]) async def list_mcp_servers() -> McpServerListResponse: - return McpServerListResponse(items=mcp_call(container.mcp_servers.list)) + return McpServerListResponse(items=await mcp_call_async(container.mcp_servers.list)) @router.post( "/mcp/servers", response_model=McpServer, status_code=201, tags=["MCP Servers"] ) async def create_mcp_server(request: McpServerCreateRequest) -> McpServer: - return mcp_call(lambda: container.mcp_servers.create(request)) + return await mcp_call_async(lambda: container.mcp_servers.create(request)) @router.get("/mcp/servers/{server_id}", response_model=McpServer, tags=["MCP Servers"]) async def get_mcp_server(server_id: str) -> McpServer: - return mcp_call(lambda: container.mcp_servers.get(server_id)) + return await mcp_call_async(lambda: container.mcp_servers.get(server_id)) @router.get( @@ -543,7 +536,7 @@ async def get_mcp_server(server_id: str) -> McpServer: ) async def list_mcp_server_tools(server_id: str) -> McpToolSummaryListResponse: return McpToolSummaryListResponse( - items=mcp_call(lambda: container.mcp_servers.list_tools(server_id)) + items=await mcp_call_async(lambda: container.mcp_servers.list_tools(server_id)) ) @@ -570,7 +563,7 @@ async def delete_mcp_server(server_id: str) -> OperationResponse: "/mcp/servers/{server_id}/trust", response_model=McpServer, tags=["MCP Servers"] ) async def trust_mcp_server(server_id: str, request: McpServerTrustRequest) -> McpServer: - return mcp_call( + return await mcp_call_async( lambda: container.mcp_servers.trust(server_id, request.command_digest) ) diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 73901a5..97d7fee 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -2,6 +2,8 @@ import asyncio import threading from types import SimpleNamespace +import pytest + from app.contracts import ( McpServerSecretStatus, McpServerSecretWriteRequest, @@ -61,9 +63,7 @@ def test_mcp_secret_routes_offload_blocking_lifecycle_work(monkeypatch) -> None: ) ) deleted = asyncio.run( - routes.delete_mcp_server_secret( - "server-1", "TOKEN", kind="environment" - ) + routes.delete_mcp_server_secret("server-1", "TOKEN", kind="environment") ) assert written.configured is True @@ -77,6 +77,172 @@ def test_health() -> None: assert response.model_dump() == {"status": "ok"} +def test_mcp_create_and_trust_are_not_executed_on_event_loop(monkeypatch) -> None: + from app import routes + from app.contracts import McpServerCreateRequest, McpServerTrustRequest + + caller = threading.get_ident() + workers = [] + + class Registry: + def create(self, request): + workers.append(threading.get_ident()) + return "created" + + def trust(self, server_id, digest): + workers.append(threading.get_ident()) + return "trusted" + + monkeypatch.setattr(routes, "container", SimpleNamespace(mcp_servers=Registry())) + assert ( + asyncio.run( + routes.create_mcp_server(McpServerCreateRequest(name="test", command="uvx")) + ) + == "created" + ) + assert ( + asyncio.run( + routes.trust_mcp_server( + "test", McpServerTrustRequest(command_digest="a" * 64) + ) + ) + == "trusted" + ) + assert len(workers) == 2 + assert all(worker != caller for worker in workers) + + +def test_mcp_split_config_and_secret_requests_persist_without_plaintext( + monkeypatch, +) -> None: + from fastapi.testclient import TestClient + + from app import routes + from app.agent.tools import ToolRegistry + from app.config import get_settings + from app.extensions.mcp_registry import McpServerRegistry + from app.main import app + from app.providers.credentials import EncryptedCredentialStore + + service = McpServerRegistry( + ToolRegistry(), + EncryptedCredentialStore(), + get_settings().data_dir, + allow_process_launch=True, + ) + monkeypatch.setattr(routes, "container", SimpleNamespace(mcp_servers=service)) + client = TestClient(app) + config = { + "name": "MiniMax configuration test", + "command": "uvx", + "environment": {"MINIMAX_API_HOST": "https://api.minimaxi.com"}, + "secret_environment_keys": ["MINIMAX_API_KEY"], + "startup_timeout_seconds": 120, + "tool_timeout_seconds": 300, + } + # Reproduce the old frontend payload. The backend still enforces separation. + invalid = client.post( + "/api/mcp/servers", + json={ + **config, + "environment": { + **config["environment"], + "MINIMAX_API_KEY": "synthetic-only", + }, + }, + ) + assert invalid.status_code == 422 + assert invalid.json()["error"]["code"] == "MCP_ENVIRONMENT_INVALID" + created = client.post("/api/mcp/servers", json=config) + assert created.status_code == 201 + server_id = created.json()["server_id"] + saved = client.put( + f"/api/mcp/servers/{server_id}/secrets/MINIMAX_API_KEY", + json={"secret": "synthetic-only"}, + ) + assert saved.status_code == 200 + current = client.get(f"/api/mcp/servers/{server_id}") + assert current.json()["secret_environment"] == {"MINIMAX_API_KEY": True} + assert "synthetic-only" not in current.text + assert "synthetic-only" not in service._path.read_text(encoding="utf-8") + _, credentials_path = service.credentials._paths() + assert "synthetic-only" not in credentials_path.read_text(encoding="utf-8") + assert not current.json()["enabled"] # Saving never starts a third-party process. + client.close() + + +@pytest.mark.parametrize("operation", ["create", "trust"]) +def test_mcp_lifecycle_lock_contention_keeps_event_loop_responsive( + monkeypatch, + operation, +) -> None: + from app import routes + from app.agent.tools import ToolRegistry + from app.config import get_settings + from app.contracts import McpServerCreateRequest, McpServerTrustRequest + from app.extensions.mcp_registry import McpServerRegistry + from app.providers.credentials import EncryptedCredentialStore + + service = McpServerRegistry( + ToolRegistry(), + EncryptedCredentialStore(), + get_settings().data_dir, + allow_process_launch=True, + ) + request = McpServerCreateRequest( + name="Lock contention fixture", command="not-executed" + ) + server = service.create(request) + monkeypatch.setattr(routes, "container", SimpleNamespace(mcp_servers=service)) + entered = threading.Event() + locked = threading.Event() + release = threading.Event() + original = getattr(service, operation) + + def observed(*args): + entered.set() + return original(*args) + + def hold_lifecycle_lock(): + with service._lifecycle_lock: + locked.set() + release.wait(timeout=5) + + monkeypatch.setattr(service, operation, observed) + holder = threading.Thread(target=hold_lifecycle_lock, daemon=True) + holder.start() + # An independent watchdog lets the test fail rather than hang if a regression + # blocks the event loop itself (an asyncio timeout alone cannot catch that). + watchdog = threading.Timer(5, release.set) + watchdog.start() + + async def exercise(): + pending = asyncio.create_task( + routes.create_mcp_server(request) + if operation == "create" + else routes.trust_mcp_server( + server.server_id, + McpServerTrustRequest(command_digest=server.command_digest), + ) + ) + try: + assert await asyncio.to_thread(entered.wait, 2) + assert not pending.done() + assert not release.is_set() + assert (await health()).status == "ok" + finally: + release.set() + await pending + + try: + assert locked.wait(timeout=2) + asyncio.run(exercise()) + finally: + release.set() + watchdog.cancel() + holder.join(timeout=2) + + def test_service_status() -> None: response = asyncio.run(service_status()) @@ -93,7 +259,9 @@ def test_core_collections_are_typed() -> None: assert notes.items == [] assert notes.page.limit == 20 - assert [skill.manifest.skill_id for skill in skills.items] == ["knowledge-assistant"] + assert [skill.manifest.skill_id for skill in skills.items] == [ + "knowledge-assistant" + ] assert skills.items[0].status == "ready" assert [plugin.manifest.plugin_id for plugin in plugins.items] == ["text-tools"] assert plugins.items[0].status == "ready" @@ -114,9 +282,15 @@ def test_provider_presets_include_openai_and_deepseek() -> None: def test_provider_presets_static_route_precedes_provider_id_route() -> None: from app.routes import router - get_paths = [route.path for route in router.routes if "GET" in getattr(route, "methods", set())] + get_paths = [ + route.path + for route in router.routes + if "GET" in getattr(route, "methods", set()) + ] - assert get_paths.index("/api/providers/presets") < get_paths.index("/api/providers/{provider_id}") + assert get_paths.index("/api/providers/presets") < get_paths.index( + "/api/providers/{provider_id}" + ) def test_openapi_contains_documented_frontend_interfaces() -> None: diff --git a/backend/tests/test_mcp_registry.py b/backend/tests/test_mcp_registry.py index 1adfa14..2d141c0 100644 --- a/backend/tests/test_mcp_registry.py +++ b/backend/tests/test_mcp_registry.py @@ -1,4 +1,5 @@ import asyncio +import hashlib import json import sys import threading @@ -111,7 +112,9 @@ def test_update_disables_server_and_revokes_command_trust() -> None: service.shutdown() -def test_update_remains_retryable_when_removed_secret_cleanup_fails(monkeypatch) -> None: +def test_update_remains_retryable_when_removed_secret_cleanup_fails( + monkeypatch, +) -> None: service = registry() created = service.create(request()) service.put_secret(created.server_id, "TEST_MCP_SECRET", "keep-until-retry") @@ -166,7 +169,9 @@ def test_unavailable_server_removes_bridge_host(monkeypatch) -> None: removed: list[str] = [] monkeypatch.setattr(service.bridge, "remove", removed.append) - service._unavailable(created.server_id, "connection lost") + generation = object() + service._generations[created.server_id] = generation + service._unavailable(created.server_id, generation, "connection lost") current = service.get(created.server_id) assert removed == [f"mcp.{created.server_id}"] @@ -175,6 +180,180 @@ def test_unavailable_server_removes_bridge_host(monkeypatch) -> None: service.shutdown() +def test_old_failure_callback_cannot_stop_replacement_host(monkeypatch) -> None: + service = registry() + callbacks = [] + original_start = service.bridge.start + + def capture_callback(*args, **kwargs): + callbacks.append(args[4]) + return original_start(*args, **kwargs) + + monkeypatch.setattr(service.bridge, "start", capture_callback) + created = service.create(request(secret_environment_keys=[])) + service.trust(created.server_id, created.command_digest) + callback_thread = None + try: + service.test(created.server_id) + service.enable(created.server_id) + old_callback = callbacks[-1] + callback_started = threading.Event() + callback_finished = threading.Event() + + def delayed_failure(): + callback_started.set() + old_callback(f"mcp.{created.server_id}", "delayed old failure") + callback_finished.set() + + # Queue the old callback while a replacement owns the lifecycle lock. + with service._lifecycle_lock: + callback_thread = threading.Thread(target=delayed_failure, daemon=True) + callback_thread.start() + assert callback_started.wait(timeout=2) + service.disable(created.server_id) + service.enable(created.server_id) + assert callback_finished.wait(timeout=2) + assert service.get(created.server_id).enabled is True + assert service.get(created.server_id).status == "ready" + assert service.tools.definitions() + callbacks[-1](f"mcp.{created.server_id}", "current failure") + assert service.get(created.server_id).enabled is False + assert service.get(created.server_id).status == "unhealthy" + finally: + service.shutdown() + if callback_thread is not None: + callback_thread.join(timeout=2) + + +def test_header_case_only_rename_preserves_secret() -> None: + service = registry() + config = { + "name": "HTTP", + "transport": "streamable_http", + "url": "https://example.test/mcp", + "secret_header_keys": ["Authorization"], + } + created = service.create(McpServerCreateRequest(**config)) + service.put_secret(created.server_id, "Authorization", "synthetic", kind="header") + config["secret_header_keys"] = ["authorization"] + updated = service.update( + created.server_id, McpServerUpdateRequest(**config, version=created.version) + ) + assert updated.secret_headers == {"authorization": True} + assert ( + service.credentials.resolve( + service._secret_id(created.server_id, "authorization", "header") + ) + == "synthetic" + ) + + +def test_environment_secrets_are_case_sensitive_and_delete_independently() -> None: + service = registry() + created = service.create(request(secret_environment_keys=["TOKEN", "token"])) + service.put_secret(created.server_id, "TOKEN", "upper") + service.put_secret(created.server_id, "token", "lower") + assert ( + service.credentials.resolve(service._secret_id(created.server_id, "TOKEN")) + == "upper" + ) + assert ( + service.credentials.resolve(service._secret_id(created.server_id, "token")) + == "lower" + ) + service.delete_secret(created.server_id, "TOKEN") + assert service.get(created.server_id).secret_environment == { + "TOKEN": False, + "token": True, + } + + +def test_legacy_environment_credential_migration_is_idempotent() -> None: + service = registry() + created = service.create(request(secret_environment_keys=["TOKEN"])) + suffix = hashlib.sha256(b"environment\0token").hexdigest()[:20] + legacy_id = f"mcp.{created.server_id}.{suffix}" + service.credentials.put(legacy_id, "legacy-value") + service._records[created.server_id]["secret_environment_version"] = 1 + service._write() + migrated = registry() + assert migrated.get(created.server_id).secret_environment == {"TOKEN": True} + assert ( + migrated.credentials.resolve(migrated._secret_id(created.server_id, "TOKEN")) + == "legacy-value" + ) + assert not migrated.credentials.has(legacy_id) + migrated.put_secret(created.server_id, "TOKEN", "new-value") + assert ( + registry().credentials.resolve(migrated._secret_id(created.server_id, "TOKEN")) + == "new-value" + ) + + +def test_ambiguous_legacy_credentials_are_not_assigned_to_two_variables() -> None: + service = registry() + created = service.create(request(secret_environment_keys=["TOKEN", "token"])) + suffix = hashlib.sha256(b"environment\0token").hexdigest()[:20] + legacy_id = f"mcp.{created.server_id}.{suffix}" + service.credentials.put(legacy_id, "cannot-reconstruct-originals") + service._records[created.server_id]["secret_environment_version"] = 1 + service._write() + migrated = registry() + current = migrated.get(created.server_id) + assert current.secret_environment == {"TOKEN": False, "token": False} + assert current.enabled is False + assert current.last_test_succeeded is None + assert migrated.credentials.has( + legacy_id + ) # Keep the original ciphertext recoverable. + migrated.put_secret(created.server_id, "TOKEN", "upper") + migrated.put_secret(created.server_id, "token", "lower") + assert registry().get(created.server_id).secret_environment == { + "TOKEN": True, + "token": True, + } + migrated.delete(created.server_id) + assert not migrated.credentials.has(legacy_id) + + +def test_credential_id_migration_keeps_new_values_and_is_atomic(monkeypatch) -> None: + credentials = EncryptedCredentialStore() + credentials.put("mcp.old", "old-value") + credentials.put("mcp.new", "new-value") + original_write = credentials._write_tokens + + def fail_write(_tokens): + raise CredentialStoreError("synthetic failure") + + monkeypatch.setattr(credentials, "_write_tokens", fail_write) + with pytest.raises(CredentialStoreError): + credentials.move_many({"mcp.old": "mcp.new"}) + assert credentials.resolve("mcp.old") == "old-value" + assert credentials.resolve("mcp.new") == "new-value" + monkeypatch.setattr(credentials, "_write_tokens", original_write) + credentials.move_many({"mcp.old": "mcp.new"}) + assert credentials.resolve("mcp.old") is None + assert credentials.resolve("mcp.new") == "new-value" + + +def test_ambiguous_legacy_secret_is_not_resurrected_after_removing_a_key() -> None: + service = registry() + created = service.create(request(secret_environment_keys=["TOKEN", "token"])) + legacy_id = service._legacy_environment_secret_id(created.server_id, "TOKEN") + service.credentials.put(legacy_id, "ambiguous-old-value") + service._records[created.server_id]["secret_environment_version"] = 1 + service._write() + migrated = registry() + migrated.update( + created.server_id, + McpServerUpdateRequest( + **request(secret_environment_keys=["token"]).model_dump(), + version=created.version, + ), + ) + assert registry().get(created.server_id).secret_environment == {"token": False} + + def test_production_rejects_process_launch_even_after_approval() -> None: service = registry(launch=False) created = service.create(request(secret_environment_keys=[])) @@ -184,6 +363,29 @@ def test_production_rejects_process_launch_even_after_approval() -> None: assert error.value.code == "MCP_SANDBOX_REQUIRED" +@pytest.mark.parametrize("startup,tool", [(120, 300), (1.5, 2.5)]) +def test_server_timeouts_survive_bridge_adaptation_and_reload(startup, tool) -> None: + service = registry() + created = service.create( + request( + secret_environment_keys=[], + startup_timeout_seconds=startup, + tool_timeout_seconds=tool, + ) + ) + service.trust(created.server_id, created.command_digest) + try: + tested = service.test(created.server_id) + assert tested.last_test_succeeded is True + assert tested.startup_timeout_seconds == startup + assert tested.tool_timeout_seconds == tool + restored = registry().get(created.server_id) + assert restored.startup_timeout_seconds == startup + assert restored.tool_timeout_seconds == tool + finally: + service.shutdown() + + def test_enable_requires_successful_test_and_update_checks_version() -> None: service = registry() created = service.create(request(secret_environment_keys=[])) @@ -307,15 +509,18 @@ def test_lifecycle_operations_are_serialized_and_tool_names_are_isolated() -> No service.test(server.server_id) with ThreadPoolExecutor(max_workers=4) as pool: - enabled = list(pool.map(lambda item: service.enable(item.server_id), servers * 2)) + enabled = list( + pool.map(lambda item: service.enable(item.server_id), servers * 2) + ) assert all(item.enabled for item in enabled) names = [ - item.name - for item in service.tools.definitions() - if item.source == "mcp_server" + item.name for item in service.tools.definitions() if item.source == "mcp_server" ] assert len(names) == len(set(names)) - assert all(any(name.startswith(f"mcp.{item.server_id}.") for name in names) for item in servers) + assert all( + any(name.startswith(f"mcp.{item.server_id}.") for name in names) + for item in servers + ) with ThreadPoolExecutor(max_workers=4) as pool: list(pool.map(lambda item: service.disable(item.server_id), servers * 2)) diff --git a/backend/tests/test_mcp_sse_limits.py b/backend/tests/test_mcp_sse_limits.py new file mode 100644 index 0000000..6c0a091 --- /dev/null +++ b/backend/tests/test_mcp_sse_limits.py @@ -0,0 +1,83 @@ +import json +from contextlib import closing + +import httpx +import pytest + +from app.extensions import mcp + + +class ChunkStream(httpx.SyncByteStream): + def __init__(self, chunks): + self.chunks = chunks + self.bytes_read = 0 + + def __iter__(self): + for chunk in self.chunks: + self.bytes_read += len(chunk) + yield chunk + + +def test_sse_rejects_unterminated_line_before_reading_entire_stream(monkeypatch): + monkeypatch.setattr(mcp, "MAX_MCP_MESSAGE_BYTES", 1024) + stream = ChunkStream([b"x" * 256] * 256) + with ( + closing(httpx.Response(200, stream=stream)) as response, + pytest.raises(mcp.McpBridgeError, match="too large"), + ): + list(mcp._iter_sse(response)) + assert stream.bytes_read == 1280 + + +def test_sse_limits_combined_event_before_partial_line_is_complete(monkeypatch): + monkeypatch.setattr(mcp, "MAX_MCP_MESSAGE_BYTES", 32) + stream = ChunkStream( + [b"data: 123456789\n", b"data: 123456789\n", b"x", b"not-read"] + ) + with ( + closing(httpx.Response(200, stream=stream)) as response, + pytest.raises(mcp.McpBridgeError, match="too large"), + ): + list(mcp._iter_sse(response)) + assert stream.bytes_read == 33 + + +@pytest.mark.parametrize("separator", [b"\n", b"\r", b"\r\n"]) +@pytest.mark.parametrize("chunk_size", [1, 2, 7, 1024]) +def test_sse_preserves_utf8_and_line_endings_across_chunks(separator, chunk_size): + payload = json.dumps( + {"jsonrpc": "2.0", "id": 1, "result": {"text": "中文"}}, ensure_ascii=False + ) + wire = b"\xef\xbb\xbf" + separator.join( + [ + b": heartbeat", + b"event: message", + b"id: replay-1", + ("data: " + payload).encode(), + b"", + b"", + ] + ) + stream = ChunkStream( + [wire[index : index + chunk_size] for index in range(0, len(wire), chunk_size)] + ) + with closing(httpx.Response(200, stream=stream)) as response: + assert list(mcp._iter_sse(response)) == [("message", "replay-1", payload)] + + +def test_sse_event_limit_resets_between_events(monkeypatch): + monkeypatch.setattr(mcp, "MAX_MCP_MESSAGE_BYTES", 16) + with closing( + httpx.Response(200, stream=ChunkStream([b"data: one\n\ndata: two\r\r"])) + ) as response: + assert list(mcp._iter_sse(response)) == [ + ("message", None, "one"), + ("message", None, "two"), + ] + + +def test_sse_preserves_multiline_data_and_final_unterminated_line(): + with closing( + httpx.Response(200, stream=ChunkStream([b"data: first\ndata: last"])) + ) as response: + assert list(mcp._iter_sse(response)) == [("message", None, "first\nlast")] diff --git a/docs/development/独立MCP-Server配置中心开发说明.md b/docs/development/独立MCP-Server配置中心开发说明.md index dcb5346..e0d0a28 100644 --- a/docs/development/独立MCP-Server配置中心开发说明.md +++ b/docs/development/独立MCP-Server配置中心开发说明.md @@ -19,7 +19,44 @@ Streamable HTTP 按 MCP 当前规范实现；SSE 仅用于兼容旧 Server，不 普通配置原子写入 `APP_DATA_DIR/mcp/servers.json`。更新请求必须携带读取到的 `version`；版本过期返回 `409 MCP_SERVER_VERSION_CONFLICT`，避免多个页面互相覆盖。改变 Transport、命令、URL、Header、环境变量或权限后，旧授权和测试结果立即失效。 -Secret 使用带类型和键名哈希的 `mcp.*` 内部 ID 写入 Fernet 凭据存储。API 只返回环境变量或 Header 是否配置，绝不返回明文。删除 Server 或移除 Secret 键会同步清理密文。前端密码框提交后立即清空，不写入 JSON 编辑器、localStorage、普通配置或日志。 +`backend/data/mcp/` 是本机运行数据，包含连接配置、授权状态和第三方进程工作目录，不属于团队共享配置。`.gitignore` 忽略整个目录以及 `server.json`、`servers.json` 文件名；不得强制添加到 Git。提交前检查暂存文件清单，不要将本地密钥、连接配置或运行数据推送到远程。 + +Secret 使用带类型和键名哈希的 `mcp.*` 内部 ID 写入 Fernet 凭据存储。查询 API 只返回环境变量或 Header 是否配置，不返回明文。删除 Server 或移除 Secret 键会同步清理密文。前端密码框提交后立即清空；用户主动粘贴到 JSON 的密钥仅在当前编辑会话中暂存，解析后从 JSON 中移除，不写入 localStorage、普通配置或日志。 + +环境变量的凭据 ID 使用区分大小写的 v2 名称规则，Header ID 保持大小写不敏感。更新配置按凭据 ID 的差集删除密文，因此 `Authorization` 改为 `authorization` 不会丢失认证信息。启动时对无歧义的旧环境变量凭据原子迁移密文，不覆盖新 ID 已有的值；若旧配置把 `TOKEN`、`token` 合并存到了同一个 ID，无法推断原来的两个值，会保留旧密文、停用连接并要求重新录入和测试。删除服务器时也会清理这些保留的旧密文。 + +### 2.1 JSON 导入与 API Key 填写 + +前端支持 NotesAgent 完整/精简配置、单个 `command / args / env` 配置，以及只含一个服务器的 `mcpServers` 包装。JSON 与表单之间切换会补齐数组、对象及超时默认值，并校验字段类型。批量导入暂不支持；后端配置接口仍只接收 NotesAgent DTO，兼容转换发生在前端。 + +可以先声明 `secret_environment_keys`，保存后在服务器卡片的密码框填写密钥；也可以把密钥放进 JSON 的 `environment` 或通用配置的 `env`。前端会将已声明的敏感变量，以及名称含 API Key、Token、Secret、Password、Authorization、Cookie、Credential 的常见字段拆出：普通配置请求只包含键名，密钥另经 Secret API 加密保存。其他敏感字段必须显式声明，不能只依赖名称识别；命令与参数中不要携带密钥。 + +例如 MiniMax 的输入结构如下，占位值需在自己的本地页面替换，不要把真实密钥贴进聊天或提交到 Git： + +```json +{ + "name": "MiniMax Coding Plan", + "command": "uvx", + "args": ["--index-url", "https://pypi.tuna.tsinghua.edu.cn/simple", "--with", "mcp<2", "minimax-coding-plan-mcp", "-y"], + "environment": { + "MINIMAX_API_HOST": "https://api.minimaxi.com", + "MINIMAX_API_KEY": "<在本地填入新密钥>" + }, + "secret_environment_keys": ["MINIMAX_API_KEY"], + "startup_timeout_seconds": 120, + "tool_timeout_seconds": 300 +} +``` + +旧版前端将 `environment.MINIMAX_API_KEY` 与 `secret_environment_keys` 原样一起发送，触发后端“普通与敏感变量不可同名”的校验。这是配置保存失败，不是模型服务返回的鉴权失败。现在在前端拆分两类请求，后端仍保留互斥校验。 + +导入兼容规则：`env` 转为 `environment`；`timeout` 作为启动超时；`sse_read_timeout` 作为工具等待上限，不保留其原客户端 SSE 读取超时语义。启动超时范围为 1–120 秒，工具超时为 1–300 秒。URL 必须是纯地址，不能使用 Markdown 链接，JSON 中不能包含 `\_` 这样的非法转义。 + +另一个已修复的失败原因是运行时适配层复用了 `PluginBackend` 的整数超时与 60 秒启动上限，导致合法的 120 秒或小数超时配置在保存返回、读取或测试时失败。独立 Server 现在使用专门的 Bridge 适配模型，保留自己的浮点超时范围，不改变 Plugin 清单原有约束。已有的 120 秒记录可直接读取，无需删库重建。 + +配置保存成功但后续 Secret 写入失败时，窗口保留服务器 ID、新版本和未写入的密钥。点击保存会更新同一服务器并重试，不重复创建记录；取消会清除未保存密钥，已经保存的服务器和凭据不会回滚。错误信息显示在配置窗口内。保存配置不会自动运行第三方进程，仍需确认、测试和启用。 + +暂存的 Header Secret 与已保存凭据使用一致的大小写规则：将 `Authorization` 改为 `authorization` 不会丢弃尚未保存的值，提交时采用当前声明名。重新输入同一 Header 的值会覆盖旧草稿；真正删除声明才清除草稿。环境变量仍区分大小写，不会把 `TOKEN` 的草稿转交给 `token`。 跨 Registry 与凭据存储的删除以“失败后可重试”为顺序约束：先原子清理密文，再提交新版本或删除 Registry 记录。凭据存储失败时保留原版本和 Server 记录，避免出现返回 500 但配置已提交、版本无法重试或密文失去清理入口的状态。 @@ -34,12 +71,16 @@ Secret 使用带类型和键名哈希的 `mcp.*` 内部 ID 写入 Fernet 凭据 3. 只有当前摘要测试成功，启用操作才会启动长期连接并注册 Tool； 4. 停用、删除、超时或异常退出会注销 Tool 并关闭连接。 -运行期连接异常或 Tool 列表变化时，Registry 会在同一生命周期临界区内标记不可用、注销 Tool，并从 Bridge 移除 Host，不保留后台事件线程或 HTTP Client。Secret 写入和删除可能触发该停用流程，因此对应 API 通过工作线程执行，不阻塞 FastAPI 事件循环。 +运行期连接异常或 Tool 列表变化时，Registry 会在同一生命周期临界区内标记不可用、注销 Tool，并从 Bridge 移除 Host。每次启动分配独立的连接代次；失败回调取得锁后先核对代次，旧连接延迟到达的回调不能停用新连接。停用、测试结束及关闭服务时撤销对应代次。 + +所有独立 MCP API 都通过工作线程执行，包括新增、确认授权和读取接口。虽然部分操作不直接访问网络，但仍可能等待正在测试或启动的连接持有的锁，不能在 FastAPI 事件循环上同步等待。 HTTP Header 中 `Host`、`Content-Type`、`MCP-Session-Id` 等协议保留项不可由配置覆盖。URL 不允许内嵌凭据或 Fragment。旧 SSE 返回的 POST endpoint 必须与初始 URL 同源，防止认证 Header 被转发到其他站点。 启动与 Tool 请求超时会同时应用于业务等待和底层 HTTP 请求；旧 SSE 的 endpoint 等待也使用启动超时。非主动结束的旧 SSE 事件流视为 Host 不可用，宿主随后注销 Tool。注册表加载时逐条校验 Server ID、配置字段、Transport 组合和摘要格式，损坏记录统一返回 `MCP_REGISTRY_INVALID`。 +两种 HTTP Transport 共用有界 SSE 行解析器：按响应字节块检查未完成行及当前事件的累计大小，再扩展缓冲区，不依赖 `iter_lines()` 先缓存完整行。持续无换行输入也会及时触发上限；解析兼容跨块 UTF-8、首行 BOM、LF/CR/CRLF、多行 data 和事件间计数重置。`tests/test_mcp_sse_limits.py` 覆盖这些边界，防止仅在完整行生成后检查大小。 + stdio 命令不经过 Shell，管道、重定向和命令拼接不会被解释。Windows 使用新进程组并通过 `taskkill /T` 回收子树；POSIX 使用独立 session/process group 并向进程组发信号。Python 阶段仍无法提供文件、网络、系统调用或操作系统版本差异下的绝对隔离保证。 `uvx` 模板使用 `--isolated`、明确的 `--from` 和固定包版本。它只能隔离依赖，不能替代安全沙箱。非开发环境仍拒绝启动 stdio Server，并返回 `403 MCP_SANDBOX_REQUIRED`；远程 HTTP Transport 不创建本机子进程，但仍要求摘要确认和成功测试。第三阶段前的 C.5 将冻结 Tauri/Rust 沙箱设计。 @@ -75,10 +116,26 @@ npm test npm run build ``` -后端测试使用无需网络或密钥的 stdio Fixture，以及 `httpx.MockTransport` 驱动的确定性 HTTP/SSE Server Fixture。覆盖摘要授权、Secret 不回显、版本冲突、生产门禁、重启恢复、Streamable HTTP Session/Header/工具调用及旧 SSE 同源校验。前端覆盖模板切换、JSON 校验、Secret 请求期输入、测试失败和删除确认。 +后端测试使用无需网络或真实密钥的 stdio Fixture，以及 `httpx.MockTransport` 驱动的确定性 HTTP/SSE Server Fixture。覆盖摘要授权、Secret 不回显、版本冲突、生产门禁、重启恢复、Streamable HTTP Session/Header/工具调用及旧 SSE 同源校验；新增覆盖旧回调隔离、路由线程卸载、凭据大小写差异与旧密文迁移。前端覆盖模板切换、JSON 默认值与格式兼容、明文拆分、模式切换、部分保存失败重试、取消清理、测试失败和删除确认。此处的 MiniMax 配置转换测试使用假密钥，不等同于真实 MiniMax 网络调用验证。 ## 6. 后续边界 +### 本轮 P1/P2 修复验收 + +| 审阅问题 | 修复方式 | 回归验证 | +| --- | --- | --- | +| P1：新增或授权等待生命周期锁时阻塞事件循环 | 独立 MCP 路由统一交给工作线程 | `test_mcp_lifecycle_lock_contention_keeps_event_loop_responsive`：分别阻塞 create/trust，在锁释放前仍能执行健康检查 | +| P2：旧失败回调误停新连接 | 回调在锁内核对连接代次 | `test_old_failure_callback_cannot_stop_replacement_host`：旧回调排队期间重启连接，释放锁后新连接仍可用，当前代次的失败仍正确停用 | +| P2：精简 JSON 切换模式或编辑保存报错 | 运行时校验、默认值补全、统一转换 | `configuration.spec.ts` 与 `McpServersView.spec.ts`：精简 JSON、编辑版本、模式切换和 Secret 保存重试 | +| P2：Header 大小写改名删除凭据 | 按规范化凭据 ID 而非原始键名计算差集 | `test_header_case_only_rename_preserves_secret` | +| P2：大小写不同的环境变量覆盖同一凭据 | 区分大小写的 v2 ID，带迁移标记的旧密文迁移 | `test_environment_secrets_are_case_sensitive_and_delete_independently` 及 legacy migration 测试，包含删除后不复活旧密钥 | +| P1：SSE 无换行输入在大小校验前无限缓冲 | 在行拼接前校验字节数与事件累计大小 | `test_mcp_sse_limits.py`，包括小块持续输入和跨块换行 | +| P2：Header 大小写改名丢失未保存密钥 | 草稿使用规范化名称匹配，并重新绑定当前声明名 | `configuration.spec.ts` 与页面保存回归测试 | + +这些修复不放宽 stdio 的 JSON-RPC 校验。第三方程序向 stdout 打印普通日志造成的握手失败，应由服务端调整输出或使用不打印日志的启动入口处理。 + +### 后续工作 + - 增加真实第三方 Server 的兼容矩阵；确定性 Fixture 只能证明宿主协议行为，不能代表所有实现兼容； - C.5 在第二阶段开发与测试完成后、第三阶段桌面端实现前冻结沙箱 Contract； - 第三阶段将 stdio 进程创建和 Secret 托管迁移至 Tauri/Rust Host 与 Stronghold/系统 Keychain。 diff --git a/frontend/src/features/mcp/McpServersView.spec.ts b/frontend/src/features/mcp/McpServersView.spec.ts index 659295f..235999c 100644 --- a/frontend/src/features/mcp/McpServersView.spec.ts +++ b/frontend/src/features/mcp/McpServersView.spec.ts @@ -92,4 +92,78 @@ describe('McpServersView', () => { expect(confirm).toHaveBeenCalledWith(expect.stringContaining('旧测试与授权会失效')) expect(service.updateMcpServer).toHaveBeenCalled() }) + + it('saves an environment API key via the encrypted endpoint, not the config body', async () => { + const wrapper = await render() + vi.mocked(service.createMcpServer).mockResolvedValue({ ...server, server_id: 'new-server' }) + vi.mocked(service.putMcpServerSecret).mockResolvedValue({}) + await wrapper.findAll('button').find(button => button.text() === '新增服务器')!.trigger('click') + await wrapper.findAll('button').find(button => button.text() === 'JSON 配置')!.trigger('click') + await wrapper.get('.json-editor').setValue(JSON.stringify({ command: 'uvx', environment: { MINIMAX_API_KEY: 'synthetic-only' }, secret_environment_keys: ['MINIMAX_API_KEY'] })) + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(service.createMcpServer).toHaveBeenCalledWith(expect.objectContaining({ environment: {}, secret_environment_keys: ['MINIMAX_API_KEY'] })) + expect(JSON.stringify(vi.mocked(service.createMcpServer).mock.calls)).not.toContain('synthetic-only') + expect(service.putMcpServerSecret).toHaveBeenCalledWith('new-server', 'MINIMAX_API_KEY', 'synthetic-only', 'environment') + expect(wrapper.find('.modal-backdrop').exists()).toBe(false) + }) + + it('retains imported keys over mode switches and retries partial saves without duplicates', async () => { + const wrapper = await render() + vi.mocked(service.createMcpServer).mockResolvedValue({ ...server, server_id: 'new-server', version: 1 }) + vi.mocked(service.updateMcpServer).mockResolvedValue({ ...server, server_id: 'new-server', version: 2 }) + vi.mocked(service.putMcpServerSecret).mockRejectedValueOnce(new Error('credential store unavailable')).mockResolvedValue({}) + await wrapper.findAll('button').find(button => button.text() === '新增服务器')!.trigger('click') + await wrapper.findAll('button').find(button => button.text() === 'JSON 配置')!.trigger('click') + await wrapper.get('.json-editor').setValue(JSON.stringify({ command: 'uvx', env: { API_KEY: 'retry-value' } })) + await wrapper.findAll('button').find(button => button.text() === '表单配置')!.trigger('click') + expect(wrapper.text()).toContain('已识别 1 项密钥') + await wrapper.findAll('button').find(button => button.text() === 'JSON 配置')!.trigger('click') + expect((wrapper.get('.json-editor').element as HTMLTextAreaElement).value).not.toContain('retry-value') + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(wrapper.get('.modal-card [role="alert"]').text()).toContain('服务器配置已保存，但密钥保存失败') + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(service.createMcpServer).toHaveBeenCalledTimes(1) + expect(service.updateMcpServer).toHaveBeenCalledWith('new-server', expect.objectContaining({ version: 1 })) + expect(service.putMcpServerSecret).toHaveBeenCalledTimes(2) + expect(wrapper.find('.modal-backdrop').exists()).toBe(false) + }) + + it('clears staged keys on cancel and accepts minimal JSON while editing', async () => { + const wrapper = await render([server]) + await wrapper.findAll('button').find(button => button.text() === '新增服务器')!.trigger('click') + await wrapper.findAll('button').find(button => button.text() === 'JSON 配置')!.trigger('click') + await wrapper.get('.json-editor').setValue('{"command":"uvx","env":{"API_KEY":"cancelled-value"}}') + await wrapper.findAll('button').find(button => button.text() === '表单配置')!.trigger('click') + await wrapper.findAll('button').find(button => button.text() === '取消')!.trigger('click') + await wrapper.findAll('button').find(button => button.text().includes('编辑'))!.trigger('click') + await wrapper.findAll('button').find(button => button.text() === 'JSON 配置')!.trigger('click') + await wrapper.get('.json-editor').setValue('{"name":"Minimal","url":"https://example.test/mcp"}') + vi.mocked(service.updateMcpServer).mockResolvedValue(server) + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(service.updateMcpServer).toHaveBeenCalledWith('server-1', expect.objectContaining({ version: 2, headers: {}, args: [] })) + expect(service.putMcpServerSecret).not.toHaveBeenCalled() + }) + + it('saves an imported Header secret after a case-only declaration rename', async () => { + const wrapper = await render() + vi.mocked(service.createMcpServer).mockResolvedValue({ ...server, secret_headers: { authorization: false } }) + vi.mocked(service.putMcpServerSecret).mockResolvedValue({}) + await wrapper.findAll('button').find(button => button.text() === '新增服务器')!.trigger('click') + await wrapper.findAll('button').find(button => button.text() === 'JSON 配置')!.trigger('click') + await wrapper.get('.json-editor').setValue(JSON.stringify({ url: 'https://example.test/mcp', headers: { Authorization: 'synthetic-draft' } })) + await wrapper.findAll('button').find(button => button.text() === '表单配置')!.trigger('click') + await wrapper.get('textarea[placeholder="Authorization"]').setValue('authorization') + await wrapper.findAll('button').find(button => button.text() === 'JSON 配置')!.trigger('click') + expect(wrapper.text()).toContain('已识别 1 项密钥') + expect((wrapper.get('.json-editor').element as HTMLTextAreaElement).value).not.toContain('synthetic-draft') + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(service.createMcpServer).toHaveBeenCalledWith(expect.objectContaining({ headers: {}, secret_header_keys: ['authorization'] })) + expect(service.putMcpServerSecret).toHaveBeenCalledWith('server-1', 'authorization', 'synthetic-draft', 'header') + expect(wrapper.find('.modal-backdrop').exists()).toBe(false) + }) }) diff --git a/frontend/src/features/mcp/McpServersView.vue b/frontend/src/features/mcp/McpServersView.vue index 1f9380f..89cff71 100644 --- a/frontend/src/features/mcp/McpServersView.vue +++ b/frontend/src/features/mcp/McpServersView.vue @@ -4,8 +4,7 @@ import { Connection, Delete, EditPen, Plus, Refresh, VideoPlay } from '@element- import AppIcon from '@/components/common/AppIcon.vue' import type { McpServer, McpServerInput, McpServerTransport } from '@/contracts' import * as service from '@/services/mcpServerService' - -type SecretKind = 'environment' | 'header' +import { emptyMcpConfig, mergeImportedSecrets, normalizeMcpConfig, parseMcpJson, type ImportedSecret, type SecretKind } from './configuration' const servers = ref([]) const busy = ref('') @@ -22,18 +21,11 @@ const secretHeaderKeysText = ref('') const permissionsText = ref('') const rawConfig = ref('') const secretDrafts = reactive>({}) -const form = reactive(emptyForm()) +const form = reactive(emptyMcpConfig()) +const importedSecrets = ref([]) const dialogTitle = computed(() => editingId.value ? '编辑 MCP 服务器' : '新增 MCP 服务器') -function emptyForm(): McpServerInput { - return { - name: '', transport: 'stdio', command: '', args: [], url: null, headers: {}, - environment: {}, secret_environment_keys: [], secret_header_keys: [], permissions: [], - startup_timeout_seconds: 15, tool_timeout_seconds: 30, - } -} - async function load() { error.value = '' try { servers.value = await service.listMcpServers() } @@ -41,7 +33,7 @@ async function load() { } function resetEditor(input: McpServerInput) { - Object.assign(form, input) + Object.assign(form, emptyMcpConfig(), { version: undefined }, input) argsText.value = input.args.join('\n') environmentText.value = JSON.stringify(input.environment, null, 2) headersText.value = JSON.stringify(input.headers, null, 2) @@ -53,13 +45,19 @@ function resetEditor(input: McpServerInput) { } function openCreate() { + if (busy.value) return + error.value = '' + importedSecrets.value = [] editingId.value = null editingOriginal.value = null - resetEditor(emptyForm()) + resetEditor(emptyMcpConfig()) dialogOpen.value = true } function openEdit(server: McpServer) { + if (busy.value) return + error.value = '' + importedSecrets.value = [] editingId.value = server.server_id editingOriginal.value = server resetEditor({ @@ -109,38 +107,69 @@ function formPayload(): McpServerInput { } } -function payload(): McpServerInput { - if (editorMode.value === 'form') return formPayload() - let parsed: unknown - try { parsed = JSON.parse(rawConfig.value) } catch { throw new Error('服务器配置不是有效 JSON') } - if (!parsed || Array.isArray(parsed) || typeof parsed !== 'object') throw new Error('服务器配置必须是 JSON 对象') - const value = parsed as McpServerInput - if (editingId.value) value.version = form.version - return value +function payload(requireConnection = true): McpServerInput { + const { config, secrets } = editorMode.value === 'form' + ? normalizeMcpConfig(formPayload(), '', requireConnection) : parseMcpJson(rawConfig.value, form.name, requireConnection) + // Keep only still-declared drafts. A mode switch must not discard imported keys, + // and editing the declaration must not later send a removed key to the Secret API. + importedSecrets.value = mergeImportedSecrets(config, importedSecrets.value, secrets) + if (editingId.value) config.version = form.version + if (editorMode.value === 'json') rawConfig.value = JSON.stringify(config, null, 2) + else { + environmentText.value = JSON.stringify(config.environment, null, 2) + headersText.value = JSON.stringify(config.headers, null, 2) + secretKeysText.value = config.secret_environment_keys.join('\n') + secretHeaderKeysText.value = config.secret_header_keys.join('\n') + } + return config } function switchMode(mode: 'form' | 'json') { try { if (mode === editorMode.value) return - if (mode === 'json') rawConfig.value = JSON.stringify(formPayload(), null, 2) - else resetEditor(payload()) + error.value = '' + if (mode === 'json') rawConfig.value = JSON.stringify(payload(false), null, 2) + else resetEditor(payload(false)) editorMode.value = mode } catch (cause) { error.value = message(cause, '配置转换失败') } } async function save() { + if (busy.value) return + let saved: McpServer | undefined try { + error.value = '' const input = payload() if (!input.name || (input.transport === 'stdio' ? !input.command : !input.url)) throw new Error('请填写服务器名称和连接地址') if (editingOriginal.value && executionChanged(editingOriginal.value, input) && !confirm('连接命令、地址或认证配置已变化，保存后旧测试与授权会失效。是否保存？')) return busy.value = 'save' - editingId.value ? await service.updateMcpServer(editingId.value, input) : await service.createMcpServer(input) - dialogOpen.value = false + saved = editingId.value ? await service.updateMcpServer(editingId.value, input) : await service.createMcpServer(input) + // Commit the returned ID/version before saving secrets so a partial failure can + // retry this server instead of creating a duplicate or sending a stale version. + editingId.value = saved.server_id + editingOriginal.value = saved + resetEditor({ ...input, version: saved.version }) + for (const item of [...importedSecrets.value]) { + await service.putMcpServerSecret(saved.server_id, item.key, item.value, item.kind) + importedSecrets.value = importedSecrets.value.filter(candidate => candidate !== item) + } + closeEditor() await load() - } catch (cause) { error.value = message(cause, '保存失败') } + } catch (cause) { + if (saved) await load() + error.value = `${saved ? '服务器配置已保存，但密钥保存失败；可点击保存重试。' : ''}${message(cause, '保存失败')}` + } finally { busy.value = '' } } +function closeEditor() { + importedSecrets.value = [] + rawConfig.value = '' + environmentText.value = '{}' + headersText.value = '{}' + dialogOpen.value = false +} + function executionChanged(server: McpServer, input: McpServerInput) { const sortedEntries = (value: Record) => Object.entries(value).sort(([left], [right]) => left.localeCompare(right)) const current = [ @@ -212,9 +241,12 @@ onMounted(load) -
+
-

{{ dialogTitle }}

+
+

{{ dialogTitle }}

+ +
已识别 {{ importedSecrets.length }} 项密钥，保存时将单独加密，不会写入普通服务器配置；取消将清除未保存密钥。
- -
+ +
+