fix(mcp): resolve session-token sub UUID to email in streamable-HTTP auth

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

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