fix: resolve proxy auth database lookup for team/admin context

- Add database lookup in proxy authentication to resolve user teams and admin status
- Proxy auth now queries DB via EmailAuthService.get_user_by_email()
- Resolves teams using _resolve_teams_from_db() (same pattern as session tokens)
- Admin users get teams=None (admin bypass), regular users get team list
- Platform admin bootstrap supported when REQUIRE_USER_IN_DB=false
- Cache enriched payload in request.state._jwt_verified_payload for downstream use
- Add comprehensive test coverage for proxy auth scenarios
- Tests cover: normal users, admin users, platform admin bootstrap, user not found

Fixes proxy authentication bug where users only had public-only access
regardless of their actual database permissions.

Signed-off-by: Bogdan-Marius-Catanus <bogdan-marius.catanus@ibm.com>

Author: Bogdan-Marius-Catanus

mcpgateway/utils/verify_credentials.py

@@ -486,7 +486,52 @@ 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:
-                return {"sub": proxy_user, "source": "proxy", "token": None}  # nosec B105 - None is not a password
+                # 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()
             # No proxy header - check auth_required (matches RBAC/WebSocket behavior)
             if settings.auth_required:
                 raise HTTPException(

tests/unit/mcpgateway/utils/test_proxy_auth.py

@@ -43,6 +43,8 @@ class MockSettings:
             proxy_user_header = "X-Authenticated-User"
             require_token_expiration = False
             docs_allow_basic_auth = False
+            require_user_in_db = True
+            platform_admin_email = "admin@example.com"
 
         return MockSettings()
 
@@ -112,10 +114,31 @@ async def test_proxy_auth_with_header(self, mock_settings, mock_request):
         mock_settings.trust_proxy_auth = True
         mock_settings.trust_proxy_auth_dangerously = True
         mock_request.headers = {"X-Authenticated-User": "proxy-user"}
+        mock_request.state = Mock()
+
+        # Mock database user lookup
+        mock_user = Mock()
+        mock_user.is_admin = False
+        mock_user.email = "proxy-user"
 
         with patch.object(vc, "settings", mock_settings):
-            result = await vc.require_auth(mock_request, None, None)
-            assert result == {"sub": "proxy-user", "source": "proxy", "token": None}
+            with patch("mcpgateway.db.get_db") as mock_get_db:
+                with patch("mcpgateway.services.email_auth_service.EmailAuthService") as mock_auth_service:
+                    with patch("mcpgateway.auth._resolve_teams_from_db", new_callable=AsyncMock) as mock_resolve_teams:
+                        # Setup mocks
+                        mock_db = Mock()
+                        mock_get_db.return_value = iter([mock_db])
+                        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(mock_request, None, None)
+
+                        assert result["sub"] == "proxy-user"
+                        assert result["source"] == "proxy"
+                        assert result["token"] is None
+                        assert result["is_admin"] is False
+                        assert result["teams"] == ["team1"]
+                        assert result["email"] == "proxy-user"
 
     @pytest.mark.asyncio
     async def test_proxy_auth_without_header_raises_when_auth_required(self, mock_settings, mock_request):
@@ -153,10 +176,31 @@ async def test_custom_proxy_header(self, mock_settings, mock_request):
         mock_settings.trust_proxy_auth_dangerously = True
         mock_settings.proxy_user_header = "X-Remote-User"
         mock_request.headers = {"X-Remote-User": "custom-user"}
+        mock_request.state = Mock()
+
+        # Mock database user lookup
+        mock_user = Mock()
+        mock_user.is_admin = False
+        mock_user.email = "custom-user"
 
         with patch.object(vc, "settings", mock_settings):
-            result = await vc.require_auth(mock_request, None, None)
-            assert result == {"sub": "custom-user", "source": "proxy", "token": None}
+            with patch("mcpgateway.db.get_db") as mock_get_db:
+                with patch("mcpgateway.services.email_auth_service.EmailAuthService") as mock_auth_service:
+                    with patch("mcpgateway.auth._resolve_teams_from_db", new_callable=AsyncMock) as mock_resolve_teams:
+                        # Setup mocks
+                        mock_db = Mock()
+                        mock_get_db.return_value = iter([mock_db])
+                        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(mock_request, None, None)
+
+                        assert result["sub"] == "custom-user"
+                        assert result["source"] == "proxy"
+                        assert result["token"] is None
+                        assert result["is_admin"] is False
+                        assert result["teams"] == ["team1"]
+                        assert result["email"] == "custom-user"
 
     @pytest.mark.asyncio
     async def test_jwt_auth_with_proxy_enabled(self, mock_settings, mock_request):
@@ -192,15 +236,90 @@ async def test_mixed_auth_scenario(self, mock_settings, mock_request):
         mock_settings.trust_proxy_auth = True
         mock_settings.trust_proxy_auth_dangerously = True
         mock_request.headers = {"X-Authenticated-User": "proxy-user"}
+        mock_request.state = Mock()
 
         # Create a valid JWT token
         token = jwt.encode({"sub": "jwt-user"}, mock_settings.jwt_secret_key, algorithm="HS256")
         creds = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
 
+        # Mock database user lookup
+        mock_user = Mock()
+        mock_user.is_admin = False
+        mock_user.email = "proxy-user"
+
+        with patch.object(vc, "settings", mock_settings):
+            with patch("mcpgateway.db.get_db") as mock_get_db:
+                with patch("mcpgateway.services.email_auth_service.EmailAuthService") as mock_auth_service:
+                    with patch("mcpgateway.auth._resolve_teams_from_db", new_callable=AsyncMock) as mock_resolve_teams:
+                        # Setup mocks
+                        mock_db = Mock()
+                        mock_get_db.return_value = iter([mock_db])
+                        mock_auth_service.return_value.get_user_by_email = AsyncMock(return_value=mock_user)
+                        mock_resolve_teams.return_value = ["team1"]
+
+                        # When MCP client auth is disabled, proxy takes precedence
+                        result = await vc.require_auth(mock_request, creds, None)
+
+                        assert result["sub"] == "proxy-user"
+                        assert result["source"] == "proxy"
+                        assert result["token"] is None
+                        assert result["is_admin"] is False
+                        assert result["teams"] == ["team1"]
+                        assert result["email"] == "proxy-user"
+
+    @pytest.mark.asyncio
+    async def test_proxy_auth_platform_admin_bootstrap(self, mock_settings, mock_request):
+        """Test proxy authentication with platform admin bootstrap when user not in DB."""
+        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.state = Mock()
+
+        with patch.object(vc, "settings", mock_settings):
+            with patch("mcpgateway.db.get_db") as mock_get_db:
+                with patch("mcpgateway.services.email_auth_service.EmailAuthService") as mock_auth_service:
+                    # Setup mocks - user NOT found in DB
+                    mock_db = Mock()
+                    mock_get_db.return_value = iter([mock_db])
+                    mock_auth_service.return_value.get_user_by_email = AsyncMock(return_value=None)
+
+                    result = await vc.require_auth(mock_request, None, None)
+
+                    # Should bootstrap platform admin
+                    assert result["sub"] == "admin@example.com"
+                    assert result["source"] == "proxy"
+                    assert result["token"] is None
+                    assert result["is_admin"] is True
+                    assert result["teams"] is None  # Admin bypass
+                    assert result["email"] == "admin@example.com"
+
+    @pytest.mark.asyncio
+    async def test_proxy_auth_user_not_found_raises_401(self, mock_settings, mock_request):
+        """Test proxy authentication raises 401 when user not in DB and not platform admin."""
+        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 = True
+        mock_settings.platform_admin_email = "admin@example.com"
+        mock_request.headers = {"X-Authenticated-User": "unknown-user@example.com"}
+        mock_request.state = Mock()
+
         with patch.object(vc, "settings", mock_settings):
-            # When MCP client auth is disabled, proxy takes precedence
-            result = await vc.require_auth(mock_request, creds, None)
-            assert result == {"sub": "proxy-user", "source": "proxy", "token": None}
+            with patch("mcpgateway.db.get_db") as mock_get_db:
+                with patch("mcpgateway.services.email_auth_service.EmailAuthService") as mock_auth_service:
+                    # Setup mocks - user NOT found in DB
+                    mock_db = Mock()
+                    mock_get_db.return_value = iter([mock_db])
+                    mock_auth_service.return_value.get_user_by_email = AsyncMock(return_value=None)
+
+                    with pytest.raises(HTTPException) as exc_info:
+                        await vc.require_auth(mock_request, None, None)
+
+                    assert exc_info.value.status_code == 401
+                    assert "User not found in database" in exc_info.value.detail
 
 
 class TestRBACProxyAuthentication: