diff --git a/examples/clients/conformance-auth-client/mcp_conformance_auth_client/__init__.py b/examples/clients/conformance-auth-client/mcp_conformance_auth_client/__init__.py index 71edc1c75a..eecd92409a 100644 --- a/examples/clients/conformance-auth-client/mcp_conformance_auth_client/__init__.py +++ b/examples/clients/conformance-auth-client/mcp_conformance_auth_client/__init__.py @@ -7,11 +7,27 @@ fetching the authorization URL and extracting the auth code from the redirect. Usage: - python -m mcp_conformance_auth_client + python -m mcp_conformance_auth_client + +Environment Variables: + MCP_CONFORMANCE_CONTEXT - JSON object containing test credentials: + { + "client_id": "...", + "client_secret": "...", # For client_secret_basic flow + "private_key_pem": "...", # For private_key_jwt flow + "signing_algorithm": "ES256" # Optional, defaults to ES256 + } + +Scenarios: + auth/* - Authorization code flow scenarios (default behavior) + auth/client-credentials-jwt - Client credentials with JWT authentication (SEP-1046) + auth/client-credentials-basic - Client credentials with client_secret_basic """ import asyncio +import json import logging +import os import sys from datetime import timedelta from urllib.parse import ParseResult, parse_qs, urlparse @@ -19,10 +35,30 @@ import httpx from mcp import ClientSession from mcp.client.auth import OAuthClientProvider, TokenStorage +from mcp.client.auth.extensions.client_credentials import ( + ClientCredentialsOAuthProvider, + PrivateKeyJWTOAuthProvider, + SignedJWTParameters, +) from mcp.client.streamable_http import streamablehttp_client from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken from pydantic import AnyUrl + +def get_conformance_context() -> dict: + """Load conformance test context from MCP_CONFORMANCE_CONTEXT environment variable.""" + context_json = os.environ.get("MCP_CONFORMANCE_CONTEXT") + if not context_json: + raise RuntimeError( + "MCP_CONFORMANCE_CONTEXT environment variable not set. " + "Expected JSON with client_id, client_secret, and/or private_key_pem." + ) + try: + return json.loads(context_json) + except json.JSONDecodeError as e: + raise RuntimeError(f"Failed to parse MCP_CONFORMANCE_CONTEXT as JSON: {e}") from e + + # Set up logging to stderr (stdout is for conformance test output) logging.basicConfig( level=logging.DEBUG, @@ -111,17 +147,17 @@ async def handle_callback(self) -> tuple[str, str | None]: return auth_code, state -async def run_client(server_url: str) -> None: +async def run_authorization_code_client(server_url: str) -> None: """ - Run the conformance test client against the given server URL. + Run the conformance test client with authorization code flow. This function: - 1. Connects to the MCP server with OAuth authentication + 1. Connects to the MCP server with OAuth authorization code flow 2. Initializes the session 3. Lists available tools 4. Calls a test tool """ - logger.debug(f"Starting conformance auth client for {server_url}") + logger.debug(f"Starting conformance auth client (authorization_code) for {server_url}") # Create callback handler that will automatically fetch auth codes callback_handler = ConformanceOAuthCallbackHandler() @@ -140,6 +176,89 @@ async def run_client(server_url: str) -> None: callback_handler=callback_handler.handle_callback, ) + await _run_session(server_url, oauth_auth) + + +async def run_client_credentials_jwt_client(server_url: str) -> None: + """ + Run the conformance test client with client credentials flow using private_key_jwt (SEP-1046). + + This function: + 1. Connects to the MCP server with OAuth client_credentials grant + 2. Uses private_key_jwt authentication with credentials from MCP_CONFORMANCE_CONTEXT + 3. Initializes the session + 4. Lists available tools + 5. Calls a test tool + """ + logger.debug(f"Starting conformance auth client (client_credentials_jwt) for {server_url}") + + # Load credentials from environment + context = get_conformance_context() + client_id = context.get("client_id") + private_key_pem = context.get("private_key_pem") + signing_algorithm = context.get("signing_algorithm", "ES256") + + if not client_id: + raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_id'") + if not private_key_pem: + raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'private_key_pem'") + + # Create JWT parameters for SDK-signed assertions + jwt_params = SignedJWTParameters( + issuer=client_id, + subject=client_id, + signing_algorithm=signing_algorithm, + signing_key=private_key_pem, + ) + + # Create OAuth provider for client_credentials with private_key_jwt + oauth_auth = PrivateKeyJWTOAuthProvider( + server_url=server_url, + storage=InMemoryTokenStorage(), + client_id=client_id, + assertion_provider=jwt_params.create_assertion_provider(), + ) + + await _run_session(server_url, oauth_auth) + + +async def run_client_credentials_basic_client(server_url: str) -> None: + """ + Run the conformance test client with client credentials flow using client_secret_basic. + + This function: + 1. Connects to the MCP server with OAuth client_credentials grant + 2. Uses client_secret_basic authentication with credentials from MCP_CONFORMANCE_CONTEXT + 3. Initializes the session + 4. Lists available tools + 5. Calls a test tool + """ + logger.debug(f"Starting conformance auth client (client_credentials_basic) for {server_url}") + + # Load credentials from environment + context = get_conformance_context() + client_id = context.get("client_id") + client_secret = context.get("client_secret") + + if not client_id: + raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_id'") + if not client_secret: + raise RuntimeError("MCP_CONFORMANCE_CONTEXT missing 'client_secret'") + + # Create OAuth provider for client_credentials with client_secret_basic + oauth_auth = ClientCredentialsOAuthProvider( + server_url=server_url, + storage=InMemoryTokenStorage(), + client_id=client_id, + client_secret=client_secret, + token_endpoint_auth_method="client_secret_basic", + ) + + await _run_session(server_url, oauth_auth) + + +async def _run_session(server_url: str, oauth_auth: OAuthClientProvider) -> None: + """Common session logic for all OAuth flows.""" # Connect using streamable HTTP transport with OAuth async with streamablehttp_client( url=server_url, @@ -168,14 +287,26 @@ async def run_client(server_url: str) -> None: def main() -> None: """Main entry point for the conformance auth client.""" - if len(sys.argv) != 2: - print(f"Usage: {sys.argv[0]} ", file=sys.stderr) + if len(sys.argv) != 3: + print(f"Usage: {sys.argv[0]} ", file=sys.stderr) + print("", file=sys.stderr) + print("Scenarios:", file=sys.stderr) + print(" auth/* - Authorization code flow (default)", file=sys.stderr) + print(" auth/client-credentials-jwt - Client credentials with JWT auth (SEP-1046)", file=sys.stderr) + print(" auth/client-credentials-basic - Client credentials with client_secret_basic", file=sys.stderr) sys.exit(1) - server_url = sys.argv[1] + scenario = sys.argv[1] + server_url = sys.argv[2] try: - asyncio.run(run_client(server_url)) + if scenario == "auth/client-credentials-jwt": + asyncio.run(run_client_credentials_jwt_client(server_url)) + elif scenario == "auth/client-credentials-basic": + asyncio.run(run_client_credentials_basic_client(server_url)) + else: + # Default to authorization code flow for all other auth/* scenarios + asyncio.run(run_authorization_code_client(server_url)) except Exception: logger.exception("Client failed") sys.exit(1) diff --git a/src/mcp/client/auth/extensions/client_credentials.py b/src/mcp/client/auth/extensions/client_credentials.py index e96554063d..e2f3f08a4d 100644 --- a/src/mcp/client/auth/extensions/client_credentials.py +++ b/src/mcp/client/auth/extensions/client_credentials.py @@ -1,6 +1,16 @@ +""" +OAuth client credential extensions for MCP. + +Provides OAuth providers for machine-to-machine authentication flows: +- ClientCredentialsOAuthProvider: For client_credentials with client_id + client_secret +- PrivateKeyJWTOAuthProvider: For client_credentials with private_key_jwt authentication + (typically using a pre-built JWT from workload identity federation) +- RFC7523OAuthClientProvider: For jwt-bearer grant (RFC 7523 Section 2.1) +""" + import time from collections.abc import Awaitable, Callable -from typing import Any +from typing import Any, Literal from uuid import uuid4 import httpx @@ -8,7 +18,321 @@ from pydantic import BaseModel, Field from mcp.client.auth import OAuthClientProvider, OAuthFlowError, OAuthTokenError, TokenStorage -from mcp.shared.auth import OAuthClientMetadata +from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata + + +class ClientCredentialsOAuthProvider(OAuthClientProvider): + """OAuth provider for client_credentials grant with client_id + client_secret. + + This provider sets client_info directly, bypassing dynamic client registration. + Use this when you already have client credentials (client_id and client_secret). + + Example: + ```python + provider = ClientCredentialsOAuthProvider( + server_url="https://api.example.com", + storage=my_token_storage, + client_id="my-client-id", + client_secret="my-client-secret", + ) + ``` + """ + + def __init__( + self, + server_url: str, + storage: TokenStorage, + client_id: str, + client_secret: str, + token_endpoint_auth_method: Literal["client_secret_basic", "client_secret_post"] = "client_secret_basic", + scopes: str | None = None, + ) -> None: + """Initialize client_credentials OAuth provider. + + Args: + server_url: The MCP server URL. + storage: Token storage implementation. + client_id: The OAuth client ID. + client_secret: The OAuth client secret. + token_endpoint_auth_method: Authentication method for token endpoint. + Either "client_secret_basic" (default) or "client_secret_post". + scopes: Optional space-separated list of scopes to request. + """ + # Build minimal client_metadata for the base class + client_metadata = OAuthClientMetadata( + redirect_uris=None, + grant_types=["client_credentials"], + token_endpoint_auth_method=token_endpoint_auth_method, + scope=scopes, + ) + super().__init__(server_url, client_metadata, storage, None, None, 300.0) + # Store client_info to be set during _initialize - no dynamic registration needed + self._fixed_client_info = OAuthClientInformationFull( + redirect_uris=None, + client_id=client_id, + client_secret=client_secret, + grant_types=["client_credentials"], + token_endpoint_auth_method=token_endpoint_auth_method, + scope=scopes, + ) + + async def _initialize(self) -> None: + """Load stored tokens and set pre-configured client_info.""" + self.context.current_tokens = await self.context.storage.get_tokens() + self.context.client_info = self._fixed_client_info + self._initialized = True + + async def _perform_authorization(self) -> httpx.Request: + """Perform client_credentials authorization.""" + return await self._exchange_token_client_credentials() + + async def _exchange_token_client_credentials(self) -> httpx.Request: + """Build token exchange request for client_credentials grant.""" + token_data: dict[str, Any] = { + "grant_type": "client_credentials", + } + + headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"} + + # Use standard auth methods (client_secret_basic, client_secret_post, none) + token_data, headers = self.context.prepare_token_auth(token_data, headers) + + if self.context.should_include_resource_param(self.context.protocol_version): + token_data["resource"] = self.context.get_resource_url() + + if self.context.client_metadata.scope: + token_data["scope"] = self.context.client_metadata.scope + + token_url = self._get_token_endpoint() + return httpx.Request("POST", token_url, data=token_data, headers=headers) + + +def static_assertion_provider(token: str) -> Callable[[str], Awaitable[str]]: + """Create an assertion provider that returns a static JWT token. + + Use this when you have a pre-built JWT (e.g., from workload identity federation) + that doesn't need the audience parameter. + + Example: + ```python + provider = PrivateKeyJWTOAuthProvider( + server_url="https://api.example.com", + storage=my_token_storage, + client_id="my-client-id", + assertion_provider=static_assertion_provider(my_prebuilt_jwt), + ) + ``` + + Args: + token: The pre-built JWT assertion string. + + Returns: + An async callback suitable for use as an assertion_provider. + """ + + async def provider(audience: str) -> str: + return token + + return provider + + +class SignedJWTParameters(BaseModel): + """Parameters for creating SDK-signed JWT assertions. + + Use `create_assertion_provider()` to create an assertion provider callback + for use with `PrivateKeyJWTOAuthProvider`. + + Example: + ```python + jwt_params = SignedJWTParameters( + issuer="my-client-id", + subject="my-client-id", + signing_key=private_key_pem, + ) + provider = PrivateKeyJWTOAuthProvider( + server_url="https://api.example.com", + storage=my_token_storage, + client_id="my-client-id", + assertion_provider=jwt_params.create_assertion_provider(), + ) + ``` + """ + + issuer: str = Field(description="Issuer for JWT assertions (typically client_id).") + subject: str = Field(description="Subject identifier for JWT assertions (typically client_id).") + signing_key: str = Field(description="Private key for JWT signing (PEM format).") + signing_algorithm: str = Field(default="RS256", description="Algorithm for signing JWT assertions.") + lifetime_seconds: int = Field(default=300, description="Lifetime of generated JWT in seconds.") + additional_claims: dict[str, Any] | None = Field(default=None, description="Additional claims.") + + def create_assertion_provider(self) -> Callable[[str], Awaitable[str]]: + """Create an assertion provider callback for use with PrivateKeyJWTOAuthProvider. + + Returns: + An async callback that takes the audience (authorization server issuer URL) + and returns a signed JWT assertion. + """ + + async def provider(audience: str) -> str: + now = int(time.time()) + claims: dict[str, Any] = { + "iss": self.issuer, + "sub": self.subject, + "aud": audience, + "exp": now + self.lifetime_seconds, + "iat": now, + "jti": str(uuid4()), + } + if self.additional_claims: + claims.update(self.additional_claims) + + return jwt.encode(claims, self.signing_key, algorithm=self.signing_algorithm) + + return provider + + +class PrivateKeyJWTOAuthProvider(OAuthClientProvider): + """OAuth provider for client_credentials grant with private_key_jwt authentication. + + Uses RFC 7523 Section 2.2 for client authentication via JWT assertion. + + The JWT assertion's audience MUST be the authorization server's issuer identifier + (per RFC 7523bis security updates). The `assertion_provider` callback receives + this audience value and must return a JWT with that audience. + + **Option 1: Pre-built JWT via Workload Identity Federation** + + In production scenarios, the JWT assertion is typically obtained from a workload + identity provider (e.g., GCP, AWS IAM, Azure AD): + + ```python + async def get_workload_identity_token(audience: str) -> str: + # Fetch JWT from your identity provider + # The JWT's audience must match the provided audience parameter + return await fetch_token_from_identity_provider(audience=audience) + + provider = PrivateKeyJWTOAuthProvider( + server_url="https://api.example.com", + storage=my_token_storage, + client_id="my-client-id", + assertion_provider=get_workload_identity_token, + ) + ``` + + **Option 2: Static pre-built JWT** + + If you have a static JWT that doesn't need the audience parameter: + + ```python + provider = PrivateKeyJWTOAuthProvider( + server_url="https://api.example.com", + storage=my_token_storage, + client_id="my-client-id", + assertion_provider=static_assertion_provider(my_prebuilt_jwt), + ) + ``` + + **Option 3: SDK-signed JWT (for testing/simple setups)** + + For testing or simple deployments, use `SignedJWTParameters.create_assertion_provider()`: + + ```python + jwt_params = SignedJWTParameters( + issuer="my-client-id", + subject="my-client-id", + signing_key=private_key_pem, + ) + provider = PrivateKeyJWTOAuthProvider( + server_url="https://api.example.com", + storage=my_token_storage, + client_id="my-client-id", + assertion_provider=jwt_params.create_assertion_provider(), + ) + ``` + """ + + def __init__( + self, + server_url: str, + storage: TokenStorage, + client_id: str, + assertion_provider: Callable[[str], Awaitable[str]], + scopes: str | None = None, + ) -> None: + """Initialize private_key_jwt OAuth provider. + + Args: + server_url: The MCP server URL. + storage: Token storage implementation. + client_id: The OAuth client ID. + assertion_provider: Async callback that takes the audience (authorization + server's issuer identifier) and returns a JWT assertion. Use + `SignedJWTParameters.create_assertion_provider()` for SDK-signed JWTs, + `static_assertion_provider()` for pre-built JWTs, or provide your own + callback for workload identity federation. + scopes: Optional space-separated list of scopes to request. + """ + # Build minimal client_metadata for the base class + client_metadata = OAuthClientMetadata( + redirect_uris=None, + grant_types=["client_credentials"], + token_endpoint_auth_method="private_key_jwt", + scope=scopes, + ) + super().__init__(server_url, client_metadata, storage, None, None, 300.0) + self._assertion_provider = assertion_provider + # Store client_info to be set during _initialize - no dynamic registration needed + self._fixed_client_info = OAuthClientInformationFull( + redirect_uris=None, + client_id=client_id, + grant_types=["client_credentials"], + token_endpoint_auth_method="private_key_jwt", + scope=scopes, + ) + + async def _initialize(self) -> None: + """Load stored tokens and set pre-configured client_info.""" + self.context.current_tokens = await self.context.storage.get_tokens() + self.context.client_info = self._fixed_client_info + self._initialized = True + + async def _perform_authorization(self) -> httpx.Request: + """Perform client_credentials authorization with private_key_jwt.""" + return await self._exchange_token_client_credentials() + + async def _add_client_authentication_jwt(self, *, token_data: dict[str, Any]) -> None: + """Add JWT assertion for client authentication to token endpoint parameters.""" + if not self.context.oauth_metadata: + raise OAuthFlowError("Missing OAuth metadata for private_key_jwt flow") # pragma: no cover + + # Audience MUST be the issuer identifier of the authorization server + # https://datatracker.ietf.org/doc/html/draft-ietf-oauth-rfc7523bis-01 + audience = str(self.context.oauth_metadata.issuer) + assertion = await self._assertion_provider(audience) + + # RFC 7523 Section 2.2: client authentication via JWT + token_data["client_assertion"] = assertion + token_data["client_assertion_type"] = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + + async def _exchange_token_client_credentials(self) -> httpx.Request: + """Build token exchange request for client_credentials grant with private_key_jwt.""" + token_data: dict[str, Any] = { + "grant_type": "client_credentials", + } + + headers: dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"} + + # Add JWT client authentication (RFC 7523 Section 2.2) + await self._add_client_authentication_jwt(token_data=token_data) + + if self.context.should_include_resource_param(self.context.protocol_version): + token_data["resource"] = self.context.get_resource_url() + + if self.context.client_metadata.scope: + token_data["scope"] = self.context.client_metadata.scope + + token_url = self._get_token_endpoint() + return httpx.Request("POST", token_url, data=token_data, headers=headers) class JWTParameters(BaseModel): @@ -64,9 +388,16 @@ def to_assertion(self, with_audience_fallback: str | None = None) -> str: class RFC7523OAuthClientProvider(OAuthClientProvider): - """OAuth client provider for RFC7532 clients.""" + """OAuth client provider for RFC 7523 jwt-bearer grant. - jwt_parameters: JWTParameters | None = None + .. deprecated:: + Use :class:`ClientCredentialsOAuthProvider` for client_credentials with + client_id + client_secret, or :class:`PrivateKeyJWTOAuthProvider` for + client_credentials with private_key_jwt authentication instead. + + This provider supports the jwt-bearer authorization grant (RFC 7523 Section 2.1) + where the JWT itself is the authorization grant. + """ def __init__( self, @@ -78,6 +409,14 @@ def __init__( timeout: float = 300.0, jwt_parameters: JWTParameters | None = None, ) -> None: + import warnings + + warnings.warn( + "RFC7523OAuthClientProvider is deprecated. Use ClientCredentialsOAuthProvider " + "or PrivateKeyJWTOAuthProvider instead.", + DeprecationWarning, + stacklevel=2, + ) super().__init__(server_url, client_metadata, storage, redirect_handler, callback_handler, timeout) self.jwt_parameters = jwt_parameters diff --git a/tests/client/auth/extensions/test_client_credentials.py b/tests/client/auth/extensions/test_client_credentials.py index 15fb9152ad..6d134af742 100644 --- a/tests/client/auth/extensions/test_client_credentials.py +++ b/tests/client/auth/extensions/test_client_credentials.py @@ -4,7 +4,14 @@ import pytest from pydantic import AnyHttpUrl, AnyUrl -from mcp.client.auth.extensions.client_credentials import JWTParameters, RFC7523OAuthClientProvider +from mcp.client.auth.extensions.client_credentials import ( + ClientCredentialsOAuthProvider, + JWTParameters, + PrivateKeyJWTOAuthProvider, + RFC7523OAuthClientProvider, + SignedJWTParameters, + static_assertion_provider, +) from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthMetadata, OAuthToken @@ -53,13 +60,17 @@ async def callback_handler() -> tuple[str, str | None]: # pragma: no cover """Mock callback handler.""" return "test_auth_code", "test_state" - return RFC7523OAuthClientProvider( - server_url="https://api.example.com/v1/mcp", - client_metadata=client_metadata, - storage=mock_storage, - redirect_handler=redirect_handler, - callback_handler=callback_handler, - ) + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + return RFC7523OAuthClientProvider( + server_url="https://api.example.com/v1/mcp", + client_metadata=client_metadata, + storage=mock_storage, + redirect_handler=redirect_handler, + callback_handler=callback_handler, + ) class TestOAuthFlowClientCredentials: @@ -161,3 +172,260 @@ async def test_token_exchange_request_jwt(self, rfc7523_oauth_provider: RFC7523O assert claims["name"] == "John Doe" assert claims["admin"] assert claims["iat"] == 1516239022 + + +class TestClientCredentialsOAuthProvider: + """Test ClientCredentialsOAuthProvider.""" + + @pytest.mark.anyio + async def test_init_sets_client_info(self, mock_storage: MockTokenStorage): + """Test that _initialize sets client_info.""" + provider = ClientCredentialsOAuthProvider( + server_url="https://api.example.com", + storage=mock_storage, + client_id="test-client-id", + client_secret="test-client-secret", + ) + + # client_info is set during _initialize + await provider._initialize() + + assert provider.context.client_info is not None + assert provider.context.client_info.client_id == "test-client-id" + assert provider.context.client_info.client_secret == "test-client-secret" + assert provider.context.client_info.grant_types == ["client_credentials"] + assert provider.context.client_info.token_endpoint_auth_method == "client_secret_basic" + + @pytest.mark.anyio + async def test_init_with_scopes(self, mock_storage: MockTokenStorage): + """Test that constructor accepts scopes.""" + provider = ClientCredentialsOAuthProvider( + server_url="https://api.example.com", + storage=mock_storage, + client_id="test-client-id", + client_secret="test-client-secret", + scopes="read write", + ) + + await provider._initialize() + assert provider.context.client_info is not None + assert provider.context.client_info.scope == "read write" + + @pytest.mark.anyio + async def test_init_with_client_secret_post(self, mock_storage: MockTokenStorage): + """Test that constructor accepts client_secret_post auth method.""" + provider = ClientCredentialsOAuthProvider( + server_url="https://api.example.com", + storage=mock_storage, + client_id="test-client-id", + client_secret="test-client-secret", + token_endpoint_auth_method="client_secret_post", + ) + + await provider._initialize() + assert provider.context.client_info is not None + assert provider.context.client_info.token_endpoint_auth_method == "client_secret_post" + + @pytest.mark.anyio + async def test_exchange_token_client_credentials(self, mock_storage: MockTokenStorage): + """Test token exchange request building.""" + provider = ClientCredentialsOAuthProvider( + server_url="https://api.example.com/v1/mcp", + storage=mock_storage, + client_id="test-client-id", + client_secret="test-client-secret", + scopes="read write", + ) + provider.context.oauth_metadata = OAuthMetadata( + issuer=AnyHttpUrl("https://api.example.com"), + authorization_endpoint=AnyHttpUrl("https://api.example.com/authorize"), + token_endpoint=AnyHttpUrl("https://api.example.com/token"), + ) + provider.context.protocol_version = "2025-06-18" + + request = await provider._perform_authorization() + + assert request.method == "POST" + assert str(request.url) == "https://api.example.com/token" + + content = urllib.parse.unquote_plus(request.content.decode()) + assert "grant_type=client_credentials" in content + assert "scope=read write" in content + assert "resource=https://api.example.com/v1/mcp" in content + + @pytest.mark.anyio + async def test_exchange_token_without_scopes(self, mock_storage: MockTokenStorage): + """Test token exchange without scopes.""" + provider = ClientCredentialsOAuthProvider( + server_url="https://api.example.com/v1/mcp", + storage=mock_storage, + client_id="test-client-id", + client_secret="test-client-secret", + ) + provider.context.oauth_metadata = OAuthMetadata( + issuer=AnyHttpUrl("https://api.example.com"), + authorization_endpoint=AnyHttpUrl("https://api.example.com/authorize"), + token_endpoint=AnyHttpUrl("https://api.example.com/token"), + ) + provider.context.protocol_version = "2024-11-05" # Old version - no resource param + + request = await provider._perform_authorization() + + content = urllib.parse.unquote_plus(request.content.decode()) + assert "grant_type=client_credentials" in content + assert "scope=" not in content + assert "resource=" not in content + + +class TestPrivateKeyJWTOAuthProvider: + """Test PrivateKeyJWTOAuthProvider.""" + + @pytest.mark.anyio + async def test_init_sets_client_info(self, mock_storage: MockTokenStorage): + """Test that _initialize sets client_info.""" + + async def mock_assertion_provider(audience: str) -> str: # pragma: no cover + return "mock-jwt" + + provider = PrivateKeyJWTOAuthProvider( + server_url="https://api.example.com", + storage=mock_storage, + client_id="test-client-id", + assertion_provider=mock_assertion_provider, + ) + + # client_info is set during _initialize + await provider._initialize() + + assert provider.context.client_info is not None + assert provider.context.client_info.client_id == "test-client-id" + assert provider.context.client_info.grant_types == ["client_credentials"] + assert provider.context.client_info.token_endpoint_auth_method == "private_key_jwt" + + @pytest.mark.anyio + async def test_exchange_token_client_credentials(self, mock_storage: MockTokenStorage): + """Test token exchange request building with assertion provider.""" + + async def mock_assertion_provider(audience: str) -> str: + return f"jwt-for-{audience}" + + provider = PrivateKeyJWTOAuthProvider( + server_url="https://api.example.com/v1/mcp", + storage=mock_storage, + client_id="test-client-id", + assertion_provider=mock_assertion_provider, + scopes="read write", + ) + provider.context.oauth_metadata = OAuthMetadata( + issuer=AnyHttpUrl("https://auth.example.com"), + authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"), + token_endpoint=AnyHttpUrl("https://auth.example.com/token"), + ) + provider.context.protocol_version = "2025-06-18" + + request = await provider._perform_authorization() + + assert request.method == "POST" + assert str(request.url) == "https://auth.example.com/token" + + content = urllib.parse.unquote_plus(request.content.decode()) + assert "grant_type=client_credentials" in content + assert "client_assertion=jwt-for-https://auth.example.com/" in content + assert "client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer" in content + assert "scope=read write" in content + + @pytest.mark.anyio + async def test_exchange_token_without_scopes(self, mock_storage: MockTokenStorage): + """Test token exchange without scopes.""" + + async def mock_assertion_provider(audience: str) -> str: + return f"jwt-for-{audience}" + + provider = PrivateKeyJWTOAuthProvider( + server_url="https://api.example.com/v1/mcp", + storage=mock_storage, + client_id="test-client-id", + assertion_provider=mock_assertion_provider, + ) + provider.context.oauth_metadata = OAuthMetadata( + issuer=AnyHttpUrl("https://auth.example.com"), + authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"), + token_endpoint=AnyHttpUrl("https://auth.example.com/token"), + ) + provider.context.protocol_version = "2024-11-05" # Old version - no resource param + + request = await provider._perform_authorization() + + content = urllib.parse.unquote_plus(request.content.decode()) + assert "grant_type=client_credentials" in content + assert "scope=" not in content + assert "resource=" not in content + + +class TestSignedJWTParameters: + """Test SignedJWTParameters.""" + + @pytest.mark.anyio + async def test_create_assertion_provider(self): + """Test that create_assertion_provider creates valid JWTs.""" + params = SignedJWTParameters( + issuer="test-issuer", + subject="test-subject", + signing_key="a-string-secret-at-least-256-bits-long", + signing_algorithm="HS256", + lifetime_seconds=300, + ) + + provider = params.create_assertion_provider() + assertion = await provider("https://auth.example.com") + + claims = jwt.decode( + assertion, + key="a-string-secret-at-least-256-bits-long", + algorithms=["HS256"], + audience="https://auth.example.com", + ) + assert claims["iss"] == "test-issuer" + assert claims["sub"] == "test-subject" + assert claims["aud"] == "https://auth.example.com" + assert "exp" in claims + assert "iat" in claims + assert "jti" in claims + + @pytest.mark.anyio + async def test_create_assertion_provider_with_additional_claims(self): + """Test that additional_claims are included in the JWT.""" + params = SignedJWTParameters( + issuer="test-issuer", + subject="test-subject", + signing_key="a-string-secret-at-least-256-bits-long", + signing_algorithm="HS256", + additional_claims={"custom": "value"}, + ) + + provider = params.create_assertion_provider() + assertion = await provider("https://auth.example.com") + + claims = jwt.decode( + assertion, + key="a-string-secret-at-least-256-bits-long", + algorithms=["HS256"], + audience="https://auth.example.com", + ) + assert claims["custom"] == "value" + + +class TestStaticAssertionProvider: + """Test static_assertion_provider helper.""" + + @pytest.mark.anyio + async def test_returns_static_token(self): + """Test that static_assertion_provider returns the same token regardless of audience.""" + token = "my-static-jwt-token" + provider = static_assertion_provider(token) + + result1 = await provider("https://auth1.example.com") + result2 = await provider("https://auth2.example.com") + + assert result1 == token + assert result2 == token