fix(auth): extend proxy auth enrichment to MCP transport path

The original PR #4320 added DB-backed team/admin resolution to require_auth,
but left require_auth_header_first returning the old minimal payload
(sub/source/token only). require_auth_header_first is the authentication
entry point used by the MCP streamable HTTP transport, so proxy-authenticated
MCP clients continued to receive public-only access via _normalize_jwt_payload
-> normalize_token_teams([]) - the exact symptom #4262 reports.

Changes:
- Extract _authenticate_proxy_user(request, proxy_user) helper owning DB
  lookup, team resolution, payload construction, platform-admin bootstrap,
  and request.state caching.
- Reuse the helper from both require_auth and require_auth_header_first so
  REST admin paths and the MCP transport path return the same enriched
  payload (sub, source, token, is_admin, teams, email).
- Fix the now-inaccurate comment on require_auth_header_first's proxy branch
  that claimed it was 'identical to require_auth'.
- Update the existing test_require_auth_header_first_proxy_auth_returns_proxy_user
  to assert the enriched payload shape.
- Add regression tests covering admin-user DB bypass (is_admin=True, teams=None),
  multi-team membership, and require_auth_header_first parity (enriched payload
  and platform-admin bootstrap).

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

Author: jonpspri

mcpgateway/utils/verify_credentials.py

@@ -399,6 +399,86 @@ async def _enforce_revocation_and_active_user(payload: dict) -> None:
         _raise_auth_401("Account disabled")
 
 
+async def _authenticate_proxy_user(request: Request, proxy_user: str) -> dict:
+    """Authenticate a proxy-identified user and build an enriched auth payload.
+
+    Performs a DB lookup for the proxy-identified user, resolves their teams
+    and admin status via ``_resolve_teams_from_db``, caches the payload on
+    ``request.state._jwt_verified_payload``, and returns it.
+
+    Supports a platform-admin bootstrap flow: when
+    ``settings.require_user_in_db`` is ``False`` **and** the proxy header
+    matches ``settings.platform_admin_email``, an admin payload is returned
+    without requiring a DB record (same policy applied to JWTs in
+    ``_enforce_revocation_and_active_user``).
+
+    This helper is shared by :func:`require_auth` and
+    :func:`require_auth_header_first` so that proxy-authenticated callers get
+    the same enriched context regardless of the entry point (REST admin paths
+    vs MCP streamable HTTP transport).
+
+    Args:
+        request: FastAPI request used to cache the payload for downstream code.
+        proxy_user: The authenticated user identifier from the configured
+            proxy header (e.g. ``X-Authenticated-User``).
+
+    Returns:
+        dict: Enriched auth payload with keys ``sub``, ``source``, ``token``,
+        ``is_admin``, ``teams``, and ``email``. ``teams`` is ``None`` for
+        admin bypass, ``[]`` for public-only, or a list of team ID strings.
+
+    Raises:
+        HTTPException: 401 when the proxy-identified user is not present in
+            the DB and the platform-admin bootstrap conditions do not apply.
+    """
+    # 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:
+            # 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 = {
+                "sub": proxy_user,
+                "source": "proxy",
+                "token": None,  # nosec B105 - None is not a password
+                "is_admin": user_info.is_admin,
+                "teams": token_teams,  # None for admin bypass, [] for public-only, or list of team IDs
+                "email": proxy_user,
+            }
+        else:
+            # User not in DB - handle based on REQUIRE_USER_IN_DB setting
+            platform_admin_email = getattr(settings, "platform_admin_email", "admin@example.com")
+            if not settings.require_user_in_db and proxy_user == platform_admin_email:
+                # Platform admin bootstrap (matches the JWT path in _enforce_revocation_and_active_user)
+                payload = {
+                    "sub": proxy_user,
+                    "source": "proxy",
+                    "token": None,  # nosec B105 - None is not a password
+                    "is_admin": True,
+                    "teams": None,  # Admin bypass
+                    "email": proxy_user,
+                }
+            else:
+                raise HTTPException(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    detail="User not found in database",
+                    headers={"WWW-Authenticate": "Bearer"},
+                )
+
+        # Cache in request state for downstream use (same pattern as JWT tokens)
+        request.state._jwt_verified_payload = (None, payload)
+        return payload
+    finally:
+        db.close()
+
+
 async def require_auth(request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(security), jwt_token: Optional[str] = Cookie(default=None)) -> str | dict:
     """Require authentication via JWT token or proxy headers.
 
@@ -486,52 +566,7 @@ async def require_auth(request: Request, credentials: Optional[HTTPAuthorization
             # Extract user from proxy header
             proxy_user = request.headers.get(settings.proxy_user_header)
             if proxy_user:
-                # First-Party
-                from mcpgateway.auth import _resolve_teams_from_db
-                from mcpgateway.db import get_db
-                from mcpgateway.services.email_auth_service import EmailAuthService
-
-                # Query database for user info
-                db = next(get_db())
-                try:
-                    auth_service = EmailAuthService(db)
-                    user_info = await auth_service.get_user_by_email(proxy_user)
-
-                    if user_info:
-                        # 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)
-                        is_admin = user_info.is_admin
-
-                        # Build enriched payload similar to session tokens
-                        payload = {
-                            "sub": proxy_user,
-                            "source": "proxy",
-                            "token": None,
-                            "is_admin": is_admin,
-                            "teams": token_teams,  # None for admin bypass, [] for public-only, or list of team IDs
-                            "email": proxy_user,
-                        }
-
-                        # Cache in request state for downstream use (same pattern as JWT tokens)
-                        request.state._jwt_verified_payload = (None, payload)
-
-                        return payload
-                    else:
-                        # User not in DB - handle based on REQUIRE_USER_IN_DB setting
-                        platform_admin_email = getattr(settings, "platform_admin_email", "admin@example.com")
-                        if not settings.require_user_in_db and proxy_user == platform_admin_email:
-                            # Platform admin bootstrap
-                            payload = {"sub": proxy_user, "source": "proxy", "token": None, "is_admin": True, "teams": None, "email": proxy_user}  # Admin bypass
-                            request.state._jwt_verified_payload = (None, payload)
-                            return payload
-                        else:
-                            raise HTTPException(
-                                status_code=status.HTTP_401_UNAUTHORIZED,
-                                detail="User not found in database",
-                                headers={"WWW-Authenticate": "Bearer"},
-                            )
-                finally:
-                    db.close()
+                return await _authenticate_proxy_user(request, proxy_user)
             # No proxy header - check auth_required (matches RBAC/WebSocket behavior)
             if settings.auth_required:
                 raise HTTPException(
@@ -1089,12 +1124,14 @@ async def require_auth_header_first(
     if request is None:
         request = Request(scope={"type": "http", "headers": []})
 
-    # Proxy auth path — identical to require_auth
+    # Proxy auth path — shares _authenticate_proxy_user() with require_auth
+    # so proxy-authenticated callers get the same enriched payload (teams,
+    # is_admin, email, request.state caching) regardless of entry point.
     if not settings.mcp_client_auth_enabled:
         if is_proxy_auth_trust_active():
             proxy_user = request.headers.get(settings.proxy_user_header)
             if proxy_user:
-                return {"sub": proxy_user, "source": "proxy", "token": None}  # nosec B105 - None is not a password
+                return await _authenticate_proxy_user(request, proxy_user)
             if settings.auth_required:
                 raise HTTPException(
                     status_code=status.HTTP_401_UNAUTHORIZED,

tests/unit/mcpgateway/utils/test_proxy_auth.py

@@ -321,6 +321,127 @@ async def test_proxy_auth_user_not_found_raises_401(self, mock_settings, mock_re
                     assert exc_info.value.status_code == 401
                     assert "User not found in database" in exc_info.value.detail
 
+    @pytest.mark.asyncio
+    async def test_proxy_auth_admin_user_in_db_yields_admin_bypass(self, mock_settings, mock_request):
+        """Admin user found in DB gets is_admin=True and teams=None (admin bypass)."""
+        mock_settings.mcp_client_auth_enabled = False
+        mock_settings.trust_proxy_auth = True
+        mock_settings.trust_proxy_auth_dangerously = True
+        mock_request.headers = {"X-Authenticated-User": "admin-user@example.com"}
+        mock_request.state = Mock()
+
+        mock_user = Mock()
+        mock_user.is_admin = True
+        mock_user.email = "admin-user@example.com"
+
+        with patch.object(vc, "settings", mock_settings), 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)
+            # _resolve_teams_from_db returns None for admins (admin bypass)
+            mock_resolve_teams.return_value = None
+
+            result = await vc.require_auth(mock_request, None, None)
+
+        assert result["sub"] == "admin-user@example.com"
+        assert result["is_admin"] is True
+        assert result["teams"] is None
+        assert result["source"] == "proxy"
+
+    @pytest.mark.asyncio
+    async def test_proxy_auth_multiple_teams(self, mock_settings, mock_request):
+        """Non-admin user with multiple team memberships returns the full list."""
+        mock_settings.mcp_client_auth_enabled = False
+        mock_settings.trust_proxy_auth = True
+        mock_settings.trust_proxy_auth_dangerously = True
+        mock_request.headers = {"X-Authenticated-User": "multi-team-user@example.com"}
+        mock_request.state = Mock()
+
+        mock_user = Mock()
+        mock_user.is_admin = False
+        mock_user.email = "multi-team-user@example.com"
+
+        with patch.object(vc, "settings", mock_settings), 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 = ["team-a", "team-b", "team-c"]
+
+            result = await vc.require_auth(mock_request, None, None)
+
+        assert result["is_admin"] is False
+        assert result["teams"] == ["team-a", "team-b", "team-c"]
+
+    @pytest.mark.asyncio
+    async def test_require_auth_header_first_proxy_returns_enriched_payload(self, mock_settings, mock_request):
+        """Parity test: require_auth_header_first returns the enriched payload (same helper)."""
+        mock_settings.mcp_client_auth_enabled = False
+        mock_settings.trust_proxy_auth = True
+        mock_settings.trust_proxy_auth_dangerously = True
+        mock_request.headers = {"X-Authenticated-User": "mcp-user@example.com"}
+        mock_request.cookies = {}
+        mock_request.state = Mock()
+
+        mock_user = Mock()
+        mock_user.is_admin = False
+        mock_user.email = "mcp-user@example.com"
+
+        with patch.object(vc, "settings", mock_settings), 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 = ["team1"]
+
+            result = await vc.require_auth_header_first(auth_header=None, jwt_token=None, request=mock_request)
+
+        # Same enriched shape as require_auth (both go through _authenticate_proxy_user)
+        assert result["sub"] == "mcp-user@example.com"
+        assert result["source"] == "proxy"
+        assert result["token"] is None
+        assert result["is_admin"] is False
+        assert result["teams"] == ["team1"]
+        assert result["email"] == "mcp-user@example.com"
+
+    @pytest.mark.asyncio
+    async def test_require_auth_header_first_proxy_admin_bootstrap(self, mock_settings, mock_request):
+        """Parity test: require_auth_header_first supports platform-admin bootstrap."""
+        mock_settings.mcp_client_auth_enabled = False
+        mock_settings.trust_proxy_auth = True
+        mock_settings.trust_proxy_auth_dangerously = True
+        mock_settings.require_user_in_db = False
+        mock_settings.platform_admin_email = "admin@example.com"
+        mock_request.headers = {"X-Authenticated-User": "admin@example.com"}
+        mock_request.cookies = {}
+        mock_request.state = Mock()
+
+        with patch.object(vc, "settings", mock_settings), patch(
+            "mcpgateway.db.get_db"
+        ) as mock_get_db, patch(
+            "mcpgateway.services.email_auth_service.EmailAuthService"
+        ) as mock_auth_service:
+            mock_get_db.return_value = iter([Mock()])
+            mock_auth_service.return_value.get_user_by_email = AsyncMock(return_value=None)
+
+            result = await vc.require_auth_header_first(auth_header=None, jwt_token=None, request=mock_request)
+
+        assert result["is_admin"] is True
+        assert result["teams"] is None
+
 
 class TestRBACProxyAuthentication:
     """Test cases for RBAC middleware proxy authentication functionality."""

tests/unit/mcpgateway/utils/test_verify_credentials.py

@@ -1653,20 +1653,40 @@ async def test_require_auth_header_first_no_request_uses_jwt_token_param(monkeyp
 
 @pytest.mark.asyncio
 async def test_require_auth_header_first_proxy_auth_returns_proxy_user(monkeypatch):
-    """Proxy user is returned when mcp_client_auth_enabled=False and trust_proxy_auth=True."""
+    """Proxy user returns the enriched payload (shared helper with require_auth)."""
     monkeypatch.setattr(vc.settings, "mcp_client_auth_enabled", False, raising=False)
     monkeypatch.setattr(vc.settings, "trust_proxy_auth", True, raising=False)
     monkeypatch.setattr(vc.settings, "trust_proxy_auth_dangerously", True, raising=False)
     monkeypatch.setattr(vc.settings, "proxy_user_header", "x-authenticated-user", raising=False)
     monkeypatch.setattr(vc.settings, "auth_required", True, raising=False)
+    monkeypatch.setattr(vc.settings, "require_user_in_db", True, raising=False)
 
     mock_request = Mock(spec=Request)
     mock_request.headers = {"x-authenticated-user": "proxy-user@example.com"}
     mock_request.cookies = {}
+    mock_request.state = Mock()
+
+    mock_user = Mock()
+    mock_user.is_admin = False
+    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 = ["team1"]
+
+        result = await vc.require_auth_header_first(auth_header=None, jwt_token=None, request=mock_request)
 
-    result = await vc.require_auth_header_first(auth_header=None, jwt_token=None, request=mock_request)
     assert result["sub"] == "proxy-user@example.com"
     assert result["source"] == "proxy"
+    assert result["token"] is None
+    assert result["is_admin"] is False
+    assert result["teams"] == ["team1"]
+    assert result["email"] == "proxy-user@example.com"
 
 
 @pytest.mark.asyncio