From c8369e37f49a5d3633f357abe9d01f6b4f2149df Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 7 Aug 2026 07:34:35 -0700 Subject: [PATCH] feat(mcp): trust-tier gating for write-capable MCP tools via readOnlyHint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a per-server `trust: full|untrusted` config key (mcp_servers..trust). On an untrusted server, every write-capable tool call — any tool whose discovery-time annotations do not carry readOnlyHint=True — routes through the existing approval surface (tools.approval.request_elicitation_consent, same lazy-import + surface-routing pattern the MCP elicitation handler uses) before the RPC fires. Denied/cancelled/errored approvals fail closed: the RPC never runs, including the lazy first-use server spawn. Design points: - Classification happens at CALL TIME from metadata captured at DISCOVERY (_record_tool_trust_metadata in _register_server_tools and the lazy cache-registration path). No toolset/schema mutation, so the toolset stays byte-stable and prompt caching is preserved. - readOnlyHint is a server-supplied HINT: on an untrusted server a lying server can at most skip approval for tools it claims read-only — it can never widen access. Trust tiering itself is operator config. - Missing/malformed annotations => write-capable (fail closed). - Unrecognized trust values => untrusted (fail closed); missing key => full (backward compatible, documented in mcp-config-reference). - The schema cache now persists readOnlyHint so lazy-registered servers gate identically on next startup without spawning. Tests: tests/tools/test_mcp_trust_gating.py (11 tests, TDD red->green): approval invoked + accept proceeds, deny/cancel blocks RPC, readOnlyHint =true skips gate, trusted/unconfigured servers skip gate, explicit readOnlyHint=false gated, approval exception fails closed, trust normalization, discovery-time capture (SDK objects and cached dicts). Ported from: cloudflare-os classifyTool() (Apache-2.0), corroborated by Claude Cowork (idea-level). --- tests/tools/test_mcp_trust_gating.py | 247 ++++++++++++++++++ tools/mcp_tool.py | 180 +++++++++++++ .../docs/reference/mcp-config-reference.md | 1 + .../current/reference/mcp-config-reference.md | 1 + 4 files changed, 429 insertions(+) create mode 100644 tests/tools/test_mcp_trust_gating.py diff --git a/tests/tools/test_mcp_trust_gating.py b/tests/tools/test_mcp_trust_gating.py new file mode 100644 index 0000000000000..e38866cc6ca5e --- /dev/null +++ b/tests/tools/test_mcp_trust_gating.py @@ -0,0 +1,247 @@ +"""Tests for MCP tool trust-tier gating via readOnlyHint annotations. + +Security boundary under test: write-capable MCP tools (anything whose +``readOnlyHint`` annotation is not exactly ``True``) on servers configured +``trust: untrusted`` must route through the existing dangerous-approval +path before the RPC fires. Read-only tools and tools on trusted servers +pass straight through. + +Adversarial notes encoded in these tests: +- ``readOnlyHint`` is a HINT supplied by the (potentially hostile) server. + It can only ever RELAX gating on a server the operator already marked + untrusted; the trust tier itself is operator-side config, so a lying + server can at worst skip approval for a tool it claims is read-only — + which is why the trust key is per-server and gating is fail-closed for + missing/unknown metadata. +- Missing annotations ⇒ write-capable (fail closed). +- Unknown/garbage ``trust`` values ⇒ treated as untrusted (fail closed). +""" + +import asyncio +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from tools import mcp_tool + + +class _FakeContentBlock: + def __init__(self, text: str, block_type: str = "text"): + self.text = text + self.type = block_type + + +class _FakeCallToolResult: + def __init__(self, content, is_error=False, structuredContent=None): + self.content = content + self.isError = is_error + self.structuredContent = structuredContent + + +def _fake_run_on_mcp_loop(coro_or_factory, timeout=30): + coro = coro_or_factory() if callable(coro_or_factory) else coro_or_factory + loop = asyncio.new_event_loop() + try: + async def _install_lock_and_run(): + for srv in list(mcp_tool._servers.values()): + if getattr(srv, "_rpc_lock", None) is None: + srv._rpc_lock = asyncio.Lock() + return await coro + return loop.run_until_complete(_install_lock_and_run()) + finally: + loop.close() + + +@pytest.fixture +def fake_session(): + """Patch a fake connected server + MCP loop; yield its session mock.""" + session = MagicMock() + session.call_tool = AsyncMock( + return_value=_FakeCallToolResult(content=[_FakeContentBlock("ok")]) + ) + server = SimpleNamespace(session=session, _rpc_lock=None) + with patch.dict(mcp_tool._servers, {"srv": server}), \ + patch("tools.mcp_tool._run_on_mcp_loop", + side_effect=_fake_run_on_mcp_loop), \ + patch.dict(mcp_tool._server_error_counts, {}, clear=True): + yield session + + +@pytest.fixture(autouse=True) +def _clean_trust_state(): + """Isolate the module-level trust metadata between tests.""" + with patch.dict(mcp_tool._server_trust_levels, {}, clear=True), \ + patch.dict(mcp_tool._tool_read_only_hints, {}, clear=True): + yield + + +def _set_trust(server: str, trust: str): + mcp_tool._server_trust_levels[server] = trust + + +def _set_read_only(server: str, tool: str, value: bool): + mcp_tool._tool_read_only_hints.setdefault(server, {})[tool] = value + + +class TestTrustGateAtCallTime: + """The handler preamble consults the approval path when required.""" + + def test_write_capable_on_untrusted_server_requires_approval( + self, fake_session + ): + """Approval consulted; 'accept' lets the RPC through.""" + _set_trust("srv", "untrusted") + # No readOnlyHint recorded for delete_repo → write-capable. + handler = mcp_tool._make_tool_handler("srv", "delete_repo", 30.0) + with patch( + "tools.approval.request_elicitation_consent", + return_value="accept", + ) as consent: + raw = handler({"repo": "x"}) + consent.assert_called_once() + assert json.loads(raw) == {"result": "ok"} + fake_session.call_tool.assert_awaited_once() + + def test_denied_approval_blocks_rpc(self, fake_session): + """'decline' blocks the call — the RPC must never fire.""" + _set_trust("srv", "untrusted") + handler = mcp_tool._make_tool_handler("srv", "delete_repo", 30.0) + with patch( + "tools.approval.request_elicitation_consent", + return_value="decline", + ): + raw = handler({"repo": "x"}) + fake_session.call_tool.assert_not_awaited() + assert "error" in json.loads(raw) + assert "did not approve" in json.loads(raw)["error"] + + def test_read_only_tool_on_untrusted_server_skips_approval( + self, fake_session + ): + """readOnlyHint=True tools pass without consulting approval.""" + _set_trust("srv", "untrusted") + _set_read_only("srv", "list_repos", True) + handler = mcp_tool._make_tool_handler("srv", "list_repos", 30.0) + with patch( + "tools.approval.request_elicitation_consent" + ) as consent: + raw = handler({}) + consent.assert_not_called() + assert json.loads(raw) == {"result": "ok"} + + def test_trusted_server_skips_approval_for_write_tools( + self, fake_session + ): + """trust: full (and the default) never consults approval.""" + _set_trust("srv", "full") + handler = mcp_tool._make_tool_handler("srv", "delete_repo", 30.0) + with patch( + "tools.approval.request_elicitation_consent" + ) as consent: + raw = handler({"repo": "x"}) + consent.assert_not_called() + assert json.loads(raw) == {"result": "ok"} + + def test_unconfigured_server_defaults_to_full_trust(self, fake_session): + """Backward compat: servers with no trust key behave as before.""" + handler = mcp_tool._make_tool_handler("srv", "delete_repo", 30.0) + with patch( + "tools.approval.request_elicitation_consent" + ) as consent: + raw = handler({"repo": "x"}) + consent.assert_not_called() + assert json.loads(raw) == {"result": "ok"} + + def test_read_only_false_hint_is_gated(self, fake_session): + """An explicit readOnlyHint=False is write-capable.""" + _set_trust("srv", "untrusted") + _set_read_only("srv", "write_file", False) + handler = mcp_tool._make_tool_handler("srv", "write_file", 30.0) + with patch( + "tools.approval.request_elicitation_consent", + return_value="decline", + ) as consent: + handler({"path": "/etc/passwd"}) + consent.assert_called_once() + fake_session.call_tool.assert_not_awaited() + + def test_approval_exception_fails_closed(self, fake_session): + """Any exception in the consent path blocks the call.""" + _set_trust("srv", "untrusted") + handler = mcp_tool._make_tool_handler("srv", "delete_repo", 30.0) + with patch( + "tools.approval.request_elicitation_consent", + side_effect=RuntimeError("approval backend down"), + ): + raw = handler({"repo": "x"}) + fake_session.call_tool.assert_not_awaited() + assert "error" in json.loads(raw) + + +class TestTrustNormalization: + def test_unknown_trust_value_treated_as_untrusted(self): + """Garbage trust strings fail closed to untrusted.""" + assert mcp_tool._normalize_server_trust("banana") == "untrusted" + + def test_known_values(self): + assert mcp_tool._normalize_server_trust("full") == "full" + assert mcp_tool._normalize_server_trust("UNTRUSTED") == "untrusted" + assert mcp_tool._normalize_server_trust(" Full ") == "full" + # Missing key → default full (backward compatible; documented). + assert mcp_tool._normalize_server_trust(None) == "full" + + +class TestAnnotationCaptureAtDiscovery: + """_register_server_tools records trust + readOnlyHint metadata.""" + + def _make_tool(self, name, annotations=None): + return SimpleNamespace( + name=name, description="", inputSchema=None, + annotations=annotations, + ) + + def test_registration_records_hints_and_trust(self): + from tools.registry import ToolRegistry + + server = mcp_tool.MCPServerTask("srv") + server.session = MagicMock() + server._tools = [ + self._make_tool( + "list_repos", SimpleNamespace(readOnlyHint=True) + ), + self._make_tool( + "delete_repo", SimpleNamespace(readOnlyHint=False) + ), + self._make_tool("no_annotations", None), + ] + config = { + "trust": "untrusted", + "tools": {"resources": False, "prompts": False}, + } + with patch("tools.registry.registry", ToolRegistry()), \ + patch("tools.mcp_tool._track_mcp_tool_server"): + mcp_tool._register_server_tools("srv", server, config) + + assert mcp_tool._server_trust_levels["srv"] == "untrusted" + hints = mcp_tool._tool_read_only_hints["srv"] + assert hints.get("list_repos") is True + # Anything not exactly True is write-capable. + assert not hints.get("delete_repo") + assert not hints.get("no_annotations") + + def test_dict_annotations_supported(self): + """Cached/JSON annotations arrive as plain dicts.""" + assert mcp_tool._annotation_read_only_hint( + SimpleNamespace(annotations={"readOnlyHint": True}) + ) is True + assert mcp_tool._annotation_read_only_hint( + SimpleNamespace(annotations={"readOnlyHint": "yes"}) + ) is False # non-bool truthy → NOT read-only (hint must be True) + assert mcp_tool._annotation_read_only_hint( + SimpleNamespace(annotations=None) + ) is False + assert mcp_tool._annotation_read_only_hint( + SimpleNamespace() + ) is False diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index df617b8be256c..ad962558a1e52 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -3712,6 +3712,150 @@ _server_breaker_opened_at: Dict[str, float] = {} _CIRCUIT_BREAKER_THRESHOLD = 3 _CIRCUIT_BREAKER_COOLDOWN_SEC = 60.0 +# --------------------------------------------------------------------------- +# Trust-tier gating state (per-server trust + per-tool readOnlyHint). +# +# ``trust: full | untrusted`` is a per-server key in the MCP server config +# (config.yaml → mcp_servers..trust). On an ``untrusted`` server, +# every WRITE-CAPABLE tool call routes through the existing dangerous- +# approval surface before the RPC fires. A tool is write-capable unless its +# discovery-time ``annotations.readOnlyHint`` is exactly ``True`` +# (missing/malformed annotations fail closed to write-capable). +# +# Security model (read this before changing defaults): +# - ``readOnlyHint`` is a HINT supplied by the server itself. A hostile +# server can lie. That is precisely why the gate is tiered per-server by +# OPERATOR config: on an untrusted server the hint can only ever exempt +# tools the server claims are read-only — the worst a lie buys is +# skipping approval for calls the operator was already warned about when +# they marked the server untrusted. It can never widen access on top of +# the approval a write-capable tool would otherwise need. +# - Default trust for servers with NO ``trust`` key is ``full`` (gate off) +# for backward compatibility — existing configs keep working unchanged. +# Operators opt servers into gating explicitly with ``trust: untrusted``. +# - Any unrecognized ``trust`` value normalizes to ``untrusted`` +# (fail closed): a typo must never silently disable the gate. +# +# Classification happens at CALL TIME from data captured at DISCOVERY — +# no toolset or schema mutation, so the conversation's toolset stays +# byte-stable and prompt caching is preserved. +_server_trust_levels: Dict[str, str] = {} +_tool_read_only_hints: Dict[str, Dict[str, bool]] = {} + +_TRUST_FULL = "full" +_TRUST_UNTRUSTED = "untrusted" + + +def _normalize_server_trust(value: Any) -> str: + """Normalize a config ``trust`` value to ``full`` or ``untrusted``. + + Missing (None) → ``full`` (backward-compatible default, documented + above). Any string other than the two known tiers → ``untrusted``: + a misspelled tier must fail closed, never silently disable gating. + """ + if value is None: + return _TRUST_FULL + text = str(value).strip().lower() + if text == _TRUST_FULL: + return _TRUST_FULL + if text == _TRUST_UNTRUSTED: + return _TRUST_UNTRUSTED + logger.warning( + "MCP trust: unrecognized trust value %r — treating as 'untrusted' " + "(valid values: full, untrusted)", value, + ) + return _TRUST_UNTRUSTED + + +def _annotation_read_only_hint(mcp_tool: Any) -> bool: + """Return True only when the tool's annotations carry readOnlyHint=True. + + Accepts both SDK annotation objects (attribute access) and plain dicts + (schema-cache JSON). Anything else — missing annotations, missing key, + non-bool truthy values — is False: unknown metadata means the tool must + be treated as write-capable. + """ + annotations = getattr(mcp_tool, "annotations", None) + if annotations is None: + return False + if isinstance(annotations, dict): + hint = annotations.get("readOnlyHint") + else: + hint = getattr(annotations, "readOnlyHint", None) + return hint is True + + +def _record_tool_trust_metadata( + server_name: str, config: dict, tools: List[Any] +) -> None: + """Capture per-server trust and per-tool readOnlyHint at discovery.""" + with _lock: + _server_trust_levels[server_name] = _normalize_server_trust( + (config or {}).get("trust") + ) + hints = _tool_read_only_hints.setdefault(server_name, {}) + for tool in tools: + name = getattr(tool, "name", None) + if name: + hints[name] = _annotation_read_only_hint(tool) + + +def _trust_gate_check(server_name: str, tool_name: str) -> Optional[str]: + """Consult the approval path for write-capable tools on untrusted servers. + + Returns None when the call may proceed, or an error string (already + formatted via ``tool_error``) when the call is blocked. Fail-closed: + approval-system errors block the call. + """ + trust = _server_trust_levels.get(server_name, _TRUST_FULL) + if trust != _TRUST_UNTRUSTED: + return None + if _tool_read_only_hints.get(server_name, {}).get(tool_name) is True: + return None + + # Lazy import mirrors the elicitation handler's pattern: tools.approval + # routes the prompt to whichever surface owns the session (CLI, TUI, + # Telegram, Slack, ...) and normalizes the answer. + try: + from tools.approval import request_elicitation_consent + + answer = request_elicitation_consent( + ( + f"MCP tool '{tool_name}' on UNTRUSTED server " + f"'{server_name}' wants to run. This tool is write-capable " + f"(no readOnlyHint=true annotation) and may modify external " + f"state." + ), + ( + f"Server '{server_name}' is configured 'trust: untrusted'. " + f"Approve to run '{tool_name}' once, or deny to block it." + ), + surface=f"mcp-trust/{server_name}", + ) + except Exception as exc: + logger.error( + "MCP trust gate: approval check failed for %s.%s: %s", + server_name, tool_name, exc, exc_info=True, + ) + return tool_error( + f"MCP tool '{tool_name}' on untrusted server '{server_name}' " + f"was blocked: the approval system was unavailable " + f"(fail-closed)." + ) + + if answer == "accept": + return None + logger.info( + "MCP trust gate: user %s '%s' on untrusted server '%s'", + "cancelled" if answer == "cancel" else "denied", + tool_name, server_name, + ) + return tool_error( + f"The user did not approve running write-capable MCP tool " + f"'{tool_name}' on untrusted server '{server_name}'. The command " + f"was NOT run. Do not retry without explicit user direction." + ) + def _bump_server_error(server_name: str) -> None: """Increment the consecutive-failure count for ``server_name``. @@ -4973,6 +5117,14 @@ def _make_tool_handler(server_name: str, tool_name: str, tool_timeout: float): """ def _handler(args: dict, **kwargs) -> str: + # Trust-tier gate (security boundary): write-capable tools on + # servers configured ``trust: untrusted`` must be approved by the + # user before ANY transport work happens — including the lazy + # first-use spawn below. A denied call never touches the server. + gate_error = _trust_gate_check(server_name, tool_name) + if gate_error is not None: + return gate_error + # Circuit breaker: if this server has failed too many times # consecutively, short-circuit with a clear message so the model # stops retrying and uses alternative approaches (#10447). @@ -5958,6 +6110,12 @@ def _register_server_tools(name: str, server: MCPServerTask, config: dict) -> Li check_fn = _make_check_fn(name) candidates: List[dict] = [] + # Trust-tier metadata (security boundary): capture the server's + # configured trust tier and each tool's readOnlyHint annotation NOW, + # at discovery, so the call-time gate in _make_tool_handler classifies + # from data we control rather than re-reading server-supplied state. + _record_tool_trust_metadata(name, config, server._tools) + for mcp_tool in server._tools: if not _should_register(mcp_tool.name): logger.debug( @@ -6110,6 +6268,12 @@ def _register_server_tools(name: str, server: MCPServerTask, config: dict) -> Li "name": mcp_tool.name, "description": mcp_tool.description or "", "inputSchema": schema_obj if isinstance(schema_obj, dict) else {}, + # Persist the trust-relevant annotation so the lazy + # (cache-registered) path gates identically on next + # startup without spawning the server. + "annotations": { + "readOnlyHint": _annotation_read_only_hint(mcp_tool), + }, }) utility_payload = [ {"schema": entry["schema"], "handler_key": entry["handler_key"]} @@ -6172,6 +6336,22 @@ def _register_from_cache_sync(name: str, config: dict, entry: dict) -> List[str] return True check_fn = _make_check_fn(name) + # Trust-tier metadata for the lazy path: the cached manifest carries + # each tool's readOnlyHint (written by the live discovery path), and + # trust comes from operator config. Recording it before registration + # keeps the call-time gate identical whether the server was spawned + # live or registered from cache. Missing "annotations" in older cache + # files fails closed to write-capable. + cached_tool_objs = [ + SimpleNamespace( + name=raw.get("name"), + annotations=raw.get("annotations") + if isinstance(raw.get("annotations"), dict) else None, + ) + for raw in tools_from_cache_entry(entry) + if isinstance(raw, dict) and raw.get("name") + ] + _record_tool_trust_metadata(name, config, cached_tool_objs) for raw in tools_from_cache_entry(entry): if not isinstance(raw, dict): continue diff --git a/website/docs/reference/mcp-config-reference.md b/website/docs/reference/mcp-config-reference.md index 36b5458ac0dc4..f0b96adfd3a64 100644 --- a/website/docs/reference/mcp-config-reference.md +++ b/website/docs/reference/mcp-config-reference.md @@ -66,6 +66,7 @@ mcp_servers: | `auth` | string | HTTP | Authentication method. Set to `oauth` to enable OAuth 2.1 with PKCE | | `sampling` | mapping | both | Server-initiated LLM request policy (see MCP guide) | | `elicitation` | mapping | both | Server-initiated user-input requests. `enabled` (default `true`) and `timeout` in seconds (default `300`). Form-mode requests route through the approval surface; URL-mode is declined (see MCP guide) | +| `trust` | string | both | Trust tier: `full` (default) or `untrusted`. On an `untrusted` server, every write-capable tool call (any tool without a `readOnlyHint: true` annotation) requires user approval through the standard approval surface before it runs. `readOnlyHint` is a server-supplied *hint* — a lying server can at most skip approval for tools it claims are read-only, never gain extra access — so mark any server you don't fully control as `untrusted`. Unrecognized values are treated as `untrusted` (fail-closed) | ## Environment variable references diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/reference/mcp-config-reference.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/reference/mcp-config-reference.md index 8207a2e2160f8..899b94999aaf4 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/reference/mcp-config-reference.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/reference/mcp-config-reference.md @@ -52,6 +52,7 @@ mcp_servers: | `tools` | mapping | 两者 | 过滤及工具策略 | | `auth` | string | HTTP | 认证方式。设为 `oauth` 可启用带 PKCE 的 OAuth 2.1 | | `sampling` | mapping | 两者 | 服务器发起的 LLM 请求策略（参见 MCP 指南） | +| `trust` | string | 两者 | 信任层级：`full`（默认）或 `untrusted`。在 `untrusted` 服务器上，所有具备写能力的工具调用（即没有 `readOnlyHint: true` 注解的工具）在执行前都需要通过标准审批界面获得用户批准。`readOnlyHint` 是服务器自报的*提示* —— 恶意服务器最多只能让自称只读的工具跳过审批，绝不会因此获得额外权限，因此对不完全受控的服务器请标记为 `untrusted`。无法识别的值按 `untrusted` 处理（失败即关闭） | ## `tools` 策略键