feat: option to add upstream claims to the FastMCP proxy JWT (#2997)

src/fastmcp/server/auth/jwt_issuer.py

@@ -103,6 +103,7 @@ class JWTIssuer:
         scopes: list[str],
         jti: str,
         expires_in: int = 3600,
+        upstream_claims: dict[str, Any] | None = None,
     ) -> str:
         """Issue a minimal FastMCP access token.
 
@@ -115,6 +116,7 @@ class JWTIssuer:
             scopes: Token scopes
             jti: Unique token identifier (maps to upstream token)
             expires_in: Token lifetime in seconds
+            upstream_claims: Optional claims from upstream IdP token to include
 
         Returns:
             Signed JWT token
@@ -122,7 +124,7 @@ class JWTIssuer:
         now = int(time.time())
 
         header = {"alg": "HS256", "typ": "JWT"}
-        payload = {
+        payload: dict[str, Any] = {
             "iss": self.issuer,
             "aud": self.audience,
             "client_id": client_id,
@@ -132,6 +134,9 @@ class JWTIssuer:
             "jti": jti,
         }
 
+        if upstream_claims:
+            payload["upstream_claims"] = upstream_claims
+
         token_bytes = self._jwt.encode(header, payload, self._signing_key)
         token = token_bytes.decode("utf-8")
 
@@ -150,6 +155,7 @@ class JWTIssuer:
         scopes: list[str],
         jti: str,
         expires_in: int,
+        upstream_claims: dict[str, Any] | None = None,
     ) -> str:
         """Issue a minimal FastMCP refresh token.
 
@@ -162,6 +168,7 @@ class JWTIssuer:
             scopes: Token scopes
             jti: Unique token identifier (maps to upstream token)
             expires_in: Token lifetime in seconds (should match upstream refresh expiry)
+            upstream_claims: Optional claims from upstream IdP token to include
 
         Returns:
             Signed JWT token
@@ -169,7 +176,7 @@ class JWTIssuer:
         now = int(time.time())
 
         header = {"alg": "HS256", "typ": "JWT"}
-        payload = {
+        payload: dict[str, Any] = {
             "iss": self.issuer,
             "aud": self.audience,
             "client_id": client_id,
@@ -180,6 +187,9 @@ class JWTIssuer:
             "token_use": "refresh",
         }
 
+        if upstream_claims:
+            payload["upstream_claims"] = upstream_claims
+
         token_bytes = self._jwt.encode(header, payload, self._signing_key)
         token = token_bytes.decode("utf-8")
 

src/fastmcp/server/auth/oauth_proxy/proxy.py

@@ -898,6 +898,9 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
         )
         logger.debug("Stored encrypted upstream tokens (jti=%s)", access_jti[:8])
 
+        # Extract upstream claims to embed in FastMCP JWT (if subclass implements)
+        upstream_claims = await self._extract_upstream_claims(idp_tokens)
+
         # Issue minimal FastMCP access token (just a reference via JTI)
         if client.client_id is None:
             raise TokenError("invalid_client", "Client ID is required")
@@ -906,6 +909,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
             scopes=authorization_code.scopes,
             jti=access_jti,
             expires_in=expires_in,
+            upstream_claims=upstream_claims,
         )
 
         # Issue minimal FastMCP refresh token if upstream provided one
@@ -917,6 +921,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
                 scopes=authorization_code.scopes,
                 jti=refresh_jti,
                 expires_in=refresh_expires_in,
+                upstream_claims=upstream_claims,
             )
 
         # Store JTI mappings
@@ -990,6 +995,33 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
         """
         return scopes
 
+    async def _extract_upstream_claims(
+        self, idp_tokens: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        """Extract upstream claims to embed in FastMCP JWT.
+
+        Override this method to decode upstream tokens, call userinfo endpoints,
+        or otherwise extract claims that should be embedded in the FastMCP JWT
+        issued to MCP clients. This enables gateways to inspect upstream identity
+        information by decoding the JWT without server-side storage lookups.
+
+        Args:
+            idp_tokens: Full token response from upstream provider. Contains
+                access_token, and for OIDC providers may include id_token,
+                refresh_token, and other response fields.
+
+        Returns:
+            Dict of claims to embed in JWT under the "upstream_claims" key,
+            or None to not embed any upstream claims.
+
+        Example:
+            For Azure/Entra ID, you might decode the access_token JWT and
+            extract claims like sub, oid, name, preferred_username, email,
+            roles, and groups.
+        """
+        _ = idp_tokens
+        return None
+
     async def load_refresh_token(
         self,
         client: OAuthClientInformationFull,
@@ -1156,6 +1188,11 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
             ),  # Keep until longest-lived token expires (min 1s for safety)
         )
 
+        # Re-extract upstream claims from refreshed token response
+        upstream_claims = await self._extract_upstream_claims(
+            upstream_token_set.raw_token_data
+        )
+
         # Issue new minimal FastMCP access token (just a reference via JTI)
         if client.client_id is None:
             raise TokenError("invalid_client", "Client ID is required")
@@ -1165,6 +1202,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
             scopes=scopes,
             jti=new_access_jti,
             expires_in=new_expires_in,
+            upstream_claims=upstream_claims,
         )
 
         # Store new access token JTI mapping
@@ -1187,6 +1225,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
             jti=new_refresh_jti,
             expires_in=new_refresh_expires_in
             or 60 * 60 * 24 * 30,  # Fallback to 30 days
+            upstream_claims=upstream_claims,
         )
 
         # Store new refresh token JTI mapping with aligned expiry

src/fastmcp/server/auth/providers/azure.py

@@ -12,7 +12,7 @@ from key_value.aio.protocols import AsyncKeyValue
 
 from fastmcp.server.auth.oauth_proxy import OAuthProxy
 from fastmcp.server.auth.providers.jwt import JWTVerifier
-from fastmcp.utilities.auth import parse_scopes
+from fastmcp.utilities.auth import decode_jwt_payload, parse_scopes
 from fastmcp.utilities.logging import get_logger
 
 if TYPE_CHECKING:
@@ -356,3 +356,75 @@ class AzureProvider(OAuthProxy):
 
         logger.debug("Scopes for Azure token endpoint: %s", deduplicated_scopes)
         return deduplicated_scopes
+
+    async def _extract_upstream_claims(
+        self, idp_tokens: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        """Extract claims from Azure token response to embed in FastMCP JWT.
+
+        Decodes the Azure access token (which is a JWT) to extract user identity
+        claims. This allows gateways to inspect upstream identity information by
+        decoding the FastMCP JWT without needing server-side storage lookups.
+
+        Azure access tokens contain claims like:
+        - sub: Subject identifier (unique per user per application)
+        - oid: Object ID (unique user identifier across Azure AD)
+        - tid: Tenant ID
+        - azp: Authorized party (client ID that requested the token)
+        - name: Display name
+        - given_name: First name
+        - family_name: Last name
+        - preferred_username: User principal name (email format)
+        - upn: User Principal Name
+        - email: Email address (if available)
+        - roles: Application roles assigned to the user
+        - groups: Group memberships (if configured)
+
+        Args:
+            idp_tokens: Full token response from Azure, containing access_token
+                and potentially id_token.
+
+        Returns:
+            Dict of extracted claims, or None if extraction fails.
+        """
+        access_token = idp_tokens.get("access_token")
+        if not access_token:
+            return None
+
+        try:
+            # Azure access tokens are JWTs - decode without verification
+            # (already validated by token_verifier during token exchange)
+            payload = decode_jwt_payload(access_token)
+
+            # Extract useful identity claims
+            claims: dict[str, Any] = {}
+            claim_keys = [
+                "sub",
+                "oid",
+                "tid",
+                "azp",
+                "name",
+                "given_name",
+                "family_name",
+                "preferred_username",
+                "upn",
+                "email",
+                "roles",
+                "groups",
+            ]
+            for claim in claim_keys:
+                if claim in payload:
+                    claims[claim] = payload[claim]
+
+            if claims:
+                logger.debug(
+                    "Extracted %d Azure claims for embedding in FastMCP JWT",
+                    len(claims),
+                )
+                return claims
+
+            return None
+
+        except Exception as e:
+            logger.debug("Failed to extract Azure claims: %s", e)
+            return None

src/fastmcp/server/auth/providers/jwt.py

@@ -15,7 +15,7 @@ from pydantic import AnyHttpUrl, SecretStr
 from typing_extensions import TypedDict
 
 from fastmcp.server.auth import AccessToken, TokenVerifier
-from fastmcp.utilities.auth import parse_scopes
+from fastmcp.utilities.auth import decode_jwt_header, parse_scopes
 from fastmcp.utilities.logging import get_logger
 
 logger = get_logger(__name__)
@@ -235,14 +235,8 @@ class JWTVerifier(TokenVerifier):
 
         # Extract kid from token header for JWKS lookup
         try:
-            import base64
-            import json
-
-            header_b64 = token.split(".")[0]
-            header_b64 += "=" * (4 - len(header_b64) % 4)  # Add padding
-            header = json.loads(base64.urlsafe_b64decode(header_b64))
+            header = decode_jwt_header(token)
             kid = header.get("kid")
-
             return await self._get_jwks_key(kid)
 
         except Exception as e:

src/fastmcp/utilities/auth.py

@@ -2,10 +2,67 @@
 
 from __future__ import annotations
 
+import base64
 import json
 from typing import Any
 
 
+def _decode_jwt_part(token: str, part_index: int) -> dict[str, Any]:
+    """Decode a JWT part (header or payload) without signature verification.
+
+    Args:
+        token: JWT token string (header.payload.signature)
+        part_index: 0 for header, 1 for payload
+
+    Returns:
+        Decoded part as a dictionary
+
+    Raises:
+        ValueError: If token is not a valid JWT format
+    """
+    parts = token.split(".")
+    if len(parts) != 3:
+        raise ValueError("Invalid JWT format (expected 3 parts)")
+
+    part_b64 = parts[part_index]
+    part_b64 += "=" * (-len(part_b64) % 4)  # Add padding
+    return json.loads(base64.urlsafe_b64decode(part_b64))
+
+
+def decode_jwt_header(token: str) -> dict[str, Any]:
+    """Decode JWT header without signature verification.
+
+    Useful for extracting the key ID (kid) for JWKS lookup.
+
+    Args:
+        token: JWT token string (header.payload.signature)
+
+    Returns:
+        Decoded header as a dictionary
+
+    Raises:
+        ValueError: If token is not a valid JWT format
+    """
+    return _decode_jwt_part(token, 0)
+
+
+def decode_jwt_payload(token: str) -> dict[str, Any]:
+    """Decode JWT payload without signature verification.
+
+    Use only for tokens received directly from trusted sources (e.g., IdP token endpoints).
+
+    Args:
+        token: JWT token string (header.payload.signature)
+
+    Returns:
+        Decoded payload as a dictionary
+
+    Raises:
+        ValueError: If token is not a valid JWT format
+    """
+    return _decode_jwt_part(token, 1)
+
+
 def parse_scopes(value: Any) -> list[str] | None:
     """Parse scopes from environment variables or settings values.
 

tests/server/auth/providers/test_azure.py

@@ -807,3 +807,244 @@ class TestOIDCScopeHandling:
         assert "profile" in result
         assert "api://my-api/openid" not in result
         assert "api://my-api/profile" not in result
+
+
+class TestAzureExtractUpstreamClaims:
+    """Tests for Azure provider's _extract_upstream_claims method."""
+
+    @staticmethod
+    def create_test_jwt(claims: dict) -> str:
+        """Create a test JWT token with the given claims."""
+        import base64
+        import json
+
+        header = base64.urlsafe_b64encode(
+            json.dumps({"alg": "RS256", "typ": "JWT"}).encode()
+        ).rstrip(b"=")
+        payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).rstrip(b"=")
+        signature = base64.urlsafe_b64encode(b"fake-signature").rstrip(b"=")
+        return f"{header.decode()}.{payload.decode()}.{signature.decode()}"
+
+    async def test_extract_claims_from_azure_jwt(self):
+        """Test that Azure identity claims are extracted from access token."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            base_url="https://myserver.com",
+            required_scopes=["read"],
+            jwt_signing_key="test-secret",
+        )
+
+        azure_jwt = self.create_test_jwt(
+            {
+                "sub": "user-subject-id",
+                "oid": "user-object-id",
+                "tid": "tenant-id-123",
+                "azp": "client-app-id",
+                "name": "Test User",
+                "given_name": "Test",
+                "family_name": "User",
+                "preferred_username": "testuser@example.com",
+                "upn": "testuser@example.com",
+                "email": "test@example.com",
+                "roles": ["Admin", "Reader"],
+                "groups": ["group-1", "group-2"],
+                "exp": 9999999999,
+                "iat": 1234567890,
+                "iss": "https://login.microsoftonline.com/test-tenant/v2.0",
+            }
+        )
+
+        idp_tokens = {
+            "access_token": azure_jwt,
+            "token_type": "Bearer",
+            "expires_in": 3600,
+        }
+
+        claims = await provider._extract_upstream_claims(idp_tokens)
+
+        assert claims is not None
+        assert claims["sub"] == "user-subject-id"
+        assert claims["oid"] == "user-object-id"
+        assert claims["tid"] == "tenant-id-123"
+        assert claims["azp"] == "client-app-id"
+        assert claims["name"] == "Test User"
+        assert claims["given_name"] == "Test"
+        assert claims["family_name"] == "User"
+        assert claims["preferred_username"] == "testuser@example.com"
+        assert claims["upn"] == "testuser@example.com"
+        assert claims["email"] == "test@example.com"
+        assert claims["roles"] == ["Admin", "Reader"]
+        assert claims["groups"] == ["group-1", "group-2"]
+
+    async def test_extract_claims_only_includes_identity_claims(self):
+        """Test that only identity claims are extracted, not all JWT claims."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            base_url="https://myserver.com",
+            required_scopes=["read"],
+            jwt_signing_key="test-secret",
+        )
+
+        azure_jwt = self.create_test_jwt(
+            {
+                "sub": "user-id",
+                "oid": "object-id",
+                "name": "Test User",
+                "exp": 9999999999,
+                "iat": 1234567890,
+                "iss": "https://issuer.example.com",
+                "aud": "test-audience",
+                "nbf": 1234567890,
+                "scp": "read write",
+                "azp": "some-client",
+            }
+        )
+
+        idp_tokens = {"access_token": azure_jwt}
+
+        claims = await provider._extract_upstream_claims(idp_tokens)
+
+        # Only identity claims should be present
+        assert claims is not None
+        assert "sub" in claims
+        assert "oid" in claims
+        assert "name" in claims
+        assert "azp" in claims  # azp is an identity claim we extract
+        # Standard JWT claims should NOT be extracted
+        assert "exp" not in claims
+        assert "iat" not in claims
+        assert "iss" not in claims
+        assert "aud" not in claims
+        assert "nbf" not in claims
+        assert "scp" not in claims
+
+    async def test_extract_claims_returns_none_for_missing_access_token(self):
+        """Test that None is returned when access_token is missing."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            base_url="https://myserver.com",
+            required_scopes=["read"],
+            jwt_signing_key="test-secret",
+        )
+
+        idp_tokens = {"token_type": "Bearer", "expires_in": 3600}
+
+        claims = await provider._extract_upstream_claims(idp_tokens)
+
+        assert claims is None
+
+    async def test_extract_claims_returns_none_for_opaque_token(self):
+        """Test that None is returned for opaque (non-JWT) tokens."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            base_url="https://myserver.com",
+            required_scopes=["read"],
+            jwt_signing_key="test-secret",
+        )
+
+        idp_tokens = {
+            "access_token": "gho_opaque_token_not_a_jwt",  # Not a JWT
+            "token_type": "Bearer",
+        }
+
+        claims = await provider._extract_upstream_claims(idp_tokens)
+
+        assert claims is None
+
+    async def test_extract_claims_returns_none_for_malformed_jwt(self):
+        """Test that None is returned for malformed JWT tokens."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            base_url="https://myserver.com",
+            required_scopes=["read"],
+            jwt_signing_key="test-secret",
+        )
+
+        # Only two parts (missing signature)
+        idp_tokens = {"access_token": "header.payload"}
+
+        claims = await provider._extract_upstream_claims(idp_tokens)
+
+        assert claims is None
+
+    async def test_extract_claims_returns_none_for_invalid_base64(self):
+        """Test that None is returned for JWT with invalid base64."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            base_url="https://myserver.com",
+            required_scopes=["read"],
+            jwt_signing_key="test-secret",
+        )
+
+        # Invalid base64 in payload
+        idp_tokens = {"access_token": "header.not-valid-base64!!!.signature"}
+
+        claims = await provider._extract_upstream_claims(idp_tokens)
+
+        assert claims is None
+
+    async def test_extract_claims_returns_none_for_empty_identity_claims(self):
+        """Test that None is returned when no identity claims are present."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            base_url="https://myserver.com",
+            required_scopes=["read"],
+            jwt_signing_key="test-secret",
+        )
+
+        # JWT with only standard claims, no identity claims
+        azure_jwt = self.create_test_jwt(
+            {
+                "exp": 9999999999,
+                "iat": 1234567890,
+                "iss": "https://issuer.example.com",
+                "aud": "test-audience",
+            }
+        )
+
+        idp_tokens = {"access_token": azure_jwt}
+
+        claims = await provider._extract_upstream_claims(idp_tokens)
+
+        assert claims is None
+
+    async def test_extract_claims_partial_identity_claims(self):
+        """Test extraction when only some identity claims are present."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            base_url="https://myserver.com",
+            required_scopes=["read"],
+            jwt_signing_key="test-secret",
+        )
+
+        # JWT with only sub and name
+        azure_jwt = self.create_test_jwt(
+            {
+                "sub": "user-id",
+                "name": "Test User",
+                "exp": 9999999999,
+            }
+        )
+
+        idp_tokens = {"access_token": azure_jwt}
+
+        claims = await provider._extract_upstream_claims(idp_tokens)
+
+        assert claims is not None
+        assert claims == {"sub": "user-id", "name": "Test User"}

tests/server/auth/test_jwt_issuer.py

@@ -228,3 +228,60 @@ class TestJWTIssuer:
 
         with pytest.raises(JoseError):
             issuer.verify_token("header.payload")  # Missing signature
+
+    def test_issue_access_token_with_upstream_claims(self, issuer):
+        """Test that upstream claims are included when provided."""
+        upstream_claims = {
+            "sub": "user-123",
+            "oid": "object-id-456",
+            "name": "Test User",
+            "email": "test@example.com",
+            "roles": ["Admin", "Reader"],
+        }
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read", "write"],
+            jti="token-id-123",
+            expires_in=3600,
+            upstream_claims=upstream_claims,
+        )
+
+        payload = issuer.verify_token(token)
+        assert "upstream_claims" in payload
+        assert payload["upstream_claims"]["sub"] == "user-123"
+        assert payload["upstream_claims"]["oid"] == "object-id-456"
+        assert payload["upstream_claims"]["name"] == "Test User"
+        assert payload["upstream_claims"]["email"] == "test@example.com"
+        assert payload["upstream_claims"]["roles"] == ["Admin", "Reader"]
+
+    def test_issue_access_token_without_upstream_claims(self, issuer):
+        """Test that upstream_claims is not present when not provided."""
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="token-id-123",
+            expires_in=3600,
+        )
+
+        payload = issuer.verify_token(token)
+        assert "upstream_claims" not in payload
+
+    def test_issue_refresh_token_with_upstream_claims(self, issuer):
+        """Test that refresh tokens also include upstream claims when provided."""
+        upstream_claims = {
+            "sub": "user-123",
+            "name": "Test User",
+        }
+        token = issuer.issue_refresh_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="refresh-token-id",
+            expires_in=60 * 60 * 24 * 30,
+            upstream_claims=upstream_claims,
+        )
+
+        payload = issuer.verify_token(token)
+        assert "upstream_claims" in payload
+        assert payload["upstream_claims"]["sub"] == "user-123"
+        assert payload["upstream_claims"]["name"] == "Test User"
+        assert payload["token_use"] == "refresh"

tests/utilities/test_auth.py

@@ -0,0 +1,177 @@
+"""Tests for authentication utility helpers."""
+
+import base64
+import json
+
+import pytest
+
+from fastmcp.utilities.auth import (
+    decode_jwt_header,
+    decode_jwt_payload,
+    parse_scopes,
+)
+
+
+def create_jwt(header: dict, payload: dict, signature: bytes = b"fake-sig") -> str:
+    """Create a test JWT token."""
+    header_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).rstrip(b"=")
+    payload_b64 = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=")
+    sig_b64 = base64.urlsafe_b64encode(signature).rstrip(b"=")
+    return f"{header_b64.decode()}.{payload_b64.decode()}.{sig_b64.decode()}"
+
+
+class TestDecodeJwtHeader:
+    """Tests for decode_jwt_header utility."""
+
+    def test_decode_basic_header(self):
+        """Test decoding a basic JWT header."""
+        header = {"alg": "RS256", "typ": "JWT"}
+        payload = {"sub": "user-123"}
+        token = create_jwt(header, payload)
+
+        result = decode_jwt_header(token)
+
+        assert result == header
+        assert result["alg"] == "RS256"
+        assert result["typ"] == "JWT"
+
+    def test_decode_header_with_kid(self):
+        """Test decoding header with key ID for JWKS lookup."""
+        header = {"alg": "RS256", "typ": "JWT", "kid": "key-id-123"}
+        payload = {"sub": "user-123"}
+        token = create_jwt(header, payload)
+
+        result = decode_jwt_header(token)
+
+        assert result["kid"] == "key-id-123"
+
+    def test_invalid_jwt_format_two_parts(self):
+        """Test that two-part token raises ValueError."""
+        with pytest.raises(ValueError, match="Invalid JWT format"):
+            decode_jwt_header("header.payload")
+
+    def test_invalid_jwt_format_one_part(self):
+        """Test that single-part token raises ValueError."""
+        with pytest.raises(ValueError, match="Invalid JWT format"):
+            decode_jwt_header("not-a-jwt")
+
+    def test_invalid_jwt_format_four_parts(self):
+        """Test that four-part token raises ValueError."""
+        with pytest.raises(ValueError, match="Invalid JWT format"):
+            decode_jwt_header("a.b.c.d")
+
+    def test_decode_header_length_divisible_by_4(self):
+        """Test decoding when base64 length is divisible by 4 (no padding needed).
+
+        This tests the edge case where len(part) % 4 == 0.
+        The padding calculation (-len % 4) correctly yields 0 in this case.
+        """
+        # Create a header that encodes to exactly 12 chars (divisible by 4)
+        header = {"x": ""}  # eyJ4IjogIiJ9 = 12 chars
+        payload = {"sub": "user"}
+        token = create_jwt(header, payload)
+
+        result = decode_jwt_header(token)
+        assert result == header
+
+
+class TestDecodeJwtPayload:
+    """Tests for decode_jwt_payload utility."""
+
+    def test_decode_basic_payload(self):
+        """Test decoding a basic JWT payload."""
+        header = {"alg": "RS256", "typ": "JWT"}
+        payload = {"sub": "user-123", "name": "Test User"}
+        token = create_jwt(header, payload)
+
+        result = decode_jwt_payload(token)
+
+        assert result == payload
+        assert result["sub"] == "user-123"
+        assert result["name"] == "Test User"
+
+    def test_decode_payload_with_claims(self):
+        """Test decoding payload with various claims."""
+        header = {"alg": "RS256"}
+        payload = {
+            "sub": "user-123",
+            "oid": "object-456",
+            "name": "Test User",
+            "email": "test@example.com",
+            "roles": ["admin", "user"],
+            "exp": 1234567890,
+        }
+        token = create_jwt(header, payload)
+
+        result = decode_jwt_payload(token)
+
+        assert result["sub"] == "user-123"
+        assert result["oid"] == "object-456"
+        assert result["name"] == "Test User"
+        assert result["email"] == "test@example.com"
+        assert result["roles"] == ["admin", "user"]
+        assert result["exp"] == 1234567890
+
+    def test_invalid_jwt_format(self):
+        """Test that invalid format raises ValueError."""
+        with pytest.raises(ValueError, match="Invalid JWT format"):
+            decode_jwt_payload("not-a-jwt")
+
+    def test_decode_payload_with_padding_edge_cases(self):
+        """Test that base64 padding is handled correctly."""
+        # Create payloads of different sizes to test padding
+        for payload_size in [1, 2, 3, 4, 5, 10, 100]:
+            payload = {"data": "x" * payload_size}
+            token = create_jwt({"alg": "RS256"}, payload)
+            result = decode_jwt_payload(token)
+            assert result == payload
+
+    def test_decode_payload_length_divisible_by_4(self):
+        """Test decoding when base64 length is divisible by 4 (no padding needed).
+
+        This tests the edge case where len(part) % 4 == 0.
+        The padding calculation (-len % 4) correctly yields 0 in this case.
+        """
+        # Create a payload that encodes to exactly 12 chars (divisible by 4)
+        header = {"alg": "RS256"}
+        payload = {"x": ""}  # eyJ4IjogIiJ9 = 12 chars
+        token = create_jwt(header, payload)
+
+        result = decode_jwt_payload(token)
+        assert result == payload
+
+
+class TestParseScopes:
+    """Tests for parse_scopes utility."""
+
+    def test_parse_none(self):
+        """Test that None returns None."""
+        assert parse_scopes(None) is None
+
+    def test_parse_empty_string(self):
+        """Test that empty string returns empty list."""
+        assert parse_scopes("") == []
+
+    def test_parse_space_separated(self):
+        """Test parsing space-separated scopes."""
+        assert parse_scopes("read write delete") == ["read", "write", "delete"]
+
+    def test_parse_comma_separated(self):
+        """Test parsing comma-separated scopes."""
+        assert parse_scopes("read,write,delete") == ["read", "write", "delete"]
+
+    def test_parse_json_array(self):
+        """Test parsing JSON array string."""
+        assert parse_scopes('["read", "write", "delete"]') == [
+            "read",
+            "write",
+            "delete",
+        ]
+
+    def test_parse_list(self):
+        """Test that list is returned as-is."""
+        assert parse_scopes(["read", "write"]) == ["read", "write"]
+
+    def test_parse_strips_whitespace(self):
+        """Test that whitespace is stripped."""
+        assert parse_scopes("  read  ,  write  ") == ["read", "write"]