From 22b07ba13c19aa678af018bed0b9b790a38d04ed Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:01:27 +0900 Subject: [PATCH 01/16] feat(mcp): add typed per-user OAuth identity scope --- tools/mcp_oauth_identity.py | 251 ++++++++++++++++++++++++++++++++++++ 1 file changed, 251 insertions(+) create mode 100644 tools/mcp_oauth_identity.py diff --git a/tools/mcp_oauth_identity.py b/tools/mcp_oauth_identity.py new file mode 100644 index 000000000000..07c439f47f30 --- /dev/null +++ b/tools/mcp_oauth_identity.py @@ -0,0 +1,251 @@ +"""Trusted identity boundary for MCP OAuth authorization. + +Issue #78174 requires OAuth credentials to follow the authenticated human who +initiated an MCP request. This module deliberately keeps that identity out of +model-visible tool arguments: the only automatic source in ``per_user`` mode is +the gateway's task-local session binding. + +Configuration:: + + mcp: + oauth: + identity_mode: shared # default, backwards compatible + # identity_mode: per_user + +``shared`` preserves the historic one-token-set-per-profile/server behaviour. +``per_user`` derives an immutable principal from the bound platform, platform +scope (Slack workspace / Discord guild / Matrix server), and user id. Missing +identity fails closed. Explicit but invalid configuration also fails closed; +a typo must never silently downgrade a multi-user deployment to shared creds. + +Raw human identifiers are never used as filesystem/cache keys. The stable +``storage_key`` is an opaque SHA-256 digest of a versioned canonical encoding. +""" + +from __future__ import annotations + +import hashlib +import json +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from typing import Iterator, Literal + +IdentityMode = Literal["shared", "per_user"] + +_SHARED = "shared" +_PER_USER = "per_user" +_VALID_MODES = frozenset({_SHARED, _PER_USER}) +_SCOPE_VERSION = "mcp-oauth-principal-v1" + + +class McpOAuthIdentityError(RuntimeError): + """Base class for MCP OAuth identity-boundary failures.""" + + +class InvalidMcpOAuthIdentityModeError(McpOAuthIdentityError, ValueError): + """Raised for an explicitly configured unknown identity mode.""" + + +class MissingMcpOAuthIdentityError(McpOAuthIdentityError): + """Raised when ``per_user`` mode has no authenticated bound principal.""" + + +@dataclass(frozen=True, slots=True) +class McpOAuthPrincipal: + """Authenticated human principal attached to one gateway request. + + ``scope_id`` is the platform-neutral account/workspace discriminator that + Hermes already captures in ``gateway.session_context``. It may be empty on + platforms that do not expose one; platform + user_id still forms a stable + principal there. + """ + + platform: str + scope_id: str + user_id: str + + def __post_init__(self) -> None: + platform = self.platform.strip().lower() + scope_id = self.scope_id.strip() + user_id = self.user_id.strip() + if not platform or not user_id: + raise MissingMcpOAuthIdentityError( + "MCP OAuth per-user identity requires a bound session platform " + "and authenticated user id. Hermes will not reuse another " + "user's OAuth credentials." + ) + object.__setattr__(self, "platform", platform) + object.__setattr__(self, "scope_id", scope_id) + object.__setattr__(self, "user_id", user_id) + + @property + def storage_key(self) -> str: + """Opaque, collision-resistant key that does not expose raw identity.""" + canonical = json.dumps( + [_SCOPE_VERSION, self.platform, self.scope_id, self.user_id], + ensure_ascii=False, + separators=(",", ":"), + ).encode("utf-8") + return "u-v1-" + hashlib.sha256(canonical).hexdigest() + + +@dataclass(frozen=True, slots=True) +class McpOAuthScope: + """Immutable credential/connection scope for an MCP OAuth operation.""" + + mode: IdentityMode + principal: McpOAuthPrincipal | None = None + + def __post_init__(self) -> None: + if self.mode not in _VALID_MODES: + raise InvalidMcpOAuthIdentityModeError( + f"invalid MCP OAuth identity mode: {self.mode!r}" + ) + if self.mode == _PER_USER and self.principal is None: + raise MissingMcpOAuthIdentityError( + "MCP OAuth per-user mode requires an authenticated principal" + ) + if self.mode == _SHARED and self.principal is not None: + raise McpOAuthIdentityError( + "shared MCP OAuth scope must not carry a user principal" + ) + + @classmethod + def shared(cls) -> "McpOAuthScope": + return cls(_SHARED) + + @property + def key(self) -> str: + """Stable internal key used by provider/connection/cache registries.""" + if self.mode == _SHARED: + return _SHARED + assert self.principal is not None + return self.principal.storage_key + + @property + def is_per_user(self) -> bool: + return self.mode == _PER_USER + + +# Explicit trusted override for administrative CLI flows and tests. This is a +# typed scope, not a string selector accepted from the model/MCP argument map. +_EXPLICIT_SCOPE: ContextVar[McpOAuthScope | None] = ContextVar( + "mcp_oauth_explicit_scope", default=None +) + + +@contextmanager +def explicit_oauth_scope(scope: McpOAuthScope) -> Iterator[None]: + """Temporarily bind a trusted, already-validated OAuth scope.""" + token = _EXPLICIT_SCOPE.set(scope) + try: + yield + finally: + _EXPLICIT_SCOPE.reset(token) + + +def get_oauth_identity_mode() -> IdentityMode: + """Return the configured mode; reject explicit unknown values. + + Absence means ``shared`` for backwards compatibility. An explicit typo is + a configuration error rather than a security downgrade. + """ + from hermes_cli.config import load_config + + config = load_config() or {} + mcp = config.get("mcp") + if mcp is None: + return _SHARED + if not isinstance(mcp, dict): + raise InvalidMcpOAuthIdentityModeError("config 'mcp' must be a mapping") + oauth = mcp.get("oauth") + if oauth is None: + return _SHARED + if not isinstance(oauth, dict): + raise InvalidMcpOAuthIdentityModeError("config 'mcp.oauth' must be a mapping") + raw = oauth.get("identity_mode") + if raw is None: + return _SHARED + mode = str(raw).strip().lower() + if mode not in _VALID_MODES: + raise InvalidMcpOAuthIdentityModeError( + "mcp.oauth.identity_mode must be 'shared' or 'per_user' " + f"(got {raw!r})" + ) + return mode # type: ignore[return-value] + + +def _bound_session_value(var_name: str) -> str: + """Read a ContextVar only when THIS task explicitly bound it. + + Security-sensitive identity resolution must not use ``get_session_env``'s + process-global ``os.environ`` fallback: a long-lived shared gateway may + retain legacy env mirrors from another request. We intentionally inspect + the gateway's task-local binding and reject ``_UNSET`` / empty values. + """ + from gateway import session_context + + var = getattr(session_context, var_name) + unset = getattr(session_context, "_UNSET") + value = var.get() + if value is unset: + return "" + return str(value or "").strip() + + +def current_bound_principal() -> McpOAuthPrincipal | None: + """Return the authenticated principal bound to the current task, if any.""" + platform = _bound_session_value("_SESSION_PLATFORM") + user_id = _bound_session_value("_SESSION_USER_ID") + scope_id = _bound_session_value("_SESSION_SCOPE_ID") + if not platform or not user_id: + return None + return McpOAuthPrincipal( + platform=platform, + scope_id=scope_id, + user_id=user_id, + ) + + +def resolve_oauth_scope(*, require_identity: bool = True) -> McpOAuthScope: + """Resolve the credential scope for the current trusted runtime context.""" + explicit = _EXPLICIT_SCOPE.get() + if explicit is not None: + return explicit + + mode = get_oauth_identity_mode() + if mode == _SHARED: + return McpOAuthScope.shared() + + principal = current_bound_principal() + if principal is None: + if require_identity: + raise MissingMcpOAuthIdentityError( + "mcp.oauth.identity_mode is 'per_user' but this request has no " + "authenticated task-local user identity. Hermes will not fall " + "back to shared or another user's MCP OAuth credentials." + ) + # Callers that only need to determine whether a scope is currently + # available can use ``try_resolve_oauth_scope`` below. Returning a + # fake/anonymous per-user scope here would create a shared backdoor, so + # even the non-requiring form never manufactures one. + raise MissingMcpOAuthIdentityError( + "no authenticated MCP OAuth principal is bound to this task" + ) + return McpOAuthScope(_PER_USER, principal) + + +def try_resolve_oauth_scope() -> McpOAuthScope | None: + """Best-effort scope lookup without manufacturing an anonymous identity.""" + try: + return resolve_oauth_scope() + except MissingMcpOAuthIdentityError: + return None + + +def connection_registry_key(server_name: str, scope: McpOAuthScope) -> str: + """Opaque exact key for a long-lived authenticated MCP connection.""" + if not scope.is_per_user: + return server_name + return f"{server_name}@@{scope.key}" From d1bda91890c9b0c7479dea26e227a3fad2d6e6ce Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:01:41 +0900 Subject: [PATCH 02/16] feat(mcp): isolate OAuth persistence by identity scope --- tools/mcp_oauth_scoped_storage.py | 56 +++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 tools/mcp_oauth_scoped_storage.py diff --git a/tools/mcp_oauth_scoped_storage.py b/tools/mcp_oauth_scoped_storage.py new file mode 100644 index 000000000000..e26d9dd82bb3 --- /dev/null +++ b/tools/mcp_oauth_scoped_storage.py @@ -0,0 +1,56 @@ +"""Scope-aware persistent storage for MCP OAuth state. + +The existing :class:`tools.mcp_oauth.HermesTokenStorage` remains the storage +implementation (permissions, atomic writes, expiry handling, snapshots). This +subclass changes only path resolution so every piece of OAuth state follows the +same immutable :class:`McpOAuthScope` boundary. + +Shared mode intentionally preserves the historical layout exactly. Per-user +mode stores all state under an opaque digest directory; legacy shared tokens are +never migrated or read as a fallback because their human owner is unknowable. +""" + +from __future__ import annotations + +from pathlib import Path + +from tools.mcp_oauth import HermesTokenStorage, _get_token_dir +from tools.mcp_oauth_identity import McpOAuthScope + + +class ScopedHermesTokenStorage(HermesTokenStorage): + """Hermes OAuth storage whose complete state is bound to one OAuth scope.""" + + def __init__( + self, + server_name: str, + scope: McpOAuthScope, + *, + hermes_home: str | Path | None = None, + ) -> None: + super().__init__(server_name, hermes_home=hermes_home) + self._oauth_scope = scope + + @property + def oauth_scope(self) -> McpOAuthScope: + return self._oauth_scope + + def _scoped_token_dir(self) -> Path: + root = _get_token_dir(self._hermes_home) + if not self._oauth_scope.is_per_user: + return root + # Opaque digest only: never expose Slack/Discord/etc identifiers in + # directory names, logs, backups, or path traversal diagnostics. + return root / "by-user" / self._oauth_scope.key + + def _tokens_path(self) -> Path: + return self._scoped_token_dir() / f"{self._server_name}.json" + + def _client_info_path(self) -> Path: + return self._scoped_token_dir() / f"{self._server_name}.client.json" + + def _meta_path(self) -> Path: + return self._scoped_token_dir() / f"{self._server_name}.meta.json" + + def _cimd_rejected_path(self) -> Path: + return self._scoped_token_dir() / f"{self._server_name}.cimd-off" From c9df1231bce8126fb74d68079805250518a25c9e Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:05:00 +0900 Subject: [PATCH 03/16] test(mcp): cover per-user OAuth identity boundary --- tests/tools/test_mcp_oauth_identity_scope.py | 231 +++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 tests/tools/test_mcp_oauth_identity_scope.py diff --git a/tests/tools/test_mcp_oauth_identity_scope.py b/tests/tools/test_mcp_oauth_identity_scope.py new file mode 100644 index 000000000000..e23b8959e53b --- /dev/null +++ b/tests/tools/test_mcp_oauth_identity_scope.py @@ -0,0 +1,231 @@ +"""Security-boundary tests for per-user MCP OAuth identity (#78174).""" + +from __future__ import annotations + +import re + +import pytest + + +@pytest.fixture(autouse=True) +def _clear_session_context(): + from gateway.session_context import clear_session_vars + + clear_session_vars() + yield + clear_session_vars() + + +def _patch_config(monkeypatch, mode_marker=...): + if mode_marker is ...: + config = {} + else: + config = {"mcp": {"oauth": {"identity_mode": mode_marker}}} + monkeypatch.setattr("hermes_cli.config.load_config", lambda: config) + + +def _bind(*, platform="slack", scope_id="T_WORKSPACE", user_id="U_USER"): + from gateway.session_context import set_session_vars + + set_session_vars( + platform=platform, + scope_id=scope_id, + user_id=user_id, + ) + + +def test_absent_config_preserves_shared_mode(monkeypatch): + from tools.mcp_oauth_identity import McpOAuthScope, resolve_oauth_scope + + _patch_config(monkeypatch) + + assert resolve_oauth_scope() == McpOAuthScope.shared() + + +def test_explicit_shared_mode_preserves_legacy_scope(monkeypatch): + from tools.mcp_oauth_identity import McpOAuthScope, resolve_oauth_scope + + _patch_config(monkeypatch, "shared") + _bind() + + assert resolve_oauth_scope() == McpOAuthScope.shared() + + +def test_invalid_explicit_mode_fails_closed(monkeypatch): + from tools.mcp_oauth_identity import ( + InvalidMcpOAuthIdentityModeError, + resolve_oauth_scope, + ) + + _patch_config(monkeypatch, "per-user") + _bind() + + with pytest.raises(InvalidMcpOAuthIdentityModeError): + resolve_oauth_scope() + + +def test_per_user_mode_requires_task_local_authenticated_identity(monkeypatch): + from tools.mcp_oauth_identity import ( + MissingMcpOAuthIdentityError, + resolve_oauth_scope, + ) + + _patch_config(monkeypatch, "per_user") + + with pytest.raises(MissingMcpOAuthIdentityError, match="will not fall back"): + resolve_oauth_scope() + + +def test_process_environment_is_not_a_trusted_principal(monkeypatch): + """Legacy process-global session mirrors cannot authorize per-user MCP.""" + from tools.mcp_oauth_identity import MissingMcpOAuthIdentityError, resolve_oauth_scope + + _patch_config(monkeypatch, "per_user") + monkeypatch.setenv("HERMES_SESSION_PLATFORM", "slack") + monkeypatch.setenv("HERMES_SESSION_SCOPE_ID", "T_STALE") + monkeypatch.setenv("HERMES_SESSION_USER_ID", "U_OTHER_USER") + + with pytest.raises(MissingMcpOAuthIdentityError): + resolve_oauth_scope() + + +def test_per_user_scope_uses_platform_scope_and_user(monkeypatch): + from tools.mcp_oauth_identity import resolve_oauth_scope + + _patch_config(monkeypatch, "per_user") + _bind(platform="slack", scope_id="T_ONE", user_id="U_123") + one = resolve_oauth_scope() + + # Same platform-local user identifier in another workspace is a distinct + # authorization principal. + _bind(platform="slack", scope_id="T_TWO", user_id="U_123") + two = resolve_oauth_scope() + + assert one != two + assert one.key != two.key + + +def test_scope_key_is_opaque_and_filesystem_safe(monkeypatch): + from tools.mcp_oauth_identity import resolve_oauth_scope + + _patch_config(monkeypatch, "per_user") + _bind( + platform="slack", + scope_id="T-sensitive-workspace", + user_id="U-sensitive-person", + ) + scope = resolve_oauth_scope() + + assert scope.key.startswith("u-v1-") + assert re.fullmatch(r"u-v1-[0-9a-f]{64}", scope.key) + assert "slack" not in scope.key + assert "sensitive" not in scope.key + + +def test_connection_registry_keys_never_collide_between_users(monkeypatch): + from tools.mcp_oauth_identity import connection_registry_key, resolve_oauth_scope + + _patch_config(monkeypatch, "per_user") + _bind(user_id="U_ALICE") + alice = resolve_oauth_scope() + _bind(user_id="U_BOB") + bob = resolve_oauth_scope() + + assert connection_registry_key("github", alice) != connection_registry_key( + "github", bob + ) + assert connection_registry_key("github", alice).startswith("github@@u-v1-") + + +def test_shared_connection_registry_key_is_backwards_compatible(): + from tools.mcp_oauth_identity import McpOAuthScope, connection_registry_key + + assert connection_registry_key("github", McpOAuthScope.shared()) == "github" + + +def test_explicit_scope_is_typed_not_model_argument(monkeypatch): + from tools.mcp_oauth_identity import ( + McpOAuthPrincipal, + McpOAuthScope, + explicit_oauth_scope, + resolve_oauth_scope, + ) + + _patch_config(monkeypatch, "per_user") + admin_scope = McpOAuthScope( + "per_user", + McpOAuthPrincipal("slack", "T_ADMIN", "U_ADMIN"), + ) + + with explicit_oauth_scope(admin_scope): + assert resolve_oauth_scope() == admin_scope + + +def test_scoped_storage_separates_all_oauth_state(monkeypatch, tmp_path): + from tools.mcp_oauth_identity import McpOAuthPrincipal, McpOAuthScope + from tools.mcp_oauth_scoped_storage import ScopedHermesTokenStorage + + alice_scope = McpOAuthScope( + "per_user", McpOAuthPrincipal("slack", "T_ONE", "U_ALICE") + ) + bob_scope = McpOAuthScope( + "per_user", McpOAuthPrincipal("slack", "T_ONE", "U_BOB") + ) + alice = ScopedHermesTokenStorage("github", alice_scope, hermes_home=tmp_path) + bob = ScopedHermesTokenStorage("github", bob_scope, hermes_home=tmp_path) + + alice_paths = { + alice._tokens_path(), + alice._client_info_path(), + alice._meta_path(), + alice._cimd_rejected_path(), + } + bob_paths = { + bob._tokens_path(), + bob._client_info_path(), + bob._meta_path(), + bob._cimd_rejected_path(), + } + + assert alice_paths.isdisjoint(bob_paths) + assert all("by-user" in path.parts for path in alice_paths | bob_paths) + assert all("U_ALICE" not in str(path) for path in alice_paths) + assert all("U_BOB" not in str(path) for path in bob_paths) + + +def test_scoped_storage_shared_mode_keeps_historical_paths(tmp_path): + from tools.mcp_oauth import HermesTokenStorage + from tools.mcp_oauth_identity import McpOAuthScope + from tools.mcp_oauth_scoped_storage import ScopedHermesTokenStorage + + legacy = HermesTokenStorage("github", hermes_home=tmp_path) + scoped = ScopedHermesTokenStorage( + "github", McpOAuthScope.shared(), hermes_home=tmp_path + ) + + assert scoped._tokens_path() == legacy._tokens_path() + assert scoped._client_info_path() == legacy._client_info_path() + assert scoped._meta_path() == legacy._meta_path() + assert scoped._cimd_rejected_path() == legacy._cimd_rejected_path() + + +def test_per_user_storage_never_falls_back_to_legacy_shared_file(tmp_path): + """Unknown ownership means shared tokens cannot be auto-migrated.""" + from tools.mcp_oauth import HermesTokenStorage + from tools.mcp_oauth_identity import McpOAuthPrincipal, McpOAuthScope + from tools.mcp_oauth_scoped_storage import ScopedHermesTokenStorage + + legacy = HermesTokenStorage("github", hermes_home=tmp_path) + legacy._tokens_path().parent.mkdir(parents=True, exist_ok=True) + legacy._tokens_path().write_text('{"access_token":"SHARED"}', encoding="utf-8") + + alice = ScopedHermesTokenStorage( + "github", + McpOAuthScope( + "per_user", McpOAuthPrincipal("slack", "T_ONE", "U_ALICE") + ), + hermes_home=tmp_path, + ) + + assert not alice._tokens_path().exists() + assert alice._tokens_path() != legacy._tokens_path() From 0bdb426eb09b098422517cc8db87d02938f75ab5 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:05:54 +0900 Subject: [PATCH 04/16] refactor(mcp): separate OAuth provider from scoped manager --- tools/mcp_oauth_provider.py | 332 ++++++++++++++++++++++++++++++++++++ 1 file changed, 332 insertions(+) create mode 100644 tools/mcp_oauth_provider.py diff --git a/tools/mcp_oauth_provider.py b/tools/mcp_oauth_provider.py new file mode 100644 index 000000000000..e2eed558c958 --- /dev/null +++ b/tools/mcp_oauth_provider.py @@ -0,0 +1,332 @@ +"""Hermes MCP OAuth provider implementation. + +Extracted from :mod:`tools.mcp_oauth_manager` so the manager can key all +mutable OAuth state by an immutable requesting-user scope without entangling +provider protocol behaviour. The provider keeps the existing MCP SDK +compatibility, metadata persistence, DCR/CIMD recovery, expiry handling, and +bidirectional auth-flow bridge. +""" + +from __future__ import annotations + +import logging +import re +from typing import Any, Optional + +logger = logging.getLogger(__name__) + + +def _same_endpoint(a: str, b: str) -> bool: + """Return True when URLs target the same endpoint, ignoring query/fragment.""" + from urllib.parse import urlsplit + + try: + pa, pb = urlsplit(a), urlsplit(b) + except ValueError: # pragma: no cover + return False + return ( + pa.scheme == pb.scheme + and pa.netloc.lower() == pb.netloc.lower() + and pa.path.rstrip("/") == pb.path.rstrip("/") + ) + + +def _make_hermes_provider_class() -> Optional[type]: + """Lazy-import the MCP SDK OAuth provider and return Hermes' subclass.""" + try: + from mcp.client.auth.oauth2 import OAuthClientProvider + except ImportError: # pragma: no cover - SDK required in CI + return None + + class HermesMCPOAuthProvider(OAuthClientProvider): + """OAuthClientProvider with Hermes persistence/recovery hooks.""" + + def __init__( + self, + *args: Any, + server_name: str = "", + oauth_scope=None, + preregistered: bool = False, + token_user_agent: "str | None" = None, + **kwargs: Any, + ): + super().__init__(*args, **kwargs) + self._hermes_server_name = server_name + self._hermes_home = "" + # Scope is captured when the provider is constructed. Never + # re-resolve ambient identity during refresh/background auth. + self._hermes_oauth_scope = oauth_scope + self._hermes_preregistered = preregistered + self._hermes_token_user_agent = token_user_agent + + def _stamp_token_user_agent(self, request): + ua = getattr(self, "_hermes_token_user_agent", None) + if ua: + request.headers["User-Agent"] = ua + return request + + def _coerce_client_secret_post(self) -> None: + info = getattr(self.context, "client_info", None) + if not info or not getattr(info, "client_secret", None): + return + method = getattr(info, "token_endpoint_auth_method", None) + if method not in (None, "none", ""): + return + from mcp.shared.auth import OAuthClientInformationFull + + data = info.model_dump(mode="json", exclude_none=True) + data["token_endpoint_auth_method"] = "client_secret_post" + self.context.client_info = OAuthClientInformationFull.model_validate(data) + + async def _exchange_token_authorization_code(self, *args: Any, **kwargs: Any): + self._coerce_client_secret_post() + request = await super()._exchange_token_authorization_code(*args, **kwargs) + return self._stamp_token_user_agent(request) + + async def _refresh_token(self): + self._coerce_client_secret_post() + request = await super()._refresh_token() + return self._stamp_token_user_agent(request) + + async def _handle_token_response(self, response): + """Accept any 2xx token response without exposing token bodies.""" + if 200 <= response.status_code < 300: + from httpx import HTTPError + from mcp.client.auth.oauth2 import OAuthTokenError + from mcp.client.auth.utils import handle_token_response_scopes + + try: + token_response = await handle_token_response_scopes(response) + except (HTTPError, OAuthTokenError): + raise OAuthTokenError("Invalid token response") from None + self.context.current_tokens = token_response + self.context.update_token_expiry(token_response) + await self.context.storage.set_tokens(token_response) + return + + from mcp.client.auth.oauth2 import OAuthTokenError + + raise OAuthTokenError(f"Token exchange failed ({response.status_code})") + + async def _handle_refresh_response(self, response) -> bool: + """Accept any 2xx refresh response without logging credential bodies.""" + if not (200 <= response.status_code < 300): + logger.warning("Token refresh failed: %s", response.status_code) + self.context.clear_tokens() + return False + + from httpx import HTTPError + from mcp.shared.auth import OAuthToken + from pydantic import ValidationError + + try: + content = await response.aread() + token_response = OAuthToken.model_validate_json(content) + self.context.current_tokens = token_response + self.context.update_token_expiry(token_response) + await self.context.storage.set_tokens(token_response) + return True + except (HTTPError, ValidationError): + logger.warning("Invalid refresh response: %s", response.status_code) + self.context.clear_tokens() + return False + + async def _initialize(self) -> None: + """Load persisted state, seed expiry, and restore/discover metadata.""" + await super()._initialize() + tokens = self.context.current_tokens + if tokens is not None and tokens.expires_in is not None: + self.context.update_token_expiry(tokens) + + storage = self.context.storage + from tools.mcp_oauth import HermesTokenStorage + + if isinstance(storage, HermesTokenStorage) and self.context.oauth_metadata is None: + meta = storage.load_oauth_metadata() + if meta is not None: + self.context.oauth_metadata = meta + logger.debug( + "MCP OAuth '%s': restored metadata from disk " + "(token_endpoint=%s)", + self._hermes_server_name, + meta.token_endpoint, + ) + + if tokens is not None and self.context.oauth_metadata is None: + try: + await self._prefetch_oauth_metadata() + except Exception as exc: # pragma: no cover - defensive + logger.debug( + "MCP OAuth '%s': pre-flight metadata discovery " + "failed (non-fatal): %s", + self._hermes_server_name, + exc, + ) + + async def _prefetch_oauth_metadata(self) -> None: + """Fetch PRM + authorization-server metadata before refresh.""" + from tools.mcp_tool import sdk_httpx + + httpx = sdk_httpx() + if httpx is None: # pragma: no cover + return + from mcp.client.auth.utils import ( + build_oauth_authorization_server_metadata_discovery_urls, + build_protected_resource_metadata_discovery_urls, + create_oauth_metadata_request, + handle_auth_metadata_response, + handle_protected_resource_response, + ) + + server_url = self.context.server_url + async with httpx.AsyncClient(timeout=10.0) as client: + for url in build_protected_resource_metadata_discovery_urls(None, server_url): + req = create_oauth_metadata_request(url) + try: + resp = await client.send(req) + except httpx.HTTPError as exc: + logger.debug( + "MCP OAuth '%s': PRM discovery to %s failed: %s", + self._hermes_server_name, + url, + exc, + ) + continue + prm = await handle_protected_resource_response(resp) + if prm: + self.context.protected_resource_metadata = prm + if prm.authorization_servers: + self.context.auth_server_url = str(prm.authorization_servers[0]) + break + + for url in build_oauth_authorization_server_metadata_discovery_urls( + self.context.auth_server_url, server_url + ): + req = create_oauth_metadata_request(url) + try: + resp = await client.send(req) + except httpx.HTTPError as exc: + logger.debug( + "MCP OAuth '%s': ASM discovery to %s failed: %s", + self._hermes_server_name, + url, + exc, + ) + continue + ok, asm = await handle_auth_metadata_response(resp) + if not ok: + break + if asm: + self.context.oauth_metadata = asm + storage = self.context.storage + from tools.mcp_oauth import HermesTokenStorage + + if isinstance(storage, HermesTokenStorage): + storage.save_oauth_metadata(asm) + logger.debug( + "MCP OAuth '%s': pre-flight ASM discovered " + "token_endpoint=%s", + self._hermes_server_name, + asm.token_endpoint, + ) + break + + def _persist_oauth_metadata_if_changed(self) -> None: + meta = self.context.oauth_metadata + if meta is None: + return + storage = self.context.storage + from tools.mcp_oauth import HermesTokenStorage + + if not isinstance(storage, HermesTokenStorage): + return + existing = storage.load_oauth_metadata() + if existing is None or str(existing.token_endpoint) != str(meta.token_endpoint): + storage.save_oauth_metadata(meta) + + async def _maybe_flag_poisoned_client(self, response: Any) -> None: + """Detect invalid_client and force safe re-registration when possible.""" + try: + if self._hermes_preregistered: + return + status = getattr(response, "status_code", None) + if status not in (400, 401): + return + meta = getattr(self.context, "oauth_metadata", None) + token_endpoint = ( + str(meta.token_endpoint) + if meta is not None and getattr(meta, "token_endpoint", None) + else None + ) + req = getattr(response, "request", None) + req_url = str(req.url) if req is not None else None + if not token_endpoint or not req_url or not _same_endpoint( + req_url, token_endpoint + ): + return + body = await response.aread() + if not re.search(rb"\binvalid_client\b", body.lower()): + return + + storage = self.context.storage + from tools.mcp_oauth import HermesTokenStorage + + cimd_url = getattr(self.context, "client_metadata_url", None) + rejected_id = getattr(self.context.client_info, "client_id", None) + if cimd_url and rejected_id == cimd_url: + logger.warning( + "MCP OAuth '%s': authorization server rejected our " + "Client ID Metadata Document (%s) with invalid_client " + "— falling back to dynamic client registration.", + self._hermes_server_name, + cimd_url, + ) + self.context.client_metadata_url = None + if isinstance(storage, HermesTokenStorage): + storage.mark_cimd_rejected() + + if isinstance(storage, HermesTokenStorage): + storage.poison_client_registration() + self.context.client_info = None + self._initialized = False + except Exception as exc: # pragma: no cover - defensive + logger.debug( + "MCP OAuth '%s': invalid_client detection failed (non-fatal): %s", + self._hermes_server_name, + exc, + ) + + async def async_auth_flow(self, request): # type: ignore[override] + # Scope is immutable provider state. Background refreshes must not + # consult whichever gateway request happens to be running now. + try: + from tools.mcp_oauth_manager import get_manager + + await get_manager().invalidate_if_disk_changed( + self._hermes_server_name, + hermes_home=self._hermes_home, + oauth_scope=self._hermes_oauth_scope, + ) + except Exception as exc: # pragma: no cover - defensive + logger.debug( + "MCP OAuth '%s': pre-flow disk-watch failed (non-fatal): %s", + self._hermes_server_name, + exc, + ) + + # Preserve the MCP SDK async-generator .asend(response) protocol. + inner = super().async_auth_flow(request) + try: + outgoing = await inner.__anext__() + while True: + incoming = yield outgoing + await self._maybe_flag_poisoned_client(incoming) + outgoing = await inner.asend(incoming) + except StopAsyncIteration: + self._persist_oauth_metadata_if_changed() + return + + return HermesMCPOAuthProvider + + +_HERMES_PROVIDER_CLS: Optional[type] = _make_hermes_provider_class() From ab679baabb6c35a1cc9b25e36ccb33f3e9c80b20 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:06:40 +0900 Subject: [PATCH 05/16] feat(mcp): scope OAuth manager state by requesting user --- tools/mcp_oauth_manager.py | 807 +++++++------------------------------ 1 file changed, 138 insertions(+), 669 deletions(-) diff --git a/tools/mcp_oauth_manager.py b/tools/mcp_oauth_manager.py index d0bca49967b5..5bcb64b7e9e6 100644 --- a/tools/mcp_oauth_manager.py +++ b/tools/mcp_oauth_manager.py @@ -1,601 +1,100 @@ #!/usr/bin/env python3 -"""Central manager for per-server MCP OAuth state. - -One instance shared across the process. Holds per-server OAuth provider -instances and coordinates: - -- **Cross-process token reload** via mtime-based disk watch. When an external - process (e.g. a user cron job) refreshes tokens on disk, the next auth flow - picks them up without requiring a process restart. -- **401 deduplication** via in-flight futures. When N concurrent tool calls - all hit 401 with the same access_token, only one recovery attempt fires; - the rest await the same result. -- **Reconnect signalling** for long-lived MCP sessions. The manager itself - does not drive reconnection — the `MCPServerTask` in `mcp_tool.py` does — - but the manager is the single source of truth that decides when reconnect - is warranted. - -Replaces what used to be scattered across eight call sites in `mcp_oauth.py`, -`mcp_tool.py`, and `hermes_cli/mcp_config.py`. This module is the ONLY place -that instantiates the MCP SDK's `OAuthClientProvider` — all other code paths -go through `get_manager()`. - -Design reference: - -- Claude Code's ``invalidateOAuthCacheIfDiskChanged`` - (``claude-code/src/utils/auth.ts:1320``, CC-1096 / GH#24317). Identical - external-refresh staleness bug class. -- Codex's ``refresh_oauth_if_needed`` / ``persist_if_needed`` - (``codex-rs/rmcp-client/src/rmcp_client.rs:805``). We lean on the MCP SDK's - lazy refresh rather than calling refresh before every op, because one - ``stat()`` per tool call is cheaper than an ``await`` + potential refresh - round-trip, and the SDK's in-memory expiry path is already correct. +"""Central manager for MCP OAuth state. + +The security key for every mutable OAuth object is now +``(Hermes home, MCP server, OAuth scope)``. In ``shared`` mode the scope key +is the backwards-compatible constant ``shared``. In ``per_user`` mode it is +an opaque digest of the authenticated gateway principal, so providers, +refreshes, locks, disk watches, and 401 deduplication can never silently cross +human-user boundaries (#78174). + +The MCP SDK provider implementation lives in :mod:`tools.mcp_oauth_provider`; +this module remains the single process-wide manager and re-exports +``_HERMES_PROVIDER_CLS`` for existing tests/callers. """ from __future__ import annotations import asyncio import logging -import re import threading from dataclasses import dataclass, field from pathlib import Path from typing import Any, Optional -logger = logging.getLogger(__name__) - - -def _same_endpoint(a: str, b: str) -> bool: - """Return True if two URLs target the same endpoint (ignoring query/fragment). - - Compares scheme, host (case-insensitive), and path. Used to confirm a - rejected response actually came from the OAuth token endpoint before we - act on an ``invalid_client`` body. - """ - from urllib.parse import urlsplit - - try: - pa, pb = urlsplit(a), urlsplit(b) - except ValueError: # pragma: no cover — malformed URL - return False - return ( - pa.scheme == pb.scheme - and pa.netloc.lower() == pb.netloc.lower() - and pa.path.rstrip("/") == pb.path.rstrip("/") - ) +from tools.mcp_oauth_identity import McpOAuthScope, resolve_oauth_scope +from tools.mcp_oauth_provider import _HERMES_PROVIDER_CLS +from tools.mcp_oauth_scoped_storage import ScopedHermesTokenStorage - -# --------------------------------------------------------------------------- -# Per-server entry -# --------------------------------------------------------------------------- +logger = logging.getLogger(__name__) @dataclass class _ProviderEntry: - """Per-server OAuth state tracked by the manager. - - Fields: - server_url: The MCP server URL used to build the provider. Tracked - so we can discard a cached provider if the URL changes. - oauth_config: Optional dict from ``mcp_servers..oauth``. - provider: The ``httpx.Auth``-compatible provider wrapping the MCP - SDK. None until first use. - last_mtime_ns: Last-seen ``st_mtime_ns`` of the on-disk tokens file. - Zero if never read. Used by :meth:`MCPOAuthManager.invalidate_if_disk_changed` - to detect external refreshes. - lock: Serialises concurrent access to this entry's state. Bound to - whichever asyncio loop first awaits it (the MCP event loop). - pending_401: In-flight 401-handler futures keyed by the failed - access_token, for deduplicating thundering-herd 401s. Mirrors - Claude Code's ``pending401Handlers`` map. - """ + """OAuth state isolated to one server and immutable credential scope.""" server_url: str oauth_config: Optional[dict] + oauth_scope: McpOAuthScope = field(default_factory=McpOAuthScope.shared) provider: Optional[Any] = None last_mtime_ns: int = 0 lock: asyncio.Lock = field(default_factory=asyncio.Lock) pending_401: dict[str, "asyncio.Future[bool]"] = field(default_factory=dict) -# --------------------------------------------------------------------------- -# HermesMCPOAuthProvider — OAuthClientProvider subclass with disk-watch -# --------------------------------------------------------------------------- - - -def _make_hermes_provider_class() -> Optional[type]: - """Lazy-import the SDK base class and return our subclass. - - Wrapped in a function so this module imports cleanly even when the - MCP SDK's OAuth module is unavailable (e.g. older mcp versions). - """ - try: - from mcp.client.auth.oauth2 import OAuthClientProvider - except ImportError: # pragma: no cover — SDK required in CI - return None - - class HermesMCPOAuthProvider(OAuthClientProvider): - """OAuthClientProvider with pre-flow disk-mtime reload. - - Before every ``async_auth_flow`` invocation, asks the manager to - check whether the tokens file on disk has been modified externally. - If so, the manager resets ``_initialized`` so the next flow - re-reads from storage. - - This makes external-process refreshes (cron, another CLI instance) - visible to the running MCP session without requiring a restart. - - Reference: Claude Code's ``invalidateOAuthCacheIfDiskChanged`` - (``src/utils/auth.ts:1320``, CC-1096 / GH#24317). - """ - - def __init__( - self, - *args: Any, - server_name: str = "", - preregistered: bool = False, - token_user_agent: "str | None" = None, - **kwargs: Any, - ): - super().__init__(*args, **kwargs) - self._hermes_server_name = server_name - self._hermes_home = "" - # When the client_id comes from config.yaml (pre-registered), an - # invalid_client rejection means the *config* is wrong — deleting - # client.json would just be re-seeded from config and re-running - # registration can't help. Only auto-heal dynamically-registered - # clients. See _maybe_flag_poisoned_client. - self._hermes_preregistered = preregistered - # oauth.user_agent — stamped onto token-endpoint requests only; - # some authorization servers/WAFs reject httpx's default (#75576). - self._hermes_token_user_agent = token_user_agent - - def _stamp_token_user_agent(self, request): - ua = getattr(self, "_hermes_token_user_agent", None) - if ua: - request.headers["User-Agent"] = ua - return request - - def _coerce_client_secret_post(self) -> None: - """Use client_secret_post when dynamic registration returned a secret. - - Some MCP OAuth providers, notably Supabase, return a - ``client_secret`` from dynamic client registration but omit - ``token_endpoint_auth_method``. The MCP SDK treats the missing - value as public-client auth (``none``), so token exchange omits the - secret and Supabase rejects it with ``Required parameter: - client_secret``. Coerce the in-memory client info before token and - refresh requests. - """ - info = getattr(self.context, "client_info", None) - if not info or not getattr(info, "client_secret", None): - return - method = getattr(info, "token_endpoint_auth_method", None) - if method not in (None, "none", ""): - return - from mcp.shared.auth import OAuthClientInformationFull - - data = info.model_dump(mode="json", exclude_none=True) - data["token_endpoint_auth_method"] = "client_secret_post" - self.context.client_info = OAuthClientInformationFull.model_validate(data) - - async def _exchange_token_authorization_code(self, *args: Any, **kwargs: Any): - self._coerce_client_secret_post() - request = await super()._exchange_token_authorization_code(*args, **kwargs) - return self._stamp_token_user_agent(request) - - async def _refresh_token(self): - self._coerce_client_secret_post() - request = await super()._refresh_token() - return self._stamp_token_user_agent(request) - - async def _handle_token_response(self, response): - """Accept any 2xx token response and avoid leaking token bodies in errors.""" - if 200 <= response.status_code < 300: - from mcp.client.auth.utils import handle_token_response_scopes - from mcp.client.auth.oauth2 import OAuthTokenError - from httpx import HTTPError - - try: - token_response = await handle_token_response_scopes(response) - except (HTTPError, OAuthTokenError): - raise OAuthTokenError("Invalid token response") from None - self.context.current_tokens = token_response - self.context.update_token_expiry(token_response) - await self.context.storage.set_tokens(token_response) - return - - from mcp.client.auth.oauth2 import OAuthTokenError - - raise OAuthTokenError(f"Token exchange failed ({response.status_code})") - - async def _handle_refresh_response(self, response) -> bool: - """Accept any 2xx refresh response and avoid logging token bodies.""" - if not (200 <= response.status_code < 300): - logger.warning("Token refresh failed: %s", response.status_code) - self.context.clear_tokens() - return False - - from mcp.shared.auth import OAuthToken - from httpx import HTTPError - from pydantic import ValidationError - - try: - content = await response.aread() - token_response = OAuthToken.model_validate_json(content) - self.context.current_tokens = token_response - self.context.update_token_expiry(token_response) - await self.context.storage.set_tokens(token_response) - return True - except (HTTPError, ValidationError): - logger.warning("Invalid refresh response: %s", response.status_code) - self.context.clear_tokens() - return False - - async def _initialize(self) -> None: - """Load stored tokens + client info AND seed token_expiry_time. - - Also eagerly fetches OAuth authorization-server metadata (PRM + - ASM) when we have stored tokens but no cached metadata, so the - SDK's ``_refresh_token`` can build the correct token_endpoint - URL on the preemptive-refresh path. Without this, the SDK - falls back to ``{mcp_server_url}/token`` (wrong for providers - whose AS is a different origin — BetterStack's MCP lives at - ``https://mcp.betterstack.com`` but its token endpoint is at - ``https://betterstack.com/oauth/token``), the refresh 404s, and - we drop through to full browser reauth. - - The SDK's base ``_initialize`` populates ``current_tokens`` but - does NOT call ``update_token_expiry``, so ``token_expiry_time`` - stays ``None`` and ``is_token_valid()`` returns True for any - loaded token regardless of actual age. After a process restart - this ships stale Bearer tokens to the server; some providers - return HTTP 401 (caught by the 401 handler), others return 200 - with an app-level auth error (invisible to the transport layer, - e.g. BetterStack returning "No teams found. Please check your - authentication."). - - Seeding ``token_expiry_time`` from the reloaded token fixes that: - ``is_token_valid()`` correctly reports False for expired tokens, - ``async_auth_flow`` takes the ``can_refresh_token()`` branch, - and the SDK quietly refreshes before the first real request. - - Paired with :class:`HermesTokenStorage` persisting an absolute - ``expires_at`` timestamp (``mcp_oauth.py:set_tokens``) so the - remaining TTL we compute here reflects real wall-clock age. - """ - await super()._initialize() - tokens = self.context.current_tokens - if tokens is not None and tokens.expires_in is not None: - self.context.update_token_expiry(tokens) - - # Cold-load: restore OAuth server metadata from disk before any - # refresh attempt. Without this, a restarted process with cached - # tokens but no in-memory metadata would fall back to the SDK's - # guessed ``{server_url}/token`` path (returns 404 on most real - # providers) and require a full browser re-authorization. - storage = self.context.storage - from tools.mcp_oauth import HermesTokenStorage - if ( - isinstance(storage, HermesTokenStorage) - and self.context.oauth_metadata is None - ): - meta = storage.load_oauth_metadata() - if meta is not None: - self.context.oauth_metadata = meta - logger.debug( - "MCP OAuth '%s': restored metadata from disk " - "(token_endpoint=%s)", - self._hermes_server_name, - meta.token_endpoint, - ) - - # Pre-flight OAuth AS discovery so ``_refresh_token`` has a - # correct ``token_endpoint`` before the first refresh attempt. - # Only runs when we have tokens on cold-load but no cached - # metadata — i.e. the exact scenario where the SDK's built-in - # 401-branch discovery hasn't had a chance to run yet. - if ( - tokens is not None - and self.context.oauth_metadata is None - ): - try: - await self._prefetch_oauth_metadata() - except Exception as exc: # pragma: no cover — defensive - # Non-fatal: if discovery fails, the SDK's normal 401- - # branch discovery will run on the next request. - logger.debug( - "MCP OAuth '%s': pre-flight metadata discovery " - "failed (non-fatal): %s", - self._hermes_server_name, exc, - ) - - async def _prefetch_oauth_metadata(self) -> None: - """Fetch PRM + ASM from the well-known endpoints, cache on context. - - Mirrors the SDK's 401-branch discovery (oauth2.py ~line 511-551) - but runs synchronously before the first request instead of - inside the httpx auth_flow generator. Uses the SDK's own URL - builders and response handlers so we track whatever the SDK - version we're pinned to expects. - """ - # The SDK's httpx flavour, not Hermes' — mcp 2.0 builds on httpx2, - # and `create_oauth_metadata_request` below returns one of *its* - # Request objects, which only its own AsyncClient can send. See - # tools.mcp_tool.sdk_httpx. - from tools.mcp_tool import sdk_httpx - httpx = sdk_httpx() - if httpx is None: # pragma: no cover — SDK import would have failed - return - from mcp.client.auth.utils import ( - build_oauth_authorization_server_metadata_discovery_urls, - build_protected_resource_metadata_discovery_urls, - create_oauth_metadata_request, - handle_auth_metadata_response, - handle_protected_resource_response, - ) - - server_url = self.context.server_url - async with httpx.AsyncClient(timeout=10.0) as client: - # Step 1: PRM discovery to learn the authorization_server URL. - for url in build_protected_resource_metadata_discovery_urls( - None, server_url - ): - req = create_oauth_metadata_request(url) - try: - resp = await client.send(req) - except httpx.HTTPError as exc: - logger.debug( - "MCP OAuth '%s': PRM discovery to %s failed: %s", - self._hermes_server_name, url, exc, - ) - continue - prm = await handle_protected_resource_response(resp) - if prm: - self.context.protected_resource_metadata = prm - if prm.authorization_servers: - self.context.auth_server_url = str( - prm.authorization_servers[0] - ) - break - - # Step 2: ASM discovery against the auth_server_url (or - # server_url fallback for legacy providers). - for url in build_oauth_authorization_server_metadata_discovery_urls( - self.context.auth_server_url, server_url - ): - req = create_oauth_metadata_request(url) - try: - resp = await client.send(req) - except httpx.HTTPError as exc: - logger.debug( - "MCP OAuth '%s': ASM discovery to %s failed: %s", - self._hermes_server_name, url, exc, - ) - continue - ok, asm = await handle_auth_metadata_response(resp) - if not ok: - break - if asm: - self.context.oauth_metadata = asm - # Persist immediately so a subsequent cold-load can - # skip discovery entirely. - storage = self.context.storage - from tools.mcp_oauth import HermesTokenStorage - if isinstance(storage, HermesTokenStorage): - storage.save_oauth_metadata(asm) - logger.debug( - "MCP OAuth '%s': pre-flight ASM discovered " - "token_endpoint=%s", - self._hermes_server_name, asm.token_endpoint, - ) - break - - def _persist_oauth_metadata_if_changed(self) -> None: - """Persist discovered OAuth metadata for future process restarts. - - Called after the SDK's normal 401-branch auth flow completes so - metadata discovered via the lazy path (not pre-flight) is also - saved. No-op when nothing to persist or metadata hasn't changed. - """ - meta = self.context.oauth_metadata - if meta is None: - return - storage = self.context.storage - from tools.mcp_oauth import HermesTokenStorage - if not isinstance(storage, HermesTokenStorage): - return - existing = storage.load_oauth_metadata() - if ( - existing is None - or str(existing.token_endpoint) != str(meta.token_endpoint) - ): - storage.save_oauth_metadata(meta) - - async def _maybe_flag_poisoned_client(self, response: Any) -> None: - """Detect a dead client registration and force re-registration. - - When the IdP rejects our ``client_id`` with ``invalid_client`` on - the token endpoint (token exchange or refresh), the cached client - registration is provably dead server-side. We delete ``client.json`` - (+ stale metadata) so the SDK's next ``async_auth_flow`` takes the - ``if not client_info`` branch and re-runs RFC 7591 dynamic client - registration. This addresses the recurring manual-reset ritual in - GH#36767 for the auto-detectable subset (token-endpoint rejection); - the browser-side "Redirect URI Mismatch" case has no HTTP signal - and is handled by ``hermes mcp reauth``. - - Conservative by construction — acts ONLY when all hold: - * status is 400/401, - * the request hit the discovered ``token_endpoint`` (the only - request carrying our ``client_id``), and - * the body carries the ``invalid_client`` error code - (word-boundary match, so RFC 7591's ``invalid_client_metadata`` - registration error does not trip it). - Pre-registered (config-supplied) clients are never poisoned. - Fully best-effort: any failure here is swallowed so a detection - miss never breaks the live auth flow. - - Covers both the authorization-code token exchange and the - preemptive refresh — but only when ``token_endpoint`` was - discovered (``_initialize`` prefetches it on cold-load). If that - discovery was skipped, the guard returns early and the user falls - back to ``hermes mcp reauth``. - """ - try: - if self._hermes_preregistered: - return - status = getattr(response, "status_code", None) - if status not in (400, 401): - return - meta = getattr(self.context, "oauth_metadata", None) - token_endpoint = ( - str(meta.token_endpoint) - if meta is not None and getattr(meta, "token_endpoint", None) - else None - ) - req = getattr(response, "request", None) - req_url = str(req.url) if req is not None else None - if not token_endpoint or not req_url: - return - if not _same_endpoint(req_url, token_endpoint): - return - body = await response.aread() - # Word-boundary match: matches `"error":"invalid_client"` but - # not the RFC 7591 registration error `invalid_client_metadata` - # (the trailing `_metadata` removes the right-hand boundary). - if not re.search(rb"\binvalid_client\b", body.lower()): - return - - storage = self.context.storage - from tools.mcp_oauth import HermesTokenStorage - - # When the rejected client_id was our Client ID Metadata - # Document URL, re-presenting it next flow would loop: the - # server has already fetched that document and refused it. - # Dropping the URL sends the retry down the DCR branch - # instead, and the marker on disk keeps the next process from - # walking back into the same refusal. `hermes mcp login` - # clears the marker, so a fixed document gets another chance. - cimd_url = getattr(self.context, "client_metadata_url", None) - rejected_id = getattr(self.context.client_info, "client_id", None) - if cimd_url and rejected_id == cimd_url: - logger.warning( - "MCP OAuth '%s': authorization server rejected our " - "Client ID Metadata Document (%s) with invalid_client " - "— falling back to dynamic client registration.", - self._hermes_server_name, cimd_url, - ) - self.context.client_metadata_url = None - if isinstance(storage, HermesTokenStorage): - storage.mark_cimd_rejected() - - if isinstance(storage, HermesTokenStorage): - storage.poison_client_registration() - # Drop the in-memory client so the SDK re-registers next flow. - self.context.client_info = None - self._initialized = False - except Exception as exc: # pragma: no cover — defensive, must not throw - logger.debug( - "MCP OAuth '%s': invalid_client detection failed (non-fatal): %s", - self._hermes_server_name, exc, - ) - - async def async_auth_flow(self, request): # type: ignore[override] - # Pre-flow hook: ask the manager to refresh from disk if needed. - # Any failure here is non-fatal — we just log and proceed with - # whatever state the SDK already has. - try: - await get_manager().invalidate_if_disk_changed( - self._hermes_server_name, - hermes_home=self._hermes_home, - ) - except Exception as exc: # pragma: no cover — defensive - logger.debug( - "MCP OAuth '%s': pre-flow disk-watch failed (non-fatal): %s", - self._hermes_server_name, exc, - ) - - # Manually bridge the bidirectional generator protocol. httpx's - # auth_flow driver (httpx._client._send_handling_auth) calls - # ``auth_flow.asend(response)`` to feed HTTP responses back into - # the generator. A naive wrapper using ``async for item in inner: - # yield item`` DISCARDS those .asend(response) values and resumes - # the inner generator with None, so the SDK's - # ``response = yield request`` branch in - # mcp/client/auth/oauth2.py sees response=None and crashes at - # ``if response.status_code == 401`` with AttributeError. - # - # The bridge below forwards each .asend() value into the inner - # generator via inner.asend(incoming), preserving the bidirectional - # contract. Regression from PR #11383 caught by - # tests/tools/test_mcp_oauth_bidirectional.py. - inner = super().async_auth_flow(request) - try: - outgoing = await inner.__anext__() - while True: - incoming = yield outgoing - # Sniff the response for a dead-client-registration signal - # before handing it back to the SDK (best-effort, GH#36767). - await self._maybe_flag_poisoned_client(incoming) - outgoing = await inner.asend(incoming) - except StopAsyncIteration: - # Persist any metadata the SDK discovered lazily during the - # 401 branch so a subsequent cold-load skips discovery. - self._persist_oauth_metadata_if_changed() - return - - return HermesMCPOAuthProvider - - -# Cached at import time. Tested and used by :class:`MCPOAuthManager`. -_HERMES_PROVIDER_CLS: Optional[type] = _make_hermes_provider_class() - - -# --------------------------------------------------------------------------- -# Manager -# --------------------------------------------------------------------------- - - class MCPOAuthManager: - """Single source of truth for per-server MCP OAuth state. + """Single source of truth for scope-isolated MCP OAuth state. - Thread-safe: the ``_entries`` dict is guarded by ``_entries_lock`` for - get-or-create semantics. Per-entry state is guarded by the entry's own - ``asyncio.Lock`` (used from the MCP event loop thread). + ``oauth_scope`` parameters are optional only for API compatibility. When + omitted they are resolved from the trusted runtime context. Therefore an + omitted scope in ``per_user`` mode fails closed when no authenticated + principal is bound; it never falls back to shared credentials. """ def __init__(self) -> None: - self._entries: dict[tuple[str, str], _ProviderEntry] = {} + self._entries: dict[tuple[str, str, str], _ProviderEntry] = {} self._entries_lock = threading.Lock() - # Holds strong references to in-flight 401 handler tasks so the - # event loop's weak-reference bookkeeping cannot GC them mid-run - # and leave `await pending` waiters hanging forever. self._inflight_tasks: set[asyncio.Task] = set() - # -- Provider construction / caching ------------------------------------- + @staticmethod + def _resolve_scope(oauth_scope: McpOAuthScope | None) -> McpOAuthScope: + return oauth_scope if oauth_scope is not None else resolve_oauth_scope() + + @classmethod + def _key( + cls, + server_name: str, + hermes_home: str | Path | None = None, + oauth_scope: McpOAuthScope | None = None, + ) -> tuple[str, str, str]: + from hermes_constants import get_hermes_home + + home = Path(hermes_home) if hermes_home is not None else get_hermes_home() + scope = cls._resolve_scope(oauth_scope) + return ( + str(home.expanduser().resolve(strict=False)), + server_name, + scope.key, + ) def get_or_build_provider( self, server_name: str, server_url: str, oauth_config: Optional[dict], + *, + oauth_scope: McpOAuthScope | None = None, ) -> Optional[Any]: - """Return a cached OAuth provider for ``server_name`` or build one. - - Idempotent: repeat calls with the same name return the same instance. - If ``server_url`` changes for a given name, the cached entry is - discarded and a fresh provider is built. - - Returns None if the MCP SDK's OAuth support is unavailable. - """ - key = self._key(server_name) + """Return/build a provider for exactly one OAuth identity scope.""" + scope = self._resolve_scope(oauth_scope) + key = self._key(server_name, oauth_scope=scope) with self._entries_lock: entry = self._entries.get(key) if entry is not None and entry.server_url != server_url: logger.info( - "MCP OAuth '%s': URL changed from %s to %s, discarding cache", - server_name, entry.server_url, server_url, + "MCP OAuth '%s': URL changed; discarding provider for scope %s", + server_name, + scope.key, ) entry = None @@ -603,6 +102,7 @@ def get_or_build_provider( entry = _ProviderEntry( server_url=server_url, oauth_config=oauth_config, + oauth_scope=scope, ) self._entries[key] = entry @@ -613,47 +113,25 @@ def get_or_build_provider( return entry.provider - @staticmethod - def _key( - server_name: str, - hermes_home: str | Path | None = None, - ) -> tuple[str, str]: - from hermes_constants import get_hermes_home - - home = Path(hermes_home) if hermes_home is not None else get_hermes_home() - return (str(home.expanduser().resolve(strict=False)), server_name) - def _build_provider( self, server_name: str, entry: _ProviderEntry, ) -> Optional[Any]: - """Build the underlying OAuth provider. - - Constructs :class:`HermesMCPOAuthProvider` directly using the helpers - extracted from ``tools.mcp_oauth``. The subclass injects a pre-flow - disk-watch hook so external token refreshes (cron, other CLI - instances) are visible to running MCP sessions. - - Returns None if the MCP SDK's OAuth support is unavailable. - """ if _HERMES_PROVIDER_CLS is None: - logger.warning( - "MCP OAuth '%s': SDK auth module unavailable", server_name, - ) + logger.warning("MCP OAuth '%s': SDK auth module unavailable", server_name) return None - # Local imports avoid circular deps at module import time. from tools.mcp_oauth import ( - HermesTokenStorage, OAuthNonInteractiveError, _OAUTH_AVAILABLE, _build_client_metadata, _configure_callback_port, _is_interactive, - _maybe_preregister_client, _make_callback_waiter, _make_redirect_handler, + _maybe_preregister_client, + apply_oauth_provider_defaults, cimd_provider_kwargs, token_request_user_agent, ) @@ -662,12 +140,15 @@ def _build_provider( return None cfg = dict(entry.oauth_config or {}) - from tools.mcp_oauth import apply_oauth_provider_defaults - apply_oauth_provider_defaults( - cfg, server_name=server_name, server_url=entry.server_url + cfg, + server_name=server_name, + server_url=entry.server_url, ) - storage = HermesTokenStorage(server_name) + + # The complete OAuth persistence surface follows the same scope: + # tokens, DCR client info, AS metadata, and CIMD refusal marker. + storage = ScopedHermesTokenStorage(server_name, entry.oauth_scope) from tools.mcp_dashboard_oauth import get_dashboard_oauth_flow @@ -678,10 +159,9 @@ def _build_provider( ): raise OAuthNonInteractiveError( "MCP OAuth for " - f"'{server_name}': non-interactive environment and no " - "cached tokens found. Run `hermes mcp login " - f"{server_name}` interactively first to complete initial " - "authorization." + f"'{server_name}': non-interactive environment and no cached " + "tokens exist for the current OAuth identity. Complete " + "authorization from an interactive surface first." ) _configure_callback_port(cfg, storage) @@ -690,15 +170,15 @@ def _build_provider( resolved_port = cfg.get("_resolved_port", 0) redirect_handler = _make_redirect_handler(resolved_port) - # mcp 2.0 removed OAuthClientProvider's `timeout` argument, so the - # configured `oauth.timeout` now bounds the callback waiter's own poll - # loop instead — that is where the browser round-trip is awaited. callback_handler = _make_callback_waiter( - resolved_port, cfg.get("_cimd_url"), timeout=float(cfg.get("timeout", 300)) + resolved_port, + cfg.get("_cimd_url"), + timeout=float(cfg.get("timeout", 300)), ) return _HERMES_PROVIDER_CLS( server_name=server_name, + oauth_scope=entry.oauth_scope, preregistered=bool(cfg.get("client_id")), server_url=entry.server_url, client_metadata=client_metadata, @@ -714,20 +194,25 @@ def remove( server_name: str, *, hermes_home: str | Path | None = None, + oauth_scope: McpOAuthScope | None = None, ) -> _ProviderEntry | None: - """Evict the provider from cache AND delete tokens from disk. - - Called by ``hermes mcp remove `` and (indirectly) by - ``hermes mcp login `` during forced re-auth. - """ + """Evict and delete OAuth state for exactly one scope.""" + scope = self._resolve_scope(oauth_scope) with self._entries_lock: - entry = self._entries.pop(self._key(server_name, hermes_home), None) + entry = self._entries.pop( + self._key(server_name, hermes_home, scope), + None, + ) - from tools.mcp_oauth import remove_oauth_tokens - remove_oauth_tokens(server_name, hermes_home=hermes_home) + ScopedHermesTokenStorage( + server_name, + scope, + hermes_home=hermes_home, + ).remove() logger.info( - "MCP OAuth '%s': evicted from cache and removed from disk", + "MCP OAuth '%s': evicted provider and persisted state for scope %s", server_name, + scope.key, ) return entry @@ -737,115 +222,103 @@ def restore_entry( entry: _ProviderEntry | None, *, hermes_home: str | Path | None = None, + oauth_scope: McpOAuthScope | None = None, ) -> None: - """Restore a provider entry removed for a failed reauthorization.""" + """Restore a removed entry without overwriting a newer scoped entry.""" if entry is None: return + scope = oauth_scope if oauth_scope is not None else entry.oauth_scope with self._entries_lock: - self._entries.setdefault(self._key(server_name, hermes_home), entry) + self._entries.setdefault( + self._key(server_name, hermes_home, scope), + entry, + ) def evict( self, server_name: str, *, hermes_home: str | Path | None = None, + oauth_scope: McpOAuthScope | None = None, ) -> None: - """Drop only the in-process provider, preserving persisted OAuth state.""" + """Drop only the in-process provider for exactly one scope.""" + scope = self._resolve_scope(oauth_scope) with self._entries_lock: - self._entries.pop(self._key(server_name, hermes_home), None) - - # -- Disk watch ---------------------------------------------------------- + self._entries.pop(self._key(server_name, hermes_home, scope), None) async def invalidate_if_disk_changed( self, server_name: str, *, hermes_home: str | Path | None = None, + oauth_scope: McpOAuthScope | None = None, ) -> bool: - """If the tokens file on disk has a newer mtime than last-seen, force - the MCP SDK provider to reload its in-memory state. - - Returns True if the cache was invalidated (mtime differed). This is - the core fix for the external-refresh workflow: a cron job writes - fresh tokens to disk, and on the next tool call the running MCP - session picks them up without a restart. - """ - from tools.mcp_oauth import _get_token_dir, _safe_filename - - entry = self._entries.get(self._key(server_name, hermes_home)) + """Reload provider state only when THIS scope's token file changed.""" + scope = self._resolve_scope(oauth_scope) + entry = self._entries.get(self._key(server_name, hermes_home, scope)) if entry is None or entry.provider is None: return False async with entry.lock: - tokens_path = _get_token_dir(hermes_home) / f"{_safe_filename(server_name)}.json" + tokens_path = ScopedHermesTokenStorage( + server_name, + scope, + hermes_home=hermes_home, + )._tokens_path() try: mtime_ns = tokens_path.stat().st_mtime_ns except (FileNotFoundError, OSError): return False - if mtime_ns != entry.last_mtime_ns: - old = entry.last_mtime_ns - entry.last_mtime_ns = mtime_ns - # Force the SDK's OAuthClientProvider to reload from storage - # on its next auth flow. `_initialized` is private API but - # stable across the MCP SDK versions we pin (>=1.26.0). - if hasattr(entry.provider, "_initialized"): - entry.provider._initialized = False # noqa: SLF001 - logger.info( - "MCP OAuth '%s': tokens file changed (mtime %d -> %d), " - "forcing reload", - server_name, old, mtime_ns, - ) - return True - return False + if mtime_ns == entry.last_mtime_ns: + return False - # -- 401 handler (dedup'd) ----------------------------------------------- + old = entry.last_mtime_ns + entry.last_mtime_ns = mtime_ns + if hasattr(entry.provider, "_initialized"): + entry.provider._initialized = False # noqa: SLF001 + logger.info( + "MCP OAuth '%s': scoped token file changed " + "(mtime %d -> %d), forcing reload", + server_name, + old, + mtime_ns, + ) + return True async def handle_401( self, server_name: str, failed_access_token: Optional[str] = None, + *, + oauth_scope: McpOAuthScope | None = None, ) -> bool: - """Handle a 401 from a tool call, deduplicated across concurrent callers. - - Returns: - True if a (possibly new) access token is now available — caller - should trigger a reconnect and retry the operation. - False if no recovery path exists — caller should surface a - ``needs_reauth`` error to the model so it stops hallucinating - manual refresh attempts. - - Thundering-herd protection: if N concurrent tool calls hit 401 with - the same ``failed_access_token``, only one recovery attempt fires. - Others await the same future. - """ - entry = self._entries.get(self._key(server_name)) + """Recover a 401 without sharing dedup/refresh state across users.""" + scope = self._resolve_scope(oauth_scope) + entry = self._entries.get(self._key(server_name, oauth_scope=scope)) if entry is None or entry.provider is None: return False - key = failed_access_token or "" + failed_key = failed_access_token or "" loop = asyncio.get_running_loop() async with entry.lock: - pending = entry.pending_401.get(key) + pending = entry.pending_401.get(failed_key) if pending is None: pending = loop.create_future() - entry.pending_401[key] = pending + entry.pending_401[failed_key] = pending async def _do_handle() -> None: try: - # Step 1: Did disk change? Picks up external refresh. disk_changed = await self.invalidate_if_disk_changed( - server_name + server_name, + oauth_scope=scope, ) if disk_changed: if not pending.done(): pending.set_result(True) return - # Step 2: No disk change — if the SDK can refresh - # in-place, let the caller retry. The SDK's httpx.Auth - # flow will issue the refresh on the next request. provider = entry.provider ctx = getattr(provider, "context", None) can_refresh = False @@ -858,15 +331,16 @@ async def _do_handle() -> None: can_refresh = False if not pending.done(): pending.set_result(can_refresh) - except Exception as exc: # pragma: no cover — defensive + except Exception as exc: # pragma: no cover - defensive logger.warning( - "MCP OAuth '%s': 401 handler failed: %s", - server_name, exc, + "MCP OAuth '%s': scoped 401 handler failed: %s", + server_name, + exc, ) if not pending.done(): pending.set_result(False) finally: - entry.pending_401.pop(key, None) + entry.pending_401.pop(failed_key, None) task = asyncio.create_task(_do_handle()) self._inflight_tasks.add(task) @@ -874,25 +348,21 @@ async def _do_handle() -> None: try: return await pending - except Exception as exc: # pragma: no cover — defensive + except Exception as exc: # pragma: no cover - defensive logger.warning( - "MCP OAuth '%s': awaiting 401 handler failed: %s", - server_name, exc, + "MCP OAuth '%s': awaiting scoped 401 handler failed: %s", + server_name, + exc, ) return False -# --------------------------------------------------------------------------- -# Module-level singleton -# --------------------------------------------------------------------------- - - _MANAGER: Optional[MCPOAuthManager] = None _MANAGER_LOCK = threading.Lock() def get_manager() -> MCPOAuthManager: - """Return the process-wide :class:`MCPOAuthManager` singleton.""" + """Return the process-wide scoped MCP OAuth manager singleton.""" global _MANAGER with _MANAGER_LOCK: if _MANAGER is None: @@ -901,7 +371,6 @@ def get_manager() -> MCPOAuthManager: def reset_manager_for_tests() -> None: - """Test-only helper: drop the singleton so fixtures start clean.""" global _MANAGER with _MANAGER_LOCK: _MANAGER = None From a02b98847a4cc1bc2abc497b51bd3d88094e194e Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:08:26 +0900 Subject: [PATCH 06/16] feat(mcp): isolate long-lived OAuth MCP connections by scope --- tools/mcp_oauth_runtime.py | 177 +++++++++++++++++++++++++++++++++++++ 1 file changed, 177 insertions(+) create mode 100644 tools/mcp_oauth_runtime.py diff --git a/tools/mcp_oauth_runtime.py b/tools/mcp_oauth_runtime.py new file mode 100644 index 000000000000..39fcb67ae70f --- /dev/null +++ b/tools/mcp_oauth_runtime.py @@ -0,0 +1,177 @@ +"""Scope-aware adapter for the MCP runtime's long-lived server registry. + +``tools.mcp_tool`` historically indexes ``_servers`` by logical server name. +Changing only token files/provider caches would therefore leave a critical +cross-user path: User B could obtain User A's already-authenticated +``MCPServerTask``. This module upgrades that registry in-place to exact +``(server, OAuthScope)`` lookup while preserving the existing dictionary API +used throughout the large MCP runtime. + +The adapter is installed only when an OAuth server is prepared. Non-OAuth MCP +servers retain their historical shared connection key. Per-user lookups have +NO "find any server with the same name" fallback: missing scope/connection +returns missing and the existing lazy-connect path can create the caller's own +connection. + +This compatibility layer keeps #78174 narrowly scoped instead of spreading +identity plumbing through every MCP handler. Scope selection remains outside +the model/tool argument surface and is driven by trusted ContextVars. +""" + +from __future__ import annotations + +import threading +from contextvars import ContextVar +from typing import Any + +from tools.mcp_oauth_identity import ( + McpOAuthScope, + MissingMcpOAuthIdentityError, + connection_registry_key, + try_resolve_oauth_scope, +) + +# A long-lived MCPServerTask keeps the context captured when it was created. +# Binding the already-resolved scope here also makes registry operations robust +# if a host's async scheduling implementation does not propagate gateway +# ContextVars exactly as CPython's run_coroutine_threadsafe currently does. +_TASK_SCOPES: ContextVar[dict[str, McpOAuthScope]] = ContextVar( + "mcp_oauth_runtime_task_scopes", + default={}, +) + +_INSTALL_LOCK = threading.Lock() + + +def _task_scope(server_name: str) -> McpOAuthScope | None: + explicit = _TASK_SCOPES.get().get(server_name) + if explicit is not None: + return explicit + return try_resolve_oauth_scope() + + +def bind_runtime_scope(server_name: str, scope: McpOAuthScope) -> None: + """Bind a resolved scope to the current server task's ContextVar state.""" + current = dict(_TASK_SCOPES.get()) + current[server_name] = scope + _TASK_SCOPES.set(current) + + +class ScopedMCPServerRegistry(dict): + """Dict-compatible registry with exact per-user keys for OAuth servers.""" + + def __init__(self, initial: dict[str, Any] | None = None) -> None: + super().__init__(initial or {}) + self.oauth_servers: set[str] = set() + + def mark_oauth_server(self, server_name: str) -> None: + self.oauth_servers.add(server_name) + + def _key_for_read(self, key: Any) -> Any: + if not isinstance(key, str) or key not in self.oauth_servers: + return key + # Encoded/internal keys pass through. They are intentionally opaque. + if "@@u-v1-" in key: + return key + scope = _task_scope(key) + if scope is None: + # Missing authenticated identity must never select a shared/other + # connection. A guaranteed-missing key gives ordinary dict callers + # fail-closed semantics without changing every call site. + return f"{key}@@" + return connection_registry_key(key, scope) + + def _key_for_write(self, key: Any) -> Any: + if not isinstance(key, str) or key not in self.oauth_servers: + return key + if "@@u-v1-" in key: + return key + scope = _task_scope(key) + if scope is None: + raise MissingMcpOAuthIdentityError( + f"refusing to register OAuth MCP server {key!r} without an " + "authenticated per-user scope" + ) + return connection_registry_key(key, scope) + + def __getitem__(self, key: Any) -> Any: + return super().__getitem__(self._key_for_read(key)) + + def __setitem__(self, key: Any, value: Any) -> None: + super().__setitem__(self._key_for_write(key), value) + + def __contains__(self, key: object) -> bool: + return super().__contains__(self._key_for_read(key)) + + def get(self, key: Any, default: Any = None) -> Any: + return super().get(self._key_for_read(key), default) + + def pop(self, key: Any, default: Any = ...): + resolved = self._key_for_read(key) + if default is ...: + return super().pop(resolved) + return super().pop(resolved, default) + + def setdefault(self, key: Any, default: Any = None) -> Any: + return super().setdefault(self._key_for_write(key), default) + + +class PersistentPerUserLazyConfigs(dict): + """Keep OAuth server config available for every user's first connection. + + ``mcp_tool._ensure_lazy_server_connected`` historically pops a lazy config + after one successful connection because there was only one global server. + Per-user OAuth has N independent long-lived connections, so that config is + reusable metadata rather than one-shot state. + """ + + def __init__(self, initial: dict[str, Any] | None = None) -> None: + super().__init__(initial or {}) + self.oauth_servers: set[str] = set() + + def mark_oauth_server(self, server_name: str) -> None: + self.oauth_servers.add(server_name) + + def pop(self, key: Any, default: Any = ...): + if isinstance(key, str) and key in self.oauth_servers and key in self: + # Return without deleting so Alice's successful lazy connect does + # not prevent Bob/Carol from creating their own scoped connection. + return self[key] + if default is ...: + return super().pop(key) + return super().pop(key, default) + + +def prepare_oauth_server_runtime(server_name: str) -> None: + """Install/mark scope-aware MCP registries before resolving identity. + + This is intentionally called before ``resolve_oauth_scope``. On a headless + gateway startup with no human bound, the provider build then fails closed + but the runtime is already prepared for a later authenticated user's lazy + connection. + """ + from tools import mcp_tool + + with _INSTALL_LOCK: + servers = mcp_tool._servers + if not isinstance(servers, ScopedMCPServerRegistry): + servers = ScopedMCPServerRegistry(dict(servers)) + mcp_tool._servers = servers + servers.mark_oauth_server(server_name) + + lazy = mcp_tool._lazy_server_configs + if not isinstance(lazy, PersistentPerUserLazyConfigs): + lazy = PersistentPerUserLazyConfigs(dict(lazy)) + mcp_tool._lazy_server_configs = lazy + lazy.mark_oauth_server(server_name) + + # Preserve the full connection config for later users. This is safe + # config metadata, not credential state; actual OAuth providers/tokens + # are still created under the caller's immutable scope. + if server_name not in lazy: + try: + config = (mcp_tool._load_mcp_config() or {}).get(server_name) + except Exception: + config = None + if isinstance(config, dict): + dict.__setitem__(lazy, server_name, config) From ba0715a8f65df8e4fb0c961f3602861c3f2ba259 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:09:02 +0900 Subject: [PATCH 07/16] feat(mcp): bind scoped OAuth providers to MCP runtime --- tools/mcp_oauth_manager.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tools/mcp_oauth_manager.py b/tools/mcp_oauth_manager.py index 5bcb64b7e9e6..617e765caa8f 100644 --- a/tools/mcp_oauth_manager.py +++ b/tools/mcp_oauth_manager.py @@ -86,7 +86,19 @@ def get_or_build_provider( oauth_scope: McpOAuthScope | None = None, ) -> Optional[Any]: """Return/build a provider for exactly one OAuth identity scope.""" + # Prepare the string-keyed MCP runtime before identity resolution. This + # matters on headless startup: resolution must fail closed without a + # human, while a later authenticated request still needs a per-user + # lazy connection rather than reusing a different user's task. + from tools.mcp_oauth_runtime import ( + bind_runtime_scope, + prepare_oauth_server_runtime, + ) + + prepare_oauth_server_runtime(server_name) scope = self._resolve_scope(oauth_scope) + bind_runtime_scope(server_name, scope) + key = self._key(server_name, oauth_scope=scope) with self._entries_lock: entry = self._entries.get(key) From 01867919dcc68ff31ab509bed0aaa85ca0a33b95 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:09:36 +0900 Subject: [PATCH 08/16] test(mcp): prove OAuth connection registry fails closed per user --- tests/tools/test_mcp_oauth_runtime_scope.py | 139 ++++++++++++++++++++ 1 file changed, 139 insertions(+) create mode 100644 tests/tools/test_mcp_oauth_runtime_scope.py diff --git a/tests/tools/test_mcp_oauth_runtime_scope.py b/tests/tools/test_mcp_oauth_runtime_scope.py new file mode 100644 index 000000000000..fbd65e72383a --- /dev/null +++ b/tests/tools/test_mcp_oauth_runtime_scope.py @@ -0,0 +1,139 @@ +"""Adversarial tests for scoped long-lived MCP server lookup (#78174).""" + +from __future__ import annotations + +import pytest + + +@pytest.fixture(autouse=True) +def _clear_session_context(): + from gateway.session_context import clear_session_vars + + clear_session_vars() + yield + clear_session_vars() + + +def _patch_per_user(monkeypatch): + monkeypatch.setattr( + "hermes_cli.config.load_config", + lambda: {"mcp": {"oauth": {"identity_mode": "per_user"}}}, + ) + + +def _bind(user_id: str, *, scope_id: str = "T_ONE") -> None: + from gateway.session_context import set_session_vars + + set_session_vars( + platform="slack", + scope_id=scope_id, + user_id=user_id, + ) + + +def test_registry_never_falls_back_to_another_users_live_server(monkeypatch): + from tools.mcp_oauth_identity import resolve_oauth_scope + from tools.mcp_oauth_runtime import ScopedMCPServerRegistry, bind_runtime_scope + + _patch_per_user(monkeypatch) + registry = ScopedMCPServerRegistry() + registry.mark_oauth_server("github") + + alice_server = object() + bob_server = object() + + _bind("U_ALICE") + alice_scope = resolve_oauth_scope() + bind_runtime_scope("github", alice_scope) + registry["github"] = alice_server + + # A fresh task/request binding must not inherit the previous explicit + # runtime map; bind Bob explicitly as a server task would after provider + # construction. + _bind("U_BOB") + bob_scope = resolve_oauth_scope() + bind_runtime_scope("github", bob_scope) + + assert registry.get("github") is None + registry["github"] = bob_server + assert registry.get("github") is bob_server + + _bind("U_ALICE") + bind_runtime_scope("github", alice_scope) + assert registry.get("github") is alice_server + + +def test_registry_missing_identity_is_not_shared_fallback(monkeypatch): + from tools.mcp_oauth_runtime import ScopedMCPServerRegistry + + _patch_per_user(monkeypatch) + registry = ScopedMCPServerRegistry() + registry.mark_oauth_server("github") + + # Seed an encoded Alice entry without binding any request principal. + dict.__setitem__(registry, "github@@u-v1-deadbeef", object()) + + assert registry.get("github") is None + assert "github" not in registry + + +def test_non_oauth_servers_keep_historical_shared_registry_semantics(monkeypatch): + from tools.mcp_oauth_runtime import ScopedMCPServerRegistry + + _patch_per_user(monkeypatch) + registry = ScopedMCPServerRegistry() + filesystem = object() + + registry["filesystem"] = filesystem + _bind("U_ALICE") + assert registry.get("filesystem") is filesystem + _bind("U_BOB") + assert registry.get("filesystem") is filesystem + + +def test_lazy_oauth_config_survives_multiple_users(): + from tools.mcp_oauth_runtime import PersistentPerUserLazyConfigs + + config = {"url": "https://mcp.example/mcp", "auth": "oauth"} + lazy = PersistentPerUserLazyConfigs({"github": config}) + lazy.mark_oauth_server("github") + + assert lazy.pop("github") == config + assert "github" in lazy + assert lazy.pop("github") == config + assert "github" in lazy + + +def test_provider_manager_cache_isolated_by_principal(monkeypatch): + """Provider entries for Alice and Bob cannot share locks/401 state.""" + from tools.mcp_oauth_identity import resolve_oauth_scope + from tools.mcp_oauth_manager import MCPOAuthManager, _ProviderEntry + + _patch_per_user(monkeypatch) + manager = MCPOAuthManager() + + _bind("U_ALICE") + alice = resolve_oauth_scope() + _bind("U_BOB") + bob = resolve_oauth_scope() + + manager._entries[manager._key("github", oauth_scope=alice)] = _ProviderEntry( + server_url="https://mcp.example/mcp", + oauth_config=None, + oauth_scope=alice, + provider=object(), + ) + manager._entries[manager._key("github", oauth_scope=bob)] = _ProviderEntry( + server_url="https://mcp.example/mcp", + oauth_config=None, + oauth_scope=bob, + provider=object(), + ) + + alice_entry = manager._entries[manager._key("github", oauth_scope=alice)] + bob_entry = manager._entries[manager._key("github", oauth_scope=bob)] + + assert alice_entry is not bob_entry + assert alice_entry.provider is not bob_entry.provider + assert alice_entry.lock is not bob_entry.lock + assert alice_entry.pending_401 is not bob_entry.pending_401 From e15a059988a2b493919ce03157207613e2646e73 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:10:25 +0900 Subject: [PATCH 09/16] fix(mcp): pin OAuth scope to server task without context bleed --- tools/mcp_oauth_runtime.py | 72 ++++++++++++++++++++++---------------- 1 file changed, 41 insertions(+), 31 deletions(-) diff --git a/tools/mcp_oauth_runtime.py b/tools/mcp_oauth_runtime.py index 39fcb67ae70f..a71dfab83f96 100644 --- a/tools/mcp_oauth_runtime.py +++ b/tools/mcp_oauth_runtime.py @@ -14,14 +14,15 @@ connection. This compatibility layer keeps #78174 narrowly scoped instead of spreading -identity plumbing through every MCP handler. Scope selection remains outside +identity plumbing through every MCP handler. Scope selection remains outside the model/tool argument surface and is driven by trusted ContextVars. """ from __future__ import annotations +import asyncio import threading -from contextvars import ContextVar +import weakref from typing import Any from tools.mcp_oauth_identity import ( @@ -31,30 +32,50 @@ try_resolve_oauth_scope, ) -# A long-lived MCPServerTask keeps the context captured when it was created. -# Binding the already-resolved scope here also makes registry operations robust -# if a host's async scheduling implementation does not propagate gateway -# ContextVars exactly as CPython's run_coroutine_threadsafe currently does. -_TASK_SCOPES: ContextVar[dict[str, McpOAuthScope]] = ContextVar( - "mcp_oauth_runtime_task_scopes", - default={}, +# Some hosts may not preserve caller ContextVars when moving work onto the +# dedicated MCP event loop. Pin the scope to the actual long-lived asyncio Task +# as a fallback. Weak keys disappear automatically when the task exits, so a +# reused request ContextVar can never inherit another user's pin. +_TASK_SCOPES: "weakref.WeakKeyDictionary[asyncio.Task, dict[str, McpOAuthScope]]" = ( + weakref.WeakKeyDictionary() ) - +_TASK_SCOPE_LOCK = threading.Lock() _INSTALL_LOCK = threading.Lock() def _task_scope(server_name: str) -> McpOAuthScope | None: - explicit = _TASK_SCOPES.get().get(server_name) - if explicit is not None: - return explicit - return try_resolve_oauth_scope() + # Prefer a currently bound authenticated request principal. This lets a + # reused host task move safely between users instead of being stuck on an + # earlier runtime pin. + ambient = try_resolve_oauth_scope() + if ambient is not None: + return ambient + + try: + task = asyncio.current_task() + except RuntimeError: + task = None + if task is None: + return None + with _TASK_SCOPE_LOCK: + scopes = _TASK_SCOPES.get(task) + return scopes.get(server_name) if scopes else None def bind_runtime_scope(server_name: str, scope: McpOAuthScope) -> None: - """Bind a resolved scope to the current server task's ContextVar state.""" - current = dict(_TASK_SCOPES.get()) - current[server_name] = scope - _TASK_SCOPES.set(current) + """Pin a resolved OAuth scope to the current long-lived server task.""" + try: + task = asyncio.current_task() + except RuntimeError: + task = None + if task is None: + # Sync callers already have their gateway ContextVars, and there is no + # safe process-global fallback. Do not manufacture one. + return + with _TASK_SCOPE_LOCK: + current = dict(_TASK_SCOPES.get(task) or {}) + current[server_name] = scope + _TASK_SCOPES[task] = current class ScopedMCPServerRegistry(dict): @@ -70,14 +91,10 @@ def mark_oauth_server(self, server_name: str) -> None: def _key_for_read(self, key: Any) -> Any: if not isinstance(key, str) or key not in self.oauth_servers: return key - # Encoded/internal keys pass through. They are intentionally opaque. if "@@u-v1-" in key: return key scope = _task_scope(key) if scope is None: - # Missing authenticated identity must never select a shared/other - # connection. A guaranteed-missing key gives ordinary dict callers - # fail-closed semantics without changing every call site. return f"{key}@@" return connection_registry_key(key, scope) @@ -134,8 +151,6 @@ def mark_oauth_server(self, server_name: str) -> None: def pop(self, key: Any, default: Any = ...): if isinstance(key, str) and key in self.oauth_servers and key in self: - # Return without deleting so Alice's successful lazy connect does - # not prevent Bob/Carol from creating their own scoped connection. return self[key] if default is ...: return super().pop(key) @@ -145,10 +160,8 @@ def pop(self, key: Any, default: Any = ...): def prepare_oauth_server_runtime(server_name: str) -> None: """Install/mark scope-aware MCP registries before resolving identity. - This is intentionally called before ``resolve_oauth_scope``. On a headless - gateway startup with no human bound, the provider build then fails closed - but the runtime is already prepared for a later authenticated user's lazy - connection. + Called before ``resolve_oauth_scope`` so headless startup can fail closed + yet retain connection config for a later authenticated user's lazy start. """ from tools import mcp_tool @@ -165,9 +178,6 @@ def prepare_oauth_server_runtime(server_name: str) -> None: mcp_tool._lazy_server_configs = lazy lazy.mark_oauth_server(server_name) - # Preserve the full connection config for later users. This is safe - # config metadata, not credential state; actual OAuth providers/tokens - # are still created under the caller's immutable scope. if server_name not in lazy: try: config = (mcp_tool._load_mcp_config() or {}).get(server_name) From 715aa550da801b42e96a472a3c0cdd732b35504d Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:12:16 +0900 Subject: [PATCH 10/16] fix(mcp): keep OAuth rollback inside user scope --- tools/mcp_oauth_scoped_storage.py | 60 +++++++++++++++++++++++++++---- 1 file changed, 54 insertions(+), 6 deletions(-) diff --git a/tools/mcp_oauth_scoped_storage.py b/tools/mcp_oauth_scoped_storage.py index e26d9dd82bb3..bc80f5ee8d8b 100644 --- a/tools/mcp_oauth_scoped_storage.py +++ b/tools/mcp_oauth_scoped_storage.py @@ -1,22 +1,28 @@ """Scope-aware persistent storage for MCP OAuth state. The existing :class:`tools.mcp_oauth.HermesTokenStorage` remains the storage -implementation (permissions, atomic writes, expiry handling, snapshots). This -subclass changes only path resolution so every piece of OAuth state follows the -same immutable :class:`McpOAuthScope` boundary. +implementation (permissions, atomic writes, expiry handling, snapshots). This +subclass changes path resolution so every piece of OAuth state follows the same +immutable :class:`McpOAuthScope` boundary. -Shared mode intentionally preserves the historical layout exactly. Per-user +Shared mode intentionally preserves the historical layout exactly. Per-user mode stores all state under an opaque digest directory; legacy shared tokens are never migrated or read as a fallback because their human owner is unknowable. """ from __future__ import annotations +import logging +import os +import stat from pathlib import Path +from hermes_constants import secure_parent_dir from tools.mcp_oauth import HermesTokenStorage, _get_token_dir from tools.mcp_oauth_identity import McpOAuthScope +logger = logging.getLogger(__name__) + class ScopedHermesTokenStorage(HermesTokenStorage): """Hermes OAuth storage whose complete state is bound to one OAuth scope.""" @@ -39,8 +45,6 @@ def _scoped_token_dir(self) -> Path: root = _get_token_dir(self._hermes_home) if not self._oauth_scope.is_per_user: return root - # Opaque digest only: never expose Slack/Discord/etc identifiers in - # directory names, logs, backups, or path traversal diagnostics. return root / "by-user" / self._oauth_scope.key def _tokens_path(self) -> Path: @@ -54,3 +58,47 @@ def _meta_path(self) -> Path: def _cimd_rejected_path(self) -> Path: return self._scoped_token_dir() / f"{self._server_name}.cimd-off" + + def restore(self, snapshot: dict[str, bytes], *, only_if_absent: bool = False) -> None: + """Restore a reauth snapshot into THIS scope, never the shared root. + + ``HermesTokenStorage.restore`` historically reconstructs its destination + directly from ``_get_token_dir`` because there was only one storage + namespace. Calling that implementation from a scoped subclass would + therefore copy Alice's rollback data into the legacy shared namespace. + Keep the same concurrency and file-permission semantics while resolving + the destination through ``_scoped_token_dir``. + """ + if only_if_absent and any( + path.exists() + for path in (self._tokens_path(), self._client_info_path(), self._meta_path()) + ): + logger.info( + "Skipping OAuth rollback for %s because newer scoped state exists", + self._server_name, + ) + return + + self.remove() + if not snapshot: + return + + token_dir = self._scoped_token_dir() + token_dir.mkdir(parents=True, exist_ok=True) + secure_parent_dir(token_dir / ".scope-permissions") + for fname, data in snapshot.items(): + # Snapshot names are produced only by ``snapshot()`` from the + # storage's own known paths. Still collapse to basename so a + # corrupted/injected snapshot cannot escape this principal dir. + safe_name = Path(fname).name + path = token_dir / safe_name + try: + fd = os.open( + str(path), + os.O_WRONLY | os.O_CREAT | os.O_TRUNC, + stat.S_IRUSR | stat.S_IWUSR, + ) + with os.fdopen(fd, "wb") as fh: + fh.write(data) + except OSError as exc: + logger.warning("Failed to restore scoped OAuth state %s: %s", safe_name, exc) From b353253be94af556fd358ae8751edfb8b00564e5 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:12:30 +0900 Subject: [PATCH 11/16] test(mcp): keep OAuth snapshot rollback user-scoped --- tests/tools/test_mcp_oauth_scoped_storage.py | 45 ++++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 tests/tools/test_mcp_oauth_scoped_storage.py diff --git a/tests/tools/test_mcp_oauth_scoped_storage.py b/tests/tools/test_mcp_oauth_scoped_storage.py new file mode 100644 index 000000000000..c38a1885d660 --- /dev/null +++ b/tests/tools/test_mcp_oauth_scoped_storage.py @@ -0,0 +1,45 @@ +"""Regression tests for per-user OAuth persistence rollback.""" + +from __future__ import annotations + + +def _scope(user_id: str): + from tools.mcp_oauth_identity import McpOAuthPrincipal, McpOAuthScope + + return McpOAuthScope( + "per_user", + McpOAuthPrincipal("slack", "T_ONE", user_id), + ) + + +def test_restore_never_writes_snapshot_into_legacy_shared_namespace(tmp_path): + from tools.mcp_oauth import HermesTokenStorage + from tools.mcp_oauth_scoped_storage import ScopedHermesTokenStorage + + alice = ScopedHermesTokenStorage("github", _scope("U_ALICE"), hermes_home=tmp_path) + legacy = HermesTokenStorage("github", hermes_home=tmp_path) + + alice.restore( + { + "github.json": b'{"access_token":"ALICE"}', + "github.client.json": b'{"client_id":"alice-client"}', + "github.meta.json": b'{"token_endpoint":"https://idp.example/token"}', + } + ) + + assert alice._tokens_path().read_bytes() == b'{"access_token":"ALICE"}' + assert alice._client_info_path().exists() + assert alice._meta_path().exists() + assert not legacy._tokens_path().exists() + assert not legacy._client_info_path().exists() + assert not legacy._meta_path().exists() + + +def test_restore_basename_guard_cannot_escape_principal_directory(tmp_path): + from tools.mcp_oauth_scoped_storage import ScopedHermesTokenStorage + + alice = ScopedHermesTokenStorage("github", _scope("U_ALICE"), hermes_home=tmp_path) + alice.restore({"../../escape.json": b"{}"}) + + assert (alice._scoped_token_dir() / "escape.json").exists() + assert not (tmp_path / "escape.json").exists() From 6c866898f470e03102a80c0cfc641b5c0e6ee19d Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:13:09 +0900 Subject: [PATCH 12/16] fix(mcp): isolate schema cache for per-user OAuth --- tools/mcp_schema_cache.py | 107 ++++++++++++++++++++++++++------------ 1 file changed, 75 insertions(+), 32 deletions(-) diff --git a/tools/mcp_schema_cache.py b/tools/mcp_schema_cache.py index 48bb5ae665ce..7568ae94fa81 100644 --- a/tools/mcp_schema_cache.py +++ b/tools/mcp_schema_cache.py @@ -1,9 +1,11 @@ """Persistent MCP tool-schema cache for lazy server startup. -Stores per-server tool manifests on disk so Hermes can register MCP tools -into the agent snapshot without spawning the stdio child process at idle -dashboard startup. Cache entries are keyed by server name + a fingerprint -of the connection config (command/args/url/tools filters). +Stores tool manifests on disk so Hermes can register MCP tools without eagerly +spawning a server. Historically this file assumed the cache was "per-user local +disk". That assumption is false for a shared Hermes gateway: many authenticated +humans use the same HERMES_HOME. In ``mcp.oauth.identity_mode: per_user``, OAuth +server cache entries therefore use the same opaque requesting-user scope as the +credential/connection boundary (#78174). """ from __future__ import annotations @@ -58,25 +60,56 @@ def _load_all() -> Dict[str, Any]: def _save_all(data: Dict[str, Any]) -> None: from utils import atomic_json_write - # Cache dir + 0o600: sibling precedent in tools/registry.py - # _save_discovery_cache; the cache file is trusted input on the lazy - # registration path, so keep it user-only. atomic_json_write(_cache_path(), data, mode=0o600) +def _is_per_user_oauth_server(server_name: str) -> bool: + """Return True only for OAuth servers under explicit per-user mode.""" + try: + from tools.mcp_oauth_identity import get_oauth_identity_mode + + if get_oauth_identity_mode() != "per_user": + return False + from hermes_cli.config import load_config + + servers = (load_config() or {}).get("mcp_servers") or {} + config = servers.get(server_name) if isinstance(servers, dict) else None + return isinstance(config, dict) and str(config.get("auth") or "").strip().lower() == "oauth" + except Exception: + # Invalid identity configuration is not a reason to read a less-scoped + # cache. The caller will surface the configuration error elsewhere. + return False + + +def _scoped_cache_key(server_name: str) -> str | None: + """Return the on-disk key, or None when per-user identity is unavailable.""" + if not _is_per_user_oauth_server(server_name): + return server_name + + from tools.mcp_oauth_identity import try_resolve_oauth_scope + + scope = try_resolve_oauth_scope() + if scope is None: + # Headless startup has no authenticated human. Serving another user's + # private schema would leak capability metadata and could register tools + # that the current principal is not entitled to see. + return None + return f"{server_name}@@{scope.key}" + + def get_cached_entry(server_name: str, fingerprint: str) -> Optional[dict]: - """Return cached entry when fingerprint matches (and TTL holds), else None. - - MCP 2026-07-28 (SEP-2549): ``tools/list`` results carry ``ttlMs`` as a - freshness hint. When the live discovery path recorded one, an entry - older than its TTL is treated as a miss so the next startup re-probes - the server instead of serving a stale manifest forever. Entries without - a recorded TTL (pre-2026 servers) keep the old never-expires behavior. - ``cacheScope`` is irrelevant here: this cache is per-user local disk, - which satisfies even ``private``. + """Return a valid entry for the exact current identity scope. + + MCP 2026-07-28 (SEP-2549) ``ttlMs`` freshness hints are preserved. In + per-user OAuth mode we scope all entries, not only ones explicitly marked + ``private``: doing so is conservative and avoids depending on an untrusted + or older server to classify user-specific capability schemas correctly. """ + cache_key = _scoped_cache_key(server_name) + if cache_key is None: + return None with _cache_lock: - entry = _load_all().get(server_name) + entry = _load_all().get(cache_key) if not isinstance(entry, dict): return None if entry.get("fingerprint") != fingerprint: @@ -102,12 +135,13 @@ def write_cache_entry( ttl_ms: Optional[float] = None, cache_scope: Optional[str] = None, ) -> None: - """Persist tool schemas after a successful live connect. + """Persist schemas under the exact current requesting-user scope.""" + cache_key = _scoped_cache_key(server_name) + if cache_key is None: + # Never write an anonymous/shared cache entry while per-user OAuth is + # configured but no authenticated principal is bound. + return - ``ttl_ms``/``cache_scope`` are the SEP-2549 hints from the server's - ``tools/list`` result (2026-07-28 servers). ``written_at`` anchors TTL - expiry in :func:`get_cached_entry`. - """ entry = { "fingerprint": fingerprint, "tools": tools, @@ -120,23 +154,32 @@ def write_cache_entry( entry["cache_scope"] = cache_scope with _cache_lock: data = _load_all() - # Write-through fires on every registration (reconnects, - # list_changed refreshes); skip the load-all+rewrite churn when the - # entry is byte-identical to what is already on disk. TTL'd entries - # always rewrite: written_at must advance or the entry would expire - # at its ORIGINAL write time no matter how many live reconnects - # confirmed it since. - if "written_at" not in entry and data.get(server_name) == entry: + if "written_at" not in entry and data.get(cache_key) == entry: return - data[server_name] = entry + data[cache_key] = entry _save_all(data) def clear_cache_entry(server_name: str) -> None: + """Clear current scoped entry, or all scoped entries from admin context.""" + cache_key = _scoped_cache_key(server_name) with _cache_lock: data = _load_all() - if server_name in data: - del data[server_name] + changed = False + if cache_key is not None: + if cache_key in data: + del data[cache_key] + changed = True + elif _is_per_user_oauth_server(server_name): + # No principal is bound (e.g. config/admin maintenance). Clearing is + # intentionally destructive across this logical server's cache + # entries but does not grant access to any cached content. + prefix = f"{server_name}@@u-v1-" + for key in list(data): + if key.startswith(prefix): + del data[key] + changed = True + if changed: _save_all(data) From 149d1898231c965d873598651e4e31b88284e914 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:13:44 +0900 Subject: [PATCH 13/16] fix(mcp): fail closed on invalid OAuth cache identity config --- tools/mcp_schema_cache.py | 37 ++++++++++++++++++------------------- 1 file changed, 18 insertions(+), 19 deletions(-) diff --git a/tools/mcp_schema_cache.py b/tools/mcp_schema_cache.py index 7568ae94fa81..062eb5aadc5f 100644 --- a/tools/mcp_schema_cache.py +++ b/tools/mcp_schema_cache.py @@ -64,22 +64,26 @@ def _save_all(data: Dict[str, Any]) -> None: def _is_per_user_oauth_server(server_name: str) -> bool: - """Return True only for OAuth servers under explicit per-user mode.""" - try: - from tools.mcp_oauth_identity import get_oauth_identity_mode - - if get_oauth_identity_mode() != "per_user": - return False - from hermes_cli.config import load_config - - servers = (load_config() or {}).get("mcp_servers") or {} - config = servers.get(server_name) if isinstance(servers, dict) else None - return isinstance(config, dict) and str(config.get("auth") or "").strip().lower() == "oauth" - except Exception: - # Invalid identity configuration is not a reason to read a less-scoped - # cache. The caller will surface the configuration error elsewhere. + """Return True only for OAuth servers under explicit per-user mode. + + Invalid identity-mode configuration deliberately propagates from + ``get_oauth_identity_mode``. A typo must not turn a scoped/private cache + read into a legacy shared-cache read. + """ + from tools.mcp_oauth_identity import get_oauth_identity_mode + + if get_oauth_identity_mode() != "per_user": return False + from hermes_cli.config import load_config + + servers = (load_config() or {}).get("mcp_servers") or {} + config = servers.get(server_name) if isinstance(servers, dict) else None + return ( + isinstance(config, dict) + and str(config.get("auth") or "").strip().lower() == "oauth" + ) + def _scoped_cache_key(server_name: str) -> str | None: """Return the on-disk key, or None when per-user identity is unavailable.""" @@ -138,8 +142,6 @@ def write_cache_entry( """Persist schemas under the exact current requesting-user scope.""" cache_key = _scoped_cache_key(server_name) if cache_key is None: - # Never write an anonymous/shared cache entry while per-user OAuth is - # configured but no authenticated principal is bound. return entry = { @@ -171,9 +173,6 @@ def clear_cache_entry(server_name: str) -> None: del data[cache_key] changed = True elif _is_per_user_oauth_server(server_name): - # No principal is bound (e.g. config/admin maintenance). Clearing is - # intentionally destructive across this logical server's cache - # entries but does not grant access to any cached content. prefix = f"{server_name}@@u-v1-" for key in list(data): if key.startswith(prefix): From ddbb189063e0522e13a6c42065bd0d03ce21bc87 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:14:06 +0900 Subject: [PATCH 14/16] test(mcp): isolate schema cache across OAuth users --- tests/tools/test_mcp_schema_cache_per_user.py | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) create mode 100644 tests/tools/test_mcp_schema_cache_per_user.py diff --git a/tests/tools/test_mcp_schema_cache_per_user.py b/tests/tools/test_mcp_schema_cache_per_user.py new file mode 100644 index 000000000000..daabcf37ca0b --- /dev/null +++ b/tests/tools/test_mcp_schema_cache_per_user.py @@ -0,0 +1,123 @@ +"""Per-user MCP schema-cache isolation tests for shared gateways.""" + +from __future__ import annotations + +import pytest + + +@pytest.fixture(autouse=True) +def _clear_session_context(): + from gateway.session_context import clear_session_vars + + clear_session_vars() + yield + clear_session_vars() + + +def _bind(user_id: str) -> None: + from gateway.session_context import set_session_vars + + set_session_vars(platform="slack", scope_id="T_ONE", user_id=user_id) + + +def _config(mode: str = "per_user") -> dict: + return { + "mcp": {"oauth": {"identity_mode": mode}}, + "mcp_servers": { + "github": { + "url": "https://mcp.example/mcp", + "auth": "oauth", + }, + "filesystem": {"command": "server-filesystem"}, + }, + } + + +def test_per_user_oauth_schema_entries_are_not_shared(monkeypatch, tmp_path): + from hermes_constants import reset_hermes_home_override, set_hermes_home_override + from tools import mcp_schema_cache + + monkeypatch.setattr("hermes_cli.config.load_config", lambda: _config()) + token = set_hermes_home_override(tmp_path) + try: + _bind("U_ALICE") + mcp_schema_cache.write_cache_entry( + "github", + "fp", + tools=[{"name": "alice-only"}], + cache_scope="private", + ) + + _bind("U_BOB") + assert mcp_schema_cache.get_cached_entry("github", "fp") is None + mcp_schema_cache.write_cache_entry( + "github", + "fp", + tools=[{"name": "bob-only"}], + cache_scope="private", + ) + + _bind("U_ALICE") + assert mcp_schema_cache.get_cached_entry("github", "fp")["tools"] == [ + {"name": "alice-only"} + ] + + _bind("U_BOB") + assert mcp_schema_cache.get_cached_entry("github", "fp")["tools"] == [ + {"name": "bob-only"} + ] + finally: + reset_hermes_home_override(token) + + +def test_headless_per_user_startup_cannot_read_scoped_cache(monkeypatch, tmp_path): + from hermes_constants import reset_hermes_home_override, set_hermes_home_override + from tools import mcp_schema_cache + + monkeypatch.setattr("hermes_cli.config.load_config", lambda: _config()) + token = set_hermes_home_override(tmp_path) + try: + _bind("U_ALICE") + mcp_schema_cache.write_cache_entry( + "github", "fp", tools=[{"name": "private-tool"}] + ) + + from gateway.session_context import clear_session_vars + + clear_session_vars() + assert mcp_schema_cache.get_cached_entry("github", "fp") is None + finally: + reset_hermes_home_override(token) + + +def test_invalid_identity_mode_does_not_fall_back_to_shared_cache(monkeypatch): + from tools.mcp_oauth_identity import InvalidMcpOAuthIdentityModeError + from tools.mcp_schema_cache import get_cached_entry + + monkeypatch.setattr( + "hermes_cli.config.load_config", + lambda: _config(mode="per-user"), + ) + _bind("U_ALICE") + + with pytest.raises(InvalidMcpOAuthIdentityModeError): + get_cached_entry("github", "fp") + + +def test_non_oauth_schema_cache_remains_shared(monkeypatch, tmp_path): + from hermes_constants import reset_hermes_home_override, set_hermes_home_override + from tools import mcp_schema_cache + + monkeypatch.setattr("hermes_cli.config.load_config", lambda: _config()) + token = set_hermes_home_override(tmp_path) + try: + _bind("U_ALICE") + mcp_schema_cache.write_cache_entry( + "filesystem", "fp", tools=[{"name": "read_file"}] + ) + _bind("U_BOB") + assert mcp_schema_cache.get_cached_entry("filesystem", "fp")["tools"] == [ + {"name": "read_file"} + ] + finally: + reset_hermes_home_override(token) From 07f2cbf4d40023d7e029dff8a9e195b800052bab Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:17:45 +0900 Subject: [PATCH 15/16] fix(mcp): scope connection health state per OAuth user --- tools/mcp_oauth_runtime.py | 280 +++++++++++++++++++++++++++++-------- 1 file changed, 218 insertions(+), 62 deletions(-) diff --git a/tools/mcp_oauth_runtime.py b/tools/mcp_oauth_runtime.py index a71dfab83f96..96f0892fc8ad 100644 --- a/tools/mcp_oauth_runtime.py +++ b/tools/mcp_oauth_runtime.py @@ -1,21 +1,19 @@ -"""Scope-aware adapter for the MCP runtime's long-lived server registry. - -``tools.mcp_tool`` historically indexes ``_servers`` by logical server name. -Changing only token files/provider caches would therefore leave a critical -cross-user path: User B could obtain User A's already-authenticated -``MCPServerTask``. This module upgrades that registry in-place to exact -``(server, OAuthScope)`` lookup while preserving the existing dictionary API -used throughout the large MCP runtime. - -The adapter is installed only when an OAuth server is prepared. Non-OAuth MCP -servers retain their historical shared connection key. Per-user lookups have -NO "find any server with the same name" fallback: missing scope/connection -returns missing and the existing lazy-connect path can create the caller's own -connection. - -This compatibility layer keeps #78174 narrowly scoped instead of spreading -identity plumbing through every MCP handler. Scope selection remains outside -the model/tool argument surface and is driven by trusted ContextVars. +"""Scope-aware adapters for the MCP runtime's process-global registries. + +``tools.mcp_tool`` was designed around one long-lived connection per logical +server name. In a shared gateway that is not a sufficient security boundary: +per-user OAuth needs the authenticated transport *and* its connection-health +state to follow the requesting human. + +This module provides dict/set-compatible adapters so existing ``mcp_tool`` +call sites continue to use logical names while storage is keyed internally by +``server@@``. There is deliberately no "find any connection +with the same server name" fallback. + +Credential identity never comes from model-visible tool arguments. It is +resolved from trusted task-local gateway ContextVars and, for long-lived MCP +loop tasks whose ambient context is no longer available, an immutable weakly +pinned task scope. """ from __future__ import annotations @@ -23,7 +21,7 @@ import asyncio import threading import weakref -from typing import Any +from typing import Any, Iterable, Iterator from tools.mcp_oauth_identity import ( McpOAuthScope, @@ -32,21 +30,20 @@ try_resolve_oauth_scope, ) -# Some hosts may not preserve caller ContextVars when moving work onto the -# dedicated MCP event loop. Pin the scope to the actual long-lived asyncio Task -# as a fallback. Weak keys disappear automatically when the task exits, so a -# reused request ContextVar can never inherit another user's pin. +# A long-lived transport task may outlive the request context that created it. +# Weak keys ensure task-bound identity disappears with the task and can never +# become a process-global credential selector. _TASK_SCOPES: "weakref.WeakKeyDictionary[asyncio.Task, dict[str, McpOAuthScope]]" = ( weakref.WeakKeyDictionary() ) _TASK_SCOPE_LOCK = threading.Lock() _INSTALL_LOCK = threading.Lock() +_MISSING_SCOPE_SUFFIX = "@@" def _task_scope(server_name: str) -> McpOAuthScope | None: - # Prefer a currently bound authenticated request principal. This lets a - # reused host task move safely between users instead of being stuck on an - # earlier runtime pin. + # Request context wins. This is what lets a shared gateway worker safely + # handle Alice and then Bob without inheriting Alice's transport identity. ambient = try_resolve_oauth_scope() if ambient is not None: return ambient @@ -69,8 +66,8 @@ def bind_runtime_scope(server_name: str, scope: McpOAuthScope) -> None: except RuntimeError: task = None if task is None: - # Sync callers already have their gateway ContextVars, and there is no - # safe process-global fallback. Do not manufacture one. + # Sync callers already carry their gateway ContextVars. Never create a + # process-global fallback just to make a lookup convenient. return with _TASK_SCOPE_LOCK: current = dict(_TASK_SCOPES.get(task) or {}) @@ -78,30 +75,84 @@ def bind_runtime_scope(server_name: str, scope: McpOAuthScope) -> None: _TASK_SCOPES[task] = current -class ScopedMCPServerRegistry(dict): - """Dict-compatible registry with exact per-user keys for OAuth servers.""" +def _is_internal_scoped_key(key: str) -> bool: + return "@@u-v1-" in key or key.endswith(_MISSING_SCOPE_SUFFIX) - def __init__(self, initial: dict[str, Any] | None = None) -> None: - super().__init__(initial or {}) - self.oauth_servers: set[str] = set() - def mark_oauth_server(self, server_name: str) -> None: +def _logical_state_key(server_name: str) -> str: + """Return exact internal state key for the current OAuth principal. + + Availability/error state may be recorded before a human is bound (for + example a headless startup probe). Such state is put in a dedicated + missing-identity bucket. It is never credential state and, critically, + never blocks or poisons a later authenticated user's bucket. + """ + scope = _task_scope(server_name) + if scope is None: + return f"{server_name}{_MISSING_SCOPE_SUFFIX}" + return connection_registry_key(server_name, scope) + + +class _ScopedNameMixin: + """Common name translation for per-user OAuth runtime state.""" + + oauth_servers: set[str] + + def _mark_oauth_server(self, server_name: str) -> None: self.oauth_servers.add(server_name) - def _key_for_read(self, key: Any) -> Any: + def _read_key(self, key: Any) -> Any: if not isinstance(key, str) or key not in self.oauth_servers: return key - if "@@u-v1-" in key: + if _is_internal_scoped_key(key): return key - scope = _task_scope(key) - if scope is None: - return f"{key}@@" - return connection_registry_key(key, scope) + return _logical_state_key(key) + + def _write_key(self, key: Any) -> Any: + return self._read_key(key) + + def _visible_logical_keys(self, raw_keys: Iterable[Any]) -> list[Any]: + """Project internal keys into the current request's logical view. + + ``mcp_tool`` takes snapshots with ``dict(_servers)`` and + ``set(_server_connecting)``. Returning logical keys here preserves + those public/status semantics without exposing another user's entries. + """ + raw = list(raw_keys) + visible: list[Any] = [] + raw_set = set(raw) + + # Ordinary non-OAuth keys remain visible exactly as before. + for key in raw: + if not isinstance(key, str): + visible.append(key) + elif key not in self.oauth_servers and not _is_internal_scoped_key(key): + visible.append(key) - def _key_for_write(self, key: Any) -> Any: + # Each OAuth logical server is visible only if THIS scope has state. + for server_name in self.oauth_servers: + if _logical_state_key(server_name) in raw_set: + visible.append(server_name) + return visible + + +class ScopedMCPServerRegistry(_ScopedNameMixin, dict): + """Exact per-user registry for long-lived OAuth ``MCPServerTask`` objects.""" + + def __init__(self, initial: dict[str, Any] | None = None) -> None: + dict.__init__(self, initial or {}) + self.oauth_servers: set[str] = set() + + def mark_oauth_server(self, server_name: str) -> None: + self._mark_oauth_server(server_name) + # A pre-install headless startup may have inserted a raw server entry. + # Ownership is unknowable, so it must never become a per-user fallback. + dict.pop(self, server_name, None) + + def _connection_key_for_write(self, key: Any) -> Any: if not isinstance(key, str) or key not in self.oauth_servers: return key - if "@@u-v1-" in key: + if _is_internal_scoped_key(key): return key scope = _task_scope(key) if scope is None: @@ -112,35 +163,111 @@ def _key_for_write(self, key: Any) -> Any: return connection_registry_key(key, scope) def __getitem__(self, key: Any) -> Any: - return super().__getitem__(self._key_for_read(key)) + return dict.__getitem__(self, self._read_key(key)) def __setitem__(self, key: Any, value: Any) -> None: - super().__setitem__(self._key_for_write(key), value) + dict.__setitem__(self, self._connection_key_for_write(key), value) def __contains__(self, key: object) -> bool: - return super().__contains__(self._key_for_read(key)) + return dict.__contains__(self, self._read_key(key)) def get(self, key: Any, default: Any = None) -> Any: - return super().get(self._key_for_read(key), default) + return dict.get(self, self._read_key(key), default) def pop(self, key: Any, default: Any = ...): - resolved = self._key_for_read(key) + resolved = self._read_key(key) if default is ...: - return super().pop(resolved) - return super().pop(resolved, default) + return dict.pop(self, resolved) + return dict.pop(self, resolved, default) def setdefault(self, key: Any, default: Any = None) -> Any: - return super().setdefault(self._key_for_write(key), default) + return dict.setdefault(self, self._connection_key_for_write(key), default) + def keys(self): + # ``dict(self)`` consults keys()+__getitem__ for dict subclasses. + return self._visible_logical_keys(dict.keys(self)) -class PersistentPerUserLazyConfigs(dict): - """Keep OAuth server config available for every user's first connection. - ``mcp_tool._ensure_lazy_server_connected`` historically pops a lazy config - after one successful connection because there was only one global server. - Per-user OAuth has N independent long-lived connections, so that config is - reusable metadata rather than one-shot state. - """ +class ScopedNameDict(_ScopedNameMixin, dict): + """Per-user view for connect errors/backoff/circuit-breaker dictionaries.""" + + def __init__(self, initial: dict[str, Any] | None = None) -> None: + dict.__init__(self, initial or {}) + self.oauth_servers: set[str] = set() + + def mark_oauth_server(self, server_name: str) -> None: + self._mark_oauth_server(server_name) + # State produced before the runtime knew this was per-user belongs to + # no authenticated principal. Drop it rather than assigning it to the + # first user who arrives. + dict.pop(self, server_name, None) + + def __getitem__(self, key: Any) -> Any: + return dict.__getitem__(self, self._read_key(key)) + + def __setitem__(self, key: Any, value: Any) -> None: + dict.__setitem__(self, self._write_key(key), value) + + def __contains__(self, key: object) -> bool: + return dict.__contains__(self, self._read_key(key)) + + def get(self, key: Any, default: Any = None) -> Any: + return dict.get(self, self._read_key(key), default) + + def pop(self, key: Any, default: Any = ...): + resolved = self._read_key(key) + if default is ...: + return dict.pop(self, resolved) + return dict.pop(self, resolved, default) + + def setdefault(self, key: Any, default: Any = None) -> Any: + return dict.setdefault(self, self._write_key(key), default) + + def keys(self): + return self._visible_logical_keys(dict.keys(self)) + + +class ScopedNameSet(_ScopedNameMixin, set): + """Per-user view for the MCP ``_server_connecting`` deduplication set.""" + + def __init__(self, initial: Iterable[str] | None = None) -> None: + set.__init__(self, initial or ()) + self.oauth_servers: set[str] = set() + + def mark_oauth_server(self, server_name: str) -> None: + self._mark_oauth_server(server_name) + set.discard(self, server_name) + + def add(self, element: Any) -> None: + set.add(self, self._write_key(element)) + + def discard(self, element: Any) -> None: + set.discard(self, self._read_key(element)) + + def remove(self, element: Any) -> None: + set.remove(self, self._read_key(element)) + + def __contains__(self, element: object) -> bool: + return set.__contains__(self, self._read_key(element)) + + def update(self, *others: Iterable[Any]) -> None: + for other in others: + for element in other: + self.add(element) + + def difference_update(self, *others: Iterable[Any]) -> None: + for other in others: + for element in other: + self.discard(element) + + def __iter__(self) -> Iterator[Any]: + # ``set(_server_connecting)`` must expose the current request's logical + # names, not every user's encoded state key. + return iter(self._visible_logical_keys(list(set.__iter__(self)))) + + +class PersistentPerUserLazyConfigs(dict): + """Keep OAuth server config available for every user's first connection.""" def __init__(self, initial: dict[str, Any] | None = None) -> None: super().__init__(initial or {}) @@ -151,17 +278,29 @@ def mark_oauth_server(self, server_name: str) -> None: def pop(self, key: Any, default: Any = ...): if isinstance(key, str) and key in self.oauth_servers and key in self: + # Per-user OAuth has N independent transports. Alice's successful + # lazy connect must not consume the config Bob needs later. return self[key] if default is ...: - return super().pop(key) - return super().pop(key, default) + return dict.pop(self, key) + return dict.pop(self, key, default) + + +def _wrap_scoped_dict(mcp_tool, attr_name: str, server_name: str) -> None: + current = getattr(mcp_tool, attr_name) + if not isinstance(current, ScopedNameDict): + current = ScopedNameDict(dict(current)) + setattr(mcp_tool, attr_name, current) + current.mark_oauth_server(server_name) def prepare_oauth_server_runtime(server_name: str) -> None: - """Install/mark scope-aware MCP registries before resolving identity. + """Install/mark all per-user runtime views before identity resolution. - Called before ``resolve_oauth_scope`` so headless startup can fail closed - yet retain connection config for a later authenticated user's lazy start. + It is safe to call repeatedly and deliberately runs before + ``resolve_oauth_scope``. Thus a headless startup can fail closed in its + anonymous bucket while a later Alice/Bob request sees independent + connection, connect-backoff, error, and circuit-breaker state. """ from tools import mcp_tool @@ -172,12 +311,29 @@ def prepare_oauth_server_runtime(server_name: str) -> None: mcp_tool._servers = servers servers.mark_oauth_server(server_name) + connecting = mcp_tool._server_connecting + if not isinstance(connecting, ScopedNameSet): + connecting = ScopedNameSet(set(connecting)) + mcp_tool._server_connecting = connecting + connecting.mark_oauth_server(server_name) + + for attr_name in ( + "_server_connect_errors", + "_server_error_counts", + "_server_breaker_opened_at", + "_server_connect_retry_after", + "_server_connect_failures", + ): + _wrap_scoped_dict(mcp_tool, attr_name, server_name) + lazy = mcp_tool._lazy_server_configs if not isinstance(lazy, PersistentPerUserLazyConfigs): lazy = PersistentPerUserLazyConfigs(dict(lazy)) mcp_tool._lazy_server_configs = lazy lazy.mark_oauth_server(server_name) + # Keep safe connection config as reusable metadata. Authentication + # material itself still lives only in the scoped provider/storage. if server_name not in lazy: try: config = (mcp_tool._load_mcp_config() or {}).get(server_name) From 41c5fa089edf41e131f706436ca5447c5538cc89 Mon Sep 17 00:00:00 2001 From: Yu Ishikawa Date: Thu, 27 Aug 2026 08:18:40 +0900 Subject: [PATCH 16/16] fix(mcp): leave shared OAuth runtime semantics untouched --- tools/mcp_oauth_runtime.py | 46 ++++++++++++++++---------------------- 1 file changed, 19 insertions(+), 27 deletions(-) diff --git a/tools/mcp_oauth_runtime.py b/tools/mcp_oauth_runtime.py index 96f0892fc8ad..ad8717181ba6 100644 --- a/tools/mcp_oauth_runtime.py +++ b/tools/mcp_oauth_runtime.py @@ -10,10 +10,9 @@ ``server@@``. There is deliberately no "find any connection with the same server name" fallback. -Credential identity never comes from model-visible tool arguments. It is -resolved from trusted task-local gateway ContextVars and, for long-lived MCP -loop tasks whose ambient context is no longer available, an immutable weakly -pinned task scope. +The adapters are installed only for explicit ``per_user`` mode. Historical +``shared`` OAuth therefore keeps the original runtime objects and semantics +byte-for-byte outside this module. """ from __future__ import annotations @@ -27,12 +26,10 @@ McpOAuthScope, MissingMcpOAuthIdentityError, connection_registry_key, + get_oauth_identity_mode, try_resolve_oauth_scope, ) -# A long-lived transport task may outlive the request context that created it. -# Weak keys ensure task-bound identity disappears with the task and can never -# become a process-global credential selector. _TASK_SCOPES: "weakref.WeakKeyDictionary[asyncio.Task, dict[str, McpOAuthScope]]" = ( weakref.WeakKeyDictionary() ) @@ -42,8 +39,8 @@ def _task_scope(server_name: str) -> McpOAuthScope | None: - # Request context wins. This is what lets a shared gateway worker safely - # handle Alice and then Bob without inheriting Alice's transport identity. + # Request context wins. This lets a shared gateway worker safely handle + # Alice and then Bob without inheriting Alice's transport identity. ambient = try_resolve_oauth_scope() if ambient is not None: return ambient @@ -61,6 +58,8 @@ def _task_scope(server_name: str) -> McpOAuthScope | None: def bind_runtime_scope(server_name: str, scope: McpOAuthScope) -> None: """Pin a resolved OAuth scope to the current long-lived server task.""" + if not scope.is_per_user: + return try: task = asyncio.current_task() except RuntimeError: @@ -122,14 +121,12 @@ def _visible_logical_keys(self, raw_keys: Iterable[Any]) -> list[Any]: visible: list[Any] = [] raw_set = set(raw) - # Ordinary non-OAuth keys remain visible exactly as before. for key in raw: if not isinstance(key, str): visible.append(key) elif key not in self.oauth_servers and not _is_internal_scoped_key(key): visible.append(key) - # Each OAuth logical server is visible only if THIS scope has state. for server_name in self.oauth_servers: if _logical_state_key(server_name) in raw_set: visible.append(server_name) @@ -145,8 +142,8 @@ def __init__(self, initial: dict[str, Any] | None = None) -> None: def mark_oauth_server(self, server_name: str) -> None: self._mark_oauth_server(server_name) - # A pre-install headless startup may have inserted a raw server entry. - # Ownership is unknowable, so it must never become a per-user fallback. + # Anything at the historical raw key predates the per-user boundary; + # ownership is unknowable, so it cannot be assigned to the first user. dict.pop(self, server_name, None) def _connection_key_for_write(self, key: Any) -> Any: @@ -197,9 +194,7 @@ def __init__(self, initial: dict[str, Any] | None = None) -> None: def mark_oauth_server(self, server_name: str) -> None: self._mark_oauth_server(server_name) - # State produced before the runtime knew this was per-user belongs to - # no authenticated principal. Drop it rather than assigning it to the - # first user who arrives. + # Pre-bound state belongs to no authenticated principal. dict.pop(self, server_name, None) def __getitem__(self, key: Any) -> Any: @@ -261,8 +256,6 @@ def difference_update(self, *others: Iterable[Any]) -> None: self.discard(element) def __iter__(self) -> Iterator[Any]: - # ``set(_server_connecting)`` must expose the current request's logical - # names, not every user's encoded state key. return iter(self._visible_logical_keys(list(set.__iter__(self)))) @@ -278,8 +271,6 @@ def mark_oauth_server(self, server_name: str) -> None: def pop(self, key: Any, default: Any = ...): if isinstance(key, str) and key in self.oauth_servers and key in self: - # Per-user OAuth has N independent transports. Alice's successful - # lazy connect must not consume the config Bob needs later. return self[key] if default is ...: return dict.pop(self, key) @@ -295,13 +286,16 @@ def _wrap_scoped_dict(mcp_tool, attr_name: str, server_name: str) -> None: def prepare_oauth_server_runtime(server_name: str) -> None: - """Install/mark all per-user runtime views before identity resolution. + """Install/mark per-user runtime views before principal resolution. - It is safe to call repeatedly and deliberately runs before - ``resolve_oauth_scope``. Thus a headless startup can fail closed in its - anonymous bucket while a later Alice/Bob request sees independent - connection, connect-backoff, error, and circuit-breaker state. + ``shared`` mode returns immediately so existing single-user CLI/profile + deployments keep the original plain dict/set objects and lifecycle. + Invalid configuration propagates from ``get_oauth_identity_mode`` rather + than silently falling back to shared semantics. """ + if get_oauth_identity_mode() != "per_user": + return + from tools import mcp_tool with _INSTALL_LOCK: @@ -332,8 +326,6 @@ def prepare_oauth_server_runtime(server_name: str) -> None: mcp_tool._lazy_server_configs = lazy lazy.mark_oauth_server(server_name) - # Keep safe connection config as reusable metadata. Authentication - # material itself still lives only in the scoped provider/storage. if server_name not in lazy: try: config = (mcp_tool._load_mcp_config() or {}).get(server_name)