fix(auth): enforce is_active check + extend DB-backed proxy auth to primary MCP path

Second follow-up addressing findings from a deeper security + design review:

SECURITY FIX (critical):
- _authenticate_proxy_user() now enforces user_info.is_active, matching the
  JWT path's check in _enforce_revocation_and_active_user (:398-399).
  Without this, a disabled user - including a disabled admin - could still
  authenticate via trusted-proxy mode and keep their pre-disable
  authorizations. Raises 401 'Account disabled' for inactive users.

PRIMARY MCP TRANSPORT GAP (critical):
- streamablehttp_transport._set_proxy_user_context() previously hardcoded
  teams=[] and is_admin=False, so on the primary MCP entry point
  (_StreamableHttpAuthHandler.authenticate) proxy-authenticated users
  still received public-only access. Only the stateful-session fallback
  path at :1953 went through the enriched helper.
- Rewritten as an async function that performs the same DB lookup,
  team/admin resolution, is_active enforcement, and platform-admin
  bootstrap as the REST helper. Returns None on success or
  {detail, headers} on failure so the caller can send a 401 via
  self._send_error().
- Call site at _StreamableHttpAuthHandler.authenticate awaits the helper
  and sends 401 on rejection.

DISPATCH-KEY FIX (S1/S2):
- Proxy payload now includes token_use='session' so downstream dispatchers
  (main.py:2870, streamablehttp_transport.py:1998) route through
  resolve_session_teams() - the canonical 'single policy point' per
  AGENTS.md - rather than treating the proxy payload as an API-token
  payload with embedded teams.

DOCS + TEST HYGIENE (S3/S4a):
- require_auth docstring updated to describe the DB-enriched proxy path,
  bootstrap flow, and request.state caching.
- Misleading test test_jwt_auth_with_proxy_enabled renamed to
  test_mcp_client_auth_mode_skips_proxy_path with accurate docstring
  and comment explaining why 'anonymous' is the expected outcome.
- Stale comment 'Proxy auth path - identical to require_auth' updated
  to accurately describe the shared-helper relationship.

DENY-PATH REGRESSION TESTS (AGENTS.md security invariant):
- test_proxy_auth_disabled_user_raises_401: non-admin disabled user -> 401
- test_proxy_auth_disabled_admin_raises_401: disabled admin -> 401 (no
  elevation via is_admin=True on a deactivated record)
- test_proxy_auth_payload_has_token_use_session: dispatch-key contract
- 4 pre-existing tests updated to supply DB mocks instead of relying on
  the old unvalidated proxy shape.

Signed-off-by: Jonathan Springer <jps@s390x.com>

Author: jonpspri

mcpgateway/transports/streamablehttp_transport.py

@@ -4527,23 +4527,71 @@ def _set_user_identity_from_dict(ctx: dict[str, Any]) -> None:
         )
 
 
-def _set_proxy_user_context(proxy_user: str) -> None:
-    """Set user context for a proxy-authenticated request (no team context, non-admin).
+async def _set_proxy_user_context(proxy_user: str) -> dict[str, Any] | None:
+    """Authenticate a proxy-identified user and set per-request transport context.
+
+    Performs a DB lookup via EmailAuthService, resolves team/admin state via
+    :func:`mcpgateway.auth._resolve_teams_from_db`, enforces ``is_active``, and
+    handles the platform-admin bootstrap (``REQUIRE_USER_IN_DB=False`` + email
+    matches ``settings.platform_admin_email``).  On success, sets
+    ``user_context_var``, user identity, and trace context.  Mirrors the REST
+    ``_authenticate_proxy_user`` helper in ``verify_credentials.py`` so that
+    trusted-proxy MCP clients receive the same DB-backed team/admin resolution
+    as REST admin/API callers (fixes #4262 on the primary MCP transport path).
 
     Args:
-        proxy_user: Email address of the proxy-authenticated user.
+        proxy_user: Email address supplied by the trusted upstream proxy via
+            ``settings.proxy_user_header``.
+
+    Returns:
+        ``None`` on success.  On failure, returns a dict with ``detail`` (str)
+        and optional ``headers`` (dict) suitable for passing to
+        ``_StreamableHttpAuthHandler._send_error`` to produce a 401 response.
     """
-    _proxy_ctx: dict[str, Any] = {
-        "email": proxy_user,
-        "teams": [],
-        "is_authenticated": True,
-        "is_admin": False,
-        "permission_is_admin": False,
-        "auth_method": "proxy",
-    }
-    user_context_var.set(_proxy_ctx)
-    _set_user_identity_from_dict(_proxy_ctx)
-    set_trace_context_from_teams([], user_email=proxy_user, is_admin=False, auth_method="proxy")
+    # First-Party
+    from mcpgateway.auth import _resolve_teams_from_db  # pylint: disable=import-outside-toplevel
+    from mcpgateway.db import get_db  # pylint: disable=import-outside-toplevel
+    from mcpgateway.services.email_auth_service import EmailAuthService  # pylint: disable=import-outside-toplevel
+
+    db = next(get_db())
+    try:
+        auth_service = EmailAuthService(db)
+        user_info = await auth_service.get_user_by_email(proxy_user)
+
+        if user_info:
+            # Enforce account-active check (matches JWT path in _enforce_revocation_and_active_user).
+            # A disabled user - including a disabled admin - must not be able to authenticate via
+            # trusted-proxy mode and inherit their pre-disable authorizations.
+            if not user_info.is_active:
+                return {"detail": "Account disabled", "headers": {"WWW-Authenticate": "Bearer"}}
+
+            token_teams = await _resolve_teams_from_db(proxy_user, user_info)
+            is_admin = user_info.is_admin
+        else:
+            platform_admin_email = getattr(settings, "platform_admin_email", "admin@example.com")
+            if not settings.require_user_in_db and proxy_user == platform_admin_email:
+                token_teams = None  # Admin bypass
+                is_admin = True
+            else:
+                return {"detail": "User not found in database", "headers": {"WWW-Authenticate": "Bearer"}}
+
+        _proxy_ctx: dict[str, Any] = {
+            "email": proxy_user,
+            "teams": token_teams,  # None for admin bypass, [] for public-only, or list of team IDs
+            "is_authenticated": True,
+            "is_admin": is_admin,
+            "permission_is_admin": is_admin,
+            "auth_method": "proxy",
+            "token_use": "session",  # nosec B105 - Not a password; JWT claim type. DB-backed team resolution.
+        }
+        user_context_var.set(_proxy_ctx)
+        _set_user_identity_from_dict(_proxy_ctx)
+        # For trace context, admin bypass (teams=None) is represented as [] to match the existing
+        # pre-authentication contract of set_trace_context_from_teams.
+        set_trace_context_from_teams(token_teams or [], user_email=proxy_user, is_admin=is_admin, auth_method="proxy")
+        return None
+    finally:
+        db.close()
 
 
 def get_streamable_http_auth_context() -> dict[str, Any]:
@@ -4673,8 +4721,12 @@ async def authenticate(self) -> bool:
 
         # Determine authentication strategy based on settings
         if proxy_trusted and proxy_user:
-            _set_proxy_user_context(proxy_user)
-            return True  # Trusted proxy supplied user
+            # DB-backed authentication of the proxy-supplied identity; returns None on success
+            # or {"detail": ..., "headers": ...} on failure (unknown user, disabled user).
+            proxy_error = await _set_proxy_user_context(proxy_user)
+            if proxy_error:
+                return await self._send_error(**proxy_error)
+            return True  # Trusted proxy supplied valid, active user
 
         # --- Standard JWT authentication flow (client auth enabled) ---
         token: str | None = None

mcpgateway/utils/verify_credentials.py

@@ -442,6 +442,16 @@ async def _authenticate_proxy_user(request: Request, proxy_user: str) -> dict:
         user_info = await auth_service.get_user_by_email(proxy_user)
 
         if user_info:
+            # Enforce account-active check (matches the JWT path in _enforce_revocation_and_active_user:398-399).
+            # Without this, a disabled user - including a disabled admin - could authenticate via trusted-proxy
+            # mode and inherit their pre-disable authorizations.
+            if not user_info.is_active:
+                raise HTTPException(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    detail="Account disabled",
+                    headers={"WWW-Authenticate": "Bearer"},
+                )
+
             # Resolve teams from DB (returns None for admin bypass, [] for no teams, or list of team IDs)
             token_teams = await _resolve_teams_from_db(proxy_user, user_info)
             payload = {
@@ -451,6 +461,10 @@ async def _authenticate_proxy_user(request: Request, proxy_user: str) -> dict:
                 "is_admin": user_info.is_admin,
                 "teams": token_teams,  # None for admin bypass, [] for public-only, or list of team IDs
                 "email": proxy_user,
+                # token_use: "session" signals DB-backed team resolution to downstream dispatchers
+                # (main.py:2870, streamablehttp_transport.py:1998) so they route via resolve_session_teams
+                # rather than treating the proxy payload as an API-token payload with embedded teams.
+                "token_use": "session",  # nosec B105 - Not a password; JWT claim type
             }
         else:
             # User not in DB - handle based on REQUIRE_USER_IN_DB setting
@@ -464,6 +478,7 @@ async def _authenticate_proxy_user(request: Request, proxy_user: str) -> dict:
                     "is_admin": True,
                     "teams": None,  # Admin bypass
                     "email": proxy_user,
+                    "token_use": "session",  # nosec B105 - Not a password; JWT claim type
                 }
             else:
                 raise HTTPException(
@@ -483,10 +498,22 @@ async def require_auth(request: Request, credentials: Optional[HTTPAuthorization
     """Require authentication via JWT token or proxy headers.
 
     FastAPI dependency that checks for authentication via:
-    1. Proxy headers (if mcp_client_auth_enabled=false and trust_proxy_auth=true)
+    1. Proxy headers (if mcp_client_auth_enabled=false and is_proxy_auth_trust_active())
     2. JWT token in Authorization header (Bearer scheme)
     3. JWT token in cookies
 
+    Proxy authentication path (see :func:`_authenticate_proxy_user`):
+        When configured, the proxy-supplied user identity is looked up in
+        the DB via ``EmailAuthService`` and their team/admin context is
+        resolved. The resulting enriched payload
+        (``sub``, ``source``, ``token``, ``is_admin``, ``teams``, ``email``)
+        is cached on ``request.state._jwt_verified_payload`` so downstream
+        middleware and handlers get the same shape used for JWT-authenticated
+        requests. When ``REQUIRE_USER_IN_DB=False`` and the proxy user matches
+        ``settings.platform_admin_email``, a platform-admin bootstrap payload
+        (``is_admin=True``, ``teams=None``) is returned without requiring a
+        DB record; otherwise an unknown proxy user raises 401.
+
     If authentication is required but no token is provided, raises an HTTP 401 error.
 
     Args:
@@ -496,11 +523,13 @@ async def require_auth(request: Request, credentials: Optional[HTTPAuthorization
 
     Returns:
         str | dict: The verified credentials payload if authenticated,
-            proxy user if proxy auth enabled, or "anonymous" if authentication is not required.
+            the enriched proxy payload if proxy auth succeeded, or
+            ``"anonymous"`` if authentication is not required.
 
     Raises:
         HTTPException: 401 status if authentication is required but no valid
-            token is provided.
+            token is provided, or if a proxy-identified user is unknown and
+            the platform-admin bootstrap conditions do not apply.
 
     Examples:
         >>> from mcpgateway.utils import verify_credentials as vc

tests/unit/mcpgateway/transports/test_streamablehttp_transport.py

@@ -27,7 +27,7 @@
 import json
 from types import SimpleNamespace
 from typing import List
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 # Third-Party
 from fastapi import HTTPException
@@ -4871,7 +4871,7 @@ async def failing_get_db():
 
 @pytest.mark.asyncio
 async def test_streamable_http_auth_proxy_user_when_client_auth_disabled(monkeypatch):
-    """Test auth sets user context for proxy user when client auth disabled (lines 1740-1750)."""
+    """Proxy user with valid DB record authenticates and gets DB-backed team/admin context."""
     monkeypatch.setattr("mcpgateway.transports.streamablehttp_transport.settings.mcp_client_auth_enabled", False)
     monkeypatch.setattr("mcpgateway.transports.streamablehttp_transport.settings.trust_proxy_auth", True)
     monkeypatch.setattr("mcpgateway.transports.streamablehttp_transport.settings.trust_proxy_auth_dangerously", True)
@@ -4888,7 +4888,22 @@ async def test_streamable_http_auth_proxy_user_when_client_auth_disabled(monkeyp
     async def send(msg):
         sent.append(msg)
 
-    result = await streamable_http_auth(scope, None, send)
+    mock_user = Mock()
+    mock_user.is_admin = False
+    mock_user.is_active = True
+    mock_user.email = "proxy_user@example.com"
+
+    with patch("mcpgateway.db.get_db") as mock_get_db, patch(
+        "mcpgateway.services.email_auth_service.EmailAuthService"
+    ) as mock_auth_service, patch(
+        "mcpgateway.auth._resolve_teams_from_db", new_callable=AsyncMock
+    ) as mock_resolve_teams:
+        mock_get_db.return_value = iter([Mock()])
+        mock_auth_service.return_value.get_user_by_email = AsyncMock(return_value=mock_user)
+        mock_resolve_teams.return_value = []
+
+        result = await streamable_http_auth(scope, None, send)
+
     assert result is True
     assert sent == []  # No 401 sent
 
@@ -4924,7 +4939,22 @@ async def test_streamable_http_auth_proxy_user_with_bearer_header(monkeypatch):
     async def send(msg):
         sent.append(msg)
 
-    result = await streamable_http_auth(scope, None, send)
+    mock_user = Mock()
+    mock_user.is_admin = False
+    mock_user.is_active = True
+    mock_user.email = "proxy_fallback@example.com"
+
+    with patch("mcpgateway.db.get_db") as mock_get_db, patch(
+        "mcpgateway.services.email_auth_service.EmailAuthService"
+    ) as mock_auth_service, patch(
+        "mcpgateway.auth._resolve_teams_from_db", new_callable=AsyncMock
+    ) as mock_resolve_teams:
+        mock_get_db.return_value = iter([Mock()])
+        mock_auth_service.return_value.get_user_by_email = AsyncMock(return_value=mock_user)
+        mock_resolve_teams.return_value = []
+
+        result = await streamable_http_auth(scope, None, send)
+
     assert result is True
     assert sent == []
 
@@ -4960,7 +4990,22 @@ async def fake_verify(token):
     async def send(msg):
         sent.append(msg)
 
-    result = await streamable_http_auth(scope, None, send)
+    mock_user = Mock()
+    mock_user.is_admin = False
+    mock_user.is_active = True
+    mock_user.email = "proxy_user@example.com"
+
+    with patch("mcpgateway.db.get_db") as mock_get_db, patch(
+        "mcpgateway.services.email_auth_service.EmailAuthService"
+    ) as mock_auth_service, patch(
+        "mcpgateway.auth._resolve_teams_from_db", new_callable=AsyncMock
+    ) as mock_resolve_teams:
+        mock_get_db.return_value = iter([Mock()])
+        mock_auth_service.return_value.get_user_by_email = AsyncMock(return_value=mock_user)
+        mock_resolve_teams.return_value = []
+
+        result = await streamable_http_auth(scope, None, send)
+
     assert result is True
 
     user_ctx = tr.user_context_var.get()