feat(mcp): trust-tier gating for write-capable MCP tools via readOnlyHint

Adds a per-server `trust: full|untrusted` config key
(mcp_servers.<name>.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).
This commit is contained in:
Teknium 2026-08-07 07:34:35 -07:00
parent 37cc999926
commit c8369e37f4
4 changed files with 429 additions and 0 deletions

View File

@ -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

View File

@ -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.<name>.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

View File

@ -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

View File

@ -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` 策略键