From fa5b136205c3f165fcdb502e83f6a01602cc5652 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jonas=20Kr=C3=BCger=20Svensson?= Date: Wed, 28 Jan 2026 21:57:24 +0100 Subject: [PATCH] feat: option to add upstream claims to the FastMCP proxy JWT (#2997) --- src/fastmcp/server/auth/jwt_issuer.py | 14 +- src/fastmcp/server/auth/oauth_proxy/proxy.py | 39 +++ src/fastmcp/server/auth/providers/azure.py | 74 +++++- src/fastmcp/server/auth/providers/jwt.py | 10 +- src/fastmcp/utilities/auth.py | 57 +++++ tests/server/auth/providers/test_azure.py | 241 +++++++++++++++++++ tests/server/auth/test_jwt_issuer.py | 57 +++++ tests/utilities/test_auth.py | 177 ++++++++++++++ 8 files changed, 658 insertions(+), 11 deletions(-) create mode 100644 tests/utilities/test_auth.py diff --git a/src/fastmcp/server/auth/jwt_issuer.py b/src/fastmcp/server/auth/jwt_issuer.py index 169eca9dc..90d7bc0cf 100644 --- a/src/fastmcp/server/auth/jwt_issuer.py +++ b/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") diff --git a/src/fastmcp/server/auth/oauth_proxy/proxy.py b/src/fastmcp/server/auth/oauth_proxy/proxy.py index 7914009a3..4938932f8 100644 --- a/src/fastmcp/server/auth/oauth_proxy/proxy.py +++ b/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 diff --git a/src/fastmcp/server/auth/providers/azure.py b/src/fastmcp/server/auth/providers/azure.py index ebedd8d0b..e57b9c04a 100644 --- a/src/fastmcp/server/auth/providers/azure.py +++ b/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 diff --git a/src/fastmcp/server/auth/providers/jwt.py b/src/fastmcp/server/auth/providers/jwt.py index 9cafd933e..fd01c2f2c 100644 --- a/src/fastmcp/server/auth/providers/jwt.py +++ b/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: diff --git a/src/fastmcp/utilities/auth.py b/src/fastmcp/utilities/auth.py index bec370ac7..03eb59bb3 100644 --- a/src/fastmcp/utilities/auth.py +++ b/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. diff --git a/tests/server/auth/providers/test_azure.py b/tests/server/auth/providers/test_azure.py index 688d0332b..e3fb13931 100644 --- a/tests/server/auth/providers/test_azure.py +++ b/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"} diff --git a/tests/server/auth/test_jwt_issuer.py b/tests/server/auth/test_jwt_issuer.py index 9b2d5cf32..79bbb02de 100644 --- a/tests/server/auth/test_jwt_issuer.py +++ b/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" diff --git a/tests/utilities/test_auth.py b/tests/utilities/test_auth.py new file mode 100644 index 000000000..fc8378161 --- /dev/null +++ b/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"]