diff --git a/cli-config.yaml.example b/cli-config.yaml.example index 1a8021ff984d2..1cc33e0c38d4e 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1297,6 +1297,15 @@ platform_toolsets: # Each server's tools are automatically discovered and registered. # See website/docs/user-guide/features/mcp.md for full documentation. # +# OAuth identity isolation (shared gateway): +# mcp: +# oauth: +# identity_mode: shared # default. One token per profile+server. +# # identity_mode: per_user # scope OAuth to the authenticated requester +# # (Slack/Discord/Telegram/…). Typos are +# # rejected; they never fall back to shared. +# # Direct CLI cannot pick a user's token. +# # Stdio servers (spawn a subprocess): # command: the executable to run # args: command-line arguments diff --git a/cli.py b/cli.py index 2600a5845ccfa..6628eb3166f63 100644 --- a/cli.py +++ b/cli.py @@ -14166,13 +14166,15 @@ def _confirm_and_reload_mcp(self, cmd_original: str = "") -> None: self._reload_mcp() def _reload_mcp(self): - """Reload MCP servers: disconnect all, re-read config.yaml, reconnect. + """Reload MCP servers: recycle connections, re-read config.yaml, reconnect. - After reconnecting, refreshes the agent's tool list so the model - sees the updated tools on the next turn. + Unbound CLI/TUI (and shared mode) disconnect every live server. + A bound ``per_user`` gateway request leaves other principals' + OAuth sessions up. After reconnecting, refreshes the agent's tool + list so the model sees the updated tools on the next turn. """ try: - from tools.mcp_tool import shutdown_mcp_servers, discover_mcp_tools, _servers, _lock + from tools.mcp_tool import reload_mcp_connections, discover_mcp_tools, _servers, _lock # Capture old server names with _lock: @@ -14181,8 +14183,7 @@ def _reload_mcp(self): if not self._command_running: print("πŸ”„ Reloading MCP servers...") - # Shutdown existing connections - shutdown_mcp_servers() + reload_mcp_connections() # Reconnect (reads config.yaml fresh) new_tools = discover_mcp_tools() diff --git a/docs/rfc/requester-scoped-mcp-oauth.md b/docs/rfc/requester-scoped-mcp-oauth.md new file mode 100644 index 0000000000000..f53ec92d0b5bb --- /dev/null +++ b/docs/rfc/requester-scoped-mcp-oauth.md @@ -0,0 +1,95 @@ +# RFC (Revised): Requester-Scoped MCP OAuth Isolation + +- **Status:** Revised after adversarial code review +- **Date:** 2026-08-26 +- **Upstream issue:** [NousResearch/hermes-agent#78174](https://github.com/NousResearch/hermes-agent/issues/78174) +- **Related (out of scope):** [#78169](https://github.com/NousResearch/hermes-agent/issues/78169) headless consent UX +- **Do not rebase:** [#79449](https://github.com/NousResearch/hermes-agent/pull/79449) + +This document **supersedes** the draft RFC where they disagree. The draft's +threat model, invariants I1–I18, and "fresh native scope" decision stand. +The sections below are the review findings and the locked implementation +decisions. + +## Adversarial findings (must change) + +1. **`scope_id` is empty on most adapters.** Telegram, WhatsApp, Feishu, + Mattermost, SMS, IRC, and Discord DMs do not populate tenant scope. + Requiring a non-empty `scope_id` would fail-closed almost every + non-Slack/Guild path. Empty bound `scope_id` is canonicalized to `"~"`. + `_UNSET` (not bound) is still a hard miss. + +2. **`get_session_env` falls back to `os.environ` when `_UNSET`.** A + per-user principal MUST use a dedicated bound-only getter. Env vars + are not a credential selector. + +3. **Gateway/CLI/cron call `discover_mcp_tools()` at process start** with + no human principal, and will connect OAuth servers from + `mcp-tokens/.json` under `suppress_interactive_oauth`. In + `per_user`, OAuth-protected servers MUST NOT pick a human credential at + startup. They are implicitly lazy until a request with a bound + principal arrives. Tool *names* may still be registered from the + unscoped schema cache so the model can see them. + +4. **`handle_401` / unpinned `HermesTokenStorage` re-resolve ambient + `get_hermes_home()`.** Scope and `hermes_home` MUST be captured at + provider/connection construction and passed explicitly into refresh, + 401, reconnect, and disk-watch. + +5. **`_lazy_server_configs` is popped on first connect.** In `per_user`, + Alice's first use must not delete the lazy config Bob still needs. + +6. **In-memory Hermes tool registry is process-global.** Closing the + confused-deputy hole does not require per-session tool schemas (that + would also fight prompt caching if done mid-conversation). Live calls + use the requester's connection; a tool Bob does not actually have fails + at the MCP server. Disk `cacheScope=private` entries are still scoped. + +7. **No metrics/telemetry in this change** (project policy: no outbound + telemetry without a user-facing opt-in). Structured log fields only, + with opaque principal keys, never tokens. + +8. **Do not implement #78169** (consent URL delivery, paste-back, gateway + message injection). Do not add `hermes mcp login --user`. + +9. **OAuth isolation applies to `auth: oauth` servers.** Stdio and static + header servers keep process-level connections. Stdio env credentials + remain shared; that is an explicit non-goal. + +10. **Subagents inherit the parent's bound principal** via ContextVar copy. + Cron blanks identity and therefore fail-closes in `per_user`. + +11. **`_run_on_mcp_loop` copies the MCP loop thread's ContextVars, not the + agent's.** `run_coroutine_threadsafe` creates the task inside the loop + thread. Without an explicit wrap, `_capture_oauth_identity` (and any + `get_bound_session_principal()` inside connect) would fail closed on a + live gateway request, or worse inherit a stale loop-thread principal. + The scheduling thread's bound principal MUST be re-applied inside the + scheduled task (same hop as `HERMES_HOME` override). Identity is also + pinned on `MCPServerTask` in `start()` before `ensure_future(run())` + so reconnects never re-resolve ambient identity. + +## Locked decisions + +| Topic | Decision | +|---|---| +| Default | `mcp.oauth.identity_mode: shared` (absent key = shared) | +| Invalid mode | Reject (`per-user`, typos). Never downgrade to shared | +| Principal | `(v1, platform, scope_id, user_id)` from bound ContextVars only | +| Empty `scope_id` | Canonical `"~"` when the field is bound-or-absent-as-empty | +| Persistence key | `u-v1-` + SHA-256 of canonical JSON array; never raw IDs in paths | +| Shared layout | Unchanged: `$HERMES_HOME/mcp-tokens/.*` | +| Per-user layout | `$HERMES_HOME/mcp-tokens/by-user//.*` | +| Migration | Never assign a legacy shared token to a requester | +| Registry key | `server_name` in shared; `server_name + \\x1f + persistence_key` in per_user | +| Lookup | Exact key only. No "any connection named github" fallback | +| CLI / TUI / desktop / cron in `per_user` | Fail closed with an actionable error when no bound principal | +| MCP loop hop | `_run_on_mcp_loop` re-binds the caller's principal; `start()` pins it on the connection | +| `hermes mcp remove` | Admin: may delete that server's artifacts across `by-user/*` | +| Idle eviction / per-server override / HMAC path keys | Deferred | + +## Core invariant + +A request authenticated as principal A MUST NOT read, refresh, select, +reuse, reconnect, disconnect, or otherwise affect any credential-bearing +object belonging to principal B. diff --git a/gateway/run.py b/gateway/run.py index 690c8811a08cb..0eb4f7ebb315b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -23904,23 +23904,27 @@ async def _restore_telegram_topic_session(self, event: MessageEvent, raw_session async def _execute_mcp_reload(self, event: MessageEvent) -> str: - """Actually disconnect, reconnect, and notify MCP tool changes. + """Recycle MCP connections, reconnect, and notify tool changes. Split out from ``_handle_reload_mcp_command`` so the confirmation wrapper can invoke the same path whether the user confirmed via button, text reply, or has the confirm gate disabled. + + Uses ``reload_mcp_connections`` so a ``per_user`` requester cannot + tear down another principal's live OAuth session. """ loop = asyncio.get_running_loop() try: - from tools.mcp_tool import shutdown_mcp_servers, discover_mcp_tools, _servers, _lock + from tools.mcp_tool import reload_mcp_connections, discover_mcp_tools, _servers, _lock # Capture old server names before shutdown with _lock: old_servers = set(_servers.keys()) - # Read new config before shutting down, so we know what will be added/removed - # Shutdown existing connections - await loop.run_in_executor(None, shutdown_mcp_servers) + # Recycle connections under the /reload-mcp policy: full wipe in + # shared/unbound mode; in per_user, other principals' OAuth + # sessions stay up (Alice must not disconnect Bob). + await loop.run_in_executor(None, reload_mcp_connections) # Reconnect by discovering tools (reads config.yaml fresh) new_tools = await loop.run_in_executor(None, discover_mcp_tools) diff --git a/gateway/session_context.py b/gateway/session_context.py index 9a6a8c4226074..c750d1634f211 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -38,7 +38,8 @@ from contextlib import contextmanager from contextvars import ContextVar -from typing import Any, Iterator +from dataclasses import dataclass +from typing import Any, Iterator, Optional # Sentinel to distinguish "never set in this context" from "explicitly set to empty". # When a contextvar holds _UNSET, we fall back to os.environ (CLI/cron compat). @@ -390,6 +391,85 @@ def reset_session_vars() -> None: pass +@dataclass(frozen=True, slots=True) +class BoundSessionPrincipal: + """Trusted requester identity bound on the current task. + + Returned only by :func:`get_bound_session_principal`. Values come from + ContextVars set by :func:`set_session_vars`; they never fall back to + ``os.environ``. ``scope_id`` may be empty on platforms that have no + tenant namespace (Telegram, Discord DMs, …). + """ + + platform: str + scope_id: str + user_id: str + + +def _bound_session_str(var: ContextVar) -> Optional[str]: + """Return a ContextVar string only when it was explicitly bound. + + ``None`` means the var is ``_UNSET`` (not bound in this task β€” env + fallback must not be consulted). An empty string means it was bound + empty via :func:`set_session_vars` / :func:`clear_session_vars`. + """ + value = var.get() + if value is _UNSET: + return None + if value is None: + return "" + return str(value) + + +def get_bound_session_principal() -> Optional[BoundSessionPrincipal]: + """Return the authenticated requester, or None if identity is not bound. + + Unlike :func:`get_session_env`, this never reads ``os.environ``. A + missing platform or user_id (unset *or* bound-empty) yields ``None`` so + MCP OAuth ``per_user`` mode can fail closed instead of impersonating a + process-global leftover. Empty ``scope_id`` is allowed: many adapters + do not have a tenant namespace. + """ + platform = _bound_session_str(_SESSION_PLATFORM) + user_id = _bound_session_str(_SESSION_USER_ID) + if platform is None or user_id is None: + return None + platform = platform.strip() + user_id = user_id.strip() + if not platform or not user_id: + return None + scope_raw = _bound_session_str(_SESSION_SCOPE_ID) + scope_id = "" if scope_raw is None else scope_raw.strip() + return BoundSessionPrincipal( + platform=platform, scope_id=scope_id, user_id=user_id + ) + + +@contextmanager +def apply_bound_session_principal( + principal: BoundSessionPrincipal, +) -> Iterator[None]: + """Re-bind a previously captured principal in this task, then restore. + + Used to carry gateway identity onto the dedicated MCP event-loop thread. + ``run_coroutine_threadsafe`` copies that thread's context, not the + scheduling thread's, so OAuth ``per_user`` capture would otherwise see + no requester (or a stale one). Tokens are reset, not cleared to ``""``, + so the loop thread does not retain the principal after the call. + """ + tokens = ( + _SESSION_PLATFORM.set(principal.platform), + _SESSION_SCOPE_ID.set(principal.scope_id), + _SESSION_USER_ID.set(principal.user_id), + ) + try: + yield + finally: + _SESSION_USER_ID.reset(tokens[2]) + _SESSION_SCOPE_ID.reset(tokens[1]) + _SESSION_PLATFORM.reset(tokens[0]) + + def get_session_env(name: str, default: str = "") -> str: """Read a session context variable by its legacy ``HERMES_SESSION_*`` name. diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index d3c36844a4d81..62fd6fb72a0c0 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -708,6 +708,15 @@ # When disabled, the watcher still detects the change and prints # guidance to apply it deliberately via /reload-mcp. "auto_reload_on_config_change": True, + # MCP OAuth identity isolation. ``shared`` (default) keeps the + # historical one-token-per-profile layout. ``per_user`` scopes + # OAuth credentials, providers, and live connections to the + # authenticated gateway requester β€” required for a shared Slack / + # Discord / Telegram gateway. Typos are rejected; they must never + # silently fall back to shared. See docs/rfc/requester-scoped-mcp-oauth.md. + "oauth": { + "identity_mode": "shared", + }, }, # Tool-output truncation thresholds. When terminal output or a diff --git a/hermes_cli/mcp_config.py b/hermes_cli/mcp_config.py index 05adb6bf5bea9..6478b61080e2c 100644 --- a/hermes_cli/mcp_config.py +++ b/hermes_cli/mcp_config.py @@ -56,6 +56,29 @@ def _error(text: str): print(color(f" βœ— {text}", Colors.RED)) +def _per_user_oauth_cli_block() -> Optional[str]: + """Return an error message when per_user mode has no bound requester. + + Direct CLI / TUI / desktop / cron cannot pick a human OAuth token, and + there is no ``--user`` selector. Gateway sessions with a bound principal + are the intended ``per_user`` path. + """ + from tools.mcp_oauth_identity import ( + IDENTITY_MODE_PER_USER, + MissingRequesterIdentity, + configured_identity_mode, + resolve_mcp_oauth_scope, + ) + + if configured_identity_mode() != IDENTITY_MODE_PER_USER: + return None + try: + resolve_mcp_oauth_scope(uses_oauth=True) + except MissingRequesterIdentity as exc: + return str(exc) + return None + + def _confirm(question: str, default: bool = True) -> bool: default_str = "Y/n" if default else "y/N" try: @@ -410,7 +433,11 @@ def _oauth_tokens_present(name: str) -> bool: """ try: from tools.mcp_oauth import HermesTokenStorage + from tools.mcp_oauth_identity import MissingRequesterIdentity + return HermesTokenStorage(name).has_cached_tokens() + except MissingRequesterIdentity: + return False except Exception as exc: # pragma: no cover β€” defensive logger.debug("Could not check OAuth tokens for '%s': %s", name, exc) # Be permissive on unexpected errors: don't block a real success. @@ -508,6 +535,16 @@ def cmd_mcp_add(args): # ── Authentication ──────────────────────────────────────────────── if url and auth_type == "oauth": + blocked = _per_user_oauth_cli_block() + if blocked: + _error(blocked) + _info( + "mcp.oauth.identity_mode is per_user. Direct CLI cannot " + "complete OAuth for a gateway requester. There is no " + "--user selector." + ) + return + print() _info(f"Starting OAuth flow for '{name}'...") oauth_ok = False @@ -664,9 +701,11 @@ def cmd_mcp_remove(args): # Clean up OAuth tokens if they exist β€” route through MCPOAuthManager so # any provider instance cached in the current process (e.g. from an # earlier `hermes mcp test` in the same session) is evicted too. + # ``all_identities=True`` is the admin path: removing the server from + # config deletes shared artifacts and every by-user namespace for it. try: from tools.mcp_oauth_manager import get_manager - get_manager().remove(name) + get_manager().remove(name, all_identities=True) _success("Cleaned up OAuth tokens") except Exception: pass @@ -824,6 +863,16 @@ def _reauth_oauth_server(name: str, server_config: dict) -> bool: _info("Use `hermes mcp remove` + `hermes mcp add` to reconfigure auth.") return False + blocked = _per_user_oauth_cli_block() + if blocked: + _error(blocked) + _info( + "mcp.oauth.identity_mode is per_user. Direct CLI cannot " + "complete OAuth for a gateway requester. There is no " + "--user selector." + ) + return False + # Wipe both disk and in-memory cache so the next probe forces a fresh # OAuth flow. try: diff --git a/tests/gateway/test_mcp_reload_refreshes_cached_agents.py b/tests/gateway/test_mcp_reload_refreshes_cached_agents.py index 4d945f03c5942..66c6663345510 100644 --- a/tests/gateway/test_mcp_reload_refreshes_cached_agents.py +++ b/tests/gateway/test_mcp_reload_refreshes_cached_agents.py @@ -106,6 +106,7 @@ async def test_reload_mcp_refreshes_cached_agent_tools(): ] with ( + patch("tools.mcp_tool.reload_mcp_connections"), patch("tools.mcp_tool.shutdown_mcp_servers"), patch("tools.mcp_tool.discover_mcp_tools", return_value=["HassTurnOn", "HassTurnOff"]), patch.dict("tools.mcp_tool._servers", {"homeassistant": object()}, clear=True), @@ -136,6 +137,7 @@ async def test_reload_mcp_handles_empty_agent_cache(): assert len(runner._agent_cache) == 0 with ( + patch("tools.mcp_tool.reload_mcp_connections"), patch("tools.mcp_tool.shutdown_mcp_servers"), patch("tools.mcp_tool.discover_mcp_tools", return_value=[]), patch.dict("tools.mcp_tool._servers", {}, clear=True), @@ -164,6 +166,7 @@ def _capture_get_tool_definitions(**kwargs): return [{"type": "function", "function": {"name": "refreshed"}}] with ( + patch("tools.mcp_tool.reload_mcp_connections"), patch("tools.mcp_tool.shutdown_mcp_servers"), patch("tools.mcp_tool.discover_mcp_tools", return_value=["refreshed"]), patch.dict("tools.mcp_tool._servers", {"homeassistant": object()}, clear=True), diff --git a/tests/tools/test_mcp_circuit_breaker.py b/tests/tools/test_mcp_circuit_breaker.py index 44b003c2a7da3..27c639d9f04c0 100644 --- a/tests/tools/test_mcp_circuit_breaker.py +++ b/tests/tools/test_mcp_circuit_breaker.py @@ -357,7 +357,7 @@ async def _call_tool_unused(*a, **kw): # pragma: no cover # Force handle_401 to claim recovery succeeded. mgr = get_manager() - async def _h401(name, token=None): + async def _h401(name, token=None, **_kwargs): return True monkeypatch.setattr(mgr, "handle_401", _h401) diff --git a/tests/tools/test_mcp_dynamic_discovery.py b/tests/tools/test_mcp_dynamic_discovery.py index d0d7083aa04c1..37c59298b66ad 100644 --- a/tests/tools/test_mcp_dynamic_discovery.py +++ b/tests/tools/test_mcp_dynamic_discovery.py @@ -75,6 +75,39 @@ async def test_nuke_and_repave(self, mock_registry): assert "mcp__live_srv__new_tool" in resolve_toolset("live_srv") assert server._registered_tool_names == ["mcp__live_srv__new_tool"] + @pytest.mark.asyncio + async def test_refresh_keeps_tools_still_advertised_by_sibling(self, mock_registry): + """Alice dropping a tool must not unregister it while Bob still serves it.""" + import tools.mcp_tool as mcp + + alice = MCPServerTask("peer_srv") + bob = MCPServerTask("peer_srv") + alice._config = {} + mock_registry.register( + name="mcp__peer_srv__old_tool", toolset="mcp-peer_srv", schema={}, + handler=lambda x: x, check_fn=lambda: True, is_async=False, + description="", emoji="", + ) + alice._registered_tool_names = ["mcp__peer_srv__old_tool"] + bob._registered_tool_names = ["mcp__peer_srv__old_tool"] + alice.session = SimpleNamespace( + list_tools=AsyncMock( + return_value=SimpleNamespace(tools=[_make_mcp_tool("new_tool", "new")]) + ) + ) + mcp._servers["peer_srv-alice"] = alice + mcp._servers["peer_srv-bob"] = bob + try: + with patch("tools.registry.registry", mock_registry): + await alice._refresh_tools() + assert "mcp__peer_srv__old_tool" in mock_registry.get_all_tool_names() + assert "mcp__peer_srv__new_tool" in mock_registry.get_all_tool_names() + assert "mcp__peer_srv__old_tool" not in alice._registered_tool_names + assert "mcp__peer_srv__old_tool" in bob._registered_tool_names + finally: + mcp._servers.pop("peer_srv-alice", None) + mcp._servers.pop("peer_srv-bob", None) + class TestMessageHandler: """Tests for MCPServerTask._make_message_handler dispatch.""" diff --git a/tests/tools/test_mcp_loop_session_principal.py b/tests/tools/test_mcp_loop_session_principal.py new file mode 100644 index 0000000000000..8203ccf9b8107 --- /dev/null +++ b/tests/tools/test_mcp_loop_session_principal.py @@ -0,0 +1,146 @@ +"""Bound requester identity must survive the MCP event-loop hop. + +``run_coroutine_threadsafe`` creates the task inside the MCP loop thread, +so it copies that thread's ContextVars β€” not the scheduling thread's. +OAuth ``per_user`` capture would fail closed (or inherit a stale principal) +without ``_wrap_with_session_principal``. Mirrors +``test_mcp_loop_profile_override.py`` for HERMES_HOME. +""" +import threading + +import pytest + +from gateway.session_context import ( + apply_bound_session_principal, + get_bound_session_principal, + reset_session_vars, + set_session_vars, +) + + +@pytest.fixture +def mcp_loop(): + import tools.mcp_tool as mcp_tool + + mcp_tool._ensure_mcp_loop() + yield mcp_tool + mcp_tool._stop_mcp_loop() + + +@pytest.fixture(autouse=True) +def _reset_principal(): + reset_session_vars() + yield + reset_session_vars() + + +async def _read_principal(): + principal = get_bound_session_principal() + if principal is None: + return None + return (principal.platform, principal.scope_id, principal.user_id) + + +def test_bound_principal_propagates_to_mcp_loop(mcp_loop): + assert mcp_loop._run_on_mcp_loop(_read_principal(), timeout=10) is None + + set_session_vars(platform="slack", scope_id="T1", user_id="U-alice") + assert mcp_loop._run_on_mcp_loop(_read_principal(), timeout=10) == ( + "slack", + "T1", + "U-alice", + ) + assert mcp_loop._run_on_mcp_loop(lambda: _read_principal(), timeout=10) == ( + "slack", + "T1", + "U-alice", + ) + + reset_session_vars() + assert mcp_loop._run_on_mcp_loop(_read_principal(), timeout=10) is None + + +def test_empty_scope_id_propagates(mcp_loop): + set_session_vars(platform="telegram", scope_id="", user_id="12345") + assert mcp_loop._run_on_mcp_loop(_read_principal(), timeout=10) == ( + "telegram", + "", + "12345", + ) + + +def test_concurrent_principals_do_not_interfere(mcp_loop): + results: dict = {} + + def scoped_call(key, platform, scope_id, user_id): + reset_session_vars() + set_session_vars(platform=platform, scope_id=scope_id, user_id=user_id) + results[key] = mcp_loop._run_on_mcp_loop(_read_principal(), timeout=10) + + threads = [ + threading.Thread( + target=scoped_call, args=("a", "slack", "T1", "U-alice") + ), + threading.Thread( + target=scoped_call, args=("b", "slack", "T1", "U-bob") + ), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=15) + + assert results == { + "a": ("slack", "T1", "U-alice"), + "b": ("slack", "T1", "U-bob"), + } + + +def test_wrap_is_noop_without_principal(mcp_loop): + async def trivial(): + return 42 + + coro = trivial() + wrapped = mcp_loop._wrap_with_session_principal(coro) + assert wrapped is coro + coro.close() + + +def test_oauth_capture_on_loop_uses_caller_principal(tmp_path, monkeypatch, mcp_loop): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / "config.yaml").write_text( + "mcp:\n oauth:\n identity_mode: per_user\n", + encoding="utf-8", + ) + from hermes_cli.config import load_config + from tools.mcp_oauth_identity import connection_registry_token, resolve_mcp_oauth_scope + + loaded = load_config() + assert loaded.get("mcp", {}).get("oauth", {}).get("identity_mode") == "per_user" + + set_session_vars(platform="slack", scope_id="T1", user_id="U-alice") + expected = connection_registry_token( + "github", + resolve_mcp_oauth_scope(uses_oauth=True), + ) + + async def capture(): + server = mcp_loop.MCPServerTask("github") + server._auth_type = "oauth" + server._capture_oauth_identity({"auth": "oauth", "url": "https://example"}) + return server._registry_key + + assert mcp_loop._run_on_mcp_loop(capture(), timeout=10) == expected + + +def test_apply_bound_session_principal_restores(): + reset_session_vars() + assert get_bound_session_principal() is None + from gateway.session_context import BoundSessionPrincipal + + principal = BoundSessionPrincipal("slack", "T1", "U-alice") + with apply_bound_session_principal(principal): + bound = get_bound_session_principal() + assert bound is not None + assert bound.user_id == "U-alice" + assert get_bound_session_principal() is None diff --git a/tests/tools/test_mcp_oauth_identity.py b/tests/tools/test_mcp_oauth_identity.py new file mode 100644 index 0000000000000..5d0c4e9de47ec --- /dev/null +++ b/tests/tools/test_mcp_oauth_identity.py @@ -0,0 +1,258 @@ +"""Behavior tests for requester-scoped MCP OAuth identity (#78174).""" + +from __future__ import annotations + +import hashlib +import json + +import pytest + +from gateway.session_context import ( + get_bound_session_principal, + reset_session_vars, + set_session_vars, +) +from tools.mcp_oauth_identity import ( + EMPTY_SCOPE_SENTINEL, + SHARED_SCOPE, + InvalidMcpOAuthIdentityModeError, + McpOAuthPrincipal, + McpOAuthScope, + MissingRequesterIdentity, + configured_identity_mode, + connection_registry_token, + is_registry_key_for_server, + parse_identity_mode, + principal_from_bound_fields, + registry_key_prefix, + resolve_mcp_oauth_scope, + schema_cache_entry_key, + server_uses_oauth, +) + + +def _principal(platform="slack", scope_id="T1", user_id="U1") -> McpOAuthPrincipal: + return principal_from_bound_fields(platform, scope_id, user_id) + + +class TestParseIdentityMode: + def test_shared_and_per_user_accepted(self): + assert parse_identity_mode("shared") == "shared" + assert parse_identity_mode("per_user") == "per_user" + + def test_absent_defaults_to_shared(self): + assert parse_identity_mode(None, explicit=False) == "shared" + + def test_typo_is_rejected_not_downgraded(self): + with pytest.raises(InvalidMcpOAuthIdentityModeError, match="per-user"): + parse_identity_mode("per-user") + with pytest.raises(InvalidMcpOAuthIdentityModeError): + parse_identity_mode("") + with pytest.raises(InvalidMcpOAuthIdentityModeError): + parse_identity_mode("PER_USER") + + def test_config_absent_key_is_shared(self): + assert configured_identity_mode({}) == "shared" + assert configured_identity_mode({"mcp": {}}) == "shared" + assert configured_identity_mode({"mcp": {"oauth": {}}}) == "shared" + + def test_config_explicit_invalid_raises(self): + with pytest.raises(InvalidMcpOAuthIdentityModeError): + configured_identity_mode( + {"mcp": {"oauth": {"identity_mode": "per-user"}}} + ) + + +class TestCanonicalPrincipal: + def test_deterministic_digest(self): + a = _principal() + b = _principal() + assert a.canonical_json() == b.canonical_json() + assert a.persistence_key() == b.persistence_key() + expected = ( + "u-v1-" + + hashlib.sha256(a.canonical_json().encode("utf-8")).hexdigest() + ) + assert a.persistence_key() == expected + assert json.loads(a.canonical_json()) == ["v1", "slack", "T1", "U1"] + + def test_platform_scope_and_user_are_distinct(self): + keys = { + _principal(platform="slack").persistence_key(), + _principal(platform="discord").persistence_key(), + _principal(scope_id="T2").persistence_key(), + _principal(user_id="U2").persistence_key(), + } + assert len(keys) == 4 + + def test_empty_scope_canonicalizes_to_sentinel(self): + p = _principal(scope_id="") + assert p.scope_id == EMPTY_SCOPE_SENTINEL + assert p.persistence_key() == _principal(scope_id="~").persistence_key() + + def test_path_hostile_ids_are_not_path_components(self): + p = _principal(user_id="../etc/passwd", scope_id="T/../../x") + key = p.persistence_key() + assert "/" not in key + assert ".." not in key + assert key.startswith("u-v1-") + assert len(key) == len("u-v1-") + 64 + + def test_long_ids_do_not_lengthen_the_key(self): + p = _principal(user_id="U" * 5000, scope_id="T" * 5000) + assert len(p.persistence_key()) == len("u-v1-") + 64 + + def test_missing_platform_or_user_rejected(self): + with pytest.raises(MissingRequesterIdentity): + principal_from_bound_fields("", "T1", "U1") + with pytest.raises(MissingRequesterIdentity): + principal_from_bound_fields("slack", "T1", "") + + +class TestBoundPrincipalGetter: + def test_unset_context_is_not_bound(self, monkeypatch): + reset_session_vars() + monkeypatch.setenv("HERMES_SESSION_PLATFORM", "slack") + monkeypatch.setenv("HERMES_SESSION_USER_ID", "UENV") + monkeypatch.setenv("HERMES_SESSION_SCOPE_ID", "TENV") + assert get_bound_session_principal() is None + + def test_bound_values_are_returned(self): + reset_session_vars() + set_session_vars(platform="slack", scope_id="T9", user_id="U9") + bound = get_bound_session_principal() + assert bound is not None + assert bound.platform == "slack" + assert bound.scope_id == "T9" + assert bound.user_id == "U9" + reset_session_vars() + + def test_empty_bound_user_is_not_a_principal(self): + reset_session_vars() + set_session_vars(platform="slack", scope_id="T9", user_id="") + assert get_bound_session_principal() is None + reset_session_vars() + + def test_telegram_empty_scope_is_still_bound(self): + reset_session_vars() + set_session_vars(platform="telegram", scope_id="", user_id="12345") + bound = get_bound_session_principal() + assert bound is not None + assert bound.scope_id == "" + assert bound.user_id == "12345" + reset_session_vars() + + +class TestResolveScope: + def test_shared_ignores_bound_principal(self): + reset_session_vars() + set_session_vars(platform="slack", scope_id="T1", user_id="U1") + scope = resolve_mcp_oauth_scope(identity_mode="shared") + assert scope == SHARED_SCOPE + reset_session_vars() + + def test_non_oauth_is_always_shared(self): + reset_session_vars() + set_session_vars(platform="slack", scope_id="T1", user_id="U1") + scope = resolve_mcp_oauth_scope(identity_mode="per_user", uses_oauth=False) + assert scope == SHARED_SCOPE + reset_session_vars() + + def test_per_user_uses_bound_principal(self): + reset_session_vars() + set_session_vars(platform="slack", scope_id="T1", user_id="U1") + scope = resolve_mcp_oauth_scope(identity_mode="per_user") + assert scope.mode == "per_user" + assert scope.principal is not None + assert scope.principal.user_id == "U1" + reset_session_vars() + + def test_per_user_without_identity_fails_closed(self): + reset_session_vars() + with pytest.raises(MissingRequesterIdentity, match="per_user"): + resolve_mcp_oauth_scope(identity_mode="per_user") + + def test_resolver_rejects_tool_argument_kwargs(self): + with pytest.raises(TypeError): + resolve_mcp_oauth_scope(user_id="U-from-tool") # type: ignore[call-arg] + with pytest.raises(TypeError): + resolve_mcp_oauth_scope("per_user") # type: ignore[misc] + + def test_explicit_principal_is_not_reread_from_context(self): + reset_session_vars() + set_session_vars(platform="slack", scope_id="T-bob", user_id="U-bob") + alice = _principal(user_id="U-alice") + scope = resolve_mcp_oauth_scope( + identity_mode="per_user", principal=alice + ) + assert scope.principal is not None + assert scope.principal.user_id == "U-alice" + reset_session_vars() + + +class TestRegistryAndCacheKeys: + def test_shared_registry_token_is_bare_server_name(self): + assert connection_registry_token("github", SHARED_SCOPE) == "github" + + def test_per_user_tokens_are_exact_and_distinct(self): + alice = resolve_mcp_oauth_scope( + identity_mode="per_user", principal=_principal(user_id="U-a") + ) + bob = resolve_mcp_oauth_scope( + identity_mode="per_user", principal=_principal(user_id="U-b") + ) + a_key = connection_registry_token("github", alice) + b_key = connection_registry_token("github", bob) + assert a_key != b_key + assert a_key != "github" + assert "U-a" not in a_key + assert "slack" not in a_key + registry = {a_key: "alice-conn"} + assert registry.get(b_key) is None + assert registry.get("github") is None + assert registry.get(a_key) == "alice-conn" + + def test_same_user_id_different_workspaces_differ(self): + a = _principal(scope_id="T-a", user_id="U1") + b = _principal(scope_id="T-b", user_id="U1") + assert a.persistence_key() != b.persistence_key() + + def test_private_schema_cache_is_scoped(self): + alice = resolve_mcp_oauth_scope( + identity_mode="per_user", principal=_principal(user_id="U-a") + ) + bob = resolve_mcp_oauth_scope( + identity_mode="per_user", principal=_principal(user_id="U-b") + ) + a_key = schema_cache_entry_key("github", alice) + b_key = schema_cache_entry_key("github", bob) + assert a_key != b_key + assert schema_cache_entry_key("github", SHARED_SCOPE) == "github" + assert schema_cache_entry_key("github", alice, cache_scope="public") == "github" + + def test_server_uses_oauth(self): + assert server_uses_oauth({"auth": "oauth", "url": "https://x"}) + assert not server_uses_oauth({"command": "npx"}) + assert not server_uses_oauth({"auth": "header"}) + assert not server_uses_oauth(None) + + def test_per_user_scope_requires_principal(self): + with pytest.raises(MissingRequesterIdentity): + McpOAuthScope(mode="per_user", principal=None) + + def test_shared_scope_strips_principal(self): + scope = McpOAuthScope(mode="shared", principal=_principal()) + assert scope.principal is None + assert scope.persistence_key() == "shared" + + def test_registry_key_prefix_match(self): + alice = resolve_mcp_oauth_scope( + identity_mode="per_user", principal=_principal(user_id="U-a") + ) + token = connection_registry_token("github", alice) + assert token.startswith(registry_key_prefix("github")) + assert is_registry_key_for_server("github", "github") + assert is_registry_key_for_server(token, "github") + assert not is_registry_key_for_server(token, "gitlab") + assert not is_registry_key_for_server("github-extra", "github") + assert schema_cache_entry_key("github", alice) == token diff --git a/tests/tools/test_mcp_oauth_integration.py b/tests/tools/test_mcp_oauth_integration.py index 8489548196b9f..6e662e0588f74 100644 --- a/tests/tools/test_mcp_oauth_integration.py +++ b/tests/tools/test_mcp_oauth_integration.py @@ -134,10 +134,10 @@ async def test_handle_401_deduplicates_concurrent_callers(tmp_path, monkeypatch) call_count = 0 real_invalidate = mgr.invalidate_if_disk_changed - async def counting(name): + async def counting(name, **_kwargs): nonlocal call_count call_count += 1 - return await real_invalidate(name) + return await real_invalidate(name, **_kwargs) monkeypatch.setattr(mgr, "invalidate_if_disk_changed", counting) diff --git a/tests/tools/test_mcp_oauth_per_user.py b/tests/tools/test_mcp_oauth_per_user.py new file mode 100644 index 0000000000000..8cfbaaaca0284 --- /dev/null +++ b/tests/tools/test_mcp_oauth_per_user.py @@ -0,0 +1,924 @@ +"""Requester-scoped MCP OAuth isolation (#78174). + +Exercises persistence, manager, live-registry, schema-cache, and CLI +fail-closed behavior against a real temp HERMES_HOME. Bound identity +comes from ``set_session_vars`` β€” never from tool arguments. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from gateway.session_context import reset_session_vars, set_session_vars +from tools.mcp_oauth_identity import ( + SHARED_SCOPE, + connection_registry_token, + principal_from_bound_fields, + resolve_mcp_oauth_scope, +) + + +def _alice(): + return principal_from_bound_fields("slack", "T1", "U-alice") + + +def _bob(): + return principal_from_bound_fields("slack", "T1", "U-bob") + + +def _alice_scope(): + return resolve_mcp_oauth_scope(identity_mode="per_user", principal=_alice()) + + +def _bob_scope(): + return resolve_mcp_oauth_scope(identity_mode="per_user", principal=_bob()) + + +def _seed_cached_tokens( + home: Path, server_name: str, oauth_scope, access_token: str = "TOK" +) -> None: + from tools.mcp_oauth import HermesTokenStorage + + storage = HermesTokenStorage( + server_name, hermes_home=home, oauth_scope=oauth_scope + ) + path = storage._tokens_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "access_token": access_token, + "token_type": "Bearer", + "expires_in": 3600, + } + ), + encoding="utf-8", + ) + + +def _write_per_user_config(home: Path) -> None: + (home / "config.yaml").write_text( + "mcp:\n oauth:\n identity_mode: per_user\n", + encoding="utf-8", + ) + from hermes_cli.config import load_config + + loaded = load_config() + assert loaded.get("mcp", {}).get("oauth", {}).get("identity_mode") == "per_user" + + +def _bind(platform: str, scope_id: str, user_id: str) -> None: + reset_session_vars() + set_session_vars(platform=platform, scope_id=scope_id, user_id=user_id) + + +@pytest.fixture(autouse=True) +def _reset_bound_principal(): + reset_session_vars() + yield + reset_session_vars() + + +class TestPerUserTokenLayout: + def test_alice_and_bob_get_distinct_paths(self, tmp_path): + from tools.mcp_oauth import HermesTokenStorage + + alice = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=_alice_scope() + ) + bob = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=_bob_scope() + ) + assert alice._tokens_path() != bob._tokens_path() + assert alice._tokens_path().parent != bob._tokens_path().parent + assert "by-user" in str(alice._tokens_path()) + assert alice._tokens_path().name == "github.json" + assert bob._tokens_path().name == "github.json" + assert _alice().persistence_key() in str(alice._tokens_path()) + assert _bob().persistence_key() not in str(alice._tokens_path()) + + def test_shared_layout_is_unchanged(self, tmp_path): + from tools.mcp_oauth import HermesTokenStorage + + shared = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=SHARED_SCOPE + ) + assert shared._tokens_path() == tmp_path / "mcp-tokens" / "github.json" + assert "by-user" not in str(shared._tokens_path()) + + def test_shared_token_is_not_assigned_to_a_requester(self, tmp_path): + from tools.mcp_oauth import HermesTokenStorage + + shared = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=SHARED_SCOPE + ) + shared._tokens_path().parent.mkdir(parents=True, exist_ok=True) + shared._tokens_path().write_text( + '{"access_token":"SHARED","token_type":"Bearer"}', encoding="utf-8" + ) + alice = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=_alice_scope() + ) + assert alice._tokens_path() != shared._tokens_path() + assert not alice._tokens_path().exists() + assert not alice.has_cached_tokens() + + def test_ids_never_appear_as_path_components(self, tmp_path): + from tools.mcp_oauth import HermesTokenStorage + + hostile = principal_from_bound_fields( + "slack", "../etc", "U/../../passwd" + ) + scope = resolve_mcp_oauth_scope(identity_mode="per_user", principal=hostile) + storage = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=scope + ) + path = str(storage._tokens_path()) + assert "../etc" not in path + assert "passwd" not in path + assert "U/" not in path + + +class TestPerUserRemove: + def test_alice_logout_does_not_wipe_bob(self, tmp_path): + from tools.mcp_oauth import HermesTokenStorage, remove_oauth_tokens + + alice = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=_alice_scope() + ) + bob = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=_bob_scope() + ) + for storage, token in ((alice, "ALICE"), (bob, "BOB")): + storage._tokens_path().parent.mkdir(parents=True, exist_ok=True) + storage._tokens_path().write_text( + json.dumps({"access_token": token, "token_type": "Bearer"}), + encoding="utf-8", + ) + remove_oauth_tokens( + "github", hermes_home=tmp_path, oauth_scope=_alice_scope() + ) + assert not alice._tokens_path().exists() + assert bob._tokens_path().exists() + assert "BOB" in bob._tokens_path().read_text(encoding="utf-8") + + def test_all_identities_removes_shared_and_by_user(self, tmp_path): + from tools.mcp_oauth import HermesTokenStorage, remove_oauth_tokens + + shared = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=SHARED_SCOPE + ) + alice = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=_alice_scope() + ) + bob = HermesTokenStorage( + "github", hermes_home=tmp_path, oauth_scope=_bob_scope() + ) + for storage in (shared, alice, bob): + storage._tokens_path().parent.mkdir(parents=True, exist_ok=True) + storage._tokens_path().write_text("{}", encoding="utf-8") + remove_oauth_tokens("github", hermes_home=tmp_path, all_identities=True) + assert not shared._tokens_path().exists() + assert not alice._tokens_path().exists() + assert not bob._tokens_path().exists() + + +class TestPerUserManager: + def test_alice_and_bob_get_distinct_providers(self, tmp_path, monkeypatch): + pytest.importorskip("mcp.client.auth.oauth2") + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from tools.mcp_oauth_manager import MCPOAuthManager + + _seed_cached_tokens(tmp_path, "github", _alice_scope(), "ALICE") + _seed_cached_tokens(tmp_path, "github", _bob_scope(), "BOB") + manager = MCPOAuthManager() + alice = manager.get_or_build_provider( + "github", + "https://mcp.example/mcp", + {}, + oauth_scope=_alice_scope(), + hermes_home=tmp_path, + ) + bob = manager.get_or_build_provider( + "github", + "https://mcp.example/mcp", + {}, + oauth_scope=_bob_scope(), + hermes_home=tmp_path, + ) + assert alice is not None and bob is not None + assert alice is not bob + assert manager._key( + "github", tmp_path, oauth_scope=_alice_scope() + ) != manager._key("github", tmp_path, oauth_scope=_bob_scope()) + + def test_key_without_scope_stays_shared(self, tmp_path): + from tools.mcp_oauth_manager import MCPOAuthManager + + manager = MCPOAuthManager() + assert manager._key("github", tmp_path) == manager._key( + "github", tmp_path, oauth_scope=SHARED_SCOPE + ) + assert manager._key("github", tmp_path) != manager._key( + "github", tmp_path, oauth_scope=_alice_scope() + ) + + def test_alice_remove_does_not_evict_bob(self, tmp_path, monkeypatch): + pytest.importorskip("mcp.client.auth.oauth2") + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from tools.mcp_oauth_manager import MCPOAuthManager + + _seed_cached_tokens(tmp_path, "github", _alice_scope(), "ALICE") + _seed_cached_tokens(tmp_path, "github", _bob_scope(), "BOB") + manager = MCPOAuthManager() + alice = manager.get_or_build_provider( + "github", + "https://mcp.example/mcp", + {}, + oauth_scope=_alice_scope(), + hermes_home=tmp_path, + ) + bob = manager.get_or_build_provider( + "github", + "https://mcp.example/mcp", + {}, + oauth_scope=_bob_scope(), + hermes_home=tmp_path, + ) + manager.remove( + "github", hermes_home=tmp_path, oauth_scope=_alice_scope() + ) + still_bob = manager.get_or_build_provider( + "github", + "https://mcp.example/mcp", + {}, + oauth_scope=_bob_scope(), + hermes_home=tmp_path, + ) + assert still_bob is bob + _seed_cached_tokens(tmp_path, "github", _alice_scope(), "ALICE2") + rebuilt_alice = manager.get_or_build_provider( + "github", + "https://mcp.example/mcp", + {}, + oauth_scope=_alice_scope(), + hermes_home=tmp_path, + ) + assert rebuilt_alice is not alice + + +class TestPerUserSchemaCache: + def test_private_cache_is_requester_scoped(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from tools.mcp_schema_cache import get_cached_entry, write_cache_entry + + write_cache_entry( + "github", + "fp-alice", + tools=[{"name": "alice_only", "description": "", "inputSchema": {}}], + cache_scope="private", + oauth_scope=_alice_scope(), + ) + assert get_cached_entry( + "github", + "fp-alice", + cache_scope="private", + oauth_scope=_alice_scope(), + ) is not None + assert get_cached_entry( + "github", + "fp-alice", + cache_scope="private", + oauth_scope=_bob_scope(), + ) is None + assert get_cached_entry("github", "fp-alice") is None + + def test_public_cache_may_stay_unscoped(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from tools.mcp_schema_cache import get_cached_entry, write_cache_entry + + write_cache_entry( + "github", + "fp-public", + tools=[{"name": "shared_tool", "description": "", "inputSchema": {}}], + cache_scope="public", + oauth_scope=_alice_scope(), + ) + assert get_cached_entry( + "github", + "fp-public", + cache_scope="public", + oauth_scope=_bob_scope(), + ) is not None + assert get_cached_entry( + "github", "fp-public", cache_scope="public" + ) is not None + + +class TestPerUserRuntimeRegistry: + def test_non_oauth_stays_on_bare_name_in_per_user(self): + from hermes_constants import get_hermes_home + import tools.mcp_tool as mcp + + _write_per_user_config(get_hermes_home()) + server = type( + "Srv", + (), + { + "name": "filesystem", + "session": object(), + "_oauth_scope": SHARED_SCOPE, + "_registry_key": "filesystem", + "_is_recycled_stdio": lambda self: False, + }, + )() + mcp._servers["filesystem"] = server + mcp._oauth_protected_servers.discard("filesystem") + try: + _bind("slack", "T1", "U-alice") + assert mcp._get_connected_server_for_call("filesystem") is server + finally: + mcp._servers.pop("filesystem", None) + + def test_exact_lookup_does_not_cross_principals(self, tmp_path, monkeypatch): + from hermes_constants import get_hermes_home + import tools.mcp_tool as mcp + + home = get_hermes_home() + _write_per_user_config(home) + mcp._oauth_protected_servers.add("github") + alice_key = connection_registry_token("github", _alice_scope()) + bob_key = connection_registry_token("github", _bob_scope()) + alice_server = type("Srv", (), {"name": "github", "session": object(), "_oauth_scope": _alice_scope(), "_registry_key": alice_key, "_is_recycled_stdio": lambda self: False})() + bob_server = type("Srv", (), {"name": "github", "session": object(), "_oauth_scope": _bob_scope(), "_registry_key": bob_key, "_is_recycled_stdio": lambda self: False})() + mcp._servers[alice_key] = alice_server + mcp._servers[bob_key] = bob_server + try: + _bind("slack", "T1", "U-alice") + assert mcp._get_connected_server_for_call("github") is alice_server + _bind("slack", "T1", "U-bob") + assert mcp._get_connected_server_for_call("github") is bob_server + assert mcp._servers.get("github") is None + finally: + mcp._servers.pop(alice_key, None) + mcp._servers.pop(bob_key, None) + mcp._oauth_protected_servers.discard("github") + + def test_missing_identity_does_not_pick_any_connection(self, tmp_path, monkeypatch): + from hermes_constants import get_hermes_home + import tools.mcp_tool as mcp + + home = get_hermes_home() + _write_per_user_config(home) + mcp._oauth_protected_servers.add("github") + alice_key = connection_registry_token("github", _alice_scope()) + alice_server = type("Srv", (), {"name": "github", "session": object(), "_oauth_scope": _alice_scope(), "_registry_key": alice_key, "_is_recycled_stdio": lambda self: False})() + mcp._servers[alice_key] = alice_server + try: + reset_session_vars() + assert mcp._get_connected_server_for_call("github") is None + err = mcp._oauth_call_target("github")[1] + assert err is not None + assert "per_user" in err + finally: + mcp._servers.pop(alice_key, None) + mcp._oauth_protected_servers.discard("github") + + def test_lazy_config_is_not_popped_in_per_user(self, tmp_path): + from hermes_constants import get_hermes_home + import tools.mcp_tool as mcp + + home = get_hermes_home() + _write_per_user_config(home) + mcp._lazy_server_configs["github"] = {"auth": "oauth", "url": "https://mcp.example"} + mcp._lazy_server_fingerprints["github"] = "fp" + mcp._lazy_server_tool_names["github"] = ["github_search"] + try: + fp, names = mcp._maybe_pop_lazy("github") + assert fp == "fp" + assert names == ["github_search"] + assert "github" in mcp._lazy_server_configs + assert mcp._lazy_server_tool_names["github"] == ["github_search"] + finally: + mcp._lazy_server_configs.pop("github", None) + mcp._lazy_server_fingerprints.pop("github", None) + mcp._lazy_server_tool_names.pop("github", None) + + def test_circuit_breaker_is_isolated(self): + import tools.mcp_tool as mcp + + alice_key = connection_registry_token("github", _alice_scope()) + bob_key = connection_registry_token("github", _bob_scope()) + mcp._server_error_counts.pop(alice_key, None) + mcp._server_error_counts.pop(bob_key, None) + mcp._server_breaker_opened_at.pop(alice_key, None) + mcp._server_breaker_opened_at.pop(bob_key, None) + try: + for _ in range(mcp._CIRCUIT_BREAKER_THRESHOLD): + mcp._bump_server_error(alice_key) + assert mcp._server_error_counts[alice_key] >= mcp._CIRCUIT_BREAKER_THRESHOLD + assert mcp._server_error_counts.get(bob_key, 0) == 0 + mcp._reset_server_error(alice_key) + assert mcp._server_error_counts[alice_key] == 0 + finally: + mcp._server_error_counts.pop(alice_key, None) + mcp._server_error_counts.pop(bob_key, None) + mcp._server_breaker_opened_at.pop(alice_key, None) + mcp._server_breaker_opened_at.pop(bob_key, None) + + def test_reconnect_is_exact_key_only(self): + import tools.mcp_tool as mcp + + alice_key = connection_registry_token("github", _alice_scope()) + bob_key = connection_registry_token("github", _bob_scope()) + + class _Srv: + def __init__(self, key): + self.name = "github" + self._registry_key = key + self.signaled = False + self._reconnect_event = type("E", (), {"set": lambda inner: None})() + + alice_server = _Srv(alice_key) + bob_server = _Srv(bob_key) + mcp._servers[alice_key] = alice_server + mcp._servers[bob_key] = bob_server + mcp._oauth_protected_servers.add("github") + from hermes_constants import get_hermes_home + + _write_per_user_config(get_hermes_home()) + try: + _bind("slack", "T1", "U-alice") + signaled = [] + + def _signal(server): + signaled.append(server) + return True + + original = mcp._signal_reconnect + mcp._signal_reconnect = _signal + try: + assert mcp.reconnect_mcp_server("github") is True + assert signaled == [alice_server] + finally: + mcp._signal_reconnect = original + finally: + mcp._servers.pop(alice_key, None) + mcp._servers.pop(bob_key, None) + mcp._oauth_protected_servers.discard("github") + + +class TestPerUserCliGuard: + def test_cli_blocks_without_bound_principal(self): + from hermes_constants import get_hermes_home + from hermes_cli.mcp_config import _per_user_oauth_cli_block + + _write_per_user_config(get_hermes_home()) + reset_session_vars() + msg = _per_user_oauth_cli_block() + assert msg is not None + assert "per_user" in msg + + def test_cli_allows_bound_gateway_principal(self): + from hermes_constants import get_hermes_home + from hermes_cli.mcp_config import _per_user_oauth_cli_block + + _write_per_user_config(get_hermes_home()) + _bind("slack", "T1", "U-alice") + assert _per_user_oauth_cli_block() is None + + def test_shared_mode_does_not_block_cli(self): + from hermes_cli.mcp_config import _per_user_oauth_cli_block + + reset_session_vars() + # Default config is shared. + assert _per_user_oauth_cli_block() is None + + +def _github_oauth_cfg(): + return {"auth": "oauth", "url": "https://mcp.example/mcp"} + + +def _cached_search_tools(): + return [{"name": "search", "description": "s", "inputSchema": {"type": "object"}}] + + +def _cleanup_github_runtime(mcp) -> None: + from tools.mcp_oauth_identity import ( + connection_registry_token, + is_registry_key_for_server, + ) + from tools.mcp_tool import mcp_prefixed_tool_name + from tools.registry import registry + + for name in list(mcp._lazy_server_tool_names.get("github") or []): + registry.deregister(name) + registry.deregister(mcp_prefixed_tool_name("github", "search")) + for key in list(mcp._servers): + if is_registry_key_for_server(key, "github"): + mcp._servers.pop(key, None) + mcp._lazy_server_configs.pop("github", None) + mcp._lazy_server_fingerprints.pop("github", None) + mcp._lazy_server_tool_names.pop("github", None) + mcp._oauth_protected_servers.discard("github") + mcp._server_connecting.discard("github") + alice_key = connection_registry_token("github", _alice_scope()) + bob_key = connection_registry_token("github", _bob_scope()) + mcp._server_connecting.discard(alice_key) + mcp._server_connecting.discard(bob_key) + + +class TestScopedCachePublicationOnRegister: + def test_unbound_startup_publishes_requester_scoped_cache(self): + from hermes_constants import get_hermes_home + from tools.mcp_schema_cache import config_fingerprint, write_cache_entry + from tools.mcp_tool import mcp_prefixed_tool_name + import tools.mcp_tool as mcp + + home = get_hermes_home() + _write_per_user_config(home) + cfg = _github_oauth_cfg() + write_cache_entry( + "github", + config_fingerprint(cfg), + tools=_cached_search_tools(), + oauth_scope=_alice_scope(), + ) + reset_session_vars() + try: + with patch.object(mcp, "_ensure_mcp_sdk", return_value=True), patch( + "tools.mcp_tool._run_on_mcp_loop" + ) as mock_run: + names = mcp.register_mcp_servers({"github": cfg}) + mock_run.assert_not_called() + tool_name = mcp_prefixed_tool_name("github", "search") + assert tool_name in (mcp._lazy_server_tool_names.get("github") or []) + assert tool_name in names + finally: + _cleanup_github_runtime(mcp) + + def test_bound_register_loads_scoped_cache_after_empty_stash(self): + from hermes_constants import get_hermes_home + from tools.mcp_schema_cache import config_fingerprint, write_cache_entry + from tools.mcp_tool import mcp_prefixed_tool_name + import tools.mcp_tool as mcp + + home = get_hermes_home() + _write_per_user_config(home) + cfg = _github_oauth_cfg() + reset_session_vars() + try: + with patch.object(mcp, "_ensure_mcp_sdk", return_value=True), patch( + "tools.mcp_tool._run_on_mcp_loop" + ) as mock_run: + mcp.register_mcp_servers({"github": cfg}) + mock_run.assert_not_called() + assert "github" in mcp._lazy_server_configs + assert not mcp._lazy_server_tool_names.get("github") + + write_cache_entry( + "github", + config_fingerprint(cfg), + tools=_cached_search_tools(), + oauth_scope=_alice_scope(), + ) + _bind("slack", "T1", "U-alice") + with patch.object(mcp, "_ensure_mcp_sdk", return_value=True), patch( + "tools.mcp_tool._run_on_mcp_loop" + ) as mock_run: + names = mcp.register_mcp_servers({"github": cfg}) + mock_run.assert_not_called() + tool_name = mcp_prefixed_tool_name("github", "search") + assert tool_name in (mcp._lazy_server_tool_names.get("github") or []) + assert tool_name in names + finally: + _cleanup_github_runtime(mcp) + + +class TestDeregisterKeepsSiblingTools: + def test_sibling_connection_keeps_shared_tool_name(self): + import tools.mcp_tool as mcp + from tools.registry import registry + from tools.mcp_tool import mcp_prefixed_tool_name + + tool_name = mcp_prefixed_tool_name("github", "search") + alice_key = connection_registry_token("github", _alice_scope()) + bob_key = connection_registry_token("github", _bob_scope()) + schema = { + "name": tool_name, + "description": "search", + "parameters": {"type": "object", "properties": {}}, + } + registry.register( + name=tool_name, + toolset="mcp-github", + schema=schema, + handler=lambda args, **kw: "{}", + ) + alice = mcp.MCPServerTask("github") + alice._registered_tool_names = [tool_name] + alice._registry_key = alice_key + bob = mcp.MCPServerTask("github") + bob._registered_tool_names = [tool_name] + bob._registry_key = bob_key + mcp._servers[alice_key] = alice + mcp._servers[bob_key] = bob + try: + alice._deregister_tools() + assert registry.get_toolset_for_tool(tool_name) == "mcp-github" + assert alice._registered_tool_names == [] + assert bob._registered_tool_names == [tool_name] + finally: + mcp._servers.pop(alice_key, None) + mcp._servers.pop(bob_key, None) + registry.deregister(tool_name) + + def test_cache_backed_names_survive_last_connection(self): + import tools.mcp_tool as mcp + from tools.registry import registry + from tools.mcp_tool import mcp_prefixed_tool_name + + tool_name = mcp_prefixed_tool_name("github", "search") + alice_key = connection_registry_token("github", _alice_scope()) + schema = { + "name": tool_name, + "description": "search", + "parameters": {"type": "object", "properties": {}}, + } + registry.register( + name=tool_name, + toolset="mcp-github", + schema=schema, + handler=lambda args, **kw: "{}", + ) + alice = mcp.MCPServerTask("github") + alice._registered_tool_names = [tool_name] + alice._registry_key = alice_key + mcp._servers[alice_key] = alice + mcp._lazy_server_tool_names["github"] = [tool_name] + try: + alice._deregister_tools() + assert registry.get_toolset_for_tool(tool_name) == "mcp-github" + finally: + mcp._servers.pop(alice_key, None) + mcp._lazy_server_tool_names.pop("github", None) + registry.deregister(tool_name) + + def test_last_connection_without_cache_deregisters(self): + import tools.mcp_tool as mcp + from tools.registry import registry + from tools.mcp_tool import mcp_prefixed_tool_name + + tool_name = mcp_prefixed_tool_name("github", "search") + alice_key = connection_registry_token("github", _alice_scope()) + schema = { + "name": tool_name, + "description": "search", + "parameters": {"type": "object", "properties": {}}, + } + registry.register( + name=tool_name, + toolset="mcp-github", + schema=schema, + handler=lambda args, **kw: "{}", + ) + alice = mcp.MCPServerTask("github") + alice._registered_tool_names = [tool_name] + alice._registry_key = alice_key + mcp._servers[alice_key] = alice + mcp._lazy_server_tool_names.pop("github", None) + try: + alice._deregister_tools() + assert registry.get_toolset_for_tool(tool_name) is None + finally: + mcp._servers.pop(alice_key, None) + registry.deregister(tool_name) + + +class TestOAuthClassificationReload: + def test_auth_change_discards_oauth_flag_and_refreshes_lazy_template(self): + import tools.mcp_tool as mcp + from tools.mcp_oauth_identity import server_uses_oauth + + mcp._oauth_protected_servers.add("github") + mcp._lazy_server_configs["github"] = _github_oauth_cfg() + mcp._lazy_server_tool_names["github"] = ["mcp__github__search"] + header_cfg = {"url": "https://mcp.example/mcp", "headers": {"Authorization": "Bearer x"}} + try: + with patch.object(mcp, "_ensure_mcp_sdk", return_value=True), patch( + "tools.mcp_tool._run_on_mcp_loop" + ) as mock_run: + mcp.register_mcp_servers({"github": header_cfg}) + mock_run.assert_not_called() + assert "github" not in mcp._oauth_protected_servers + assert not server_uses_oauth(mcp._lazy_server_configs["github"]) + assert not mcp._mcp_server_uses_oauth("github") + finally: + _cleanup_github_runtime(mcp) + + def test_partial_register_does_not_clear_other_oauth_servers(self): + import tools.mcp_tool as mcp + + mcp._oauth_protected_servers.add("github") + mcp._oauth_protected_servers.add("other") + mcp._lazy_server_configs["filesystem"] = {"command": "npx", "args": []} + mcp._lazy_server_tool_names["filesystem"] = ["mcp__filesystem__ls"] + try: + with patch.object(mcp, "_ensure_mcp_sdk", return_value=True): + mcp.register_mcp_servers( + {"filesystem": {"command": "npx", "args": []}} + ) + assert "github" in mcp._oauth_protected_servers + assert "other" in mcp._oauth_protected_servers + finally: + mcp._oauth_protected_servers.discard("github") + mcp._oauth_protected_servers.discard("other") + mcp._lazy_server_configs.pop("filesystem", None) + mcp._lazy_server_tool_names.pop("filesystem", None) + + def test_shutdown_clears_oauth_protected_servers(self): + import tools.mcp_tool as mcp + + mcp._oauth_protected_servers.add("github") + try: + mcp.shutdown_mcp_servers() + assert not mcp._oauth_protected_servers + finally: + mcp._oauth_protected_servers.discard("github") + + +class TestScopedMcpReload: + def test_reload_shutdown_keys_spares_other_principal(self): + from tools.mcp_tool import _reload_shutdown_keys + + alice_key = connection_registry_token("github", _alice_scope()) + bob_key = connection_registry_token("github", _bob_scope()) + retired_key = connection_registry_token("retired", _bob_scope()) + servers = { + alice_key: SimpleNamespace(name="github", _oauth_scope=_alice_scope()), + bob_key: SimpleNamespace(name="github", _oauth_scope=_bob_scope()), + "filesystem": SimpleNamespace(name="filesystem", _oauth_scope=None), + retired_key: SimpleNamespace(name="retired", _oauth_scope=_bob_scope()), + } + keys = _reload_shutdown_keys( + {"github", "filesystem"}, servers, _alice_scope() + ) + assert keys is not None + assert alice_key in keys + assert "filesystem" in keys + assert retired_key in keys + assert bob_key not in keys + + def test_reload_shutdown_keys_full_wipe_when_unbound(self): + from tools.mcp_tool import _reload_shutdown_keys + + servers = { + connection_registry_token("github", _bob_scope()): SimpleNamespace( + name="github", _oauth_scope=_bob_scope() + ) + } + assert _reload_shutdown_keys({"github"}, servers, None) is None + + def test_shutdown_with_only_lazy_templates_deregisters_tools(self): + import tools.mcp_tool as mcp + from tools.mcp_tool import mcp_prefixed_tool_name + from tools.registry import registry + + _cleanup_github_runtime(mcp) + tool_name = mcp_prefixed_tool_name("github", "search") + registry.register( + name=tool_name, + toolset="mcp-github", + schema={ + "name": tool_name, + "description": "search", + "parameters": {"type": "object", "properties": {}}, + }, + handler=lambda args, **kw: "{}", + ) + mcp._lazy_server_configs["github"] = _github_oauth_cfg() + mcp._lazy_server_tool_names["github"] = [tool_name] + mcp._oauth_protected_servers.add("github") + try: + with patch.object(mcp, "_stop_mcp_loop"): + mcp.shutdown_mcp_servers() + assert registry.get_toolset_for_tool(tool_name) is None + assert "github" not in mcp._lazy_server_configs + assert "github" not in mcp._lazy_server_tool_names + assert "github" not in mcp._oauth_protected_servers + finally: + registry.deregister(tool_name) + _cleanup_github_runtime(mcp) + + def test_alice_reload_leaves_bob_connected_and_purges_removed_lazy(self): + import tools.mcp_tool as mcp + from hermes_constants import get_hermes_home + from tools.mcp_tool import mcp_prefixed_tool_name + from tools.registry import registry + + _write_per_user_config(get_hermes_home()) + _bind("slack", "T1", "U-alice") + _cleanup_github_runtime(mcp) + + alice_key = connection_registry_token("github", _alice_scope()) + bob_key = connection_registry_token("github", _bob_scope()) + alice = mcp.MCPServerTask("github") + bob = mcp.MCPServerTask("github") + alice._oauth_scope = _alice_scope() + bob._oauth_scope = _bob_scope() + alice._registry_key = alice_key + bob._registry_key = bob_key + mcp._servers[alice_key] = alice + mcp._servers[bob_key] = bob + + gone_tool = mcp_prefixed_tool_name("retired", "x") + registry.register( + name=gone_tool, + toolset="mcp-retired", + schema={ + "name": gone_tool, + "description": "x", + "parameters": {"type": "object", "properties": {}}, + }, + handler=lambda args, **kw: "{}", + ) + mcp._lazy_server_configs["retired"] = {"url": "https://gone.example/mcp"} + mcp._lazy_server_tool_names["retired"] = [gone_tool] + mcp._lazy_server_configs["github"] = _github_oauth_cfg() + mcp._lazy_server_tool_names["github"] = [ + mcp_prefixed_tool_name("github", "search") + ] + try: + with patch.object( + mcp, "_load_mcp_config", return_value={"github": _github_oauth_cfg()} + ), patch.object(mcp, "_close_mcp_tasks"), patch.object( + mcp, "_stop_mcp_loop" + ): + mcp.reload_mcp_connections() + assert bob_key in mcp._servers + assert alice_key not in mcp._servers + assert "retired" not in mcp._lazy_server_configs + assert "retired" not in mcp._lazy_server_tool_names + assert "github" in mcp._lazy_server_configs + assert registry.get_toolset_for_tool(gone_tool) is None + finally: + registry.deregister(gone_tool) + mcp._lazy_server_configs.pop("retired", None) + mcp._lazy_server_tool_names.pop("retired", None) + mcp._servers.pop(alice_key, None) + mcp._servers.pop(bob_key, None) + _cleanup_github_runtime(mcp) + + @pytest.mark.asyncio + async def test_refresh_keeps_tool_still_advertised_by_sibling(self): + import tools.mcp_tool as mcp + from tools.mcp_tool import mcp_prefixed_tool_name + from tools.registry import registry + + tool_name = mcp_prefixed_tool_name("github", "search") + new_name = mcp_prefixed_tool_name("github", "other") + alice_key = connection_registry_token("github", _alice_scope()) + bob_key = connection_registry_token("github", _bob_scope()) + registry.register( + name=tool_name, + toolset="mcp-github", + schema={ + "name": tool_name, + "description": "search", + "parameters": {"type": "object", "properties": {}}, + }, + handler=lambda args, **kw: "{}", + ) + alice = mcp.MCPServerTask("github") + bob = mcp.MCPServerTask("github") + alice._registered_tool_names = [tool_name] + bob._registered_tool_names = [tool_name] + alice._config = {} + alice.session = SimpleNamespace( + list_tools=AsyncMock( + return_value=SimpleNamespace( + tools=[SimpleNamespace(name="other", description="o", inputSchema=None)] + ) + ) + ) + mcp._servers[alice_key] = alice + mcp._servers[bob_key] = bob + try: + await alice._refresh_tools() + assert registry.get_toolset_for_tool(tool_name) == "mcp-github" + assert new_name in registry.get_all_tool_names() + assert tool_name not in alice._registered_tool_names + assert tool_name in bob._registered_tool_names + finally: + mcp._servers.pop(alice_key, None) + mcp._servers.pop(bob_key, None) + for name in list(getattr(alice, "_registered_tool_names", []) or []): + registry.deregister(name) + registry.deregister(tool_name) + registry.deregister(new_name) + _cleanup_github_runtime(mcp) diff --git a/tests/tools/test_mcp_schema_cache.py b/tests/tools/test_mcp_schema_cache.py index cc6df9a6d29cd..be9bff4c7856d 100644 --- a/tests/tools/test_mcp_schema_cache.py +++ b/tests/tools/test_mcp_schema_cache.py @@ -110,3 +110,67 @@ def _counting_save(data): # Changed payload β†’ rewrite. msc.write_cache_entry("srv", "fp2", tools=tools, utility_tools=[]) assert len(saves) == 2 + + +class TestStartupCachedEntry: + def _isolate(self, monkeypatch, tmp_path): + monkeypatch.setattr(msc, "_cache_path", lambda: tmp_path / "cache.json") + + def test_unscoped_miss_falls_back_to_requester_scoped_entry( + self, monkeypatch, tmp_path + ): + self._isolate(monkeypatch, tmp_path) + from tools.mcp_oauth_identity import ( + principal_from_bound_fields, + resolve_mcp_oauth_scope, + ) + + scope = resolve_mcp_oauth_scope( + identity_mode="per_user", + principal=principal_from_bound_fields("slack", "T1", "U-alice"), + ) + tools = [{"name": "t1", "description": "d", "inputSchema": {"type": "object"}}] + msc.write_cache_entry("github", "fp1", tools=tools, oauth_scope=scope) + assert msc.get_cached_entry("github", "fp1") is None + entry = msc.get_startup_cached_entry("github", "fp1") + assert entry is not None + assert msc.tools_from_cache_entry(entry) == tools + + def test_unscoped_entry_wins_over_scoped(self, monkeypatch, tmp_path): + self._isolate(monkeypatch, tmp_path) + from tools.mcp_oauth_identity import ( + principal_from_bound_fields, + resolve_mcp_oauth_scope, + ) + + scope = resolve_mcp_oauth_scope( + identity_mode="per_user", + principal=principal_from_bound_fields("slack", "T1", "U-alice"), + ) + public = [{"name": "public", "description": "", "inputSchema": {}}] + private = [{"name": "private", "description": "", "inputSchema": {}}] + msc.write_cache_entry("github", "fp1", tools=private, oauth_scope=scope) + msc.write_cache_entry("github", "fp1", tools=public) + entry = msc.get_startup_cached_entry("github", "fp1") + assert msc.tools_from_cache_entry(entry) == public + + def test_fingerprint_mismatch_does_not_return_scoped_entry( + self, monkeypatch, tmp_path + ): + self._isolate(monkeypatch, tmp_path) + from tools.mcp_oauth_identity import ( + principal_from_bound_fields, + resolve_mcp_oauth_scope, + ) + + scope = resolve_mcp_oauth_scope( + identity_mode="per_user", + principal=principal_from_bound_fields("slack", "T1", "U-alice"), + ) + msc.write_cache_entry( + "github", + "fp1", + tools=[{"name": "t1", "description": "", "inputSchema": {}}], + oauth_scope=scope, + ) + assert msc.get_startup_cached_entry("github", "OTHER") is None diff --git a/tests/tools/test_mcp_tool_401_handling.py b/tests/tools/test_mcp_tool_401_handling.py index 386cfc4dc5cc6..1cfd5943d0335 100644 --- a/tests/tools/test_mcp_tool_401_handling.py +++ b/tests/tools/test_mcp_tool_401_handling.py @@ -58,7 +58,7 @@ async def _call_tool_raises(*a, **kw): # Force handle_401 to return False (no recovery available) mgr = get_manager() - async def _h401(name, token=None): + async def _h401(name, token=None, **_kwargs): return False monkeypatch.setattr(mgr, "handle_401", _h401) diff --git a/tests/tui_gateway/test_mcp_reload_rev.py b/tests/tui_gateway/test_mcp_reload_rev.py index 4370c7f35e79d..16d4cd95d4341 100644 --- a/tests/tui_gateway/test_mcp_reload_rev.py +++ b/tests/tui_gateway/test_mcp_reload_rev.py @@ -33,6 +33,7 @@ def reload_env(monkeypatch): calls = {"discover": 0, "shutdown": 0} rev_box = {"rev": "rev-a"} + monkeypatch.setattr(mcp_tool, "reload_mcp_connections", lambda: calls.__setitem__("shutdown", calls["shutdown"] + 1)) monkeypatch.setattr(mcp_tool, "shutdown_mcp_servers", lambda: calls.__setitem__("shutdown", calls["shutdown"] + 1)) monkeypatch.setattr(mcp_tool, "discover_mcp_tools", lambda: calls.__setitem__("discover", calls["discover"] + 1)) monkeypatch.setattr(srv, "_compute_mcp_rev", lambda: rev_box["rev"]) diff --git a/tools/mcp_oauth.py b/tools/mcp_oauth.py index 134b260a26687..8435fb663148d 100644 --- a/tools/mcp_oauth.py +++ b/tools/mcp_oauth.py @@ -58,9 +58,10 @@ from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path -from typing import Any +from typing import Any, Optional from urllib.parse import parse_qs, urlparse from hermes_constants import secure_parent_dir +from tools.mcp_oauth_identity import McpOAuthScope, PERSISTENCE_KEY_PREFIX logger = logging.getLogger(__name__) @@ -457,29 +458,61 @@ def _write_json(path: Path, data: dict) -> None: class HermesTokenStorage: """Persist OAuth tokens and client registration to JSON files. - File layout:: + File layout (``identity_mode: shared``, the default):: HERMES_HOME/mcp-tokens/.json -- tokens HERMES_HOME/mcp-tokens/.client.json -- client info HERMES_HOME/mcp-tokens/.meta.json -- oauth server metadata HERMES_HOME/mcp-tokens/.cimd-off -- CIMD refused here + + File layout (``identity_mode: per_user``):: + + HERMES_HOME/mcp-tokens/by-user//.json + ... """ - def __init__(self, server_name: str, *, hermes_home: str | Path | None = None): + def __init__( + self, + server_name: str, + *, + hermes_home: str | Path | None = None, + oauth_scope: Optional[McpOAuthScope] = None, + ): + from hermes_constants import get_hermes_home + from tools.mcp_oauth_identity import resolve_mcp_oauth_scope + self._server_name = _safe_filename(server_name) - self._hermes_home = Path(hermes_home) if hermes_home is not None else None + # Pin home at construction so later file ops never re-resolve ambient + # HERMES_HOME (a confused-deputy risk under concurrent profiles). + self._hermes_home = ( + Path(hermes_home) if hermes_home is not None else Path(get_hermes_home()) + ) + self._oauth_scope = ( + oauth_scope + if oauth_scope is not None + else resolve_mcp_oauth_scope(uses_oauth=True) + ) + + def _namespace_dir(self) -> Path: + from tools.mcp_oauth_identity import IDENTITY_MODE_SHARED + + base = _get_token_dir(self._hermes_home) + key = self._oauth_scope.persistence_key() + if key == IDENTITY_MODE_SHARED: + return base + return base / "by-user" / key def _tokens_path(self) -> Path: - return _get_token_dir(self._hermes_home) / f"{self._server_name}.json" + return self._namespace_dir() / f"{self._server_name}.json" def _client_info_path(self) -> Path: - return _get_token_dir(self._hermes_home) / f"{self._server_name}.client.json" + return self._namespace_dir() / f"{self._server_name}.client.json" def _meta_path(self) -> Path: - return _get_token_dir(self._hermes_home) / f"{self._server_name}.meta.json" + return self._namespace_dir() / f"{self._server_name}.meta.json" def _cimd_rejected_path(self) -> Path: - return _get_token_dir(self._hermes_home) / f"{self._server_name}.cimd-off" + return self._namespace_dir() / f"{self._server_name}.cimd-off" # -- tokens ------------------------------------------------------------ @@ -1274,13 +1307,46 @@ def remove_oauth_tokens( server_name: str, *, hermes_home: str | Path | None = None, + oauth_scope: Optional[McpOAuthScope] = None, + all_identities: bool = False, ) -> None: - """Delete stored OAuth tokens and client info for a server.""" - storage = HermesTokenStorage(server_name, hermes_home=hermes_home) + """Delete stored OAuth tokens and client info for a server. + + Default: the current OAuth scope only (shared layout, or one per-user + namespace). ``all_identities=True`` is the administrative path used when + removing a server from config β€” it deletes the shared artifacts *and* + that server's files under ``by-user/*/`` without enumerating principals. + """ + if all_identities: + _remove_oauth_tokens_all_identities(server_name, hermes_home=hermes_home) + return + storage = HermesTokenStorage( + server_name, hermes_home=hermes_home, oauth_scope=oauth_scope + ) storage.remove() logger.info("OAuth tokens removed for '%s'", server_name) +def _remove_oauth_tokens_all_identities( + server_name: str, + *, + hermes_home: str | Path | None = None, +) -> None: + safe = _safe_filename(server_name) + suffixes = (".json", ".client.json", ".meta.json", ".cimd-off") + base = _get_token_dir(hermes_home) + for suffix in suffixes: + (base / f"{safe}{suffix}").unlink(missing_ok=True) + by_user = base / "by-user" + if by_user.is_dir(): + for ns in by_user.iterdir(): + if not ns.is_dir() or not ns.name.startswith(PERSISTENCE_KEY_PREFIX): + continue + for suffix in suffixes: + (ns / f"{safe}{suffix}").unlink(missing_ok=True) + logger.info("OAuth tokens removed for '%s' (all identities)", server_name) + + # --------------------------------------------------------------------------- # Extracted helpers (Task 3 of MCP OAuth consolidation) # diff --git a/tools/mcp_oauth_identity.py b/tools/mcp_oauth_identity.py new file mode 100644 index 0000000000000..26091be262368 --- /dev/null +++ b/tools/mcp_oauth_identity.py @@ -0,0 +1,235 @@ +"""Requester-scoped MCP OAuth identity types and fail-closed resolution. + +See docs/rfc/requester-scoped-mcp-oauth.md and issue #78174. + +This module is the single source of truth for: + +- ``shared`` vs ``per_user`` identity mode +- the immutable OAuth principal/scope +- opaque persistence keys +- live-connection registry tokens + +Credential identity is derived only from trusted bound session context. +Tool arguments and model output must never reach these APIs as selectors. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Literal, Optional + +IDENTITY_MODE_SHARED = "shared" +IDENTITY_MODE_PER_USER = "per_user" +VALID_IDENTITY_MODES = frozenset({IDENTITY_MODE_SHARED, IDENTITY_MODE_PER_USER}) + +PRINCIPAL_VERSION = "v1" +EMPTY_SCOPE_SENTINEL = "~" +PERSISTENCE_KEY_PREFIX = f"u-{PRINCIPAL_VERSION}-" +REGISTRY_SEPARATOR = "\x1f" + + +class InvalidMcpOAuthIdentityModeError(ValueError): + """Raised when ``mcp.oauth.identity_mode`` is present but not a valid value.""" + + +class MissingRequesterIdentity(RuntimeError): + """Raised in ``per_user`` mode when no trusted bound principal exists.""" + + +@dataclass(frozen=True, slots=True) +class McpOAuthPrincipal: + """Immutable requester identity for MCP OAuth isolation.""" + + version: Literal["v1"] + platform: str + scope_id: str + user_id: str + + def canonical_json(self) -> str: + return json.dumps( + [self.version, self.platform, self.scope_id, self.user_id], + separators=(",", ":"), + ensure_ascii=True, + ) + + def persistence_key(self) -> str: + digest = hashlib.sha256(self.canonical_json().encode("utf-8")).hexdigest() + return f"{PERSISTENCE_KEY_PREFIX}{digest}" + + +@dataclass(frozen=True, slots=True) +class McpOAuthScope: + """Authorization isolation mode plus optional principal.""" + + mode: Literal["shared", "per_user"] + principal: Optional[McpOAuthPrincipal] = None + + def __post_init__(self) -> None: + if self.mode == IDENTITY_MODE_SHARED: + if self.principal is not None: + object.__setattr__(self, "principal", None) + return + if self.principal is None: + raise MissingRequesterIdentity( + "per_user MCP OAuth scope requires a bound requester principal." + ) + + def persistence_key(self) -> str: + if self.mode == IDENTITY_MODE_SHARED: + return IDENTITY_MODE_SHARED + # ``__post_init__`` requires a principal in per_user mode. + assert self.principal is not None + return self.principal.persistence_key() + + +SHARED_SCOPE = McpOAuthScope(mode="shared", principal=None) + + +def parse_identity_mode(value: Any, *, explicit: bool = True) -> str: + """Parse ``mcp.oauth.identity_mode``. + + Absent/None when ``explicit`` is False defaults to ``shared``. + Any other value, including typos such as ``per-user``, is an error. + """ + if value is None and not explicit: + return IDENTITY_MODE_SHARED + if isinstance(value, str) and value in VALID_IDENTITY_MODES: + return value + raise InvalidMcpOAuthIdentityModeError( + "mcp.oauth.identity_mode must be 'shared' or 'per_user', " + f"got {value!r}. A typo must not silently fall back to shared mode." + ) + + +def configured_identity_mode(config: Optional[dict] = None) -> str: + """Read identity mode from a config dict or the loaded Hermes config.""" + if config is None: + try: + from hermes_cli.config import load_config_readonly + + config = load_config_readonly() + except Exception: + return IDENTITY_MODE_SHARED + mcp = config.get("mcp") if isinstance(config, dict) else None + if not isinstance(mcp, dict): + return IDENTITY_MODE_SHARED + oauth = mcp.get("oauth") + if not isinstance(oauth, dict) or "identity_mode" not in oauth: + return IDENTITY_MODE_SHARED + return parse_identity_mode(oauth.get("identity_mode"), explicit=True) + + +def server_uses_oauth(config: Optional[dict]) -> bool: + if not isinstance(config, dict): + return False + return str(config.get("auth") or "").lower().strip() == "oauth" + + +def principal_from_bound_fields( + platform: str, + scope_id: str, + user_id: str, +) -> McpOAuthPrincipal: + platform = (platform or "").strip() + user_id = (user_id or "").strip() + scope_id = (scope_id or "").strip() or EMPTY_SCOPE_SENTINEL + if not platform or not user_id: + raise MissingRequesterIdentity( + "MCP OAuth requires an authenticated requester identity in " + "per_user mode." + ) + if "\x00" in platform or "\x00" in scope_id or "\x00" in user_id: + raise MissingRequesterIdentity( + "MCP OAuth requester identity contains invalid NUL bytes." + ) + return McpOAuthPrincipal( + version="v1", + platform=platform, + scope_id=scope_id, + user_id=user_id, + ) + + +def resolve_mcp_oauth_scope( + *, + identity_mode: Optional[str] = None, + uses_oauth: bool = True, + principal: Optional[McpOAuthPrincipal] = None, + config: Optional[dict] = None, +) -> McpOAuthScope: + """Resolve the OAuth isolation scope for this request. + + Keyword-only on purpose: callers cannot pass MCP tool arguments such as + ``user_id`` into this function. + """ + if not uses_oauth: + return SHARED_SCOPE + + mode = identity_mode + if mode is None: + mode = configured_identity_mode(config) + else: + mode = parse_identity_mode(mode, explicit=True) + + if mode == IDENTITY_MODE_SHARED: + return SHARED_SCOPE + + resolved = principal + if resolved is None: + from gateway.session_context import get_bound_session_principal + + bound = get_bound_session_principal() + if bound is not None: + resolved = principal_from_bound_fields( + bound.platform, bound.scope_id, bound.user_id + ) + if resolved is None: + raise MissingRequesterIdentity( + "MCP OAuth requires an authenticated requester identity in " + "per_user mode. Direct CLI, TUI, desktop, and cron paths " + "without a bound gateway principal cannot use a shared " + "credential as a fallback." + ) + return McpOAuthScope(mode="per_user", principal=resolved) + + +def connection_registry_token(server_name: str, scope: McpOAuthScope) -> str: + """Exact live-registry key. Shared mode preserves the bare server name.""" + if scope.mode == IDENTITY_MODE_SHARED: + return server_name + return f"{server_name}{REGISTRY_SEPARATOR}{scope.persistence_key()}" + + +def registry_key_prefix(server_name: str) -> str: + return f"{server_name}{REGISTRY_SEPARATOR}" + + +def is_registry_key_for_server(key: str, server_name: str) -> bool: + """True if ``key`` is the bare name or a per-user token for that server. + + Status/discovery aggregation only. Credential paths must use + :func:`connection_registry_token` for an exact match. + """ + return key == server_name or key.startswith(registry_key_prefix(server_name)) + + +def schema_cache_entry_key( + server_name: str, + scope: Optional[McpOAuthScope] = None, + *, + cache_scope: Optional[str] = None, +) -> str: + """Disk/memory key for list/schema cache entries. + + Unscoped/shared lookups keep the historical server-name key so existing + caches and tests keep working. In ``per_user``, private (and unknown) + cache entries are principal-scoped. Explicit ``cacheScope=public`` may + stay unscoped. + """ + if scope is None or scope.mode == IDENTITY_MODE_SHARED: + return server_name + if (cache_scope or "").lower() == "public": + return server_name + return connection_registry_token(server_name, scope) diff --git a/tools/mcp_oauth_manager.py b/tools/mcp_oauth_manager.py index d0bca49967b59..c0682a512fefd 100644 --- a/tools/mcp_oauth_manager.py +++ b/tools/mcp_oauth_manager.py @@ -42,9 +42,16 @@ from pathlib import Path from typing import Any, Optional +from tools.mcp_oauth_identity import McpOAuthScope, SHARED_SCOPE, resolve_mcp_oauth_scope + logger = logging.getLogger(__name__) +def _resolved_oauth_scope(oauth_scope: Optional[McpOAuthScope] = None) -> McpOAuthScope: + """Resolve identity at a public API edge. ``_key`` itself stays pure.""" + return oauth_scope if oauth_scope is not None else resolve_mcp_oauth_scope(uses_oauth=True) + + def _same_endpoint(a: str, b: str) -> bool: """Return True if two URLs target the same endpoint (ignoring query/fragment). @@ -96,6 +103,8 @@ class _ProviderEntry: last_mtime_ns: int = 0 lock: asyncio.Lock = field(default_factory=asyncio.Lock) pending_401: dict[str, "asyncio.Future[bool]"] = field(default_factory=dict) + oauth_scope: Optional[McpOAuthScope] = None + hermes_home: Optional[str] = None # --------------------------------------------------------------------------- @@ -140,6 +149,7 @@ def __init__( super().__init__(*args, **kwargs) self._hermes_server_name = server_name self._hermes_home = "" + self._hermes_oauth_scope = None # 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 @@ -509,6 +519,7 @@ async def async_auth_flow(self, request): # type: ignore[override] await get_manager().invalidate_if_disk_changed( self._hermes_server_name, hermes_home=self._hermes_home, + oauth_scope=getattr(self, "_hermes_oauth_scope", None) or SHARED_SCOPE, ) except Exception as exc: # pragma: no cover β€” defensive logger.debug( @@ -566,7 +577,7 @@ class MCPOAuthManager: """ 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 @@ -580,16 +591,24 @@ def get_or_build_provider( server_name: str, server_url: str, oauth_config: Optional[dict], + *, + oauth_scope: Optional[McpOAuthScope] = None, + hermes_home: str | Path | 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. + Idempotent: repeat calls with the same name *and OAuth scope* return + the same instance. If ``server_url`` changes for a given key, 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) + from hermes_constants import get_hermes_home + + scope = _resolved_oauth_scope(oauth_scope) + home = Path(hermes_home) if hermes_home is not None else get_hermes_home() + home_s = str(home.expanduser().resolve(strict=False)) + key = self._key(server_name, home_s, oauth_scope=scope) with self._entries_lock: entry = self._entries.get(key) if entry is not None and entry.server_url != server_url: @@ -603,13 +622,16 @@ def get_or_build_provider( entry = _ProviderEntry( server_url=server_url, oauth_config=oauth_config, + oauth_scope=scope, + hermes_home=home_s, ) self._entries[key] = entry if entry.provider is None: entry.provider = self._build_provider(server_name, entry) if entry.provider is not None: - entry.provider._hermes_home = key[0] + entry.provider._hermes_home = home_s + entry.provider._hermes_oauth_scope = scope return entry.provider @@ -617,11 +639,17 @@ def get_or_build_provider( def _key( server_name: str, hermes_home: str | Path | None = None, - ) -> tuple[str, str]: + oauth_scope: Optional[McpOAuthScope] = 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() - return (str(home.expanduser().resolve(strict=False)), server_name) + scope = oauth_scope if oauth_scope is not None else SHARED_SCOPE + return ( + str(home.expanduser().resolve(strict=False)), + server_name, + scope.persistence_key(), + ) def _build_provider( self, @@ -667,7 +695,11 @@ def _build_provider( apply_oauth_provider_defaults( cfg, server_name=server_name, server_url=entry.server_url ) - storage = HermesTokenStorage(server_name) + storage = HermesTokenStorage( + server_name, + hermes_home=entry.hermes_home, + oauth_scope=entry.oauth_scope, + ) from tools.mcp_dashboard_oauth import get_dashboard_oauth_flow @@ -714,22 +746,49 @@ def remove( server_name: str, *, hermes_home: str | Path | None = None, + oauth_scope: Optional[McpOAuthScope] = None, + all_identities: bool = False, ) -> _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. """ - with self._entries_lock: - entry = self._entries.pop(self._key(server_name, hermes_home), None) - - from tools.mcp_oauth import remove_oauth_tokens - remove_oauth_tokens(server_name, hermes_home=hermes_home) + popped: _ProviderEntry | None = None + if all_identities: + prefix = None + if hermes_home is not None: + prefix = str(Path(hermes_home).expanduser().resolve(strict=False)) + else: + from hermes_constants import get_hermes_home + prefix = str(get_hermes_home().expanduser().resolve(strict=False)) + with self._entries_lock: + drop = [ + key for key in self._entries + if key[0] == prefix and key[1] == server_name + ] + for key in drop: + popped = self._entries.pop(key, popped) + from tools.mcp_oauth import remove_oauth_tokens + remove_oauth_tokens( + server_name, hermes_home=hermes_home, all_identities=True + ) + else: + scope = _resolved_oauth_scope(oauth_scope) + with self._entries_lock: + popped = self._entries.pop( + self._key(server_name, hermes_home, oauth_scope=scope), + None, + ) + from tools.mcp_oauth import remove_oauth_tokens + remove_oauth_tokens( + server_name, hermes_home=hermes_home, oauth_scope=scope + ) logger.info( "MCP OAuth '%s': evicted from cache and removed from disk", server_name, ) - return entry + return popped def restore_entry( self, @@ -737,22 +796,31 @@ def restore_entry( entry: _ProviderEntry | None, *, hermes_home: str | Path | None = None, + oauth_scope: Optional[McpOAuthScope] = None, ) -> None: """Restore a provider entry removed for a failed reauthorization.""" if entry is None: return + scope = oauth_scope if oauth_scope is not None else getattr(entry, "oauth_scope", None) + home = hermes_home if hermes_home is not None else getattr(entry, "hermes_home", None) with self._entries_lock: - self._entries.setdefault(self._key(server_name, hermes_home), entry) + self._entries.setdefault( + self._key(server_name, home, oauth_scope=scope), entry + ) def evict( self, server_name: str, *, hermes_home: str | Path | None = None, + oauth_scope: Optional[McpOAuthScope] = None, ) -> None: """Drop only the in-process provider, preserving persisted OAuth state.""" + scope = _resolved_oauth_scope(oauth_scope) with self._entries_lock: - self._entries.pop(self._key(server_name, hermes_home), None) + self._entries.pop( + self._key(server_name, hermes_home, oauth_scope=scope), None + ) # -- Disk watch ---------------------------------------------------------- @@ -761,6 +829,7 @@ async def invalidate_if_disk_changed( server_name: str, *, hermes_home: str | Path | None = None, + oauth_scope: Optional[McpOAuthScope] = 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. @@ -770,14 +839,22 @@ async def invalidate_if_disk_changed( 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 + from tools.mcp_oauth import HermesTokenStorage - entry = self._entries.get(self._key(server_name, hermes_home)) + scope = _resolved_oauth_scope(oauth_scope) + entry = self._entries.get( + self._key(server_name, hermes_home, oauth_scope=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" + storage = HermesTokenStorage( + server_name, + hermes_home=hermes_home if hermes_home is not None else entry.hermes_home, + oauth_scope=scope, + ) + tokens_path = storage._tokens_path() try: mtime_ns = tokens_path.stat().st_mtime_ns except (FileNotFoundError, OSError): @@ -805,6 +882,9 @@ async def handle_401( self, server_name: str, failed_access_token: Optional[str] = None, + *, + oauth_scope: Optional[McpOAuthScope] = None, + hermes_home: str | Path | None = None, ) -> bool: """Handle a 401 from a tool call, deduplicated across concurrent callers. @@ -817,11 +897,20 @@ async def handle_401( 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. + Others await the same future. Alice's 401 never shares Bob's pending + future because entries are keyed by OAuth scope. """ - entry = self._entries.get(self._key(server_name)) + captured_scope = oauth_scope + captured_home = hermes_home + entry = self._entries.get( + self._key(server_name, captured_home, oauth_scope=captured_scope) + ) if entry is None or entry.provider is None: return False + if captured_scope is None: + captured_scope = entry.oauth_scope + if captured_home is None: + captured_home = entry.hermes_home key = failed_access_token or "" loop = asyncio.get_running_loop() @@ -834,9 +923,12 @@ async def handle_401( async def _do_handle() -> None: try: - # Step 1: Did disk change? Picks up external refresh. + # Use the captured scope/home β€” never re-resolve the + # ambient requester (Bob must not refresh Alice). disk_changed = await self.invalidate_if_disk_changed( - server_name + server_name, + hermes_home=captured_home, + oauth_scope=captured_scope, ) if disk_changed: if not pending.done(): diff --git a/tools/mcp_schema_cache.py b/tools/mcp_schema_cache.py index 48bb5ae665ce8..f173480a4c8e0 100644 --- a/tools/mcp_schema_cache.py +++ b/tools/mcp_schema_cache.py @@ -64,7 +64,27 @@ def _save_all(data: Dict[str, Any]) -> None: atomic_json_write(_cache_path(), data, mode=0o600) -def get_cached_entry(server_name: str, fingerprint: str) -> Optional[dict]: +def _entry_is_fresh(entry: Any, fingerprint: str) -> bool: + """True when *entry* is a dict whose fingerprint matches and TTL holds.""" + if not isinstance(entry, dict): + return False + if entry.get("fingerprint") != fingerprint: + return False + ttl_ms = entry.get("ttl_ms") + written_at = entry.get("written_at") + if isinstance(ttl_ms, (int, float)) and isinstance(written_at, (int, float)): + if (time.time() - written_at) * 1000.0 >= float(ttl_ms): + return False + return True + + +def get_cached_entry( + server_name: str, + fingerprint: str, + *, + oauth_scope: Optional[Any] = None, + cache_scope: Optional[str] = None, +) -> 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 @@ -72,25 +92,64 @@ def get_cached_entry(server_name: str, fingerprint: str) -> Optional[dict]: 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``. + + In ``per_user`` OAuth mode, ``cacheScope=private`` (and missing hints) + are keyed by requester so a shared gateway cannot serve Alice's list + to Bob. Explicit ``cacheScope=public`` stays on the unscoped key. """ + from tools.mcp_oauth_identity import schema_cache_entry_key + + key = schema_cache_entry_key( + server_name, oauth_scope, cache_scope=cache_scope + ) with _cache_lock: - entry = _load_all().get(server_name) - if not isinstance(entry, dict): - return None - if entry.get("fingerprint") != fingerprint: - return None - ttl_ms = entry.get("ttl_ms") - written_at = entry.get("written_at") - if isinstance(ttl_ms, (int, float)) and isinstance(written_at, (int, float)): - if (time.time() - written_at) * 1000.0 >= float(ttl_ms): - return None - return entry + entry = _load_all().get(key) + return entry if _entry_is_fresh(entry, fingerprint) else None + + +def get_startup_cached_entry(server_name: str, fingerprint: str) -> Optional[dict]: + """Return a schema-cache entry suitable for process-start tool publication. + Tries the unscoped/public key first, then any requester-scoped entry + for this logical server whose fingerprint and TTL still match. + + This is schemas only β€” tool *names* are process-global in memory so the + model can see them after a gateway restart. It must never be used to + select OAuth credentials or a live connection. + """ + unscoped = get_cached_entry(server_name, fingerprint) + if unscoped is not None: + return unscoped -def has_cached_entry(server_name: str, fingerprint: str) -> bool: - return get_cached_entry(server_name, fingerprint) is not None + from tools.mcp_oauth_identity import is_registry_key_for_server + + with _cache_lock: + data = _load_all() + for key, entry in data.items(): + if not isinstance(key, str): + continue + if key == server_name: + continue + if not is_registry_key_for_server(key, server_name): + continue + if _entry_is_fresh(entry, fingerprint): + return entry + return None + + +def has_cached_entry( + server_name: str, + fingerprint: str, + *, + oauth_scope: Optional[Any] = None, + cache_scope: Optional[str] = None, +) -> bool: + return get_cached_entry( + server_name, + fingerprint, + oauth_scope=oauth_scope, + cache_scope=cache_scope, + ) is not None def write_cache_entry( @@ -101,6 +160,7 @@ def write_cache_entry( utility_tools: Optional[List[dict]] = None, ttl_ms: Optional[float] = None, cache_scope: Optional[str] = None, + oauth_scope: Optional[Any] = None, ) -> None: """Persist tool schemas after a successful live connect. @@ -108,6 +168,11 @@ def write_cache_entry( ``tools/list`` result (2026-07-28 servers). ``written_at`` anchors TTL expiry in :func:`get_cached_entry`. """ + from tools.mcp_oauth_identity import schema_cache_entry_key + + key = schema_cache_entry_key( + server_name, oauth_scope, cache_scope=cache_scope + ) entry = { "fingerprint": fingerprint, "tools": tools, @@ -126,17 +191,27 @@ def write_cache_entry( # 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(key) == entry: return - data[server_name] = entry + data[key] = entry _save_all(data) -def clear_cache_entry(server_name: str) -> None: +def clear_cache_entry( + server_name: str, + *, + oauth_scope: Optional[Any] = None, + cache_scope: Optional[str] = None, +) -> None: + from tools.mcp_oauth_identity import schema_cache_entry_key + + key = schema_cache_entry_key( + server_name, oauth_scope, cache_scope=cache_scope + ) with _cache_lock: data = _load_all() - if server_name in data: - del data[server_name] + if key in data: + del data[key] _save_all(data) diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index c1c80e5adb18a..7003acc9e5761 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -2397,6 +2397,7 @@ class MCPServerTask: "_inflight_tasks", "_reconnecting", "_suspect_reason", "_teardown_race", "_permanent_grace_used", "_stdio_child_pids", "_ever_connected", + "_oauth_scope", "_registry_key", "_hermes_home", ) def __init__(self, name: str): @@ -2462,6 +2463,11 @@ def __init__(self, name: str): # (#81995). self._stdio_child_pids: Set[int] = set() self._auth_type: str = "" + # Captured at connect so refresh/401/reconnect never re-resolve the + # ambient requester (Alice's 401 must not refresh Bob's token). + self._oauth_scope = None + self._registry_key: str = name + self._hermes_home: Optional[str] = None self._refresh_lock = asyncio.Lock() # MCP stdio sessions are a single JSON-RPC stream. Some servers emit # list_changed notifications during startup; if the notification @@ -2795,11 +2801,17 @@ async def _refresh_tools(self): mcp_prefixed_tool_name(self.name, tool.name) for tool in new_mcp_tools } + keep = _mcp_tool_names_held_elsewhere( + self.name, self, include_lazy=False + ) for tool_name in stale_tool_names: # Never let one server's refresh remove a colliding name that - # is currently owned by another server. + # is currently owned by another server, or a tool another + # principal's live connection still advertises. if registry.get_toolset_for_tool(tool_name) != toolset_name: continue + if tool_name in keep: + continue registry.deregister(tool_name) _forget_mcp_tool_server(tool_name) @@ -2818,6 +2830,8 @@ async def _refresh_tools(self): for tool_name in old_tool_names - registered_name_set: if registry.get_toolset_for_tool(tool_name) != toolset_name: continue + if tool_name in keep: + continue registry.deregister(tool_name) _forget_mcp_tool_server(tool_name) self._registered_tool_names = registered_names @@ -3346,7 +3360,7 @@ async def _run_stdio(self, config: dict): # Session is live again: clear any breaker state from a # prior outage so the first call after recovery isn't # gated on a stale consecutive-failure count (#16788). - _reset_server_error(self.name) + _reset_server_error(self._registry_key) # A completed handshake alone is NOT proof of health: a # flapping transport can handshake fine and drop moments # later, forever (#62212). The session must prove itself @@ -3616,6 +3630,8 @@ async def _run_http(self, config: dict): from tools.mcp_oauth_manager import get_manager _oauth_auth = get_manager().get_or_build_provider( self.name, url, config.get("oauth"), + oauth_scope=self._oauth_scope, + hermes_home=self._hermes_home, ) except Exception as exc: logger.warning("MCP OAuth setup failed for '%s': %s", self.name, exc) @@ -3718,7 +3734,7 @@ def _mcp_http_client_factory( # Session is live again: clear any breaker state from a # prior outage so the first call after recovery isn't # gated on a stale consecutive-failure count (#16788). - _reset_server_error(self.name) + _reset_server_error(self._registry_key) # Unproven until keepalive/tool-call success (#62212). self._session_proven = False reason = await self._wait_for_lifecycle_event() @@ -3784,7 +3800,7 @@ def _mcp_http_client_factory( # Session is live again: clear any breaker state from # a prior outage so the first call after recovery # isn't gated on a stale failure count (#16788). - _reset_server_error(self.name) + _reset_server_error(self._registry_key) # Unproven until keepalive/tool-call success (#62212). self._session_proven = False reason = await self._wait_for_lifecycle_event() @@ -3832,7 +3848,7 @@ def _mcp_http_client_factory( # Session is live again: clear any breaker state from a # prior outage so the first call after recovery isn't # gated on a stale consecutive-failure count (#16788). - _reset_server_error(self.name) + _reset_server_error(self._registry_key) # Unproven until keepalive/tool-call success (#62212). self._session_proven = False reason = await self._wait_for_lifecycle_event() @@ -3897,7 +3913,7 @@ def _register_discovered_tools_if_needed(self) -> None: return if not self._ready.is_set(): with _lock: - if _servers.get(self.name) is not self: + if _servers.get(self._registry_key) is not self: return self._registered_tool_names = _register_server_tools( self.name, self, self._config @@ -3906,8 +3922,29 @@ def _register_discovered_tools_if_needed(self) -> None: # recovered: drop its stale connect error so status surfaces stop # reporting it as failed. with _lock: - if _servers.get(self.name) is self: - _server_connect_errors.pop(self.name, None) + if _servers.get(self._registry_key) is self: + _server_connect_errors.pop(self._registry_key, None) + + def _capture_oauth_identity(self, config: dict) -> None: + """Pin OAuth scope, registry key, and HERMES_HOME for this connection. + + Identity is taken from the bound session principal only. Tool + arguments never participate. Non-OAuth servers stay on the shared + (bare-name) registry key even in ``per_user`` mode. + """ + from hermes_constants import get_hermes_home + from tools.mcp_oauth_identity import ( + connection_registry_token, + resolve_mcp_oauth_scope, + server_uses_oauth, + ) + + self._hermes_home = str(get_hermes_home()) + uses_oauth = self._auth_type == "oauth" or server_uses_oauth(config) + if uses_oauth: + _oauth_protected_servers.add(self.name) + self._oauth_scope = resolve_mcp_oauth_scope(uses_oauth=uses_oauth) + self._registry_key = connection_registry_token(self.name, self._oauth_scope) async def run(self, config: dict): """Long-lived coroutine: connect, discover tools, wait, disconnect. @@ -3918,6 +3955,11 @@ async def run(self, config: dict): self._config = config self.tool_timeout = _resolve_tool_timeout(config) self._auth_type = (config.get("auth") or "").lower().strip() + # Prefer the identity pinned in start() on the wrapped MCP-loop + # task. Re-resolve only when run() is entered without start() + # (tests). Never overwrite a captured scope on reconnect. + if self._oauth_scope is None: + self._capture_oauth_identity(config) self._idle_timeout_seconds = _get_lifecycle_seconds(config, "idle_timeout_seconds") self._max_lifetime_seconds = _get_lifecycle_seconds(config, "max_lifetime_seconds") @@ -4371,6 +4413,13 @@ async def run(self, config: dict): async def start(self, config: dict): """Create the background Task and wait until ready (or failed).""" + # Pin OAuth identity on this (wrapped) task before ensure_future. + # run() copies this task's context, but the connection object must + # own the scope even if a later hop drops ContextVars. + self._config = config + self._auth_type = (config.get("auth") or "").lower().strip() + if self._oauth_scope is None: + self._capture_oauth_identity(config) self._task = asyncio.ensure_future(self.run(config)) try: await self._ready.wait() @@ -4426,10 +4475,20 @@ def _deregister_tools(self) -> None: reconnect budget is exhausted, so a dead server never leaves phantom tool definitions bloating the prompt cache and producing "not connected" errors on every turn. + + Names still served by another live connection for this logical + server, or by the cache-backed lazy template, stay registered. """ from tools.registry import registry - for tool_name in list(getattr(self, "_registered_tool_names", [])): + mine = list(getattr(self, "_registered_tool_names", []) or []) + if not mine: + return + + keep = _mcp_tool_names_held_elsewhere(self.name, self, include_lazy=True) + for tool_name in mine: + if tool_name in keep: + continue registry.deregister(tool_name) _forget_mcp_tool_server(tool_name) self._registered_tool_names = [] @@ -4460,9 +4519,15 @@ async def _wait_for_lazy_reconnect(self) -> None: _servers: Dict[str, MCPServerTask] = {} _server_connecting: set[str] = set() _server_connect_errors: Dict[str, str] = {} +# Server names (bare) whose config is ``auth: oauth``. Used to decide +# whether a live-registry lookup must be requester-scoped before a +# connection object exists. Not a credential selector. +_oauth_protected_servers: set[str] = set() # Lazy MCP startup (#56832): servers whose tools were registered from the -# on-disk schema cache without spawning/connecting. Keyed by server name; -# entries are popped once a real connection is established on first use. +# on-disk schema cache without spawning/connecting. Keyed by bare server +# name (shared config templates). In ``per_user`` mode these templates are +# NOT popped on first connect β€” Alice must not delete the config Bob still +# needs. Live connections in ``_servers`` are keyed by registry token. _lazy_server_configs: Dict[str, dict] = {} _lazy_server_fingerprints: Dict[str, str] = {} _lazy_server_tool_names: Dict[str, List[str]] = {} @@ -4528,6 +4593,183 @@ def _connect_cooldown_active(server_name: str) -> bool: deadline = _server_connect_retry_after.get(server_name) return deadline is not None and time.monotonic() < deadline + +def _mcp_server_uses_oauth(server_name: str, *, server=None, config=None) -> bool: + """Whether this logical server is OAuth-protected. + + Uses captured task state, the supplied config, the lazy-config stash, + or the process-level oauth-protected set. Never reads tool arguments. + """ + from tools.mcp_oauth_identity import server_uses_oauth + + if server is not None and getattr(server, "_auth_type", "") == "oauth": + return True + if config is not None and server_uses_oauth(config): + return True + if server_name in _oauth_protected_servers: + return True + lazy = _lazy_server_configs.get(server_name) + return bool(lazy is not None and server_uses_oauth(lazy)) + + +def _oauth_call_target( + server_name: str, + *, + config=None, +) -> Tuple[Optional[str], Optional[str]]: + """Return ``(registry_key, error_json)``. Error is set iff the key is missing. + + Per-user OAuth without a bound principal fails closed. Shared and + non-OAuth always resolve to a key (the bare server name). + """ + from tools.mcp_oauth_identity import ( + MissingRequesterIdentity, + connection_registry_token, + resolve_mcp_oauth_scope, + ) + + try: + uses = _mcp_server_uses_oauth(server_name, config=config) + scope = resolve_mcp_oauth_scope(uses_oauth=uses) + except MissingRequesterIdentity: + return None, tool_error( + "MCP OAuth requires an authenticated requester identity in " + "per_user mode. Direct CLI, TUI, desktop, and cron paths " + "without a bound gateway principal cannot use a shared " + f"credential as a fallback. Server '{server_name}' was not called." + ) + return connection_registry_token(server_name, scope), None + + +def _task_registry_key(server) -> str: + return getattr(server, "_registry_key", None) or getattr(server, "name", "") + + +def _any_live_named(server_name: str) -> bool: + """True if any live connection has this logical server name. + + Used for tool-name aggregation and status, not credential selection. + """ + from tools.mcp_oauth_identity import is_registry_key_for_server + + return any( + is_registry_key_for_server(k, server_name) + or getattr(srv, "name", None) == server_name + for k, srv in _servers.items() + ) + + +def _find_named_in_map(mapping, server_name: str): + """Prefer the current requester's entry; else any key for this server. + + Credential-bearing call paths must use ``_oauth_call_target`` instead. + This helper is for status/banner aggregation only. + """ + from tools.mcp_oauth_identity import is_registry_key_for_server + + if not isinstance(mapping, dict): + return None + rk, _err = _oauth_call_target(server_name) + if rk is not None and rk in mapping: + return mapping[rk] + for key, value in mapping.items(): + if is_registry_key_for_server(key, server_name): + return value + if getattr(value, "name", None) == server_name: + return value + return None + + +def _name_in_keyed_set(container, server_name: str) -> bool: + from tools.mcp_oauth_identity import is_registry_key_for_server + + return any(is_registry_key_for_server(k, server_name) for k in container) + + +def _maybe_pop_lazy(server_name: str): + """Pop lazy templates only in shared mode. Returns (fingerprint, names).""" + from tools.mcp_oauth_identity import IDENTITY_MODE_PER_USER, configured_identity_mode + + if configured_identity_mode() == IDENTITY_MODE_PER_USER: + return ( + _lazy_server_fingerprints.get(server_name), + list(_lazy_server_tool_names.get(server_name) or []), + ) + fingerprint = _lazy_server_fingerprints.pop(server_name, None) + names = _lazy_server_tool_names.pop(server_name, None) or [] + _lazy_server_configs.pop(server_name, None) + return fingerprint, names + + +def _mcp_tool_names_held_elsewhere( + logical_name: str, + holder, + *, + include_lazy: bool, +) -> set: + """Registry names still served by a peer connection (and optionally cache). + + ``include_lazy`` is for connection teardown: cache-backed names must stay + registered so first-use can reconnect. Live ``tools/list_changed`` refresh + must NOT keep a name solely because the cache still lists it β€” the live + sibling's ``_registered_tool_names`` is the ownership signal. + """ + from tools.mcp_oauth_identity import is_registry_key_for_server + + keep: set = set() + if include_lazy: + keep.update(_lazy_server_tool_names.get(logical_name) or []) + for key, other in list(_servers.items()): + if other is holder: + continue + if not ( + is_registry_key_for_server(key, logical_name) + or getattr(other, "name", None) == logical_name + ): + continue + keep.update(getattr(other, "_registered_tool_names", None) or []) + return keep + + +def _logical_mcp_server_name(key: str, server) -> str: + return getattr(server, "name", None) or str(key).split("\x1f", 1)[0] + + +def _reload_shutdown_keys( + configured_names: set, + servers: dict, + requester_scope, +) -> Optional[set]: + """Return registry keys a /reload-mcp pass may close, or None for a full wipe. + + ``None`` means every live connection plus lazy templates (shared mode, + unbound CLI/TUI, process exit). A set means close only those keys: + this requester's per-user OAuth connections, process-level (non-OAuth) + connections, and every identity of a server no longer in config. + Other principals' OAuth sessions are omitted. + """ + from tools.mcp_oauth_identity import ( + IDENTITY_MODE_PER_USER, + connection_registry_token, + ) + + if requester_scope is None or getattr(requester_scope, "mode", None) != IDENTITY_MODE_PER_USER: + return None + recycle: set = set() + for key, srv in servers.items(): + logical = _logical_mcp_server_name(key, srv) + if logical not in configured_names: + recycle.add(key) + continue + scope = getattr(srv, "_oauth_scope", None) + if scope is None or getattr(scope, "mode", None) != IDENTITY_MODE_PER_USER: + recycle.add(key) + continue + expected = connection_registry_token(logical, requester_scope) + if key == expected: + recycle.add(key) + return recycle + # Circuit breaker: consecutive error counts per server. After # _CIRCUIT_BREAKER_THRESHOLD consecutive failures, the handler returns # a "server unreachable" message that tells the model to stop retrying, @@ -4747,9 +4989,16 @@ def _signal_reconnect(server: Any) -> bool: def reconnect_mcp_server(server_name: str) -> bool: - """Ask a currently-live MCP server to rebuild after external re-auth.""" + """Ask a currently-live MCP server to rebuild after external re-auth. + + Looks up the current requester's connection only. There is no + "any connection named X" fallback. + """ + rk, _err = _oauth_call_target(server_name) + if rk is None: + return False with _lock: - server = _servers.get(server_name) + server = _servers.get(rk) if server is None: return False return _signal_reconnect(server) @@ -4967,8 +5216,27 @@ def _handle_auth_error_and_retry( from tools.mcp_oauth_manager import get_manager manager = get_manager() + rk, identity_error = _oauth_call_target(server_name) + if rk is None: + return identity_error + with _lock: + srv = _servers.get(rk) + if srv is None: + # Exact-key miss: do not recover against ambient/shared credentials. + return None + captured_scope = getattr(srv, "_oauth_scope", None) + if captured_scope is None: + from tools.mcp_oauth_identity import SHARED_SCOPE + captured_scope = SHARED_SCOPE + captured_home = getattr(srv, "_hermes_home", None) + async def _recover(): - return await manager.handle_401(server_name, None) + return await manager.handle_401( + server_name, + None, + oauth_scope=captured_scope, + hermes_home=captured_home, + ) try: recovered = _run_on_mcp_loop(_recover, timeout=10) @@ -4981,7 +5249,7 @@ async def _recover(): if recovered: with _lock: - srv = _servers.get(server_name) + srv = _servers.get(rk) reconnected = False if srv is not None and hasattr(srv, "_reconnect_event"): reconnected = _signal_reconnect_and_wait( @@ -4999,17 +5267,17 @@ async def _recover(): # _bump_server_error on failure, so a genuinely broken server will # re-trip the breaker as normal. if reconnected: - _reset_server_error(server_name) + _reset_server_error(rk) try: result = retry_call() try: parsed = json.loads(result) if "error" not in parsed: - _reset_server_error(server_name) + _reset_server_error(rk) return result except (json.JSONDecodeError, TypeError): - _reset_server_error(server_name) + _reset_server_error(rk) return result except Exception as retry_exc: logger.warning( @@ -5020,7 +5288,7 @@ async def _recover(): # No recovery available, or retry also failed: surface a structured # needs_reauth error. Bumps the circuit breaker so the model stops # retrying the tool. - _bump_server_error(server_name) + _bump_server_error(rk) return tool_error( f"MCP server '{server_name}' requires re-authentication. " f"Run `hermes mcp login {server_name}` (or delete the tokens " @@ -5165,8 +5433,11 @@ def _handle_session_expired_and_retry( if not _is_session_expired_error(exc): return None + rk, _err = _oauth_call_target(server_name) + if rk is None: + return None with _lock: - srv = _servers.get(server_name) + srv = _servers.get(rk) if srv is None or not hasattr(srv, "_reconnect_event"): return None @@ -5200,10 +5471,10 @@ def _handle_session_expired_and_retry( try: parsed = json.loads(result) if "error" not in parsed: - _reset_server_error(server_name) + _reset_server_error(rk) return result except (json.JSONDecodeError, TypeError): - _reset_server_error(server_name) + _reset_server_error(rk) return result except Exception as retry_exc: logger.warning( @@ -5519,6 +5790,34 @@ async def _scoped(): return _scoped() +def _wrap_with_session_principal(coro: "Coroutine") -> "Coroutine": + """Carry the caller's bound requester identity onto the MCP loop. + + Same hop as :func:`_wrap_with_home_override`: tasks scheduled via + ``run_coroutine_threadsafe`` copy the loop thread's ContextVars, not + the agent thread's. OAuth ``per_user`` capture in ``MCPServerTask`` + would fail closed (or inherit a stale loop-thread principal) without + this wrap. No-op when no principal is bound. + """ + try: + from gateway.session_context import ( + apply_bound_session_principal, + get_bound_session_principal, + ) + + principal = get_bound_session_principal() + except Exception: + return coro + if principal is None: + return coro + + async def _scoped(): + with apply_bound_session_principal(principal): + return await coro + + return _scoped() + + def _wrap_with_dashboard_oauth_flow(coro): """Propagate a dashboard OAuth flow onto the dedicated MCP loop task.""" try: @@ -5563,18 +5862,18 @@ def _run_on_mcp_loop(coro_or_factory, timeout: float = 30): coro = coro_or_factory() if callable(coro_or_factory) else coro_or_factory - # Propagate the context-local HERMES_HOME override onto the MCP loop. - # Tasks scheduled via run_coroutine_threadsafe are created INSIDE the - # loop thread, so they copy the loop thread's context β€” not the - # scheduling thread's. A per-request profile scope (the dashboard's - # ?profile= endpoints, e.g. the MCP "Test server" probe) would silently - # vanish here: OAuth token stores and any other get_hermes_home() - # resolution inside the coroutine would read the process home instead - # of the selected profile's. Re-establish the override inside the - # task's own context (task-local β€” concurrent calls carrying different - # scopes don't interfere). No-op when no override is active. + # Propagate the context-local HERMES_HOME override AND the bound + # session principal onto the MCP loop. Tasks scheduled via + # run_coroutine_threadsafe are created INSIDE the loop thread, so they + # copy the loop thread's context β€” not the scheduling thread's. A + # per-request profile scope (dashboard ?profile=) or a gateway + # requester identity would silently vanish here: OAuth token stores + # would read the process home, and per_user capture would fail closed + # (or inherit a stale loop-thread principal). Wrappers are task-local + # so concurrent calls carrying different scopes don't interfere. coro = _wrap_with_home_override(coro) coro = _wrap_with_dashboard_oauth_flow(coro) + coro = _wrap_with_session_principal(coro) future = safe_schedule_threadsafe( coro, loop, @@ -5897,18 +6196,21 @@ def _ensure_lazy_server_connected(server_name: str) -> bool: session is available afterwards. """ with _lock: - server = _servers.get(server_name) + rk, _err = _oauth_call_target(server_name) + if rk is None: + return False + server = _servers.get(rk) if server is not None and server.session is not None: return True config = _lazy_server_configs.get(server_name) if not config: return False - if _connect_cooldown_active(server_name): + if _connect_cooldown_active(rk): return False - if server_name in _server_connecting: + if rk in _server_connecting: return False - _server_connecting.add(server_name) - _server_connect_errors.pop(server_name, None) + _server_connecting.add(rk) + _server_connect_errors.pop(rk, None) logger.info("MCP server '%s': lazy start on first use", server_name) _ensure_mcp_loop() @@ -5922,21 +6224,19 @@ async def _connect(): except BaseException as exc: message = _format_connect_error(exc) with _lock: - _server_connecting.discard(server_name) - _server_connect_errors[server_name] = message - _record_connect_failure(server_name) + _server_connecting.discard(rk) + _server_connect_errors[rk] = message + _record_connect_failure(rk) logger.warning( "Lazy MCP connect failed for '%s': %s", server_name, message, ) return False with _lock: - _server_connecting.discard(server_name) - _clear_connect_failure(server_name) - _lazy_server_configs.pop(server_name, None) - stale_fingerprint = _lazy_server_fingerprints.pop(server_name, None) - cached_names = _lazy_server_tool_names.pop(server_name, None) or [] - server = _servers.get(server_name) + _server_connecting.discard(rk) + _clear_connect_failure(rk) + stale_fingerprint, cached_names = _maybe_pop_lazy(server_name) + server = _servers.get(rk) live_names = set( getattr(server, "_registered_tool_names", []) or [] ) @@ -5947,9 +6247,18 @@ async def _connect(): if phantom_names: from tools.registry import registry + keep = _mcp_tool_names_held_elsewhere( + server_name, server, include_lazy=False + ) + dropped = [] for tool_name in phantom_names: + if tool_name in keep: + continue registry.deregister(tool_name) _forget_mcp_tool_server(tool_name) + dropped.append(tool_name) + phantom_names = dropped + if phantom_names: logger.info( "MCP server '%s': deregistered %d phantom cached tool(s) not " "served live (stale schema-cache fingerprint %s): %s", @@ -5959,25 +6268,42 @@ async def _connect(): return server is not None and server.session is not None -def _get_connected_server_for_call(server_name: str) -> Optional[MCPServerTask]: +def _get_connected_server_for_call( + server_name: str, + registry_key: Optional[str] = None, +) -> Optional[MCPServerTask]: """Return a connected server, lazily reconnecting recycled stdio state. Also the single first-use connect point for lazy (schema-cache registered) servers, so raw tool calls AND the resource/prompt utility handlers all trigger the deferred spawn (#56832). + + Pass ``registry_key`` when the caller already resolved the fail-closed + target so this lookup does not re-read identity config. """ with _lock: - server = _servers.get(server_name) - is_lazy = server_name in _lazy_server_configs + rk = registry_key + if rk is None: + rk, _err = _oauth_call_target(server_name) + if rk is None: + # per_user OAuth without a bound principal: do not pick any + # live connection. The handler surfaces a fail-closed error. + is_lazy = server_name in _lazy_server_configs + server = None + else: + server = _servers.get(rk) + is_lazy = server_name in _lazy_server_configs if is_lazy and (server is None or server.session is None): + if rk is None: + return None _ensure_lazy_server_connected(server_name) with _lock: - server = _servers.get(server_name) + server = _servers.get(rk) return server if server is not None and server.session is None and server._is_recycled_stdio(): _request_lazy_reconnect(server_name, server) with _lock: - server = _servers.get(server_name) + server = _servers.get(rk) return server @@ -6062,6 +6388,10 @@ def _handler(args: dict, **kwargs) -> str: if gate_error is not None: return gate_error + rk, identity_error = _oauth_call_target(server_name) + if identity_error is not None: + return identity_error + # Circuit breaker: if this server has failed too many times # consecutively, short-circuit with a clear message so the model # stops retrying and uses alternative approaches (#10447). @@ -6072,23 +6402,23 @@ def _handler(args: dict, **kwargs) -> str: # failure the error paths below bump the count again, which # re-stamps the open-time via _bump_server_error (re-arming # the cooldown). - if _server_error_counts.get(server_name, 0) >= _CIRCUIT_BREAKER_THRESHOLD: - opened_at = _server_breaker_opened_at.get(server_name, 0.0) + if _server_error_counts.get(rk, 0) >= _CIRCUIT_BREAKER_THRESHOLD: + opened_at = _server_breaker_opened_at.get(rk, 0.0) age = time.monotonic() - opened_at if age < _CIRCUIT_BREAKER_COOLDOWN_SEC: remaining = max(1, int(_CIRCUIT_BREAKER_COOLDOWN_SEC - age)) return tool_error( f"MCP server '{server_name}' is unreachable after " - f"{_server_error_counts[server_name]} consecutive " + f"{_server_error_counts[rk]} consecutive " f"failures. Auto-retry available in ~{remaining}s. " f"Do NOT retry this tool yet β€” use alternative " f"approaches or ask the user to check the MCP server." ) # Cooldown elapsed β†’ fall through as a half-open probe. - server = _get_connected_server_for_call(server_name) + server = _get_connected_server_for_call(server_name, rk) if not server: - _bump_server_error(server_name) + _bump_server_error(rk) return tool_error(f"MCP server '{server_name}' is not connected") if not server.session: @@ -6112,7 +6442,7 @@ def _handler(args: dict, **kwargs) -> str: # without burning iterations. The breaker resets once the # fresh session initializes (_run_stdio/_run_http call # _reset_server_error). - _bump_server_error(server_name) + _bump_server_error(rk) if _signal_reconnect(server): return tool_error( f"MCP server '{server_name}' transport is down; " @@ -6335,11 +6665,11 @@ def _call_once(): try: parsed = json.loads(result) if "error" in parsed: - _bump_server_error(server_name) + _bump_server_error(rk) else: - _reset_server_error(server_name) # success β€” reset + _reset_server_error(rk) # success β€” reset except (json.JSONDecodeError, TypeError): - _reset_server_error(server_name) # non-JSON = success + _reset_server_error(rk) # non-JSON = success return result except InterruptedError: return _interrupted_call_result() @@ -6364,7 +6694,7 @@ def _call_once(): if recovered is not None: return recovered - _bump_server_error(server_name) + _bump_server_error(rk) logger.error( "MCP tool %s/%s call failed: %s", server_name, tool_name, exc, @@ -6380,7 +6710,10 @@ def _make_list_resources_handler(server_name: str, tool_timeout: float): """Return a sync handler that lists resources from an MCP server.""" def _handler(args: dict, **kwargs) -> str: - server = _get_connected_server_for_call(server_name) + rk, identity_error = _oauth_call_target(server_name) + if identity_error is not None: + return identity_error + server = _get_connected_server_for_call(server_name, rk) if not server or not server.session: return tool_error(f"MCP server '{server_name}' is not connected") @@ -6439,7 +6772,10 @@ def _make_read_resource_handler(server_name: str, tool_timeout: float): """Return a sync handler that reads a resource by URI from an MCP server.""" def _handler(args: dict, **kwargs) -> str: - server = _get_connected_server_for_call(server_name) + rk, identity_error = _oauth_call_target(server_name) + if identity_error is not None: + return identity_error + server = _get_connected_server_for_call(server_name, rk) if not server or not server.session: return tool_error(f"MCP server '{server_name}' is not connected") @@ -6500,7 +6836,10 @@ def _make_list_prompts_handler(server_name: str, tool_timeout: float): """Return a sync handler that lists prompts from an MCP server.""" def _handler(args: dict, **kwargs) -> str: - server = _get_connected_server_for_call(server_name) + rk, identity_error = _oauth_call_target(server_name) + if identity_error is not None: + return identity_error + server = _get_connected_server_for_call(server_name, rk) if not server or not server.session: return tool_error(f"MCP server '{server_name}' is not connected") @@ -6561,7 +6900,10 @@ def _make_get_prompt_handler(server_name: str, tool_timeout: float): """Return a sync handler that gets a prompt by name from an MCP server.""" def _handler(args: dict, **kwargs) -> str: - server = _get_connected_server_for_call(server_name) + rk, identity_error = _oauth_call_target(server_name) + if identity_error is not None: + return identity_error + server = _get_connected_server_for_call(server_name, rk) if not server or not server.session: return tool_error(f"MCP server '{server_name}' is not connected") @@ -6627,13 +6969,17 @@ def _make_check_fn(server_name: str): def _check() -> bool: with _lock: - server = _servers.get(server_name) - if server is not None and ( - server.session is not None or server._is_recycled_stdio() - ): - return True + rk, _err = _oauth_call_target(server_name) + if rk is not None: + server = _servers.get(rk) + if server is not None and ( + server.session is not None or server._is_recycled_stdio() + ): + return True # Lazy (schema-cache registered) servers are available: the - # first real call spawns/connects them (#56832). + # first real call spawns/connects them (#56832). Per-user + # OAuth without a bound principal still reports available so + # the model can see the tool; the call itself fail-closes. return server_name in _lazy_server_configs return _check @@ -7107,7 +7453,7 @@ def _existing_tool_names() -> List[str]: lazy_names = [ n for sname, tool_names in _lazy_server_tool_names.items() - if sname not in _servers + if not _any_live_named(sname) for n in tool_names ] names.extend(lazy_names) @@ -7373,6 +7719,7 @@ def _should_register(tool_name: str) -> bool: utility_tools=utility_payload, ttl_ms=(getattr(server, "_list_cache_meta", None) or {}).get("ttl_ms"), cache_scope=(getattr(server, "_list_cache_meta", None) or {}).get("cache_scope"), + oauth_scope=getattr(server, "_oauth_scope", None), ) except Exception as exc: logger.debug("MCP schema cache write failed for '%s': %s", name, exc) @@ -7567,17 +7914,20 @@ def _claim_server(created: MCPServerTask) -> None: # Recoverable park: the run task deliberately stays alive to # self-probe, so adopt it into the registry for shutdown/revival. with _lock: - _servers[name] = server + _servers[_task_registry_key(server)] = server elif server is not None: await server.shutdown() raise finally: _connect_server_claim.reset(claim_token) + live_key = _task_registry_key(server) with _lock: + _server_connecting.discard(live_key) _server_connecting.discard(name) + _server_connect_errors.pop(live_key, None) _server_connect_errors.pop(name, None) - _servers[name] = server + _servers[live_key] = server registered_names = _register_server_tools(name, server, config) server._registered_tool_names = list(registered_names) @@ -7591,6 +7941,60 @@ def _claim_server(created: MCPServerTask) -> None: return registered_names +def _stash_per_user_oauth_without_principal(deferred: Dict[str, dict]) -> tuple: + """Register cached tool names for OAuth servers lacking a principal. + + ``deferred`` is already the fail-closed set (per_user OAuth, no bound + requester). Does not re-resolve identity. Returns + ``(lazy_tool_count, lazy_server_count)``. + """ + if not deferred: + return 0, 0 + + try: + from tools.mcp_schema_cache import ( + config_fingerprint, + get_startup_cached_entry, + ) + except Exception: + config_fingerprint = None # type: ignore[assignment] + get_startup_cached_entry = None # type: ignore[assignment] + + tools = 0 + count = 0 + for name, cfg in deferred.items(): + already_lazy = name in _lazy_server_configs + with _lock: + # Refresh the template so a later /reload-mcp auth-mode change + # is not stuck on the original oauth/header setting. + _lazy_server_configs[name] = dict(cfg) + if ( + config_fingerprint is not None + and get_startup_cached_entry is not None + and name not in _lazy_server_tool_names + ): + try: + entry = get_startup_cached_entry(name, config_fingerprint(cfg)) + except Exception: + entry = None + if entry: + try: + names = _register_from_cache_sync(name, cfg, entry) + tools += len(names) + except Exception as exc: + logger.warning( + "Failed lazy MCP registration for '%s': %s", name, exc, + ) + if not already_lazy: + count += 1 + logger.info( + "MCP server '%s': deferred OAuth connect until a bound " + "requester is present (per_user mode)", + name, + ) + return tools, count + + # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- @@ -7616,42 +8020,61 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: logger.debug("No explicit MCP servers provided") return [] + from tools.mcp_oauth_identity import server_uses_oauth + # Only attempt servers that aren't already connected (or currently # connecting) and are enabled. Checking ``_server_connecting`` prevents # duplicate subprocess spawns when ``discover_mcp_tools()`` is called # from multiple entry-points before the first batch finishes (#58862). + connecting_by_name: Dict[str, str] = {} + deferred_oauth: Dict[str, dict] = {} with _lock: connecting = set(_server_connecting) - new_servers = { - k: v - for k, v in servers.items() - if k not in _servers - and k not in connecting - # Servers already lazily registered from the schema cache are - # not re-registered; they connect on first tool use (#56832). - and k not in _lazy_server_configs - and _parse_boolish(v.get("enabled", True), default=True) - # Skip a server still serving its post-failure backoff. Without - # this, a server that fails to connect (and is therefore never - # recorded in ``_servers``) would be re-spawned on every worker - # session's discovery pass -- the #50394 restart storm. The - # cooldown is cleared automatically on the next successful - # connect or by a manual /mcp refresh. - and not _connect_cooldown_active(k) - } + for srv_name, srv_cfg in servers.items(): + if server_uses_oauth(srv_cfg): + _oauth_protected_servers.add(srv_name) + else: + # Auth-mode changes in this batch must not keep a stale + # oauth classification from a previous register/reload. + _oauth_protected_servers.discard(srv_name) + new_servers: Dict[str, dict] = {} + resolved_keys: Dict[str, str] = {} + for k, v in servers.items(): + if not _parse_boolish(v.get("enabled", True), default=True): + continue + rk, _err = _oauth_call_target(k, config=v) + if rk is None: + # per_user OAuth without a bound principal: never eager-connect + # with a shared token. Stashed as lazy below. + deferred_oauth[k] = v + continue + resolved_keys[k] = rk + if k in _lazy_server_configs: + _lazy_server_configs[k] = dict(v) + if k in _lazy_server_tool_names: + continue + # Bound requester, tools never published: fall through so + # this pass can load *their* schema cache (or connect). + if rk in _servers or rk in connecting: + continue + if _connect_cooldown_active(rk): + continue + new_servers[k] = v + connecting_by_name[k] = rk # Cached entries with no live session are parked or mid-reconnect. # Their tools are deregistered, so nothing else can reach # _signal_reconnect β€” without this nudge a new session silently # waits up to _PARKED_RETRY_INTERVAL for the next self-probe # (#50170). Wake them now so their tools come back promptly. - stale_cached = [ - _servers[k] - for k in servers - if k in _servers and getattr(_servers[k], "session", None) is None - ] - _server_connecting.update(new_servers) - for srv_name in new_servers: - _server_connect_errors.pop(srv_name, None) + # Exact requester key only β€” do not wake Bob's connection for Alice. + stale_cached = [] + for k, rk in resolved_keys.items(): + srv = _servers.get(rk) + if srv is not None and getattr(srv, "session", None) is None: + stale_cached.append(srv) + _server_connecting.update(connecting_by_name.values()) + for srv_name, rk in connecting_by_name.items(): + _server_connect_errors.pop(rk, None) # Track which servers opt-in to parallel tool calls (idempotent). for srv_name, srv_cfg in servers.items(): if _parse_boolish(srv_cfg.get("supports_parallel_tool_calls", False), default=False): @@ -7662,7 +8085,21 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: for srv in stale_cached: _signal_reconnect(srv) + lazy_registered = 0 + lazy_server_count = 0 + oauth_lazy_tools, oauth_lazy_servers = _stash_per_user_oauth_without_principal( + deferred_oauth + ) + lazy_registered += oauth_lazy_tools + lazy_server_count += oauth_lazy_servers + if not new_servers: + if lazy_registered: + logger.info( + "MCP: registered %d lazy tool(s) from schema cache " + "(no processes spawned)", + lazy_registered, + ) return _existing_tool_names() # Lazy startup (#56832): servers gated with ``lazy: true`` whose config @@ -7671,22 +8108,39 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: # entry falls back to the normal eager connect below (which write-through # refreshes the cache for next time). eager_servers: Dict[str, dict] = dict(new_servers) - lazy_registered = 0 - lazy_server_count = 0 try: - from tools.mcp_schema_cache import config_fingerprint, get_cached_entry + from tools.mcp_schema_cache import ( + config_fingerprint, + get_startup_cached_entry, + ) + from tools.mcp_oauth_identity import ( + IDENTITY_MODE_PER_USER, + configured_identity_mode, + ) except Exception: # pragma: no cover - cache module missing config_fingerprint = None # type: ignore[assignment] - get_cached_entry = None # type: ignore[assignment] - if config_fingerprint is not None and get_cached_entry is not None: + get_startup_cached_entry = None # type: ignore[assignment] + IDENTITY_MODE_PER_USER = None # type: ignore[assignment] + configured_identity_mode = None # type: ignore[assignment] + if config_fingerprint is not None and get_startup_cached_entry is not None: + per_user_oauth = ( + configured_identity_mode is not None + and configured_identity_mode() == IDENTITY_MODE_PER_USER + ) for name, cfg in new_servers.items(): - if not _resolve_server_lazy(name, cfg): + is_lazy = _resolve_server_lazy(name, cfg) + is_oauth = _mcp_server_uses_oauth(name, config=cfg) + # Shared-mode oauth still eager-connects unless marked lazy. + # per_user oauth is implicitly lazy for *schema* publication so + # a bound /reload-mcp can republish names from a scoped cache + # without spawning. + if not is_lazy and not (per_user_oauth and is_oauth): continue - entry = get_cached_entry(name, config_fingerprint(cfg)) + entry = get_startup_cached_entry(name, config_fingerprint(cfg)) if not entry: continue with _lock: - _server_connecting.discard(name) + _server_connecting.discard(connecting_by_name.get(name, name)) try: names = _register_from_cache_sync(name, cfg, entry) except Exception as exc: @@ -7694,7 +8148,7 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: "Failed lazy MCP registration for '%s': %s", name, exc, ) with _lock: - _server_connecting.add(name) + _server_connecting.add(connecting_by_name.get(name, name)) continue eager_servers.pop(name, None) lazy_registered += len(names) @@ -7729,13 +8183,14 @@ async def _discover_all(): command = new_servers.get(name, {}).get("command") message = _format_connect_error(result) with _lock: - _server_connecting.discard(name) - _server_connect_errors[name] = message + rk = connecting_by_name.get(name, name) + _server_connecting.discard(rk) + _server_connect_errors[rk] = message # Arm the per-server backoff so the next discovery pass # doesn't immediately re-spawn this failing server # (#50394). Isolated to this server -- healthy servers # in the same batch are unaffected. - _record_connect_failure(name) + _record_connect_failure(rk) logger.warning( "Failed to connect to MCP server '%s'%s: %s", name, @@ -7744,9 +8199,10 @@ async def _discover_all(): ) else: with _lock: - _server_connecting.discard(name) - _server_connect_errors.pop(name, None) - _clear_connect_failure(name) + rk = connecting_by_name.get(name, name) + _server_connecting.discard(rk) + _server_connect_errors.pop(rk, None) + _clear_connect_failure(rk) # Per-server timeouts are handled inside _discover_and_register_server. # The outer timeout is generous: 120s total for parallel discovery. @@ -7766,19 +8222,23 @@ async def _discover_all(): # entries stranded in _server_connecting. Those stale # entries would block future reconnection attempts (#58862). with _lock: - stale = [n for n in new_servers if n in _server_connecting] - if stale: + stale_keys = [ + connecting_by_name.get(n, n) + for n in new_servers + if connecting_by_name.get(n, n) in _server_connecting + ] + if stale_keys: logger.warning( "MCP discovery %s while %d server(s) were still " "connecting; clearing stale connecting set: %s", "timed out" if isinstance(_e, TimeoutError) else "interrupted", - len(stale), - ", ".join(stale), + len(stale_keys), + ", ".join(stale_keys), ) - _server_connecting.difference_update(stale) - for _sn in stale: + _server_connecting.difference_update(stale_keys) + for _sk in stale_keys: _server_connect_errors.setdefault( - _sn, + _sk, f"Connection attempt {'timed out' if isinstance(_e, TimeoutError) else 'interrupted'} during discovery", ) raise @@ -7791,10 +8251,11 @@ async def _discover_all(): connected = [ n for n in new_servers - if n in _servers and n not in _server_connect_errors + if connecting_by_name.get(n, n) in _servers + and connecting_by_name.get(n, n) not in _server_connect_errors ] new_tool_count = sum( - len(getattr(_servers[n], "_registered_tool_names", [])) + len(getattr(_servers[connecting_by_name.get(n, n)], "_registered_tool_names", [])) for n in connected ) failed = len(new_servers) - len(connected) @@ -7862,8 +8323,9 @@ def discover_mcp_tools() -> List[str]: new_server_names = [ name for name, cfg in servers.items() - if name not in _servers - and name not in connecting + if not _any_live_named(name) + and not _name_in_keyed_set(connecting, name) + and name not in _lazy_server_configs and _parse_boolish(cfg.get("enabled", True), default=True) ] @@ -7875,11 +8337,13 @@ def discover_mcp_tools() -> List[str]: connected_server_names = [ name for name in new_server_names - if name in _servers and name not in _server_connect_errors + if _any_live_named(name) or name in _lazy_server_configs ] new_tool_count = sum( - len(getattr(_servers[name], "_registered_tool_names", [])) + len(getattr(srv, "_registered_tool_names", []) or []) for name in connected_server_names + for srv in [_find_named_in_map(_servers, name)] + if srv is not None ) failed_count = len(new_server_names) - len(connected_server_names) @@ -7936,7 +8400,7 @@ def get_mcp_status() -> List[dict]: for name, cfg in configured.items(): transport = cfg.get("transport", "http") if "url" in cfg else "stdio" enabled = _parse_boolish(cfg.get("enabled", True), default=True) - server = active_servers.get(name) + server = _find_named_in_map(active_servers, name) if server and server.session is not None: entry = { "name": name, @@ -7961,7 +8425,7 @@ def get_mcp_status() -> List[dict]: "disabled": True, "status": "disabled", }) - elif name in connecting: + elif _name_in_keyed_set(connecting, name): result.append({ "name": name, "transport": transport, @@ -7970,7 +8434,7 @@ def get_mcp_status() -> List[dict]: "disabled": False, "status": "connecting", }) - elif name in connect_errors: + elif _name_in_keyed_set(connect_errors, name): result.append({ "name": name, "transport": transport, @@ -7978,7 +8442,7 @@ def get_mcp_status() -> List[dict]: "connected": False, "disabled": False, "status": "failed", - "error": connect_errors[name], + "error": _find_named_in_map(connect_errors, name), }) else: result.append({ @@ -8287,46 +8751,57 @@ def _add(schema: dict) -> bool: return staged_engine_names -def shutdown_mcp_servers(): - """Close all MCP server connections and stop the background loop. +def _purge_lazy_mcp_templates(logical_names: Optional[set] = None) -> None: + """Drop cache-backed templates and deregister names no live peer still owns. - Each server Task is signalled to exit its ``async with`` block so that - the anyio cancel-scope cleanup happens in the same Task that opened it. - All servers are shut down in parallel via ``asyncio.gather``. + ``None`` purges every lazy server. A set purges only those logical names + (servers deleted from config.yaml). """ + from tools.registry import registry + with _lock: - servers_snapshot = list(_servers.values()) + if logical_names is None: + targets = list(dict.fromkeys( + list(_lazy_server_configs) + + list(_lazy_server_tool_names) + + list(_lazy_server_fingerprints) + )) + else: + targets = list(logical_names) + names_by_server = { + name: list(_lazy_server_tool_names.pop(name, None) or []) + for name in targets + } + for name in targets: + _lazy_server_configs.pop(name, None) + _lazy_server_fingerprints.pop(name, None) + _oauth_protected_servers.discard(name) + live_held: set = set() + for srv in _servers.values(): + live_held.update(getattr(srv, "_registered_tool_names", None) or []) + for tool_names in names_by_server.values(): + for tool_name in tool_names: + if tool_name in live_held: + continue + registry.deregister(tool_name) + _forget_mcp_tool_server(tool_name) - # Fast path: nothing to shut down. The connect-cooldown maps can still - # be populated here β€” a server that failed to connect is never recorded - # in ``_servers`` (that is the very premise of the #50394 cooldown), so - # "no live servers" is the MOST likely state in which stale backoff - # entries exist. Clear them so a post-shutdown restart re-attempts every - # configured server immediately. - if not servers_snapshot: - with _lock: - _server_connect_retry_after.clear() - _server_connect_failures.clear() - _stop_mcp_loop() + +def _close_mcp_tasks(servers: list) -> None: + """Signal each task to exit its transport context. Does not clear ``_servers``.""" + if not servers: return async def _shutdown(): results = await asyncio.gather( - *(server.shutdown() for server in servers_snapshot), + *(server.shutdown() for server in servers), return_exceptions=True, ) - for server, result in zip(servers_snapshot, results): + for server, result in zip(servers, results): if isinstance(result, Exception): logger.debug( "Error closing MCP server '%s': %s", server.name, result, ) - with _lock: - _servers.clear() - # Drop connect-retry cooldowns too: a full shutdown/restart - # should re-attempt every server immediately, not honour a - # stale per-server backoff from before the restart (#50394). - _server_connect_retry_after.clear() - _server_connect_failures.clear() with _lock: loop = _mcp_loop @@ -8343,17 +8818,123 @@ async def _shutdown(): except BaseException as exc: logger.debug("Error during MCP shutdown: %s", exc) - # Unconditional final sweep: whether the async ``_shutdown`` ran, - # timed out, or was never scheduled (loop already stopped), a full - # shutdown must leave no stale connect-cooldown state behind β€” the - # next start should re-attempt every server immediately (#50394). + +def _pop_mcp_server_keys(keys) -> None: + with _lock: + for key in keys: + _servers.pop(key, None) + _server_connecting.discard(key) + _server_connect_errors.pop(key, None) + _server_connect_retry_after.pop(key, None) + _server_connect_failures.pop(key, None) + + +def shutdown_mcp_servers(): + """Close all MCP server connections and stop the background loop. + + Each server Task is signalled to exit its ``async with`` block so that + the anyio cancel-scope cleanup happens in the same Task that opened it. + All servers are shut down in parallel via ``asyncio.gather``. + + Also purges cache-backed lazy templates so a server deleted from + config.yaml cannot remain callable after ``/reload-mcp``. + """ with _lock: + servers_snapshot = list(_servers.values()) + keys_snapshot = list(_servers.keys()) + # Auth classification is rebuilt on the next register/discover pass. + # Clearing here so a reload that drops ``auth: oauth`` (or removes + # the server) cannot keep serving requester-scoped connections. + _oauth_protected_servers.clear() + + # Fast path: nothing to shut down. The connect-cooldown maps can still + # be populated here β€” a server that failed to connect is never recorded + # in ``_servers`` (that is the very premise of the #50394 cooldown), so + # "no live servers" is the MOST likely state in which stale backoff + # entries exist. Clear them so a post-shutdown restart re-attempts every + # configured server immediately. Lazy templates must still be purged: + # a cache-only OAuth server has no live task. + if not servers_snapshot: + with _lock: + _server_connect_retry_after.clear() + _server_connect_failures.clear() + _purge_lazy_mcp_templates() + _stop_mcp_loop() + return + + _close_mcp_tasks(servers_snapshot) + _pop_mcp_server_keys(keys_snapshot) + with _lock: + _servers.clear() _server_connect_retry_after.clear() _server_connect_failures.clear() - + _purge_lazy_mcp_templates() _stop_mcp_loop() +def reload_mcp_connections() -> None: + """Shutdown policy for ``/reload-mcp``. + + Shared mode and unbound callers take the full ``shutdown_mcp_servers`` + path. In ``per_user`` with a bound requester, other principals' OAuth + sessions stay up; this requester's OAuth connections, process-level + (non-OAuth) connections, and every identity of a server no longer in + config.yaml are recycled. + """ + from tools.mcp_oauth_identity import ( + IDENTITY_MODE_PER_USER, + configured_identity_mode, + principal_from_bound_fields, + resolve_mcp_oauth_scope, + ) + + configured = set(_load_mcp_config() or {}) + requester_scope = None + if configured_identity_mode() == IDENTITY_MODE_PER_USER: + try: + from gateway.session_context import get_bound_session_principal + + bound = get_bound_session_principal() + except Exception: + bound = None + if bound is not None: + try: + requester_scope = resolve_mcp_oauth_scope( + identity_mode=IDENTITY_MODE_PER_USER, + principal=principal_from_bound_fields( + bound.platform, bound.scope_id, bound.user_id + ), + uses_oauth=True, + ) + except Exception: + requester_scope = None + + with _lock: + live = dict(_servers) + lazy_names = set(_lazy_server_configs) + keys = _reload_shutdown_keys(configured, live, requester_scope) + if keys is None: + shutdown_mcp_servers() + return + + to_close = [live[k] for k in keys if k in live] + _close_mcp_tasks(to_close) + _pop_mcp_server_keys(keys) + removed = { + _logical_mcp_server_name(k, live[k]) + for k in keys + if k in live and _logical_mcp_server_name(k, live[k]) not in configured + } + removed.update(name for name in lazy_names if name not in configured) + if removed: + _purge_lazy_mcp_templates(removed) + if not _servers: + with _lock: + _server_connect_retry_after.clear() + _server_connect_failures.clear() + _stop_mcp_loop() + + def _kill_orphaned_mcp_children( include_active: bool = False, server_name: Optional[str] = None, diff --git a/tui_gateway/methods_tools.py b/tui_gateway/methods_tools.py index f153bc98a0208..40155f6f8ac9a 100644 --- a/tui_gateway/methods_tools.py +++ b/tui_gateway/methods_tools.py @@ -133,7 +133,7 @@ def _(rid, params: dict) -> dict: return _err(rid, 5019, f"compute-host reload_mcp failed: {exc}") return _ok(rid, {"status": "reloaded", "turn_isolation": True, "host_ack": ack}) - from tools.mcp_tool import shutdown_mcp_servers, discover_mcp_tools + from tools.mcp_tool import reload_mcp_connections, discover_mcp_tools def _refresh_session_agent() -> None: """Rebuild THIS session's cached tool snapshot from the live @@ -170,7 +170,7 @@ def _refresh_session_agent() -> None: req_rev = str(params.get("rev") or "") def _do_full_reload() -> None: - """shutdown+discover+refresh under the lock, then mark a completed + """reload+discover+refresh under the lock, then mark a completed generation. The lock spans the refresh too: releasing after discover would let a second reload tear the registry down while this one is still reading it to rebuild the session snapshot. @@ -183,7 +183,7 @@ def _do_full_reload() -> None: loaded = _compute_mcp_rev() for _ in range(_MCP_RELOAD_MAX_PASSES): - shutdown_mcp_servers() + reload_mcp_connections() discover_mcp_tools() after = _compute_mcp_rev() if after == loaded: diff --git a/website/docs/user-guide/features/mcp.md b/website/docs/user-guide/features/mcp.md index 274c2173c4ba1..beedccba8bccb 100644 --- a/website/docs/user-guide/features/mcp.md +++ b/website/docs/user-guide/features/mcp.md @@ -271,6 +271,24 @@ mcp_servers: On first connect, Hermes prints an authorize URL, opens your browser when possible, and waits for the OAuth callback on a local loopback port. Tokens are cached at `~/.hermes/mcp-tokens/.json` with 0o600 perms; subsequent runs reuse them silently until refresh fails. +### Shared gateway: per-requester OAuth (`mcp.oauth.identity_mode`) + +By default MCP OAuth is **shared** per Hermes profile: one token file per server under `~/.hermes/mcp-tokens/`. That is correct for a single-user CLI. On a **shared messaging gateway** (Slack / Discord / Telegram / …), turn on requester isolation so Alice's GitHub token cannot be used for Bob: + +```yaml +mcp: + oauth: + identity_mode: per_user # default: shared. Typos are rejected, never silently downgraded. +``` + +In `per_user` mode: + +- OAuth credentials, refresh, 401 recovery, and live connections are keyed by the **authenticated gateway requester** (platform + tenant scope + user id). Empty tenant scope (Telegram DMs, Discord DMs, …) is canonicalized; a missing bound identity fail-closes. +- Tokens live under `~/.hermes/mcp-tokens/by-user//.json`. The key is a digest β€” raw user ids never appear in paths. A legacy shared `mcp-tokens/.json` is **never** assigned to a requester. +- Direct CLI, TUI, desktop, and cron **cannot** complete or reuse OAuth without a bound gateway principal. There is no `hermes mcp login --user` selector. `hermes mcp remove ` is an admin path and deletes that server's artifacts across every `by-user/` namespace. +- Stdio and static-header MCP servers stay process-level (not multi-user-safe). Isolation applies to `auth: oauth` servers. +- Headless consent-URL delivery for messaging is a separate issue ([#78169](https://github.com/NousResearch/hermes-agent/issues/78169)) and is not part of this setting. + **Remote / headless hosts.** When Hermes runs on a different machine than your browser, the loopback callback can't reach your laptop. Two ways to complete the flow: - **Paste-back (no setup):** on an interactive terminal Hermes prints "Or paste the redirect URL here…" alongside the authorize URL. Open the URL in your browser, approve, copy the full URL the browser ends up on (the redirect will show a connection error β€” that's expected), paste it at the prompt. Bare `?code=…&state=…` query strings work too.