diff --git a/README.md b/README.md index 2ff26cf3..4e3bf8f3 100644 --- a/README.md +++ b/README.md @@ -84,6 +84,29 @@ The key must be a PKCS8 PEM private key. Register its public key on your Auth0 a > [!IMPORTANT] > Private keys must not be committed to source control. Load them from a secure secret store or an environment-provided file. +#### Authenticating with Mutual TLS (mTLS) + +The SDK supports mTLS client authentication (RFC 8705): the client presents a TLS certificate during the handshake instead of a client secret. Pass `use_mtls=True` and a caller-built `ssl.SSLContext` that already has the certificate loaded: + +```python +import ssl + +ssl_context = ssl.create_default_context() +ssl_context.load_cert_chain("client.crt", "client.key") + +auth0 = ServerClient( + domain="login.example.com", # self_managed_certs custom domain + client_id="", + use_mtls=True, + ssl_context=ssl_context, + secret="", +) +``` + +`use_mtls=True` requires an Enterprise tenant with the Highly Regulated Identity add-on, a `self_managed_certs` custom domain, and mTLS endpoint aliases enabled. It cannot be combined with `client_secret`, `client_assertion_signing_key`, or a per-call `dpop_key`. Each raises `ConfigurationError`. See the [Auth0 mTLS configuration docs](https://auth0.com/docs/get-started/applications/configure-mtls) for tenant-side setup steps. + +See [examples/MutualTLS.md](examples/MutualTLS.md) for the full setup guide, certificate generation, and token sender-constraining details. + ### 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/examples/MutualTLS.md b/examples/MutualTLS.md new file mode 100644 index 00000000..dba59b21 --- /dev/null +++ b/examples/MutualTLS.md @@ -0,0 +1,86 @@ +# Mutual TLS (mTLS) Client Authentication + +Authenticate to Auth0 with a TLS client certificate instead of a client secret (RFC 8705). The certificate is presented during the TLS handshake. No credential travels in the request body. + +## Prerequisites + +- Auth0 **Enterprise** tenant with the **Highly Regulated Identity** add-on +- A `self_managed_certs` **custom domain** configured on the tenant +- **Allow mTLS Endpoint Aliases** enabled on the tenant (Dashboard → Settings → Advanced) +- Client application's authentication method set to **mTLS** in Dashboard → Applications → Settings → Credentials + +## Generating a client certificate (development) + +```bash +# Self-signed CA + client cert (development only - use your PKI in production) +openssl req -x509 -newkey rsa:4096 -keyout ca.key -out ca.crt -days 365 -nodes \ + -subj "/CN=dev-ca" +openssl req -newkey rsa:2048 -keyout client.key -out client.csr -nodes \ + -subj "/CN=my-app-client" +openssl x509 -req -in client.csr -CA ca.crt -CAkey ca.key -CAcreateserial \ + -out client.crt -days 365 +``` + +## Wiring into `ServerClient` + +```python +import ssl +from auth0_server_python.auth_server.server_client import ServerClient + +ssl_context = ssl.create_default_context() # trusts system/public CAs for the server side +ssl_context.load_cert_chain("client.crt", "client.key") # attaches the client identity + +auth0 = ServerClient( + domain="login.example.com", # self_managed_certs custom domain + client_id="", + use_mtls=True, + ssl_context=ssl_context, + secret="", + authorization_params={ + "audience": "", + "scope": "openid profile email offline_access", + }, +) +``` + +The SDK passes `ssl_context` as `verify=ssl_context` to every `httpx.AsyncClient` it constructs, including the authlib client used for the authorization-code exchange. You never call `load_cert_chain` inside the SDK - the caller owns the TLS material. + +## Mutual exclusion + +`use_mtls=True` cannot be combined with: + +| Parameter | Reason | +|-----------|--------| +| `client_secret` | One client-auth method only - Auth0 rejects requests carrying both. | +| `client_assertion_signing_key` | Same - one method only. | +| `dpop_key` (per-call on `signin_with_passkey` / `mfa.verify`) | DPoP binds to its own key (`cnf.jkt`) and suppresses `cnf.x5t#S256`. Combining them silently defeats mTLS token binding. | + +All three raise `ConfigurationError` immediately (constructor for the first two, at the call site for DPoP). + +## Token sender-constraining + +When the target API has **Token Sender-Constraining (mTLS)** enabled, issued access tokens carry a `cnf.x5t#S256` claim binding the token to the certificate thumbprint. + +The SDK logs a warning at the `auth0_server_python.auth_server.server_client` logger whenever an access token returned by the authorization-code or refresh-token flow does not contain `cnf.x5t#S256`. If you see that warning, enable **Token Sender-Constraining (mTLS)** on the API resource server in the Auth0 Dashboard. + +## MFA under mTLS + +The client certificate is presented on all MFA API calls. The token-endpoint call inside `mfa.verify` is routed through the mTLS alias automatically. Challenge and enrollment calls (`/mfa/challenge`, `/mfa/associate`) go to the standard host. The certificate is still included in the TLS handshake, but whether it reaches the Auth0 backend depends on the proxy configuration. + +```python +await auth0.mfa.verify( + {"mfa_token": encrypted_token, "otp": "123456"}, +) +``` + +## Passkeys under mTLS + +The client certificate is presented on all passkey calls. The token-exchange step (`signin_with_passkey`) calls the token endpoint, which is served on the mTLS alias and routed correctly. + +Challenge and enrollment calls (`/passkey/challenge`, `/passkey/register`) go to the standard host - they are not listed in `mtls_endpoint_aliases`. The certificate is still included in the TLS handshake, but whether it reaches the Auth0 backend depends on the proxy configuration - the same behaviour as `/mfa/challenge`. + +## Passwordless under mTLS + +By default, Auth0 does not require client authentication on `/passwordless/start`. If the `enforce_client_authentication_on_passwordless_start` tenant flag is enabled on your tenant, the call will fail because `/passwordless/start` does not support certificate-based client authentication. + +The verify step (`passwordless_client.verify`) calls the token endpoint, which is served on the mTLS alias and routed correctly. diff --git a/references/flow-map.md b/references/flow-map.md index 6f9c8c19..c3d6b15f 100644 --- a/references/flow-map.md +++ b/references/flow-map.md @@ -17,6 +17,7 @@ Before working on a flow, read its entry points and supporting modules. Every fl | Passkeys | `passkey_signup_challenge`, `passkey_login_challenge`, `signin_with_passkey` | `auth_schemes/dpop_auth.py` — passkey sign-in is the DPoP-bound path | `examples/Passkeys.md` | | My Account | `MyAccountClient` (factors, authentication methods, enroll/verify) | `auth_schemes/dpop_auth.py`; stateless — every call takes a user token | `examples/MyAccountAuthenticationMethods.md` | | MCD | any flow — `domain` may be an async resolver | `_resolve_current_domain`, pitfall 5 in `references/pitfalls.md` | `examples/MultipleCustomDomains.md` | +| mTLS client auth | constructor `use_mtls` + `ssl_context` | `_resolve_token_endpoint`, `_apply_client_authentication`, `_warn_if_not_cert_bound`, `mfa_client.py` (`use_mtls`, `ssl_context`, `verify`, `token_endpoint_resolver`) | `examples/MutualTLS.md` | Two rules cut across every flow above, so check them on any change here: resolve the domain through `await self._resolve_current_domain(store_options)` rather than reading `self._domain`, and accept diff --git a/src/auth0_server_python/auth_server/mfa_client.py b/src/auth0_server_python/auth_server/mfa_client.py index a4e1dd73..943923d1 100644 --- a/src/auth0_server_python/auth_server/mfa_client.py +++ b/src/auth0_server_python/auth_server/mfa_client.py @@ -4,6 +4,7 @@ """ import json +import ssl import time from collections.abc import Awaitable, Callable from typing import TYPE_CHECKING, Any, Optional, Union @@ -74,6 +75,9 @@ def __init__( ] = None, mfa_token_ttl: int = DEFAULT_MFA_TOKEN_TTL, apply_client_authentication: Optional[Callable] = None, + use_mtls: bool = False, + ssl_context: Optional[ssl.SSLContext] = None, + token_endpoint_resolver: Optional[Callable[..., Awaitable[str]]] = None, ): if callable(domain): self._domain = None @@ -92,10 +96,15 @@ def __init__( raise ConfigurationError("mfa_token_ttl must be a positive number of seconds") self._mfa_token_ttl = mfa_token_ttl self._apply_client_authentication = apply_client_authentication + self._use_mtls = use_mtls + self._ssl_context = ssl_context + self._token_endpoint_resolver = token_endpoint_resolver def _get_http_client(self, **kwargs) -> httpx.AsyncClient: """Return an httpx.AsyncClient with default headers injected.""" headers = {**kwargs.pop("headers", {}), **self._headers} + if self._use_mtls and "verify" not in kwargs: + kwargs["verify"] = self._ssl_context return httpx.AsyncClient(headers=headers, **kwargs) def _apply_mfa_client_authentication(self, body: dict, base_url: str) -> None: @@ -502,8 +511,16 @@ 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 dpop_key is combined with use_mtls. DPoP and mTLS + use incompatible token-binding mechanisms and cannot be used together. ConfigurationError: If neither client_secret nor client_assertion_signing_key is configured. """ + if self._use_mtls and dpop_key is not None: + raise ConfigurationError( + "dpop_key cannot be combined with use_mtls. DPoP and mTLS bind tokens " + "differently. DPoP would take precedence and the token would not be " + "certificate-bound." + ) mfa_token = options.get("mfa_token") if not mfa_token: raise MfaTokenInvalidError() @@ -534,7 +551,10 @@ async def verify( ) try: - token_endpoint = f"{base_url}/oauth/token" + if self._use_mtls and self._token_endpoint_resolver: + token_endpoint = await self._token_endpoint_resolver(store_options) + else: + token_endpoint = f"{base_url}/oauth/token" async with self._get_http_client() as client: headers = {"Content-Type": "application/x-www-form-urlencoded"} diff --git a/src/auth0_server_python/auth_server/my_account_client.py b/src/auth0_server_python/auth_server/my_account_client.py index e4e10f23..d17f842c 100644 --- a/src/auth0_server_python/auth_server/my_account_client.py +++ b/src/auth0_server_python/auth_server/my_account_client.py @@ -1,4 +1,5 @@ import json +import ssl from typing import TYPE_CHECKING, Optional from urllib.parse import quote, unquote, urlparse @@ -46,20 +47,31 @@ class MyAccountClient: Client for interacting with the Auth0 MyAccount API. """ - def __init__(self, domain: str, headers: Optional[dict[str, str]] = None): + def __init__( + self, + domain: str, + headers: Optional[dict[str, str]] = None, + ssl_context: Optional[ssl.SSLContext] = None, + ): """ Initialize the MyAccount API client. Args: domain: Auth0 domain (e.g., '..auth0.com') headers: Optional default headers to include on every request + ssl_context: Optional SSL context for mTLS. When provided, the client + certificate is presented on every request so the My Account API can + verify cnf.x5t#S256 binding on cert-bound access tokens. """ self._domain = domain self._headers = headers or {} + self._ssl_context = ssl_context def _get_http_client(self, **kwargs) -> httpx.AsyncClient: """Return an httpx.AsyncClient with default headers injected.""" headers = {**kwargs.pop("headers", {}), **self._headers} + if self._ssl_context is not None and "verify" not in kwargs: + kwargs["verify"] = self._ssl_context return httpx.AsyncClient(headers=headers, **kwargs) @property diff --git a/src/auth0_server_python/auth_server/passwordless_client.py b/src/auth0_server_python/auth_server/passwordless_client.py index 5d4ed230..ba900dac 100644 --- a/src/auth0_server_python/auth_server/passwordless_client.py +++ b/src/auth0_server_python/auth_server/passwordless_client.py @@ -209,7 +209,12 @@ async def verify( e, ) - token_endpoint = metadata["token_endpoint"] + token_endpoint = client._resolve_token_endpoint(metadata) + if not token_endpoint: + raise PasswordlessVerifyError( + PasswordlessErrorCode.DISCOVERY_ERROR, + "Token endpoint missing in OIDC metadata", + ) origin_issuer = metadata.get("issuer") default_scope = ( @@ -393,7 +398,7 @@ async def _verify_id_token( if not id_token: raise PasswordlessVerifyError( PasswordlessErrorCode.VERIFY_FAILED, - "Token response did not include an ID token; ensure 'openid' scope is requested", + "Token response did not include an ID token. Ensure 'openid' scope is requested.", ) jwks = await client._get_jwks_cached(origin_domain, metadata) diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index fb984b5b..8ab5dcba 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -5,6 +5,8 @@ import asyncio import json +import logging +import ssl import time from collections import OrderedDict from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union @@ -101,6 +103,8 @@ # actor_token_type URN when the actor is sourced from the agent session's ID token. ID_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:id_token" +logger = logging.getLogger(__name__) + class ServerClient(Generic[TStoreOptions]): """ @@ -133,6 +137,8 @@ def __init__( pushed_authorization_requests: bool = False, organization: Optional[str] = None, mfa_token_ttl: int = DEFAULT_MFA_TOKEN_TTL, + use_mtls: bool = False, + ssl_context: Optional[ssl.SSLContext] = None, ): """ Initialize the Auth0 server client. @@ -158,6 +164,12 @@ def __init__( `mfa.verify()`/`mfa.challenge_authenticator()` reject it as expired. Defaults to 300 (5 minutes). Increase for authenticator flows that need more time (e.g. OOB push approval on a slow connection). + use_mtls: Enable mTLS (RFC 8705) client authentication. When True, the + client certificate in ssl_context is the sole credential - no + client_secret or client_assertion is sent in the request body. + ssl_context: TLS context carrying the client certificate and key. + Required when use_mtls=True. Build with ssl.create_default_context() + and load_cert_chain(). Raises: ConfigurationError: If `mfa_token_ttl` is not a positive number of seconds. @@ -189,6 +201,25 @@ def __init__( self._domain = domain_str self._domain_resolver = None + self._use_mtls = use_mtls + self._ssl_context = ssl_context + if use_mtls: + if ssl_context is None: + raise ConfigurationError( + "ssl_context is required when use_mtls=True. Create an ssl.SSLContext " + "and call load_cert_chain() to load the client certificate." + ) + if client_secret: + raise ConfigurationError( + "use_mtls cannot be combined with client_secret. The client " + "certificate is the sole credential under mTLS." + ) + if client_assertion_signing_key: + raise ConfigurationError( + "use_mtls cannot be combined with client_assertion_signing_key. " + "The client certificate is the sole credential under mTLS." + ) + self._client_id = client_id self._client_secret = client_secret self._client_assertion_signing_key = client_assertion_signing_key @@ -218,10 +249,13 @@ def __init__( client_id=client_id, client_secret=None if client_assertion_signing_key else client_secret, headers=self._telemetry_headers, + **({"verify": self._ssl_context} if self._use_mtls else {}), ) self._my_account_client = MyAccountClient( - domain=domain, headers=self._telemetry_headers + domain=domain, + headers=self._telemetry_headers, + **({"ssl_context": self._ssl_context} if self._use_mtls else {}), ) # Unified cache for OIDC metadata and JWKS per domain (LRU eviction + TTL) @@ -241,6 +275,9 @@ def __init__( session_establisher=self._establish_session_from_mfa_verify_response, mfa_token_ttl=mfa_token_ttl, apply_client_authentication=self._apply_client_authentication, + use_mtls=self._use_mtls, + ssl_context=self._ssl_context, + token_endpoint_resolver=self._resolve_mfa_token_endpoint if self._use_mtls else None, ) self._passwordless_client = PasswordlessClient(self) @@ -248,8 +285,56 @@ def __init__( def _get_http_client(self, **kwargs) -> httpx.AsyncClient: """Return an httpx.AsyncClient with telemetry headers injected.""" headers = {**kwargs.pop("headers", {}), **self._telemetry_headers} + if self._use_mtls and "verify" not in kwargs: + kwargs["verify"] = self._ssl_context return httpx.AsyncClient(headers=headers, **kwargs) + async def _resolve_mfa_token_endpoint(self, store_options) -> str: + """Resolve the token endpoint for MfaClient, applying the mTLS alias when enabled.""" + domain = await self._resolve_current_domain(store_options) + metadata = await self._get_oidc_metadata_cached(domain) + return self._resolve_token_endpoint(metadata) + + def _resolve_token_endpoint(self, metadata: dict) -> Optional[str]: + """Return the token endpoint, routed to the mTLS alias when mTLS is enabled. + + Returns: + The token endpoint URL, or None if not present and mTLS is not enabled. + + Raises: + ConfigurationError: If mTLS is enabled but the discovery document does not + advertise mtls_endpoint_aliases.token_endpoint. + """ + if self._use_mtls: + aliases = metadata.get("mtls_endpoint_aliases") or {} + endpoint = aliases.get("token_endpoint") + if not endpoint: + raise ConfigurationError( + "use_mtls is enabled but the authorization server discovery document " + "does not advertise mtls_endpoint_aliases.token_endpoint. Ensure mTLS " + "endpoint aliases are enabled on your Auth0 tenant." + ) + return endpoint + return metadata.get("token_endpoint") + + def _resolve_par_endpoint(self, metadata: dict) -> Optional[str]: + """Return the PAR endpoint, routed to the mTLS alias when mTLS is enabled. + + Under mTLS, raises ConfigurationError immediately if the alias is absent. + Under standard auth, returns None if the endpoint is missing (caller's guard handles it). + """ + if self._use_mtls: + aliases = metadata.get("mtls_endpoint_aliases") or {} + endpoint = aliases.get("pushed_authorization_request_endpoint") + if not endpoint: + raise ConfigurationError( + "use_mtls is enabled but the authorization server discovery document " + "does not advertise mtls_endpoint_aliases.pushed_authorization_request_endpoint. " + "Ensure mTLS endpoint aliases are enabled on your Auth0 tenant." + ) + return endpoint + return metadata.get("pushed_authorization_request_endpoint") + def _apply_client_authentication( self, params: dict, issuer: str, in_body: bool = False ) -> Optional[tuple[str, str]]: @@ -270,6 +355,11 @@ def _apply_client_authentication( for reserved in ("client_secret", "client_assertion", "client_assertion_type"): params.pop(reserved, None) + if self._use_mtls: + # The client certificate presented in the TLS handshake is the sole + # credential. No body credential or HTTP basic auth is sent. + return None + if self._client_assertion_signing_key: params["client_assertion"] = build_client_assertion( self._client_assertion_signing_key, @@ -537,6 +627,26 @@ async def _get_jwks_cached(self, domain: str, metadata: dict = None) -> dict: return jwks + def _warn_if_not_cert_bound(self, token_response: dict) -> None: + """Warn if the access token lacks a cnf.x5t#S256 claim.""" + access_token = token_response.get("access_token") + if not access_token: + return + try: + claims = jwt.decode( + access_token, + options={"verify_signature": False}, + ) + except jwt.InvalidTokenError: + return # opaque or unparseable token - nothing to assert + cnf = claims.get("cnf") if isinstance(claims, dict) else None + if not (isinstance(cnf, dict) and cnf.get("x5t#S256")): + logger.warning( + "mTLS is enabled but the access token does not contain a cnf.x5t#S256 " + "claim. The token is not certificate-bound. Configure Token " + "Sender-Constraining (mTLS) on the API resource server." + ) + # ============================================================================ # INTERACTIVE LOGIN FLOW # Handles browser-based authentication using the Authorization Code flow @@ -638,8 +748,7 @@ async def start_interactive_login( self._oauth.metadata = metadata # If PAR is enabled, use the PAR endpoint if self._pushed_authorization_requests: - par_endpoint = self._oauth.metadata.get( - "pushed_authorization_request_endpoint") + par_endpoint = self._resolve_par_endpoint(self._oauth.metadata) if not par_endpoint: raise ApiError( "configuration_error", "PAR is enabled but pushed_authorization_request_endpoint is missing in metadata") @@ -752,7 +861,9 @@ async def complete_interactive_login( ) try: - token_endpoint = self._oauth.metadata["token_endpoint"] + token_endpoint = self._resolve_token_endpoint(self._oauth.metadata) + if not token_endpoint: + raise ApiError("configuration_error", "Token endpoint missing in OIDC metadata") token_response = await self._oauth.fetch_token( token_endpoint, code=code, @@ -765,6 +876,9 @@ async def complete_interactive_login( raise ApiError( "token_error", f"Token exchange failed: {str(e)}", e) + if self._use_mtls: + self._warn_if_not_cert_bound(token_response) + # Use the userinfo field from the token_response for user claims user_info = token_response.get("userinfo") user_claims = None @@ -773,8 +887,6 @@ async def complete_interactive_login( # ID token `iat`, used to detect a ceiling that is already past at login. issued_at = None id_token = token_response.get("id_token") - # Verified ID token claims, retained so the session `sid` can be sourced - # from them (back-channel logout matches on `sid`). id_token_claims = None expected_org = transaction_data.organization @@ -973,7 +1085,7 @@ async def _establish_session_from_mfa_verify_response( id_token = token_response.get("id_token") if not id_token: raise MfaVerifyError( - "MFA verification response did not include an ID token; cannot create a session" + "MFA verification response did not include an ID token. Cannot create a session." ) origin_domain = await self._resolve_current_domain(store_options) @@ -1382,7 +1494,7 @@ async def get_token_by_refresh_token(self, options: dict[str, Any]) -> dict[str, # Fetch OIDC metadata from the correct domain metadata = await self._get_oidc_metadata_cached(domain) - token_endpoint = metadata.get("token_endpoint") + token_endpoint = self._resolve_token_endpoint(metadata) if not token_endpoint: raise ApiError("configuration_error", "Token endpoint missing in OIDC metadata") @@ -1448,6 +1560,9 @@ async def get_token_by_refresh_token(self, options: dict[str, Any]) -> dict[str, token_response["expires_at"] = int( time.time()) + token_response["expires_in"] + if self._use_mtls: + self._warn_if_not_cert_bound(token_response) + return token_response except Exception as e: @@ -1792,7 +1907,7 @@ async def backchannel_authentication_grant( domain = await self._resolve_current_domain(store_options) metadata = await self._get_oidc_metadata_cached(domain) - token_endpoint = metadata.get("token_endpoint") + token_endpoint = self._resolve_token_endpoint(metadata) if not token_endpoint: raise ApiError("configuration_error", "Token endpoint missing in OIDC metadata") @@ -2241,7 +2356,7 @@ async def get_token_for_connection(self, options: dict[str, Any]) -> dict[str, A # Fetch OIDC metadata from the correct domain metadata = await self._get_oidc_metadata_cached(domain) - token_endpoint = metadata.get("token_endpoint") + token_endpoint = self._resolve_token_endpoint(metadata) if not token_endpoint: raise ApiError("configuration_error", "Token endpoint missing in OIDC metadata") @@ -2621,7 +2736,7 @@ async def custom_token_exchange( domain = await self._resolve_current_domain(store_options) metadata = await self._get_oidc_metadata_cached(domain) - token_endpoint = metadata.get("token_endpoint") + token_endpoint = self._resolve_token_endpoint(metadata) if not token_endpoint: raise ApiError("configuration_error", "Token endpoint missing in OIDC metadata") @@ -3281,6 +3396,8 @@ async def signin_with_passkey( Raises: MissingRequiredArgumentError: If auth_session or authn_response is missing. + ConfigurationError: If dpop_key is combined with use_mtls. DPoP and mTLS + use incompatible token-binding mechanisms and cannot be used together. PasskeyError: If token exchange or session creation fails. OrganizationTokenValidationError: If an organization was requested but the token response included no ID token, or the ID token's org claim does @@ -3290,12 +3407,18 @@ async def signin_with_passkey( raise MissingRequiredArgumentError("auth_session") if authn_response is None: raise MissingRequiredArgumentError("authn_response") + if self._use_mtls and dpop_key is not None: + raise ConfigurationError( + "dpop_key cannot be combined with use_mtls. DPoP and mTLS bind tokens " + "differently. DPoP would take precedence and the token would not be " + "certificate-bound." + ) try: domain = await self._resolve_current_domain(store_options) metadata = await self._get_oidc_metadata_cached(domain) - token_endpoint = metadata.get("token_endpoint") + token_endpoint = self._resolve_token_endpoint(metadata) if not token_endpoint: raise PasskeyError(PasskeyErrorCode.TOKEN_EXCHANGE_FAILED, "Token endpoint missing in OIDC metadata") diff --git a/src/auth0_server_python/tests/test_mfa_client.py b/src/auth0_server_python/tests/test_mfa_client.py index 8db33945..b8bcb1bd 100644 --- a/src/auth0_server_python/tests/test_mfa_client.py +++ b/src/auth0_server_python/tests/test_mfa_client.py @@ -3,6 +3,7 @@ """ import json +import ssl from unittest.mock import AsyncMock, MagicMock import pytest @@ -1093,3 +1094,66 @@ async def mock_post(self_client, url, **kwargs): result = await client.verify({"mfa_token": _enc(), "otp": "123456"}) assert result.token_type == "Bearer" assert "DPoP" not in captured_request["kwargs"]["headers"] + + +# ============================================================================ +# mTLS - MfaClient SSLContext threading + DPoP exclusion + endpoint override +# ============================================================================ + + +def _mtls_mfa_client() -> MfaClient: + return MfaClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=None, + secret=SECRET, + use_mtls=True, + ssl_context=ssl.create_default_context(), + ) + + +@pytest.mark.asyncio +async def test_mfa_get_http_client_passes_ssl_context(mocker): + mfa = _mtls_mfa_client() + spy = mocker.patch("auth0_server_python.auth_server.mfa_client.httpx.AsyncClient") + mfa._get_http_client() + _, kwargs = spy.call_args + assert kwargs.get("verify") is mfa._ssl_context + + +@pytest.mark.asyncio +async def test_mfa_verify_rejects_dpop_under_mtls(): + mfa = _mtls_mfa_client() + with pytest.raises(ConfigurationError): + await mfa.verify({"mfa_token": _enc(), "otp": "123456"}, dpop_key=object()) + + +@pytest.mark.asyncio +async def test_mfa_verify_uses_token_endpoint_resolver_under_mtls(mocker): + async def resolver(store_options): + return "https://mtls.auth0.local/oauth/token" + + mfa = MfaClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=None, + secret=SECRET, + use_mtls=True, + ssl_context=ssl.create_default_context(), + token_endpoint_resolver=resolver, + ) + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={ + "access_token": "at", "token_type": "Bearer", "expires_in": 3600 + }) + captured = {} + + async def mock_post(self_client, url, **kwargs): + captured["url"] = url + return response + + mocker.patch("httpx.AsyncClient.post", new=mock_post) + + await mfa.verify({"mfa_token": _enc(), "otp": "123456"}) + assert captured["url"] == "https://mtls.auth0.local/oauth/token" diff --git a/src/auth0_server_python/tests/test_my_account_client.py b/src/auth0_server_python/tests/test_my_account_client.py index f917eeff..fc314f14 100644 --- a/src/auth0_server_python/tests/test_my_account_client.py +++ b/src/auth0_server_python/tests/test_my_account_client.py @@ -1,5 +1,6 @@ import base64 import json +import ssl from unittest.mock import ANY, AsyncMock, MagicMock import httpx @@ -1422,3 +1423,24 @@ def test_dpop_auth_flow_no_retry_on_non_401(): assert not retried + +# ============================================================================= +# mTLS ssl_context wiring +# ============================================================================= + + +def test_get_http_client_passes_ssl_context_when_set(mocker): + ctx = ssl.create_default_context() + client = MyAccountClient(domain="auth0.local", ssl_context=ctx) + spy = mocker.patch("auth0_server_python.auth_server.my_account_client.httpx.AsyncClient") + client._get_http_client() + _, kwargs = spy.call_args + assert kwargs.get("verify") is ctx + + +def test_get_http_client_omits_verify_without_ssl_context(mocker): + client = MyAccountClient(domain="auth0.local") + spy = mocker.patch("auth0_server_python.auth_server.my_account_client.httpx.AsyncClient") + client._get_http_client() + _, kwargs = spy.call_args + assert "verify" not in kwargs diff --git a/src/auth0_server_python/tests/test_passwordless_client.py b/src/auth0_server_python/tests/test_passwordless_client.py index c875c991..954de277 100644 --- a/src/auth0_server_python/tests/test_passwordless_client.py +++ b/src/auth0_server_python/tests/test_passwordless_client.py @@ -2,6 +2,7 @@ Tests for PasswordlessClient embedded passwordless (OTP + magic link). """ +import ssl from unittest.mock import AsyncMock, MagicMock import jwt @@ -1182,3 +1183,118 @@ async def test_verify_no_client_auth_configured_raises_configuration_error(self, connection="email", email="user@example.com", verification_code="123456" ) ) + + +# ── mTLS ───────────────────────────────────────────────────────────────────── + +MTLS_METADATA = { + "token_endpoint": f"https://{DOMAIN}/oauth/token", + "issuer": ISSUER, + "mtls_endpoint_aliases": {"token_endpoint": f"https://mtls.{DOMAIN}/oauth/token"}, +} + + +def _make_mtls_client(**overrides) -> ServerClient: + kwargs = { + "domain": DOMAIN, + "client_id": CLIENT_ID, + "use_mtls": True, + "ssl_context": ssl.create_default_context(), + "secret": SECRET, + "redirect_uri": REDIRECT_URI, + "transaction_store": AsyncMock(), + "state_store": AsyncMock(), + } + kwargs.update(overrides) + return ServerClient(**kwargs) + + +def _mock_http_via_httpx(mocker, mock_response): + """Patch httpx.AsyncClient directly and return (spy, mock_http).""" + mock_http = AsyncMock() + mock_http.post = AsyncMock(return_value=mock_response) + mock_ctx_mgr = MagicMock() + mock_ctx_mgr.__aenter__ = AsyncMock(return_value=mock_http) + mock_ctx_mgr.__aexit__ = AsyncMock(return_value=False) + spy = mocker.patch("httpx.AsyncClient", return_value=mock_ctx_mgr) + return spy, mock_http + + +class TestMtls: + def _patch_verify_deps(self, client, mocker, claims): + mocker.patch.object(client, "_get_oidc_metadata_cached", return_value=MTLS_METADATA) + mocker.patch.object( + client, "_get_jwks_cached", return_value={"keys": [{"kty": "RSA", "kid": "k1"}]} + ) + mocker.patch.object(client, "_verify_and_decode_jwt", return_value=claims) + + def _token_response(self): + return MagicMock( + status_code=200, + json=MagicMock( + return_value={"access_token": "at", "id_token": "idt", "expires_in": 3600, "scope": "openid"} + ), + ) + + @pytest.mark.asyncio + async def test_verify_routes_token_endpoint_through_mtls_alias(self, mocker): + client = _make_mtls_client() + claims = {"iss": ISSUER, "sub": "auth0|1", "sid": "s1", "iat": 1_000} + self._patch_verify_deps(client, mocker, claims) + http = _mock_http(client, 200, {"access_token": "at", "id_token": "idt", "expires_in": 3600, "scope": "openid"}) + + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + + assert http.post.call_args.args[0] == f"https://mtls.{DOMAIN}/oauth/token" + + @pytest.mark.asyncio + async def test_start_omits_client_secret_and_uses_ssl_context_under_mtls(self, mocker): + ctx = ssl.create_default_context() + client = _make_mtls_client(ssl_context=ctx) + spy, mock_http = _mock_http_via_httpx(mocker, MagicMock(status_code=200, json=MagicMock(return_value={"_id": "req_1"}))) + + await client.passwordless.start( + StartPasswordlessEmailOptions(email="user@example.com", send="code") + ) + + _, kwargs = spy.call_args + assert kwargs.get("verify") is ctx + body = mock_http.post.call_args.kwargs["json"] + assert "client_secret" not in body + + @pytest.mark.asyncio + async def test_verify_omits_client_secret_under_mtls(self, mocker): + client = _make_mtls_client() + claims = {"iss": ISSUER, "sub": "auth0|1", "sid": "s1", "iat": 1_000} + self._patch_verify_deps(client, mocker, claims) + http = _mock_http(client, 200, {"access_token": "at", "id_token": "idt", "expires_in": 3600, "scope": "openid"}) + + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + + body = http.post.call_args.kwargs["data"] + assert "client_secret" not in body + + @pytest.mark.asyncio + async def test_verify_presents_ssl_context_under_mtls(self, mocker): + ctx = ssl.create_default_context() + client = _make_mtls_client(ssl_context=ctx) + claims = {"iss": ISSUER, "sub": "auth0|1", "sid": "s1", "iat": 1_000} + self._patch_verify_deps(client, mocker, claims) + spy, _ = _mock_http_via_httpx(mocker, self._token_response()) + + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + + _, kwargs = spy.call_args + assert kwargs.get("verify") is ctx diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 02b8d534..bcce0616 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -1,5 +1,6 @@ import base64 import json +import ssl import time import unicodedata from unittest.mock import ANY, AsyncMock, MagicMock, patch @@ -277,6 +278,43 @@ async def test_par_request_caller_cannot_inject_client_assertion(mocker): assert "client_assertion_type" not in posted +@pytest.mark.asyncio +async def test_par_request_uses_mtls_alias_endpoint(mocker): + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="", + pushed_authorization_requests=True, + authorization_params={"redirect_uri": "https://app/cb"}, + state_store=AsyncMock(), + transaction_store=AsyncMock(), + ) + mtls_metadata = { + "issuer": "https://auth0.local/", + "authorization_endpoint": "https://auth0.local/authorize", + "pushed_authorization_request_endpoint": "https://auth0.local/oauth/par", + "mtls_endpoint_aliases": { + "pushed_authorization_request_endpoint": "https://mtls.auth0.local/oauth/par", + }, + } + mocker.patch.object(client, "_get_oidc_metadata_cached", AsyncMock(return_value=mtls_metadata)) + mocker.patch.object(client._oauth, "metadata", mtls_metadata) + + 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() + + called_url = mock_post.call_args[0][0] + assert called_url == "https://mtls.auth0.local/oauth/par" + assert "client_secret" not in mock_post.call_args.kwargs["data"] + + @pytest.mark.asyncio async def test_complete_interactive_login_no_transaction(): mock_transaction_store = AsyncMock() @@ -2417,6 +2455,34 @@ async def test_backchannel_authentication_grant_json_decode_error(mocker): assert exc.value.code == "invalid_response" assert "Failed to parse token response as JSON" in str(exc.value) +@pytest.mark.asyncio +async def test_backchannel_authentication_grant_uses_mtls_token_endpoint(mocker): + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="some-secret", + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "token_endpoint": "https://auth0.local/oauth/token", + "mtls_endpoint_aliases": {"token_endpoint": "https://mtls.auth0.local/oauth/token"}, + }, + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + mock_response = AsyncMock() + mock_response.status_code = 200 + mock_response.json = MagicMock(return_value={"access_token": "at", "expires_in": 3600}) + mock_post.return_value = mock_response + + await client.backchannel_authentication_grant("auth_req_123") + + assert mock_post.call_args[0][0] == "https://mtls.auth0.local/oauth/token" + assert "client_secret" not in mock_post.call_args.kwargs["data"] + @pytest.mark.asyncio async def test_get_token_for_connection_success(mocker): client = ServerClient( @@ -2502,6 +2568,35 @@ async def test_get_token_for_connection_exchange_failed(mocker): mock_post.assert_awaited_once() +@pytest.mark.asyncio +async def test_get_token_for_connection_uses_mtls_token_endpoint(mocker): + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="some-secret", + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "token_endpoint": "https://auth0.local/oauth/token", + "mtls_endpoint_aliases": {"token_endpoint": "https://mtls.auth0.local/oauth/token"}, + }, + ) + 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}) + success_response.headers = {} + mock_post.return_value = success_response + + await client.get_token_for_connection({"connection": "github", "refresh_token": "rt"}) + + assert mock_post.call_args[0][0] == "https://mtls.auth0.local/oauth/token" + assert "client_secret" not in mock_post.call_args.kwargs["data"] + @pytest.mark.asyncio async def test_get_token_by_refresh_token_success(mocker): client = ServerClient( @@ -2615,6 +2710,67 @@ async def test_get_token_by_refresh_token_mfa_required_raises_api_error_with_raw assert exc.value.mfa_token == "raw_server_mfa_token" assert exc.value.mfa_requirements is None +@pytest.mark.asyncio +async def test_get_token_by_refresh_token_uses_mtls_token_endpoint(mocker): + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="some-secret", + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "token_endpoint": "https://auth0.local/oauth/token", + "mtls_endpoint_aliases": {"token_endpoint": "https://mtls.auth0.local/oauth/token"}, + }, + ) + 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"}) + + assert mock_post.call_args[0][0] == "https://mtls.auth0.local/oauth/token" + assert "client_secret" not in mock_post.call_args.kwargs["data"] + + +@pytest.mark.asyncio +async def test_get_token_by_refresh_token_warns_when_token_not_cert_bound(mocker): + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="", + ) + mocker.patch.object( + client, "_get_oidc_metadata_cached", + return_value={ + "token_endpoint": "https://auth0.local/oauth/token", + "mtls_endpoint_aliases": {"token_endpoint": "https://mtls.auth0.local/oauth/token"}, + }, + ) + mock_post = mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock( + return_value={ + "access_token": jwt.encode({"sub": "user123"}, "s", algorithm="HS256"), + "expires_in": 3600, + } + ) + mock_post.return_value = response + mock_logger = mocker.patch("auth0_server_python.auth_server.server_client.logger") + + await client.get_token_by_refresh_token({"refresh_token": "abc"}) + + mock_logger.warning.assert_called_once() + # ============================================================================= # Private Key JWT (client assertion) Client Authentication @@ -4264,6 +4420,47 @@ async def test_custom_token_exchange_act_dropped_on_issuer_mismatch(mocker): assert result.act is None +@pytest.mark.asyncio +async def test_custom_token_exchange_uses_mtls_token_endpoint(mocker): + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="some-secret", + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "token_endpoint": "https://auth0.local/oauth/token", + "mtls_endpoint_aliases": {"token_endpoint": "https://mtls.auth0.local/oauth/token"}, + }, + ) + 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) + + await client.custom_token_exchange(CustomTokenExchangeOptions( + subject_token="custom-token", + subject_token_type="urn:acme:token", + audience="https://api.example.com", + )) + + assert mock_httpx_client.post.call_args[0][0] == "https://mtls.auth0.local/oauth/token" + assert "client_secret" not in mock_httpx_client.post.call_args.kwargs["data"] + # ============================================================================= # Login with Custom Token Exchange Tests @@ -9001,6 +9198,23 @@ async def test_signin_with_passkey_client_default_org_is_validated_against_id_to state_store.set.assert_not_awaited() +@pytest.mark.asyncio +async def test_signin_with_passkey_rejects_dpop_under_mtls(mocker): + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="", + ) + with pytest.raises(ConfigurationError): + await client.signin_with_passkey( + auth_session="sess", + authn_response=mocker.Mock(), + dpop_key=object(), + ) + + # ============================================================================= # IPSIE session_expiry enforcement # ============================================================================= @@ -9677,3 +9891,291 @@ async def test_complete_interactive_login_milliseconds_ceiling_fails_open(mocker mock_state_store.set.assert_awaited_once() stored_state = mock_state_store.set.call_args.args[1] assert stored_state.internal.session_expires_at is None + + +@pytest.mark.asyncio +async def test_complete_interactive_login_uses_mtls_token_endpoint(mocker): + mock_tx_store = AsyncMock() + mock_tx_store.get.return_value = TransactionData( + code_verifier="cv", + domain="auth0.local", + app_state=None, + ) + mock_tx_store.delete = AsyncMock() + mock_state_store = AsyncMock() + mock_state_store.get = AsyncMock(return_value=None) + mock_state_store.set = AsyncMock() + + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="", + redirect_uri="https://app/cb", + transaction_store=mock_tx_store, + state_store=mock_state_store, + ) + + mtls_metadata = { + "issuer": "https://auth0.local/", + "token_endpoint": "https://auth0.local/oauth/token", + "mtls_endpoint_aliases": {"token_endpoint": "https://mtls.auth0.local/oauth/token"}, + } + mocker.patch.object(client, "_get_oidc_metadata_cached", AsyncMock(return_value=mtls_metadata)) + mocker.patch.object(client._oauth, "metadata", mtls_metadata) + + fetch_token = AsyncMock(return_value={"access_token": "at", "expires_in": 3600}) + mocker.patch.object(client._oauth, "fetch_token", fetch_token) + + await client.complete_interactive_login("https://app/cb?code=abc&state=xyz") + + called_endpoint = fetch_token.call_args[0][0] + assert called_endpoint == "https://mtls.auth0.local/oauth/token" + + +@pytest.mark.asyncio +async def test_complete_interactive_login_warns_when_token_not_cert_bound(mocker): + mock_tx_store = AsyncMock() + mock_tx_store.get.return_value = TransactionData( + code_verifier="cv", domain="auth0.local", app_state=None + ) + mock_tx_store.delete = AsyncMock() + mock_state_store = AsyncMock() + mock_state_store.get = AsyncMock(return_value=None) + mock_state_store.set = AsyncMock() + + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ssl.create_default_context(), + secret="", + redirect_uri="https://app/cb", + transaction_store=mock_tx_store, + state_store=mock_state_store, + ) + mtls_metadata = { + "issuer": "https://auth0.local/", + "token_endpoint": "https://auth0.local/oauth/token", + "mtls_endpoint_aliases": {"token_endpoint": "https://mtls.auth0.local/oauth/token"}, + } + mocker.patch.object(client, "_get_oidc_metadata_cached", AsyncMock(return_value=mtls_metadata)) + mocker.patch.object(client._oauth, "metadata", mtls_metadata) + mocker.patch.object( + client._oauth, "fetch_token", + AsyncMock(return_value={ + "access_token": jwt.encode({"sub": "user123"}, "s", algorithm="HS256"), + "expires_in": 3600, + }) + ) + mock_logger = mocker.patch("auth0_server_python.auth_server.server_client.logger") + + await client.complete_interactive_login("https://app/cb?code=abc&state=xyz") + + mock_logger.warning.assert_called_once() + + +# ============================================================================ +# mTLS CLIENT AUTHENTICATION +# ============================================================================ + + +def _dummy_ssl_context(): + return ssl.create_default_context() + + +@pytest.mark.asyncio +async def test_mtls_requires_ssl_context(): + with pytest.raises(ConfigurationError): + ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + secret="", + ) + + +@pytest.mark.asyncio +async def test_mtls_rejects_client_secret(): + with pytest.raises(ConfigurationError): + ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + use_mtls=True, + ssl_context=_dummy_ssl_context(), + secret="", + ) + + +@pytest.mark.asyncio +async def test_mtls_rejects_client_assertion_signing_key(): + with pytest.raises(ConfigurationError): + ServerClient( + domain="auth0.local", + client_id="", + client_assertion_signing_key="", + use_mtls=True, + ssl_context=_dummy_ssl_context(), + secret="", + ) + + +@pytest.mark.asyncio +async def test_mtls_happy_path_constructs(): + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=_dummy_ssl_context(), + secret="", + ) + assert client._use_mtls is True + assert client._ssl_context is not None + + +@pytest.mark.asyncio +async def test_mtls_oauth_client_constructed_with_no_credential_and_ssl_context(mocker): + ctx = _dummy_ssl_context() + spy = mocker.patch("auth0_server_python.auth_server.server_client.AsyncOAuth2Client") + ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ctx, + secret="", + ) + _, kwargs = spy.call_args + assert kwargs.get("client_secret") is None + assert kwargs.get("verify") is ctx + + +@pytest.mark.asyncio +async def test_mtls_get_http_client_passes_ssl_context(mocker): + ctx = _dummy_ssl_context() + client = ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=ctx, + secret="", + ) + spy = mocker.patch("auth0_server_python.auth_server.server_client.httpx.AsyncClient") + client._get_http_client() + _, kwargs = spy.call_args + assert kwargs.get("verify") is ctx + + +def _mtls_client(): + return ServerClient( + domain="auth0.local", + client_id="", + use_mtls=True, + ssl_context=_dummy_ssl_context(), + secret="", + ) + + +@pytest.mark.asyncio +async def test_resolve_token_endpoint_uses_alias_under_mtls(): + client = _mtls_client() + metadata = { + "token_endpoint": "https://auth0.local/oauth/token", + "mtls_endpoint_aliases": {"token_endpoint": "https://mtls.auth0.local/oauth/token"}, + } + assert client._resolve_token_endpoint(metadata) == "https://mtls.auth0.local/oauth/token" + + +@pytest.mark.asyncio +async def test_resolve_token_endpoint_raises_when_alias_missing(): + client = _mtls_client() + with pytest.raises(ConfigurationError): + client._resolve_token_endpoint({"token_endpoint": "https://auth0.local/oauth/token"}) + + +@pytest.mark.asyncio +async def test_resolve_token_endpoint_standard_when_not_mtls(): + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + secret="", + ) + metadata = {"token_endpoint": "https://auth0.local/oauth/token"} + assert client._resolve_token_endpoint(metadata) == "https://auth0.local/oauth/token" + + +def test_resolve_par_endpoint_uses_alias_under_mtls(): + client = _mtls_client() + metadata = { + "pushed_authorization_request_endpoint": "https://auth0.local/oauth/par", + "mtls_endpoint_aliases": {"pushed_authorization_request_endpoint": "https://mtls.auth0.local/oauth/par"}, + } + assert client._resolve_par_endpoint(metadata) == "https://mtls.auth0.local/oauth/par" + + +def test_resolve_par_endpoint_raises_when_mtls_alias_missing(): + client = _mtls_client() + with pytest.raises(ConfigurationError): + client._resolve_par_endpoint({"pushed_authorization_request_endpoint": "https://auth0.local/oauth/par"}) + + +def test_resolve_par_endpoint_standard_when_not_mtls(): + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + secret="", + ) + metadata = {"pushed_authorization_request_endpoint": "https://auth0.local/oauth/par"} + assert client._resolve_par_endpoint(metadata) == "https://auth0.local/oauth/par" + + +@pytest.mark.asyncio +async def test_apply_client_auth_mtls_returns_none_and_strips_creds(): + client = _mtls_client() + params = {"grant_type": "refresh_token", "client_secret": "leaked", "client_assertion": "x"} + result = client._apply_client_authentication(params, "https://auth0.local/") + assert result is None + assert "client_secret" not in params + assert "client_assertion" not in params + assert "client_assertion_type" not in params + + +def _make_access_token(cnf=None): + payload = {"sub": "user123"} + if cnf is not None: + payload["cnf"] = cnf + return jwt.encode(payload, "test-secret", algorithm="HS256") + + +def test_warn_if_not_cert_bound_warns_when_cnf_absent(mocker): + client = _mtls_client() + mock_logger = mocker.patch("auth0_server_python.auth_server.server_client.logger") + client._warn_if_not_cert_bound({"access_token": _make_access_token()}) + mock_logger.warning.assert_called_once() + + +def test_warn_if_not_cert_bound_no_warn_when_cert_bound(mocker): + client = _mtls_client() + mock_logger = mocker.patch("auth0_server_python.auth_server.server_client.logger") + client._warn_if_not_cert_bound({"access_token": _make_access_token(cnf={"x5t#S256": "abc123"})}) + mock_logger.warning.assert_not_called() + + +def test_warn_if_not_cert_bound_silent_on_opaque_token(mocker): + client = _mtls_client() + mock_logger = mocker.patch("auth0_server_python.auth_server.server_client.logger") + client._warn_if_not_cert_bound({"access_token": "opaque-token"}) + mock_logger.warning.assert_not_called() + + +def test_warn_if_not_cert_bound_silent_when_no_access_token(mocker): + client = _mtls_client() + mock_logger = mocker.patch("auth0_server_python.auth_server.server_client.logger") + client._warn_if_not_cert_bound({}) + mock_logger.warning.assert_not_called() + + +