From 44640a07e9a10e9b0cf4930d338756e661b93735 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Sun, 9 Aug 2026 22:50:12 +0530 Subject: [PATCH 1/3] feat: add Private Key JWT (private_key_jwt) client authentication Add private_key_jwt client authentication alongside the existing client secret. When a signing key is configured, the SDK signs a short-lived client assertion (RFC 7523) instead of sending a client secret. Client authentication is resolved in one place and applied uniformly across every token-endpoint call site. Also include the required response_type on Pushed Authorization Requests, which was previously missing and caused the request to be rejected. --- README.md | 26 ++ .../auth_schemes/client_assertion.py | 42 ++ .../auth_server/server_client.py | 111 ++++- .../tests/test_server_client.py | 426 ++++++++++++++++++ 4 files changed, 591 insertions(+), 14 deletions(-) create mode 100644 src/auth0_server_python/auth_schemes/client_assertion.py diff --git a/README.md b/README.md index 7e6c165..72610d7 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,32 @@ The `AUTH0_SECRET` is the key used to encrypt the session and transaction cookie openssl rand -hex 64 ``` +#### Authenticating with Private Key JWT + +The SDK authenticates to Auth0 with either a client secret or a Private Key JWT (`private_key_jwt`). To use Private Key JWT, pass your private signing key as `client_assertion_signing_key` instead of `client_secret`: + +```python +from auth0_server_python.auth_server.server_client import ServerClient + +with open('private_key.pem') as f: + private_key = f.read() + +auth0 = ServerClient( + domain='', + client_id='', + client_assertion_signing_key=private_key, + secret='', + authorization_params={ + 'redirect_uri': '', + } +) +``` + +The key must be a PKCS8 PEM private key whose public key is registered on your Auth0 application. The signing algorithm defaults to `RS256` and can be overridden with `client_assertion_signing_alg`. + +> [!IMPORTANT] +> Private keys must not be committed to source control. Load them from a secure secret store or an environment-provided file. + ### 3. Add login to your Application (interactive) Before using redirect-based login, ensure the `redirect_uri` is configured when initializing the SDK: diff --git a/src/auth0_server_python/auth_schemes/client_assertion.py b/src/auth0_server_python/auth_schemes/client_assertion.py new file mode 100644 index 0000000..56617d5 --- /dev/null +++ b/src/auth0_server_python/auth_schemes/client_assertion.py @@ -0,0 +1,42 @@ +import secrets +import time +from typing import Union + +import jwt + +# RFC 7523 client-assertion type for private_key_jwt authentication. +CLIENT_ASSERTION_TYPE = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + +# Assertion lifetime in seconds. +_ASSERTION_LIFETIME = 60 + + +def build_client_assertion( + private_key: Union[str, bytes], + client_id: str, + issuer: str, + alg: str = "RS256", +) -> str: + """ + Mint a private_key_jwt client assertion for token-endpoint authentication (RFC 7523). + + Args: + private_key: The client's private signing key (PKCS8 PEM string or bytes). + client_id: The Auth0 client ID, used as both the issuer and subject claim. + issuer: The authorization server issuer identifier, used as the audience claim. + alg: The signing algorithm (defaults to "RS256"). + + Returns: + The signed client-assertion JWT. + """ + now = int(time.time()) + payload = { + "iss": client_id, + "sub": client_id, + "aud": issuer, + "iat": now, + "nbf": now, + "exp": now + _ASSERTION_LIFETIME, + "jti": secrets.token_urlsafe(32), + } + return jwt.encode(payload, private_key, algorithm=alg) diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index c8eb6b3..d170f1f 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -19,6 +19,10 @@ from authlib.integrations.httpx_client import AsyncOAuth2Client from pydantic import ValidationError +from auth0_server_python.auth_schemes.client_assertion import ( + CLIENT_ASSERTION_TYPE, + build_client_assertion, +) from auth0_server_python.auth_schemes.dpop_auth import make_dpop_proof_for_token_endpoint from auth0_server_python.auth_server.mfa_client import MfaClient from auth0_server_python.auth_server.my_account_client import MyAccountClient @@ -113,6 +117,8 @@ def __init__( domain: Union[str, Callable[[Optional[dict[str, Any]]], str]] = None, client_id: str = None, client_secret: str = None, + client_assertion_signing_key: Optional[str] = None, + client_assertion_signing_alg: Optional[str] = None, redirect_uri: Optional[str] = None, secret: str = None, transaction_store=None, @@ -130,6 +136,8 @@ def __init__( domain: Auth0 domain - either a static string (e.g., 'tenant.auth0.com') or a callable that resolves domain dynamically. client_id: Auth0 client ID client_secret: Auth0 client secret + client_assertion_signing_key: Private key (PKCS8 PEM) for private_key_jwt client authentication. When set, token requests authenticate with a signed client assertion instead of the client secret. + client_assertion_signing_alg: Signing algorithm for the client assertion (defaults to "RS256"). redirect_uri: Default redirect URI for authentication secret: Secret used for encryption transaction_store: Custom transaction store (defaults to MemoryTransactionStore) @@ -171,6 +179,8 @@ def __init__( self._client_id = client_id self._client_secret = client_secret + self._client_assertion_signing_key = client_assertion_signing_key + self._client_assertion_signing_alg = client_assertion_signing_alg or "RS256" self._redirect_uri = redirect_uri self._secret = secret self._default_authorization_params = authorization_params or {} @@ -219,6 +229,43 @@ def _get_http_client(self, **kwargs) -> httpx.AsyncClient: headers = {**kwargs.pop("headers", {}), **self._telemetry_headers} return httpx.AsyncClient(headers=headers, **kwargs) + def _apply_client_authentication( + self, params: dict, issuer: str, in_body: bool = False + ) -> Optional[tuple[str, str]]: + """ + Apply client authentication to an outgoing token request. + + Args: + params: The outgoing token request body, mutated in place when a client assertion is injected. + issuer: The authorization server issuer identifier, used as the assertion audience. + in_body: When True, place the client secret in params instead of returning it for HTTP basic auth (for endpoints that authenticate the client in the request body). + + Returns: + The (client_id, client_secret) tuple for HTTP basic auth, or None when the client was authenticated in params. + + Raises: + ConfigurationError: If neither client_secret nor client_assertion_signing_key is configured. + """ + if self._client_assertion_signing_key: + params["client_assertion"] = build_client_assertion( + self._client_assertion_signing_key, + self._client_id, + issuer, + self._client_assertion_signing_alg, + ) + params["client_assertion_type"] = CLIENT_ASSERTION_TYPE + return None + + if self._client_secret: + if in_body: + params["client_secret"] = self._client_secret + return None + return (self._client_id, self._client_secret) + + raise ConfigurationError( + "Client authentication is not configured. Provide either client_secret or client_assertion_signing_key." + ) + def _normalize_url(self, value: str) -> str: """ Normalize a URL-like value (domain or issuer) for comparison. @@ -571,12 +618,16 @@ async def start_interactive_login( "configuration_error", "PAR is enabled but pushed_authorization_request_endpoint is missing in metadata") auth_params["client_id"] = self._client_id + # authlib supplies response_type on the non-PAR path, but the PAR post is built by hand. + auth_params["response_type"] = "code" + issuer = metadata.get("issuer") or f"https://{origin_domain}/" + client_auth = self._apply_client_authentication(auth_params, issuer) # Post the auth_params to the PAR endpoint async with self._get_http_client() as client: par_response = await client.post( par_endpoint, data=auth_params, - auth=(self._client_id, self._client_secret) + auth=client_auth ) if par_response.status_code not in (200, 201): error_data = par_response.json() @@ -663,6 +714,13 @@ async def complete_interactive_login( # Exchange the code for tokens # Use redirect_uri from transaction if available, otherwise fall back to default token_redirect_uri = transaction_data.redirect_uri or self._redirect_uri + + # client_secret is applied by self._oauth. private_key_jwt is passed to fetch_token as params. + client_auth_params: dict = {} + self._apply_client_authentication( + client_auth_params, origin_issuer or f"https://{origin_domain}/" + ) + try: token_endpoint = self._oauth.metadata["token_endpoint"] token_response = await self._oauth.fetch_token( @@ -670,6 +728,7 @@ async def complete_interactive_login( code=code, code_verifier=transaction_data.code_verifier, redirect_uri=token_redirect_uri, + **client_auth_params, ) except OAuthError as e: # Raise a custom error (or handle it as appropriate) @@ -1158,6 +1217,7 @@ async def get_token_by_refresh_token(self, options: dict[str, Any]) -> dict[str, Raises: AccessTokenError: If there was an issue requesting the access token. + ConfigurationError: If no client authentication is configured. Returns: A dictionary containing the token response from Auth0. @@ -1198,12 +1258,15 @@ async def get_token_by_refresh_token(self, options: dict[str, Any]) -> dict[str, if merged_scope: token_params["scope"] = merged_scope + issuer = metadata.get("issuer") or f"https://{domain}/" + client_auth = self._apply_client_authentication(token_params, issuer) + # Exchange the refresh token for an access token async with self._get_http_client() as client: response = await client.post( token_endpoint, data=token_params, - auth=(self._client_id, self._client_secret) + auth=client_auth ) if response.status_code != 200: @@ -1239,7 +1302,7 @@ async def get_token_by_refresh_token(self, options: dict[str, Any]) -> dict[str, return token_response except Exception as e: - if isinstance(e, ApiError): + if isinstance(e, (ApiError, ConfigurationError)): raise raise AccessTokenError( AccessTokenErrorCode.REFRESH_TOKEN_ERROR, @@ -1515,12 +1578,14 @@ async def initiate_backchannel_authentication( if authorization_params: params.update(authorization_params) + client_auth = self._apply_client_authentication(params, issuer) + # Make the backchannel authentication request async with self._get_http_client() as client: backchannel_response = await client.post( backchannel_endpoint, data=params, - auth=(self._client_id, self._client_secret) + auth=client_auth ) if backchannel_response.status_code != 200: @@ -1543,7 +1608,7 @@ async def initiate_backchannel_authentication( return backchannel_data except Exception as e: - if isinstance(e, ApiError): + if isinstance(e, (ApiError, ConfigurationError)): raise raise ApiError( "backchannel_error", @@ -1565,6 +1630,7 @@ async def backchannel_authentication_grant( Raises: AccessTokenError: If there was an issue requesting the access token. + ConfigurationError: If no client authentication is configured. Returns: A dictionary containing the token response from Auth0. @@ -1587,15 +1653,17 @@ async def backchannel_authentication_grant( "grant_type": "urn:openid:params:grant-type:ciba", "auth_req_id": auth_req_id, "client_id": self._client_id, - "client_secret": self._client_secret } + issuer = metadata.get("issuer") or f"https://{domain}/" + client_auth = self._apply_client_authentication(token_params, issuer) + # Exchange the auth_req_id for an access token async with self._get_http_client() as client: response = await client.post( token_endpoint, data=token_params, - auth=(self._client_id, self._client_secret) + auth=client_auth ) if response.status_code != 200: @@ -1625,7 +1693,7 @@ async def backchannel_authentication_grant( return token_response except Exception as e: - if isinstance(e, (ApiError, PollingApiError)): + if isinstance(e, (ApiError, PollingApiError, ConfigurationError)): raise raise AccessTokenError( AccessTokenErrorCode.AUTH_REQ_ID_ERROR, @@ -2008,6 +2076,7 @@ async def get_token_for_connection(self, options: dict[str, Any]) -> dict[str, A Raises: AccessTokenForConnectionError: If there was an issue requesting the access token. + ConfigurationError: If no client authentication is configured. Returns: Dictionary containing the token response with accessToken, expiresAt, and scope. @@ -2042,12 +2111,15 @@ async def get_token_for_connection(self, options: dict[str, Any]) -> dict[str, A if "login_hint" in options and options["login_hint"]: params["login_hint"] = options["login_hint"] + issuer = metadata.get("issuer") or f"https://{domain}/" + client_auth = self._apply_client_authentication(params, issuer) + # Make the request async with self._get_http_client() as client: response = await client.post( token_endpoint, data=params, - auth=(self._client_id, self._client_secret) + auth=client_auth ) if response.status_code != 200: @@ -2068,6 +2140,8 @@ async def get_token_for_connection(self, options: dict[str, Any]) -> dict[str, A } except Exception as e: + if isinstance(e, ConfigurationError): + raise if isinstance(e, ApiError): raise AccessTokenForConnectionError( AccessTokenForConnectionErrorCode.API_ERROR, @@ -2336,6 +2410,7 @@ async def custom_token_exchange( Raises: CustomTokenExchangeError: If token exchange fails MissingRequiredArgumentError: If required parameters are missing + ConfigurationError: If no client authentication is configured Example: ```python @@ -2426,17 +2501,23 @@ async def custom_token_exchange( # Merge additional authorization params if options.authorization_params: # Prevent override of critical parameters - forbidden_params = {"grant_type", "client_id", "subject_token", "subject_token_type"} + forbidden_params = { + "grant_type", "client_id", "subject_token", "subject_token_type", + "client_assertion", "client_assertion_type", + } for key, value in options.authorization_params.items(): if key not in forbidden_params: params[key] = value + issuer = metadata.get("issuer") or f"https://{domain}/" + client_auth = self._apply_client_authentication(params, issuer) + # Make the token exchange request async with self._get_http_client() as client: response = await client.post( token_endpoint, data=params, - auth=(self._client_id, self._client_secret) + auth=client_auth ) if response.status_code != 200: @@ -2482,7 +2563,7 @@ async def custom_token_exchange( f"Token validation failed: {str(e)}" ) except Exception as e: - if isinstance(e, (CustomTokenExchangeError, ApiError)): + if isinstance(e, (CustomTokenExchangeError, ApiError, ConfigurationError)): raise raise CustomTokenExchangeError( CustomTokenExchangeErrorCode.TOKEN_EXCHANGE_FAILED, @@ -3076,8 +3157,10 @@ async def signin_with_passkey( "auth_session": auth_session, "authn_response": authn_response.model_dump(by_alias=True, exclude_none=True), } - if self._client_secret: - body["client_secret"] = self._client_secret + # Passkey signin allows public clients, so only authenticate when configured. + if self._client_secret or self._client_assertion_signing_key: + issuer = metadata.get("issuer") or f"https://{domain}/" + self._apply_client_authentication(body, issuer, in_body=True) if connection: body["realm"] = connection if resolved_org: diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index c1c012a..e3ef2cb 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -6,7 +6,10 @@ from urllib.parse import parse_qs, urlparse import httpx +import jwt import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa from jwcrypto import jwk from auth0_server_python.auth_schemes.dpop_auth import DPoPAuth @@ -137,6 +140,78 @@ async def test_start_interactive_login_builds_auth_url(mocker): mock_transaction_store.set.assert_awaited() mock_oauth.assert_called_once() + +@pytest.mark.asyncio +async def test_par_request_uses_private_key_jwt_assertion(mocker): + """The pushed authorization request posts a client assertion when a signing key is set.""" + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=_generate_rsa_private_key_pem(), + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret", + pushed_authorization_requests=True, + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "issuer": "https://auth0.local/", + "authorization_endpoint": "https://auth0.local/authorize", + "pushed_authorization_request_endpoint": "https://auth0.local/oauth/par", + }, + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + par_response = AsyncMock() + par_response.status_code = 201 + par_response.json = MagicMock(return_value={"request_uri": "urn:req:abc", "expires_in": 60}) + mock_post.return_value = par_response + + await client.start_interactive_login() + + _, kwargs = mock_post.call_args + assert kwargs["auth"] is None + assert kwargs["data"]["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert len(kwargs["data"]["client_assertion"].split(".")) == 3 + + +@pytest.mark.asyncio +async def test_par_request_includes_response_type(mocker): + """The SDK supplies response_type=code on the PAR post; callers do not pass it.""" + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_secret="my_secret", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret", + pushed_authorization_requests=True, + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "issuer": "https://auth0.local/", + "authorization_endpoint": "https://auth0.local/authorize", + "pushed_authorization_request_endpoint": "https://auth0.local/oauth/par", + }, + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + par_response = AsyncMock() + par_response.status_code = 201 + par_response.json = MagicMock(return_value={"request_uri": "urn:req:abc", "expires_in": 60}) + mock_post.return_value = par_response + + final_url = await client.start_interactive_login() + + _, kwargs = mock_post.call_args + assert kwargs["data"]["response_type"] == "code" + assert "response_type=code" in final_url + + @pytest.mark.asyncio async def test_complete_interactive_login_no_transaction(): mock_transaction_store = AsyncMock() @@ -1184,6 +1259,36 @@ async def test_get_access_token_for_connection_no_refresh(): assert "A refresh token was not found" in str(exc.value) +@pytest.mark.asyncio +async def test_get_token_for_connection_uses_private_key_jwt_assertion(mocker): + """The connection token request posts a client assertion when a signing key is set.""" + private_key = _generate_rsa_private_key_pem() + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=private_key, + secret="some-secret", + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"token_endpoint": "https://auth0.local/token", "issuer": "https://auth0.local/"}, + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + resp = AsyncMock() + resp.status_code = 200 + resp.headers.get.return_value = "application/json" + resp.json = MagicMock(return_value={"access_token": "conn_at", "expires_in": 3600, "scope": "read"}) + mock_post.return_value = resp + + await client.get_token_for_connection({"connection": "google-oauth2", "refresh_token": "rt_abc"}) + + _, kwargs = mock_post.call_args + assert kwargs["auth"] is None + assert kwargs["data"]["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert len(kwargs["data"]["client_assertion"].split(".")) == 3 + + @pytest.mark.asyncio async def test_get_access_token_for_connection_domain_mismatch(): """Test that get_access_token_for_connection raises error on domain mismatch.""" @@ -1925,6 +2030,45 @@ async def test_backchannel_auth_rar(mocker): assert result["authorization_details"][0]["type"] == "accepted" assert mock_post.await_count == 2 + +@pytest.mark.asyncio +async def test_backchannel_auth_uses_private_key_jwt_assertion(mocker): + """Both CIBA requests (bc-authorize and the token grant) post a client assertion.""" + private_key = _generate_rsa_private_key_pem() + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=private_key, + secret="some-secret", + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "issuer": "https://auth0.local/", + "backchannel_authentication_endpoint": "https://auth0.local/bc-authorize", + "token_endpoint": "https://auth0.local/token", + }, + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + init_response = AsyncMock() + init_response.status_code = 200 + init_response.json = MagicMock(return_value={"auth_req_id": "req_123", "interval": 0.1, "expires_in": 60}) + grant_response = AsyncMock() + grant_response.status_code = 200 + grant_response.json = MagicMock(return_value={"access_token": "ciba_at", "expires_in": 60}) + mock_post.side_effect = [init_response, grant_response] + + await client.backchannel_authentication({"login_hint": {"sub": ""}}) + + assert mock_post.await_count == 2 + for call in mock_post.call_args_list: + data = call.kwargs["data"] + assert call.kwargs["auth"] is None + assert data["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert len(data["client_assertion"].split(".")) == 3 + + @pytest.mark.asyncio async def test_backchannel_auth_token_exchange_failed(mocker): client = ServerClient( @@ -2370,6 +2514,152 @@ async def test_get_token_by_refresh_token_mfa_required_raises_api_error_with_raw assert exc.value.mfa_requirements is None +# ============================================================================= +# Private Key JWT (client assertion) Client Authentication +# ============================================================================= + +def _generate_rsa_private_key_pem() -> str: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("ascii") + + +def _public_key_pem(private_key_pem: str) -> str: + private_key = serialization.load_pem_private_key( + private_key_pem.encode("ascii"), password=None + ) + return private_key.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ).decode("ascii") + + +@pytest.mark.asyncio +async def test_refresh_token_uses_client_secret_basic_auth(mocker): + """With client_secret configured, the refresh request authenticates via HTTP basic auth.""" + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + secret="some-secret" + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"token_endpoint": "https://auth0.local/token", "issuer": "https://auth0.local/"} + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + success_response = AsyncMock() + success_response.status_code = 200 + success_response.json = MagicMock(return_value={"access_token": "at", "expires_in": 3600}) + mock_post.return_value = success_response + + await client.get_token_by_refresh_token({"refresh_token": "abc"}) + + _, kwargs = mock_post.call_args + assert kwargs["auth"] == ("", "") + assert "client_assertion" not in kwargs["data"] + + +@pytest.mark.asyncio +async def test_refresh_token_uses_private_key_jwt_assertion(mocker): + """With a signing key configured, the refresh request posts a valid client assertion.""" + private_key = _generate_rsa_private_key_pem() + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=private_key, + secret="some-secret" + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"token_endpoint": "https://auth0.local/token", "issuer": "https://auth0.local/"} + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + success_response = AsyncMock() + success_response.status_code = 200 + success_response.json = MagicMock(return_value={"access_token": "at", "expires_in": 3600}) + mock_post.return_value = success_response + + await client.get_token_by_refresh_token({"refresh_token": "abc"}) + + _, kwargs = mock_post.call_args + assert kwargs["auth"] is None + assert kwargs["data"]["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + + assertion = kwargs["data"]["client_assertion"] + assert len(assertion.split(".")) == 3 + + claims = jwt.decode( + assertion, + _public_key_pem(private_key), + algorithms=["RS256"], + audience="https://auth0.local/", + ) + assert claims["iss"] == "my_client" + assert claims["sub"] == "my_client" + assert claims["aud"] == "https://auth0.local/" + assert "iat" in claims + assert "nbf" in claims + assert "exp" in claims + assert "jti" in claims + assert claims["exp"] - claims["iat"] == 60 + + +@pytest.mark.asyncio +async def test_client_assertion_signing_alg_override(mocker): + """The configured signing algorithm is used to mint the assertion.""" + private_key = _generate_rsa_private_key_pem() + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=private_key, + client_assertion_signing_alg="RS384", + secret="some-secret" + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"token_endpoint": "https://auth0.local/token", "issuer": "https://auth0.local/"} + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + success_response = AsyncMock() + success_response.status_code = 200 + success_response.json = MagicMock(return_value={"access_token": "at", "expires_in": 3600}) + mock_post.return_value = success_response + + await client.get_token_by_refresh_token({"refresh_token": "abc"}) + + assertion = mock_post.call_args[1]["data"]["client_assertion"] + header = jwt.get_unverified_header(assertion) + assert header["alg"] == "RS384" + + +@pytest.mark.asyncio +async def test_refresh_token_no_client_auth_raises_configuration_error(mocker): + """With neither client_secret nor a signing key, a token request raises ConfigurationError.""" + client = ServerClient( + domain="auth0.local", + client_id="my_client", + secret="some-secret" + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"token_endpoint": "https://auth0.local/token", "issuer": "https://auth0.local/"} + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + + with pytest.raises(ConfigurationError): + await client.get_token_by_refresh_token({"refresh_token": "abc"}) + + mock_post.assert_not_awaited() + + # ============================================================================= # Connected Accounts Tests (My Account Client) # ============================================================================= @@ -3483,6 +3773,98 @@ async def test_custom_token_exchange_forbidden_params_filtered(mocker): assert call_args[1]["data"]["allowed_param"] == "value" +@pytest.mark.asyncio +async def test_custom_token_exchange_uses_private_key_jwt_assertion(mocker): + """Custom token exchange authenticates with a client assertion when a signing key is set.""" + private_key = _generate_rsa_private_key_pem() + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=private_key, + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret" + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"token_endpoint": "https://auth0.local/oauth/token", "issuer": "https://auth0.local/"} + ) + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "access_token": "at", + "token_type": "Bearer", + "expires_in": 3600, + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token" + } + mock_response.headers.get.return_value = "application/json" + mock_httpx_client = AsyncMock() + mock_httpx_client.__aenter__.return_value = mock_httpx_client + mock_httpx_client.__aexit__.return_value = None + mock_httpx_client.post.return_value = mock_response + mocker.patch("httpx.AsyncClient", return_value=mock_httpx_client) + + options = CustomTokenExchangeOptions( + subject_token="custom-token-123", + subject_token_type="urn:acme:mcp-token", + audience="https://api.example.com", + ) + await client.custom_token_exchange(options) + + call_args = mock_httpx_client.post.call_args + assert call_args[1]["auth"] is None + assert call_args[1]["data"]["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert len(call_args[1]["data"]["client_assertion"].split(".")) == 3 + + +@pytest.mark.asyncio +async def test_custom_token_exchange_caller_cannot_inject_client_assertion(mocker): + """The denylist drops caller-supplied client_assertion params so they never reach the token endpoint.""" + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret" + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"token_endpoint": "https://auth0.local/oauth/token", "issuer": "https://auth0.local/"} + ) + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "access_token": "at", + "token_type": "Bearer", + "expires_in": 3600, + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token" + } + mock_response.headers.get.return_value = "application/json" + mock_httpx_client = AsyncMock() + mock_httpx_client.__aenter__.return_value = mock_httpx_client + mock_httpx_client.__aexit__.return_value = None + mock_httpx_client.post.return_value = mock_response + mocker.patch("httpx.AsyncClient", return_value=mock_httpx_client) + + options = CustomTokenExchangeOptions( + subject_token="custom-token-123", + subject_token_type="urn:acme:mcp-token", + audience="https://api.example.com", + authorization_params={ + "client_assertion": "attacker-supplied", + "client_assertion_type": "attacker-type", + }, + ) + await client.custom_token_exchange(options) + + posted = mock_httpx_client.post.call_args[1]["data"] + assert "client_assertion" not in posted + assert "client_assertion_type" not in posted + + # Delegation Support @@ -7050,6 +7432,50 @@ async def test_signin_with_passkey_success(mocker): assert "raw_id" not in body["authn_response"] +@pytest.mark.asyncio +async def test_signin_with_passkey_uses_private_key_jwt_assertion(mocker): + """Passkey signin authenticates with a client assertion in the body when a signing key is set.""" + private_key = _generate_rsa_private_key_pem() + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=private_key, + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="test-secret-value", + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"token_endpoint": "https://auth0.local/oauth/token", "issuer": "https://auth0.local/"}, + ) + mocker.patch.object(client, "_get_jwks_cached", return_value={}) + mocker.patch.object(client, "_verify_and_decode_jwt", return_value={ + "sub": "auth0|user123", "iss": "https://auth0.local/" + }) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + mock_response = AsyncMock() + mock_response.status_code = 200 + mock_response.json = MagicMock(return_value=_PASSKEY_TOKEN_RESPONSE) + mock_post.return_value = mock_response + + await client.signin_with_passkey(auth_session="session_xyz", authn_response=_make_passkey_authn_response()) + + _, kwargs = mock_post.call_args + body = kwargs["json"] + assert "client_secret" not in body + assert body["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert len(body["client_assertion"].split(".")) == 3 + claims = jwt.decode( + body["client_assertion"], + _public_key_pem(private_key), + algorithms=["RS256"], + audience="https://auth0.local/", + ) + assert claims["iss"] == "my_client" + assert claims["sub"] == "my_client" + + @pytest.mark.asyncio async def test_signin_with_passkey_uses_json_content_type(mocker): client = ServerClient( From c1a897f3a3260cef93769ee609d11c5bf6108f13 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Fri, 14 Aug 2026 14:25:11 +0530 Subject: [PATCH 2/3] fix: apply client authentication uniformly and validate the signing key Address review feedback on private_key_jwt client authentication: - Route MFA challenge and verify through the client-auth resolver so private_key_jwt works for MFA step-up, not only client_secret. - Withhold the client secret from the OAuth client when a signing key is set, so the code exchange sends a single client authentication method. - Validate the signing key and algorithm at construction and raise ConfigurationError, instead of failing on the first token request. - Strip caller-supplied client-auth keys in the resolver so they cannot be injected through PAR or backchannel authorization params. - Document ConfigurationError on the interactive login entry points and the Private Key JWT limitations in the README. --- README.md | 5 +- .../auth_schemes/client_assertion.py | 21 +++ .../auth_server/mfa_client.py | 18 +- .../auth_server/server_client.py | 21 ++- .../tests/test_mfa_client.py | 74 ++++++++ .../tests/test_server_client.py | 168 +++++++++++++++++- 6 files changed, 298 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 72610d7..b0de499 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,10 @@ auth0 = ServerClient( ) ``` -The key must be a PKCS8 PEM private key whose public key is registered on your Auth0 application. The signing algorithm defaults to `RS256` and can be overridden with `client_assertion_signing_alg`. +The key must be a PKCS8 PEM private key. Register its public key on your Auth0 application under **Settings → Credentials**, and set the application's authentication method to Private Key JWT. The signing algorithm defaults to `RS256` and can be overridden with `client_assertion_signing_alg`. The algorithm must match the key type and the algorithm chosen when the public key credential was created. + +> [!NOTE] +> The passkey challenge endpoints (`register` and `challenge`) accept only a client secret, so a client configured with just a signing key cannot use them. > [!IMPORTANT] > Private keys must not be committed to source control. Load them from a secure secret store or an environment-provided file. diff --git a/src/auth0_server_python/auth_schemes/client_assertion.py b/src/auth0_server_python/auth_schemes/client_assertion.py index 56617d5..d5a9f52 100644 --- a/src/auth0_server_python/auth_schemes/client_assertion.py +++ b/src/auth0_server_python/auth_schemes/client_assertion.py @@ -4,6 +4,8 @@ import jwt +from auth0_server_python.error import ConfigurationError + # RFC 7523 client-assertion type for private_key_jwt authentication. CLIENT_ASSERTION_TYPE = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" @@ -11,6 +13,25 @@ _ASSERTION_LIFETIME = 60 +def validate_client_assertion_key(private_key: Union[str, bytes], alg: str = "RS256") -> None: + """ + Verify the signing key and algorithm can produce a client assertion. + + Args: + private_key: The client's private signing key (PKCS8 PEM string or bytes). + alg: The signing algorithm (defaults to "RS256"). + + Raises: + ConfigurationError: If the key is malformed or does not match the algorithm. + """ + try: + jwt.encode({"probe": True}, private_key, algorithm=alg) + except Exception as e: + raise ConfigurationError( + f"Invalid client_assertion_signing_key for algorithm {alg}: {e}" + ) + + def build_client_assertion( private_key: Union[str, bytes], client_id: str, diff --git a/src/auth0_server_python/auth_server/mfa_client.py b/src/auth0_server_python/auth_server/mfa_client.py index 18e198f..07b25c7 100644 --- a/src/auth0_server_python/auth_server/mfa_client.py +++ b/src/auth0_server_python/auth_server/mfa_client.py @@ -66,7 +66,8 @@ def __init__( secret: str, state_store=None, state_identifier: str = "_a0_session", - headers: Optional[dict[str, str]] = None + headers: Optional[dict[str, str]] = None, + apply_client_authentication: Optional[Callable] = None ): if callable(domain): self._domain = None @@ -80,12 +81,20 @@ def __init__( self._state_store = state_store self._state_identifier = state_identifier self._headers = headers or {} + self._apply_client_authentication = apply_client_authentication def _get_http_client(self, **kwargs) -> httpx.AsyncClient: """Return an httpx.AsyncClient with default headers injected.""" headers = {**kwargs.pop("headers", {}), **self._headers} return httpx.AsyncClient(headers=headers, **kwargs) + def _apply_mfa_client_authentication(self, body: dict, base_url: str) -> None: + """Add client authentication to an MFA request body (client_secret or client assertion).""" + if self._apply_client_authentication: + self._apply_client_authentication(body, f"{base_url}/", in_body=True) + elif self._client_secret: + body["client_secret"] = self._client_secret + async def _resolve_base_url( self, store_options: Optional[dict[str, Any]] = None @@ -415,9 +424,9 @@ async def challenge_authenticator( body: dict[str, Any] = { "mfa_token": context.mfa_token, "client_id": self._client_id, - "client_secret": self._client_secret, "challenge_type": challenge_type } + self._apply_mfa_client_authentication(body, base_url) if "authenticator_id" in options and options["authenticator_id"]: body["authenticator_id"] = options["authenticator_id"] @@ -488,11 +497,13 @@ async def verify( raise MfaTokenInvalidError() context = self.decrypt_mfa_token(mfa_token) + base_url = await self._resolve_base_url(store_options) + body: dict[str, Any] = { "client_id": self._client_id, - "client_secret": self._client_secret, "mfa_token": context.mfa_token } + self._apply_mfa_client_authentication(body, base_url) if "otp" in options: body["grant_type"] = "http://auth0.com/oauth/grant-type/mfa-otp" @@ -511,7 +522,6 @@ async def verify( ) try: - base_url = await self._resolve_base_url(store_options) token_endpoint = f"{base_url}/oauth/token" async with self._get_http_client() as client: diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index d170f1f..fe16364 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -22,6 +22,7 @@ from auth0_server_python.auth_schemes.client_assertion import ( CLIENT_ASSERTION_TYPE, build_client_assertion, + validate_client_assertion_key, ) from auth0_server_python.auth_schemes.dpop_auth import make_dpop_proof_for_token_endpoint from auth0_server_python.auth_server.mfa_client import MfaClient @@ -181,6 +182,10 @@ def __init__( self._client_secret = client_secret self._client_assertion_signing_key = client_assertion_signing_key self._client_assertion_signing_alg = client_assertion_signing_alg or "RS256" + if client_assertion_signing_key: + validate_client_assertion_key( + client_assertion_signing_key, self._client_assertion_signing_alg + ) self._redirect_uri = redirect_uri self._secret = secret self._default_authorization_params = authorization_params or {} @@ -197,10 +202,10 @@ def __init__( self._telemetry = Telemetry.default() self._telemetry_headers = self._telemetry.headers - # Initialize OAuth client + # A signing key takes precedence, so the secret is withheld to keep client auth single. self._oauth = AsyncOAuth2Client( client_id=client_id, - client_secret=client_secret, + client_secret=None if client_assertion_signing_key else client_secret, headers=self._telemetry_headers, ) @@ -222,6 +227,7 @@ def __init__( state_store=self._state_store, state_identifier=self._state_identifier, headers=self._telemetry_headers, + apply_client_authentication=self._apply_client_authentication, ) def _get_http_client(self, **kwargs) -> httpx.AsyncClient: @@ -236,7 +242,7 @@ def _apply_client_authentication( Apply client authentication to an outgoing token request. Args: - params: The outgoing token request body, mutated in place when a client assertion is injected. + params: The outgoing token request body. Any caller-supplied client-auth keys are removed, then the client assertion is added (or the client secret when in_body is True). issuer: The authorization server issuer identifier, used as the assertion audience. in_body: When True, place the client secret in params instead of returning it for HTTP basic auth (for endpoints that authenticate the client in the request body). @@ -246,6 +252,9 @@ def _apply_client_authentication( Raises: ConfigurationError: If neither client_secret nor client_assertion_signing_key is configured. """ + for reserved in ("client_secret", "client_assertion", "client_assertion_type"): + params.pop(reserved, None) + if self._client_assertion_signing_key: params["client_assertion"] = build_client_assertion( self._client_assertion_signing_key, @@ -533,6 +542,9 @@ async def start_interactive_login( Returns: Authorization URL to redirect the user to + + Raises: + ConfigurationError: If no client authentication is configured. """ options = options or StartInteractiveLoginOptions() @@ -675,6 +687,9 @@ async def complete_interactive_login( Returns: Dictionary containing session data and app state + + Raises: + ConfigurationError: If no client authentication is configured. """ # Parse the URL to get query parameters parsed_url = urlparse(url) diff --git a/src/auth0_server_python/tests/test_mfa_client.py b/src/auth0_server_python/tests/test_mfa_client.py index ac820df..1c2834e 100644 --- a/src/auth0_server_python/tests/test_mfa_client.py +++ b/src/auth0_server_python/tests/test_mfa_client.py @@ -6,9 +6,12 @@ from unittest.mock import AsyncMock, MagicMock import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa from jwcrypto import jwk from auth0_server_python.auth_server.mfa_client import DEFAULT_MFA_TOKEN_TTL, MfaClient +from auth0_server_python.auth_server.server_client import ServerClient from auth0_server_python.auth_types import ( AuthenticatorResponse, ChallengeResponse, @@ -35,6 +38,18 @@ SECRET = "test-secret-long-enough-for-encryption" +def _make_pkjwt_key() -> str: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("ascii") + + +_PKJWT_KEY = _make_pkjwt_key() + + def _make_client() -> MfaClient: return MfaClient( domain=DOMAIN, @@ -535,6 +550,35 @@ async def test_challenge_sms_with_authenticator_id(self, mocker): assert result.challenge_type == "oob" assert result.oob_code == "oob_sms_challenge_456" + @pytest.mark.asyncio + async def test_challenge_uses_private_key_jwt_assertion(self, mocker): + """With a signing-key resolver, the challenge body carries a client assertion, not a secret.""" + server = ServerClient( + domain=DOMAIN, client_id=CLIENT_ID, + client_assertion_signing_key=_PKJWT_KEY, secret=SECRET, + ) + client = MfaClient( + domain=DOMAIN, client_id=CLIENT_ID, client_secret=None, secret=SECRET, + apply_client_authentication=server._apply_client_authentication, + ) + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={"challenge_type": "oob", "oob_code": "x"}) + captured = {} + + async def mock_post(self_client, url, **kwargs): + captured["kwargs"] = kwargs + return response + + mocker.patch("httpx.AsyncClient.post", new=mock_post) + + await client.challenge_authenticator({"mfa_token": _enc(), "factor_type": "otp"}) + + body = captured["kwargs"]["json"] + assert body["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert len(body["client_assertion"].split(".")) == 3 + assert "client_secret" not in body + # ── verify ─────────────────────────────────────────────────────────────────── @@ -595,6 +639,36 @@ async def test_verify_recovery_code_success(self, mocker): }) assert isinstance(result, MfaVerifyResponse) + @pytest.mark.asyncio + async def test_verify_uses_private_key_jwt_assertion(self, mocker): + """With a signing-key resolver, the verify body carries a client assertion, not a secret.""" + server = ServerClient( + domain=DOMAIN, client_id=CLIENT_ID, + client_assertion_signing_key=_PKJWT_KEY, secret=SECRET, + ) + client = MfaClient( + domain=DOMAIN, client_id=CLIENT_ID, client_secret=None, secret=SECRET, + apply_client_authentication=server._apply_client_authentication, + ) + response = AsyncMock() + response.status_code = 200 + response.headers = {} + response.json = MagicMock(return_value={"access_token": "at", "token_type": "Bearer", "expires_in": 3600}) + captured = {} + + async def mock_post(self_client, url, **kwargs): + captured["kwargs"] = kwargs + return response + + mocker.patch("httpx.AsyncClient.post", new=mock_post) + + await client.verify({"mfa_token": _enc(), "otp": "123456"}) + + body = captured["kwargs"]["data"] + assert body["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert len(body["client_assertion"].split(".")) == 3 + assert "client_secret" not in body + @pytest.mark.asyncio async def test_verify_no_credential_raises(self): client = _make_client() diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index e3ef2cb..4536455 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -8,8 +8,9 @@ import httpx import jwt import pytest +from authlib.integrations.httpx_client import AsyncOAuth2Client from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.asymmetric import ec, rsa from jwcrypto import jwk from auth0_server_python.auth_schemes.dpop_auth import DPoPAuth @@ -212,6 +213,45 @@ async def test_par_request_includes_response_type(mocker): assert "response_type=code" in final_url +@pytest.mark.asyncio +async def test_par_request_caller_cannot_inject_client_assertion(mocker): + """A caller-supplied client_assertion in authorization_params never reaches the PAR body.""" + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_secret="my_secret", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret", + pushed_authorization_requests=True, + authorization_params={"redirect_uri": "/test_redirect_uri", "response_type": "code"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "issuer": "https://auth0.local/", + "authorization_endpoint": "https://auth0.local/authorize", + "pushed_authorization_request_endpoint": "https://auth0.local/oauth/par", + }, + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + par_response = AsyncMock() + par_response.status_code = 201 + par_response.json = MagicMock(return_value={"request_uri": "urn:req:abc", "expires_in": 60}) + mock_post.return_value = par_response + + await client.start_interactive_login( + options=StartInteractiveLoginOptions( + authorization_params={"client_assertion": "attacker", "client_assertion_type": "attacker-type"} + ) + ) + + posted = mock_post.call_args[1]["data"] + assert "client_assertion" not in posted + assert "client_assertion_type" not in posted + + @pytest.mark.asyncio async def test_complete_interactive_login_no_transaction(): mock_transaction_store = AsyncMock() @@ -2069,6 +2109,43 @@ async def test_backchannel_auth_uses_private_key_jwt_assertion(mocker): assert len(data["client_assertion"].split(".")) == 3 +@pytest.mark.asyncio +async def test_backchannel_auth_caller_cannot_inject_client_assertion(mocker): + """A caller-supplied client_assertion in authorization_params never reaches the bc-authorize body.""" + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_secret="my_secret", + secret="some-secret", + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "issuer": "https://auth0.local/", + "backchannel_authentication_endpoint": "https://auth0.local/bc-authorize", + "token_endpoint": "https://auth0.local/token", + }, + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + init_response = AsyncMock() + init_response.status_code = 200 + init_response.json = MagicMock(return_value={"auth_req_id": "req_123", "interval": 0.1, "expires_in": 60}) + grant_response = AsyncMock() + grant_response.status_code = 200 + grant_response.json = MagicMock(return_value={"access_token": "ciba_at", "expires_in": 60}) + mock_post.side_effect = [init_response, grant_response] + + await client.backchannel_authentication({ + "login_hint": {"sub": ""}, + "authorization_params": {"client_assertion": "attacker", "client_assertion_type": "attacker-type"}, + }) + + bc_authorize_body = mock_post.call_args_list[0].kwargs["data"] + assert "client_assertion" not in bc_authorize_body + assert "client_assertion_type" not in bc_authorize_body + + @pytest.mark.asyncio async def test_backchannel_auth_token_exchange_failed(mocker): client = ServerClient( @@ -2660,6 +2737,95 @@ async def test_refresh_token_no_client_auth_raises_configuration_error(mocker): mock_post.assert_not_awaited() +def test_malformed_signing_key_raises_at_construction(): + """A malformed PEM fails at construction, not on the first token request.""" + with pytest.raises(ConfigurationError): + ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key="-----BEGIN PRIVATE KEY-----\nnope\n-----END PRIVATE KEY-----", + secret="some-secret", + ) + + +def test_signing_key_algorithm_mismatch_raises_at_construction(): + """A key that does not match the configured algorithm fails at construction.""" + ec_key = ec.generate_private_key(ec.SECP256R1()) + ec_pem = ec_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("ascii") + + with pytest.raises(ConfigurationError): + ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=ec_pem, + client_assertion_signing_alg="RS256", + secret="some-secret", + ) + + +def test_unknown_signing_algorithm_raises_at_construction(): + """An unsupported algorithm fails at construction.""" + with pytest.raises(ConfigurationError): + ServerClient( + domain="auth0.local", + client_id="my_client", + client_assertion_signing_key=_generate_rsa_private_key_pem(), + client_assertion_signing_alg="NOPE999", + secret="some-secret", + ) + + +@pytest.mark.asyncio +async def test_login_with_both_secret_and_key_sends_only_assertion(): + """With both a secret and a signing key, the code exchange sends the assertion and no basic auth header.""" + private_key = _generate_rsa_private_key_pem() + client = ServerClient( + domain="auth0.local", + client_id="my_client", + client_secret="", + client_assertion_signing_key=private_key, + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret", + ) + + # The signing key must win, so the OAuth client holds no secret to add a basic-auth header. + assert client._oauth.client_secret is None + + captured = {} + + def handler(request): + captured["authorization"] = request.headers.get("authorization") + captured["body"] = request.content.decode() + return httpx.Response(200, json={"access_token": "at", "token_type": "Bearer", "expires_in": 3600}) + + # Mirror the real OAuth client's auth config onto a transport-backed one to inspect the wire. + wire_oauth = AsyncOAuth2Client( + client_id=client._oauth.client_id, + client_secret=client._oauth.client_secret, + transport=httpx.MockTransport(handler), + ) + params = {} + client._apply_client_authentication(params, "https://auth0.local/") + + await wire_oauth.fetch_token( + "https://auth0.local/oauth/token", + grant_type="authorization_code", + code="abc", + code_verifier="v", + redirect_uri="https://app/cb", + **params, + ) + + assert captured["authorization"] is None + assert "client_assertion=" in captured["body"] + assert "client_secret=" not in captured["body"] + + # ============================================================================= # Connected Accounts Tests (My Account Client) # ============================================================================= From 8651e532ba29a97d7010ba914baa286f15f5ad60 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Fri, 14 Aug 2026 15:39:45 +0530 Subject: [PATCH 3/3] test: pin client auth at the SDK call sites and finish pkjwt docs Drive the login and MFA private_key_jwt tests through complete_interactive_login and the ServerClient-wired MFA client so a regression at the SDK call site fails the suite, not just the helper. Add the passkey signing-key caveat to examples/Passkeys.md, list the accepted signing algorithms in the README, document ConfigurationError on the MFA challenge and verify methods, and chain the client assertion validation error. --- README.md | 2 +- examples/Passkeys.md | 3 ++ .../auth_schemes/client_assertion.py | 2 +- .../auth_server/mfa_client.py | 2 + .../tests/test_mfa_client.py | 14 ++----- .../tests/test_server_client.py | 37 +++++++++++-------- 6 files changed, 32 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index b0de499..421db2e 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ auth0 = ServerClient( ) ``` -The key must be a PKCS8 PEM private key. Register its public key on your Auth0 application under **Settings → Credentials**, and set the application's authentication method to Private Key JWT. The signing algorithm defaults to `RS256` and can be overridden with `client_assertion_signing_alg`. The algorithm must match the key type and the algorithm chosen when the public key credential was created. +The key must be a PKCS8 PEM private key. Register its public key on your Auth0 application under **Settings → Credentials**, and set the application's authentication method to Private Key JWT. The signing algorithm defaults to `RS256` and can be overridden with `client_assertion_signing_alg`. Auth0 accepts `RS256`, `RS384`, and `PS256`, all of which use an RSA key. The algorithm must match the key type and the algorithm chosen when the public key credential was created. > [!NOTE] > The passkey challenge endpoints (`register` and `challenge`) accept only a client secret, so a client configured with just a signing key cannot use them. diff --git a/examples/Passkeys.md b/examples/Passkeys.md index 0ad39bb..c1eed9a 100644 --- a/examples/Passkeys.md +++ b/examples/Passkeys.md @@ -41,6 +41,9 @@ server_client = ServerClient( The **Passkey** grant (`urn:okta:params:oauth:grant-type:webauthn`) must be enabled for your application under **Applications → Your App → Grant Types**. +> [!NOTE] +> The passkey challenge endpoints accept a client secret only, not a client assertion. A client configured with just a `client_assertion_signing_key` (Private Key JWT) cannot use the passkey flows, so configure a `client_secret` to use them. + ## 1. Passkey Signup ### Step 1 — Request a signup challenge diff --git a/src/auth0_server_python/auth_schemes/client_assertion.py b/src/auth0_server_python/auth_schemes/client_assertion.py index d5a9f52..f7d69b5 100644 --- a/src/auth0_server_python/auth_schemes/client_assertion.py +++ b/src/auth0_server_python/auth_schemes/client_assertion.py @@ -29,7 +29,7 @@ def validate_client_assertion_key(private_key: Union[str, bytes], alg: str = "RS except Exception as e: raise ConfigurationError( f"Invalid client_assertion_signing_key for algorithm {alg}: {e}" - ) + ) from e def build_client_assertion( diff --git a/src/auth0_server_python/auth_server/mfa_client.py b/src/auth0_server_python/auth_server/mfa_client.py index 07b25c7..4df7baa 100644 --- a/src/auth0_server_python/auth_server/mfa_client.py +++ b/src/auth0_server_python/auth_server/mfa_client.py @@ -402,6 +402,7 @@ async def challenge_authenticator( Raises: MfaChallengeError: When the challenge fails. + ConfigurationError: If neither client_secret nor client_assertion_signing_key is configured. """ mfa_token = options.get("mfa_token") if not mfa_token: @@ -491,6 +492,7 @@ async def verify( MfaVerifyError: When verification fails, or when dpop_key was supplied but the server returned an unbound (Bearer) token. MfaRequiredError: When chained MFA is required. + ConfigurationError: If neither client_secret nor client_assertion_signing_key is configured. """ mfa_token = options.get("mfa_token") if not mfa_token: diff --git a/src/auth0_server_python/tests/test_mfa_client.py b/src/auth0_server_python/tests/test_mfa_client.py index 1c2834e..c2f444d 100644 --- a/src/auth0_server_python/tests/test_mfa_client.py +++ b/src/auth0_server_python/tests/test_mfa_client.py @@ -552,15 +552,12 @@ async def test_challenge_sms_with_authenticator_id(self, mocker): @pytest.mark.asyncio async def test_challenge_uses_private_key_jwt_assertion(self, mocker): - """With a signing-key resolver, the challenge body carries a client assertion, not a secret.""" + """The ServerClient-wired MFA client carries a client assertion, not a secret, on challenge.""" server = ServerClient( domain=DOMAIN, client_id=CLIENT_ID, client_assertion_signing_key=_PKJWT_KEY, secret=SECRET, ) - client = MfaClient( - domain=DOMAIN, client_id=CLIENT_ID, client_secret=None, secret=SECRET, - apply_client_authentication=server._apply_client_authentication, - ) + client = server._mfa_client response = AsyncMock() response.status_code = 200 response.json = MagicMock(return_value={"challenge_type": "oob", "oob_code": "x"}) @@ -641,15 +638,12 @@ async def test_verify_recovery_code_success(self, mocker): @pytest.mark.asyncio async def test_verify_uses_private_key_jwt_assertion(self, mocker): - """With a signing-key resolver, the verify body carries a client assertion, not a secret.""" + """The ServerClient-wired MFA client carries a client assertion, not a secret, on verify.""" server = ServerClient( domain=DOMAIN, client_id=CLIENT_ID, client_assertion_signing_key=_PKJWT_KEY, secret=SECRET, ) - client = MfaClient( - domain=DOMAIN, client_id=CLIENT_ID, client_secret=None, secret=SECRET, - apply_client_authentication=server._apply_client_authentication, - ) + client = server._mfa_client response = AsyncMock() response.status_code = 200 response.headers = {} diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 4536455..c7db1f8 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -2780,16 +2780,22 @@ def test_unknown_signing_algorithm_raises_at_construction(): @pytest.mark.asyncio -async def test_login_with_both_secret_and_key_sends_only_assertion(): - """With both a secret and a signing key, the code exchange sends the assertion and no basic auth header.""" +async def test_login_with_both_secret_and_key_sends_only_assertion(mocker): + """complete_interactive_login sends the assertion and no basic-auth header when both a secret and a signing key are set.""" private_key = _generate_rsa_private_key_pem() + mock_tx_store = AsyncMock() + mock_tx_store.get.return_value = TransactionData( + code_verifier="v", + redirect_uri="https://app/cb", + domain="auth0.local", + ) client = ServerClient( domain="auth0.local", client_id="my_client", client_secret="", client_assertion_signing_key=private_key, state_store=AsyncMock(), - transaction_store=AsyncMock(), + transaction_store=mock_tx_store, secret="some-secret", ) @@ -2801,26 +2807,25 @@ async def test_login_with_both_secret_and_key_sends_only_assertion(): def handler(request): captured["authorization"] = request.headers.get("authorization") captured["body"] = request.content.decode() - return httpx.Response(200, json={"access_token": "at", "token_type": "Bearer", "expires_in": 3600}) + return httpx.Response(200, json={ + "access_token": "at", "token_type": "Bearer", "expires_in": 3600, + "userinfo": {"sub": "user123"}, + }) - # Mirror the real OAuth client's auth config onto a transport-backed one to inspect the wire. - wire_oauth = AsyncOAuth2Client( + # Back the real OAuth client with a mock transport, mirroring its auth config, to inspect the wire. + client._oauth = AsyncOAuth2Client( client_id=client._oauth.client_id, client_secret=client._oauth.client_secret, transport=httpx.MockTransport(handler), ) - params = {} - client._apply_client_authentication(params, "https://auth0.local/") - - await wire_oauth.fetch_token( - "https://auth0.local/oauth/token", - grant_type="authorization_code", - code="abc", - code_verifier="v", - redirect_uri="https://app/cb", - **params, + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"issuer": "https://auth0.local/", "token_endpoint": "https://auth0.local/oauth/token"}, ) + await client.complete_interactive_login("https://app/cb?code=abc&state=xyz") + assert captured["authorization"] is None assert "client_assertion=" in captured["body"] assert "client_secret=" not in captured["body"]