diff --git a/mcpgateway/transports/streamablehttp_transport.py b/mcpgateway/transports/streamablehttp_transport.py index 3d6a226985..3e295ac474 100644 --- a/mcpgateway/transports/streamablehttp_transport.py +++ b/mcpgateway/transports/streamablehttp_transport.py @@ -40,7 +40,7 @@ import re from typing import Any, assert_never, AsyncGenerator, ContextManager, Dict, Iterable, List, Optional, Pattern, Tuple, Union from urllib.parse import urlsplit, urlunsplit -from uuid import uuid4 +from uuid import UUID, uuid4 # Third-Party import anyio @@ -2203,6 +2203,63 @@ async def _get_request_context_or_default() -> Tuple[str, dict[str, Any], dict[s return s_id, request_headers_var.get(), user_context_var.get() +async def _resolve_subject_email(payload: dict[str, Any]) -> Optional[str]: + """Resolve a JWT subject claim to the user's email address. + + Session tokens (``token_use: "session"``) carry ``sub`` as the + ``EmailUser.id`` UUID; legacy/API tokens carry the email directly. This + mirrors the resolution already performed by + ``mcpgateway.auth.get_current_user`` and + ``mcpgateway.auth.resolve_session_teams`` so that the MCP transport keys its + user lookups, auth-cache entries and RBAC context on the same identity as + the REST paths. + + Per the repository's canonical email-over-sub precedence (see + ``mcpgateway.auth_context.get_user_email``), a non-empty string ``email`` + claim always wins over ``sub`` when both are present, so the resolved + identity can never be spoofed by a conflicting ``sub`` claim. + + When the subject is a UUID that cannot be resolved (no matching user row), + the raw subject is returned unchanged so the caller's existing + ``require_user_in_db`` checks reject the request rather than silently + downgrading it to anonymous access. + + Args: + payload: Decoded JWT payload. + + Returns: + The user's email address, or ``None`` when the token carries no subject. + + Examples: + >>> import asyncio + >>> asyncio.run(_resolve_subject_email({"sub": "user@example.com"})) + 'user@example.com' + >>> asyncio.run(_resolve_subject_email({"email": "fallback@example.com"})) + 'fallback@example.com' + >>> asyncio.run(_resolve_subject_email({"email": "canonical@example.com", "sub": "ignored-sub"})) + 'canonical@example.com' + >>> asyncio.run(_resolve_subject_email({})) is None + True + """ + email_claim = payload.get("email") + subject: Optional[str] = email_claim if isinstance(email_claim, str) and email_claim else payload.get("sub") + if not subject: + return None + + try: + UUID(subject) + except (ValueError, AttributeError, TypeError): + # Already an email (legacy/API token, or a session token issued before + # the user row existed) — no database round-trip needed. + return subject + + # First-Party + from mcpgateway.auth import _get_email_by_id_sync # pylint: disable=import-outside-toplevel + + resolved = await asyncio.to_thread(_get_email_by_id_sync, subject) + return resolved or subject + + async def _normalize_jwt_payload(payload: dict[str, Any]) -> dict[str, Any]: """Normalize a raw JWT payload to the canonical user context shape. @@ -2218,7 +2275,7 @@ async def _normalize_jwt_payload(payload: dict[str, Any]) -> dict[str, Any]: Returns: Canonical user context dict with keys email, teams, is_admin, is_authenticated, token_use. """ - email = payload.get("sub") or payload.get("email") + email = await _resolve_subject_email(payload) jwt_is_admin = payload.get("is_admin", False) if not jwt_is_admin: user_info = payload.get("user", {}) @@ -5086,7 +5143,9 @@ async def _auth_jwt(self, *, token: str) -> bool: # noqa: PLR0911 from mcpgateway.cache.auth_cache import CachedAuthContext, get_auth_cache # pylint: disable=import-outside-toplevel jti = user_payload.get("jti") - user_email = user_payload.get("sub") or user_payload.get("email") + # Resolve sub (a UUID for session tokens) to the user's email before any + # cache lookup or DB query, so MCP keys the same identity as the REST paths. + user_email = await _resolve_subject_email(user_payload) nested_user = user_payload.get("user", {}) nested_is_admin = nested_user.get("is_admin", False) if isinstance(nested_user, dict) else False is_admin = user_payload.get("is_admin", False) or nested_is_admin diff --git a/tests/unit/mcpgateway/transports/test_streamablehttp_transport.py b/tests/unit/mcpgateway/transports/test_streamablehttp_transport.py index 0095029f67..7b224cecf6 100644 --- a/tests/unit/mcpgateway/transports/test_streamablehttp_transport.py +++ b/tests/unit/mcpgateway/transports/test_streamablehttp_transport.py @@ -13277,6 +13277,271 @@ async def test_normalize_jwt_payload_session_admin_no_email_no_bypass(): assert result["is_admin"] is True +# --------------------------------------------------------------------------- +# _resolve_subject_email tests (issue #5215) +# +# Session tokens carry sub = EmailUser.id (a UUID); legacy/API tokens carry the +# email. The MCP transport used to treat sub as an email unconditionally, so +# session tokens produced a 401 "User not found in database" on +# /servers/{id}/mcp and empty-team public-only visibility on /mcp. +# --------------------------------------------------------------------------- + +SESSION_SUB_UUID = "387e3345-7996-4496-aa2d-09729ec8b3be" + + +@pytest.mark.asyncio +async def test_resolve_subject_email_uuid_sub_resolves_to_email(): + """A session token's UUID sub is resolved to the user's email via the DB.""" + lookup = Mock(return_value="user@example.com") + with patch("mcpgateway.auth._get_email_by_id_sync", lookup): + result = await tr._resolve_subject_email({"sub": SESSION_SUB_UUID, "token_use": "session"}) + + assert result == "user@example.com" + lookup.assert_called_once_with(SESSION_SUB_UUID) + + +@pytest.mark.asyncio +async def test_resolve_subject_email_legacy_sub_skips_db_lookup(): + """An email sub short-circuits before any DB round-trip (hot-path guard).""" + lookup = Mock(return_value="should-not-be-used@example.com") + with patch("mcpgateway.auth._get_email_by_id_sync", lookup): + result = await tr._resolve_subject_email({"sub": "legacy@example.com", "token_use": "api"}) + + assert result == "legacy@example.com" + lookup.assert_not_called() + + +@pytest.mark.asyncio +async def test_resolve_subject_email_session_token_with_email_sub(): + """Session tokens issued without a user row carry an email sub (admin.py:4273).""" + lookup = Mock(return_value="should-not-be-used@example.com") + with patch("mcpgateway.auth._get_email_by_id_sync", lookup): + result = await tr._resolve_subject_email({"sub": "admin@example.com", "token_use": "session"}) + + assert result == "admin@example.com" + lookup.assert_not_called() + + +@pytest.mark.asyncio +async def test_resolve_subject_email_unresolvable_uuid_returns_raw_subject(): + """An unresolvable UUID keeps the raw subject so require_user_in_db still denies. + + SECURITY: the request must fail closed on the caller's existing + "User not found in database" branch, never silently downgrade to anonymous. + """ + with patch("mcpgateway.auth._get_email_by_id_sync", Mock(return_value=None)): + result = await tr._resolve_subject_email({"sub": SESSION_SUB_UUID, "token_use": "session"}) + + assert result == SESSION_SUB_UUID + + +@pytest.mark.asyncio +async def test_resolve_subject_email_falls_back_to_email_claim(): + """Falls back to the 'email' claim when 'sub' is absent.""" + result = await tr._resolve_subject_email({"email": "fallback@example.com"}) + assert result == "fallback@example.com" + + +@pytest.mark.asyncio +async def test_resolve_subject_email_prefers_email_over_conflicting_sub(): + """SECURITY: a non-empty 'email' claim wins over 'sub' when both are present. + + Mirrors the repository's canonical email-over-sub precedence + (mcpgateway.auth_context.get_user_email). Without this, a token with a + forged/stale 'sub' UUID could resolve to a different principal than the + canonical email, mismatching cache keys, team visibility, and RBAC + (CWE-287/CWE-863). + """ + lookup = Mock(return_value="should-not-be-used@example.com") + with patch("mcpgateway.auth._get_email_by_id_sync", lookup): + result = await tr._resolve_subject_email({"email": "canonical@example.com", "sub": SESSION_SUB_UUID, "token_use": "session"}) + + assert result == "canonical@example.com" + lookup.assert_not_called() + + +@pytest.mark.asyncio +async def test_resolve_subject_email_non_string_email_falls_back_to_sub(): + """A malformed (non-string) 'email' claim is ignored in favor of 'sub'.""" + result = await tr._resolve_subject_email({"email": ["not", "a", "string"], "sub": "legacy@example.com", "token_use": "api"}) + assert result == "legacy@example.com" + + +@pytest.mark.asyncio +async def test_resolve_subject_email_empty_email_falls_back_to_sub(): + """An empty-string 'email' claim is treated as absent and falls back to 'sub'.""" + result = await tr._resolve_subject_email({"email": "", "sub": "legacy@example.com", "token_use": "api"}) + assert result == "legacy@example.com" + + +@pytest.mark.asyncio +async def test_resolve_subject_email_no_subject_returns_none(): + """A payload with no subject yields None rather than raising.""" + assert await tr._resolve_subject_email({}) is None + + +@pytest.mark.asyncio +async def test_resolve_subject_email_db_error_propagates(): + """A DB failure propagates so _auth_jwt's SQLAlchemyError handler returns 503. + + SECURITY: identity resolution must fail closed. This deliberately differs + from the fail-open revocation/user lookups later in _auth_jwt — an + unresolvable subject means we do not know who the caller is. + """ + # Third-Party + from sqlalchemy.exc import SQLAlchemyError + + with patch("mcpgateway.auth._get_email_by_id_sync", Mock(side_effect=SQLAlchemyError("db down"))): + with pytest.raises(SQLAlchemyError): + await tr._resolve_subject_email({"sub": SESSION_SUB_UUID, "token_use": "session"}) + + +@pytest.mark.asyncio +async def test_normalize_jwt_payload_resolves_uuid_sub(monkeypatch): + """The stateful-session fallback path resolves a UUID sub to the email too.""" + monkeypatch.setattr("mcpgateway.auth._get_email_by_id_sync", lambda user_id: "dev@example.com") + + raw = {"sub": SESSION_SUB_UUID, "token_use": "session"} + with patch("mcpgateway.auth.resolve_session_teams", new_callable=AsyncMock, return_value=["team-x"]): + result = await _normalize_jwt_payload_ref()(raw) + + assert result["email"] == "dev@example.com" + assert result["teams"] == ["team-x"] + + +def _normalize_jwt_payload_ref(): + """Return the module's _normalize_jwt_payload (kept out of the test body for import hygiene). + + Returns: + The _normalize_jwt_payload coroutine function under test. + """ + return tr._normalize_jwt_payload + + +# --------------------------------------------------------------------------- +# _auth_jwt: session-token identity resolution (issue #5215 / #5750) +# --------------------------------------------------------------------------- + + +def _auth_handler(): + """Build a _StreamableHttpAuthHandler over a minimal ASGI triple. + + Returns: + A handler instance whose scope/receive/send are test doubles. + """ + return tr._StreamableHttpAuthHandler({"type": "http", "headers": []}, AsyncMock(), AsyncMock()) + + +@pytest.mark.asyncio +async def test_auth_jwt_session_token_keys_cache_and_lookups_by_email(monkeypatch): + """A session token's UUID sub must never reach the auth cache or the user lookup. + + Regression guard for the sticky 401: before the fix the transport missed on a + UUID cache key, fell through to a UUID-keyed DB lookup, then wrote the empty + result back under that same UUID key. + """ + handler = _auth_handler() + + monkeypatch.setattr(tr, "verify_credentials", AsyncMock(return_value={"sub": SESSION_SUB_UUID, "token_use": "session", "jti": "jti-1"})) + monkeypatch.setattr(tr._StreamableHttpAuthHandler, "_route_idp_issued_token", AsyncMock(return_value=None)) + monkeypatch.setattr("mcpgateway.auth._get_email_by_id_sync", lambda user_id: "dev@example.com") + + auth_cache = MagicMock() + auth_cache.get_auth_context = AsyncMock(return_value=None) + auth_cache.get_user_teams = AsyncMock(return_value=None) + auth_cache.set_auth_context = AsyncMock() + auth_cache.set_user_teams = AsyncMock() + monkeypatch.setattr("mcpgateway.cache.auth_cache.get_auth_cache", lambda: auth_cache) + monkeypatch.setattr(tr.settings, "auth_cache_enabled", True) + monkeypatch.setattr(tr.settings, "auth_cache_batch_queries", True) + + monkeypatch.setattr( + "mcpgateway.auth._get_auth_context_batched_sync", + lambda email, jti: {"user": {"email": email, "is_admin": False, "is_active": True}, "team_ids": ["team-x"], "is_token_revoked": False, "personal_team_id": None}, + ) + monkeypatch.setattr("mcpgateway.auth.resolve_session_teams", AsyncMock(return_value=["team-x"])) + monkeypatch.setattr("mcpgateway.auth.resolve_trace_team_name", AsyncMock(return_value=None)) + + assert await handler._auth_jwt(token="tok") is True + + # Every identity-keyed call must use the email, never the UUID. + assert auth_cache.get_auth_context.await_args.args[0] == "dev@example.com" + assert auth_cache.set_auth_context.await_args.args[0] == "dev@example.com" + assert auth_cache.set_user_teams.await_args.args[0] == "dev@example.com:True" + + ctx = tr.user_context_var.get() + assert ctx["email"] == "dev@example.com" + assert ctx["teams"] == ["team-x"] + + +@pytest.mark.asyncio +async def test_auth_jwt_session_token_unknown_user_still_denied(monkeypatch): + """Deny path preserved: an unresolvable UUID sub still 401s under require_user_in_db.""" + handler = _auth_handler() + sent = [] + monkeypatch.setattr(tr._StreamableHttpAuthHandler, "_send_error", AsyncMock(side_effect=lambda **kw: (sent.append(kw), False)[1])) + + monkeypatch.setattr(tr, "verify_credentials", AsyncMock(return_value={"sub": SESSION_SUB_UUID, "token_use": "session", "jti": "jti-2"})) + monkeypatch.setattr(tr._StreamableHttpAuthHandler, "_route_idp_issued_token", AsyncMock(return_value=None)) + monkeypatch.setattr("mcpgateway.auth._get_email_by_id_sync", lambda user_id: None) + monkeypatch.setattr(tr.settings, "auth_cache_enabled", False) + monkeypatch.setattr(tr.settings, "auth_cache_batch_queries", False) + monkeypatch.setattr(tr.settings, "require_user_in_db", True) + monkeypatch.setattr("mcpgateway.auth._check_token_revoked_sync", lambda jti: False) + monkeypatch.setattr("mcpgateway.auth._get_user_by_email_sync", lambda email: None) + + assert await handler._auth_jwt(token="tok") is False + assert sent and sent[0]["detail"] == "User not found in database" + + +@pytest.mark.asyncio +async def test_auth_jwt_session_token_platform_admin_escape_hatch(monkeypatch): + """Resolving sub lets the platform-admin escape hatch apply to session tokens. + + SECURITY: this is the one place the fix widens access. Before the fix a + platform-admin session token carried a UUID that never equalled + platform_admin_email, so the hatch could not apply. + """ + handler = _auth_handler() + + monkeypatch.setattr(tr, "verify_credentials", AsyncMock(return_value={"sub": SESSION_SUB_UUID, "token_use": "session", "is_admin": True, "jti": "jti-3"})) + monkeypatch.setattr(tr._StreamableHttpAuthHandler, "_route_idp_issued_token", AsyncMock(return_value=None)) + monkeypatch.setattr("mcpgateway.auth._get_email_by_id_sync", lambda user_id: "admin@example.com") + monkeypatch.setattr(tr.settings, "auth_cache_enabled", False) + monkeypatch.setattr(tr.settings, "auth_cache_batch_queries", False) + monkeypatch.setattr(tr.settings, "require_user_in_db", True) + monkeypatch.setattr(tr.settings, "platform_admin_email", "admin@example.com") + monkeypatch.setattr("mcpgateway.auth._check_token_revoked_sync", lambda jti: False) + monkeypatch.setattr("mcpgateway.auth._get_user_by_email_sync", lambda email: None) + monkeypatch.setattr("mcpgateway.auth.resolve_session_teams", AsyncMock(return_value=None)) + monkeypatch.setattr("mcpgateway.auth.resolve_trace_team_name", AsyncMock(return_value=None)) + + assert await handler._auth_jwt(token="tok") is True + assert tr.user_context_var.get()["email"] == "admin@example.com" + + +@pytest.mark.asyncio +async def test_auth_jwt_legacy_token_performs_no_subject_lookup(monkeypatch): + """Legacy/API tokens must not incur an extra DB read on the auth hot path.""" + handler = _auth_handler() + lookup = Mock(return_value="should-not-be-used@example.com") + + monkeypatch.setattr(tr, "verify_credentials", AsyncMock(return_value={"sub": "legacy@example.com", "token_use": "api", "jti": "jti-4"})) + monkeypatch.setattr(tr._StreamableHttpAuthHandler, "_route_idp_issued_token", AsyncMock(return_value=None)) + monkeypatch.setattr("mcpgateway.auth._get_email_by_id_sync", lookup) + monkeypatch.setattr(tr.settings, "auth_cache_enabled", False) + monkeypatch.setattr(tr.settings, "auth_cache_batch_queries", False) + monkeypatch.setattr(tr.settings, "require_user_in_db", False) + monkeypatch.setattr("mcpgateway.auth._check_token_revoked_sync", lambda jti: False) + monkeypatch.setattr("mcpgateway.auth._get_user_by_email_sync", lambda email: None) + monkeypatch.setattr("mcpgateway.auth.normalize_token_teams", lambda payload: []) + monkeypatch.setattr("mcpgateway.auth.resolve_trace_team_name", AsyncMock(return_value=None)) + + assert await handler._auth_jwt(token="tok") is True + lookup.assert_not_called() + assert tr.user_context_var.get()["email"] == "legacy@example.com" + + # --------------------------------------------------------------------------- # call_tool: recovered context propagation regression test # ---------------------------------------------------------------------------