feat(#3003): implement DB-first intersection for session token team scoping

Implement the intersection approach for session token team resolution:
DB teams are always resolved first (revoked memberships enforced
immediately), then narrowed to the intersection with any JWT-embedded
teams claim. This delivers the #3003 user story — users can scope a
session to a subset of their memberships — without risking stale grants.

Key changes:
- resolve_session_teams() now applies intersection policy with
  preresolved_db_teams support for the batched queries path
- Add resolve_session_teams_sync() for StreamableHTTP transport
- Route all 6 session-token call sites through the policy point
- Skip _check_team_membership for session tokens (DB resolution
  already validates; raw JWT check conflicts with fallback semantics)
- Cache raw DB teams (not narrowed intersection) to prevent
  cross-session cache poisoning
- Document intersection behavior and empty-intersection fallback

Closes #3003

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

Author: jonpspri

docs/docs/architecture/multitenancy.md

@@ -319,7 +319,7 @@ Token scoping is the mechanism that determines which resources a user can **see*
 **Key functions** in `mcpgateway/auth.py`:
 
 - `normalize_token_teams()` — The **single source of truth** for interpreting JWT team claims into a canonical form. Used directly by API tokens and legacy tokens.
-- `resolve_session_teams()` — The **single policy point** for session-token team resolution. Session tokens (`token_use: "session"`) always resolve teams from the database via `_resolve_teams_from_db()` so that revoked memberships take effect immediately.
+- `resolve_session_teams()` / `resolve_session_teams_sync()` — The **single policy point** for session-token team resolution. Teams are always resolved from the database first so that revoked memberships take effect immediately. If the JWT carries a `teams` claim, the result is narrowed to the intersection of DB teams and JWT teams — letting callers scope a session to a subset of their memberships without risking stale grants. If the intersection is empty (e.g. all JWT-claimed teams have been revoked), the full DB team list is returned as a fallback so the user is not locked out.
 - API tokens and legacy tokens always use `normalize_token_teams()` directly.
 
 ### Secure-First Defaults

docs/docs/manage/rbac.md

@@ -138,7 +138,7 @@ Protected entities:
 
 ## Token Scoping Model
 
-Token scoping controls what resources a token can access based on the `teams` claim in the JWT payload. The `normalize_token_teams()` function is the **single source of truth** for interpreting JWT team claims into a canonical form. For session tokens, `resolve_session_teams()` is the **single policy point** — it always resolves teams from the database so that membership changes take effect immediately.
+Token scoping controls what resources a token can access based on the `teams` claim in the JWT payload. The `normalize_token_teams()` function is the **single source of truth** for interpreting JWT team claims into a canonical form. For session tokens, `resolve_session_teams()` is the **single policy point** — it resolves teams from the database first (so revoked memberships take effect immediately), then narrows the result to the intersection with any JWT-embedded `teams` claim (so callers can scope a session to a subset of their memberships). If the intersection is empty (e.g. all JWT-claimed teams have been revoked), the full DB team list is returned as a fallback.
 
 ### Token Scoping Contract
 

mcpgateway/auth.py

@@ -257,6 +257,37 @@ def _resolve_teams_from_db_sync(email: str, is_admin: bool) -> Optional[List[str
     return team_ids
 
 
+def resolve_session_teams_sync(payload: Dict[str, Any], email: str, is_admin: bool) -> Optional[List[str]]:
+    """Synchronous companion to :func:`resolve_session_teams`.
+
+    Used by the StreamableHTTP transport which runs in a sync context.
+    Applies the same DB-first + JWT intersection policy.
+
+    Args:
+        payload: The decoded JWT payload dict.
+        email: User email address (for the DB lookup).
+        is_admin: Whether the user is an admin.
+
+    Returns:
+        None (admin bypass), [] (public-only), or list of team ID strings.
+    """
+    db_teams = _resolve_teams_from_db_sync(email, is_admin)
+
+    # Admin bypass — no narrowing possible
+    if db_teams is None:
+        return None
+
+    # Narrow to intersection when the JWT carries an explicit teams claim
+    jwt_teams = payload.get("teams")
+    if isinstance(jwt_teams, list) and jwt_teams:
+        jwt_team_set = set(normalize_token_teams({"teams": jwt_teams}) or [])
+        narrowed = [t for t in db_teams if t in jwt_team_set]
+        if narrowed:
+            return narrowed
+
+    return db_teams
+
+
 async def _resolve_teams_from_db(email: str, user_info) -> Optional[List[str]]:
     """Resolve teams for session tokens from DB/cache.
 
@@ -300,25 +331,67 @@ async def _resolve_teams_from_db(email: str, user_info) -> Optional[List[str]]:
     return team_ids
 
 
-async def resolve_session_teams(payload: Dict[str, Any], email: str, user_info) -> Optional[List[str]]:  # pylint: disable=unused-argument
-    """Resolve teams for a session token from the database.
+_UNSET: Any = object()  # sentinel distinguishing "not supplied" from explicit None
+
+
+async def resolve_session_teams(
+    payload: Dict[str, Any],
+    email: str,
+    user_info,
+    *,
+    preresolved_db_teams: Optional[List[str]] = _UNSET,
+) -> Optional[List[str]]:
+    """Resolve teams for a session token, using DB as the authority.
+
+    The database is always consulted first so that revoked team memberships
+    take effect immediately.  If the JWT carries a ``teams`` claim, the
+    result is narrowed to the **intersection** of the DB teams and the JWT
+    teams — this lets callers scope a session to a subset of their actual
+    memberships (e.g. single-team mode) without risking stale grants.
+
+    This is the **single policy point** for session-token team resolution.
+    All code paths that need teams for a session token should call this
+    function rather than inlining the decision.
 
-    Session tokens always resolve team membership from the DB/cache so that
-    revoked memberships take effect immediately rather than persisting until
-    token expiry.  This is the **single policy point** for session-token team
-    resolution — all code paths that need teams for a session token should call
-    this function rather than inlining the decision.
+    Policy:
+        1. Resolve teams from DB/cache (``_resolve_teams_from_db``), or
+           use *preresolved_db_teams* when the caller already fetched them
+           (e.g. via a batched query).
+        2. If DB returns ``None`` (admin bypass), return ``None``.
+        3. If the JWT ``teams`` claim is a non-empty list, intersect with
+           DB teams.  If the intersection is non-empty, return it;
+           otherwise fall back to the full DB result.
+        4. Otherwise return the full DB result.
 
     Args:
-        payload: The decoded JWT payload dict (unused today, reserved for
-            future policy checks).
+        payload: The decoded JWT payload dict.
         email: User email address (for the DB lookup).
         user_info: User dict or EmailUser instance (for admin detection).
+        preresolved_db_teams: If the caller already resolved DB teams (e.g.
+            from a batched query), pass them here to skip the DB call.
+            Pass ``None`` to indicate admin bypass was already determined.
 
     Returns:
         None (admin bypass), [] (public-only), or list of team ID strings.
     """
-    return await _resolve_teams_from_db(email, user_info)
+    if preresolved_db_teams is not _UNSET:
+        db_teams: Optional[List[str]] = preresolved_db_teams
+    else:
+        db_teams = await _resolve_teams_from_db(email, user_info)
+
+    # Admin bypass — no narrowing possible
+    if db_teams is None:
+        return None
+
+    # Narrow to intersection when the JWT carries an explicit teams claim
+    jwt_teams = payload.get("teams")
+    if isinstance(jwt_teams, list) and jwt_teams:
+        jwt_team_set = set(normalize_token_teams({"teams": jwt_teams}) or [])
+        narrowed = [t for t in db_teams if t in jwt_team_set]
+        if narrowed:
+            return narrowed
+
+    return db_teams
 
 
 def normalize_token_teams(payload: Dict[str, Any]) -> Optional[List[str]]:
@@ -1183,13 +1256,11 @@ async def _set_auth_method_from_payload(payload: dict) -> None:
                 # Resolve teams based on token_use
                 token_use = payload.get("token_use")
                 if token_use == "session":  # nosec B105 - Not a password; token_use is a JWT claim type
-                    # Session token: use team_ids from batched query
+                    # Session token: use team_ids from batched query via resolve_session_teams
                     user_dict = auth_ctx.get("user")
                     is_admin = user_dict.get("is_admin", False) if user_dict else False
-                    if is_admin:
-                        teams = None  # Admin bypass
-                    else:
-                        teams = auth_ctx.get("team_ids", [])
+                    batch_teams = None if is_admin else auth_ctx.get("team_ids", [])
+                    teams = await resolve_session_teams(payload, email, {"is_admin": is_admin}, preresolved_db_teams=batch_teams)
                 else:
                     # API token or legacy: use embedded teams
                     teams = normalize_token_teams(payload)
@@ -1225,8 +1296,11 @@ async def _set_auth_method_from_payload(payload: dict) -> None:
                         )
                         # Also populate teams-list cache so cached-path requests
                         # don't need an extra DB query via _resolve_teams_from_db()
-                        if token_use == "session" and teams is not None:  # nosec B105
-                            await auth_cache.set_user_teams(f"{email}:True", teams)
+                        # Cache the raw DB teams (batch_teams), not the narrowed
+                        # intersection (teams), so that other sessions for the same
+                        # user see the full membership and can narrow independently.
+                        if token_use == "session" and batch_teams is not None:  # nosec B105
+                            await auth_cache.set_user_teams(f"{email}:True", batch_teams)
                     except Exception as cache_set_error:
                         logger.debug(f"Failed to cache auth context: {cache_set_error}")
 

mcpgateway/main.py

@@ -66,7 +66,7 @@
 # First-Party
 from mcpgateway import __version__
 from mcpgateway.admin import admin_router, set_logging_service
-from mcpgateway.auth import _check_token_revoked_sync, _lookup_api_token_sync, _resolve_teams_from_db, get_current_user, get_user_team_roles, normalize_token_teams
+from mcpgateway.auth import _check_token_revoked_sync, _lookup_api_token_sync, get_current_user, get_user_team_roles, normalize_token_teams, resolve_session_teams
 from mcpgateway.bootstrap_db import main as bootstrap_db
 from mcpgateway.cache import ResourceCache, SessionRegistry
 from mcpgateway.common.models import InitializeResult
@@ -1760,7 +1760,7 @@ async def dispatch(self, request: Request, call_next):  # pylint: disable=too-ma
                     token_use = payload.get("token_use")
                     if token_use == "session":  # nosec B105 - Not a password; token_use is a JWT claim type
                         is_admin = payload.get("is_admin", False) or payload.get("user", {}).get("is_admin", False)
-                        token_teams = await _resolve_teams_from_db(username, {"is_admin": is_admin})
+                        token_teams = await resolve_session_teams(payload, username, {"is_admin": is_admin})
                     else:
                         # API token or legacy path: embedded teams claim semantics
                         token_teams = normalize_token_teams(payload)

mcpgateway/middleware/token_scoping.py

@@ -1199,8 +1199,13 @@ async def __call__(self, request: Request, call_next):
 
                 db = next(get_db())
                 try:
-                    # Check team membership with shared session
-                    if not self._check_team_membership(payload, db=db):
+                    # Check team membership — only for API/legacy tokens whose teams
+                    # come from JWT claims and may be stale.  Session tokens skip this
+                    # because resolve_session_teams() already resolved membership from
+                    # the DB; re-checking the raw JWT claim here would conflict with
+                    # the intersection-fallback semantics (stale JWT teams would cause
+                    # a 403 even though the user has valid DB teams).
+                    if token_use != "session" and not self._check_team_membership(payload, db=db):  # nosec B105 - Not a password; token_use is a JWT claim type
                         logger.warning("Token rejected: User no longer member of associated team(s)")
                         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Token is invalid: User is no longer a member of the associated team")
 

mcpgateway/transports/streamablehttp_transport.py

@@ -1331,16 +1331,14 @@ def _normalize_jwt_payload(payload: dict[str, Any]) -> dict[str, Any]:
 
     token_use = payload.get("token_use")
     if token_use == "session":  # nosec B105 - Not a password; token_use is a JWT claim type
-        # Session token: resolve teams from DB/cache
-        if is_admin:
-            final_teams = None  # Admin bypass
-        elif email:
+        # Session token: resolve teams from DB/cache via single policy point
+        if email:
             # First-Party
-            from mcpgateway.auth import _resolve_teams_from_db_sync  # pylint: disable=import-outside-toplevel
+            from mcpgateway.auth import resolve_session_teams_sync  # pylint: disable=import-outside-toplevel
 
-            final_teams = _resolve_teams_from_db_sync(email, is_admin=False)
+            final_teams = resolve_session_teams_sync(payload, email, is_admin)
         else:
-            final_teams = []  # No email — public-only
+            final_teams = None if is_admin else []  # No email — admin bypass or public-only
     else:
         # API token or legacy: use embedded teams from JWT
         # First-Party
@@ -2815,19 +2813,16 @@ async def _auth_jwt(self, *, token: str) -> bool:
             # Resolve teams based on token_use claim
             token_use = user_payload.get("token_use")
             if token_use == "session":  # nosec B105 - Not a password; token_use is a JWT claim type
-                # Session token: resolve teams from DB/cache
+                # Session token: resolve teams from DB/cache via single policy point
                 user_email_for_teams = user_payload.get("sub") or user_payload.get("email")
                 is_admin_flag = user_payload.get("is_admin", False) or user_payload.get("user", {}).get("is_admin", False)
-                if is_admin_flag:
-                    final_teams = None  # Admin bypass
-                elif user_email_for_teams:
-                    # Resolve teams synchronously with L1 cache (StreamableHTTP uses sync context)
+                if user_email_for_teams:
                     # First-Party
-                    from mcpgateway.auth import _resolve_teams_from_db_sync  # pylint: disable=import-outside-toplevel
+                    from mcpgateway.auth import resolve_session_teams_sync  # pylint: disable=import-outside-toplevel
 
-                    final_teams = _resolve_teams_from_db_sync(user_email_for_teams, is_admin=False)
+                    final_teams = resolve_session_teams_sync(user_payload, user_email_for_teams, is_admin_flag)
                 else:
-                    final_teams = []  # No email — public-only
+                    final_teams = None if is_admin_flag else []  # No email — admin bypass or public-only
             else:
                 # API token or legacy: use embedded teams from JWT
                 # First-Party

tests/unit/mcpgateway/middleware/test_token_scoping.py

@@ -700,6 +700,85 @@ async def test_legacy_token_without_token_use_uses_embedded_teams(self, middlewa
                             # Verify _resolve_teams_from_db was NOT called
                             mock_resolve_teams.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_session_token_calls_resolve_session_teams(self, middleware, mock_request):
+        """Verify middleware calls the public resolve_session_teams policy point, not _resolve_teams_from_db directly."""
+        mock_request.url.path = "/servers"
+        mock_request.method = "GET"
+        mock_request.headers = {"Authorization": "Bearer session_token"}
+
+        session_payload = {
+            "sub": "user@example.com",
+            "token_use": "session",
+            "teams": ["team-1"],
+            "scopes": {"permissions": ["*"]},
+        }
+
+        with patch.object(middleware, "_extract_token_scopes", return_value=session_payload):
+            with patch("mcpgateway.middleware.token_scoping.resolve_session_teams", new=AsyncMock(return_value=["team-1"])) as mock_resolve:
+                with patch.object(middleware, "_check_resource_team_ownership", return_value=True):
+                    call_next = AsyncMock(return_value="success")
+
+                    result = await middleware(mock_request, call_next)
+
+                    assert result == "success"
+                    mock_resolve.assert_awaited_once_with(session_payload, "user@example.com", {"is_admin": False})
+
+    @pytest.mark.asyncio
+    async def test_session_token_skips_membership_check_on_stale_jwt_teams(self, middleware, mock_request):
+        """Session tokens skip _check_team_membership so stale JWT teams don't cause a spurious 403."""
+        mock_request.url.path = "/servers"
+        mock_request.method = "GET"
+        mock_request.headers = {"Authorization": "Bearer session_token"}
+
+        # JWT claims stale team "revoked-team"; DB only has "db-team"
+        session_payload = {
+            "sub": "user@example.com",
+            "token_use": "session",
+            "teams": ["revoked-team"],
+            "scopes": {"permissions": ["*"]},
+        }
+
+        with patch.object(middleware, "_extract_token_scopes", return_value=session_payload):
+            # resolve_session_teams already handles intersection & fallback
+            with patch("mcpgateway.auth._resolve_teams_from_db", return_value=["db-team"]) as mock_resolve:
+                with patch.object(middleware, "_check_team_membership", return_value=False) as mock_membership:
+                    with patch.object(middleware, "_check_resource_team_ownership", return_value=True):
+                        call_next = AsyncMock(return_value="success")
+
+                        result = await middleware(mock_request, call_next)
+
+                        assert result == "success"
+                        call_next.assert_called_once()
+                        mock_resolve.assert_called_once()
+                        # Session tokens must NOT call _check_team_membership
+                        mock_membership.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_api_token_still_checks_membership(self, middleware, mock_request):
+        """API tokens must still go through _check_team_membership validation."""
+        mock_request.url.path = "/servers"
+        mock_request.method = "GET"
+        mock_request.headers = {"Authorization": "Bearer api_token"}
+
+        api_payload = {
+            "sub": "user@example.com",
+            "token_use": "api",
+            "teams": ["stale-team"],
+            "scopes": {"permissions": ["*"]},
+        }
+
+        with patch.object(middleware, "_extract_token_scopes", return_value=api_payload):
+            with patch("mcpgateway.middleware.token_scoping.normalize_token_teams", return_value=["stale-team"]):
+                with patch.object(middleware, "_check_team_membership", return_value=False) as mock_membership:
+                    call_next = AsyncMock(return_value="success")
+
+                    result = await middleware(mock_request, call_next)
+
+                    # Should be a 403 response, not "success"
+                    assert result != "success"
+                    mock_membership.assert_called_once()
+
     def test_check_team_membership_missing_user_email_denies(self, middleware):
         """Team-scoped tokens without a user email should be rejected."""
         payload = {"teams": ["team-1"]}