diff --git a/README.md b/README.md index 2ff26cf..4e3bf8f 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 0000000..96e97fc --- /dev/null +++ b/examples/MutualTLS.md @@ -0,0 +1,87 @@ +# 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. If your tokens do not contain this claim, enable **Token Sender-Constraining (mTLS)** on the API resource server in the Auth0 dashboard. + +To verify the thumbprint yourself: + +```bash +openssl x509 -in client.crt -outform DER | openssl dgst -sha256 -binary | openssl enc -base64 | tr '+/' '-_' | tr -d '=' +# Compare the output to the cnf.x5t#S256 claim in the decoded access token. +``` + +## 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 stay on the standard host, which does not request a client certificate. + +```python +await auth0.mfa.verify( + {"mfa_token": encrypted_token, "otp": "123456"}, +) +``` + +## Passkeys under mTLS + +`/passkey/challenge` and `/passkey/register` are not served on the mTLS endpoint aliases. Auth0 only accepts `client_secret` as the credential on those endpoints - the client certificate is not a valid credential there. + +Because `use_mtls=True` forbids `client_secret` at construction time, an mTLS-configured client has no valid credential for `passkey_login_challenge` and `passkey_signup_challenge`. Those calls will be rejected by Auth0 if the application is registered as a confidential client. + +`signin_with_passkey` (the token-exchange step) is not affected - it 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 6f9c8c1..9567948 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_override`) | `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 a4e1dd7..943923d 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 e4e10f2..d17f842 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 5d4ed23..4f8fe54 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 = ( diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index fb984b5..aa40943 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -5,6 +5,7 @@ import asyncio import json +import ssl import time from collections import OrderedDict from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union @@ -133,6 +134,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 +161,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 +198,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 +246,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 +272,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 +282,52 @@ 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. + + Under mTLS, raises ConfigurationError immediately if the alias is absent. + Under standard auth, returns None if token_endpoint is missing (caller's guard handles it). + """ + 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 +348,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, @@ -638,8 +721,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 +834,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, @@ -1382,7 +1466,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") @@ -1792,7 +1876,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 +2325,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 +2705,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 +3365,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 +3376,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 8db3394..8c301ad 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 f917eef..fc314f1 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_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 02b8d53..846f48e 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,42 @@ 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" + + @pytest.mark.asyncio async def test_complete_interactive_login_no_transaction(): mock_transaction_store = AsyncMock() @@ -2417,6 +2454,33 @@ 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" + @pytest.mark.asyncio async def test_get_token_for_connection_success(mocker): client = ServerClient( @@ -2502,6 +2566,34 @@ 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" + @pytest.mark.asyncio async def test_get_token_by_refresh_token_success(mocker): client = ServerClient( @@ -2615,6 +2707,33 @@ 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" + # ============================================================================= # Private Key JWT (client assertion) Client Authentication @@ -4264,6 +4383,46 @@ 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" + # ============================================================================= # Login with Custom Token Exchange Tests @@ -9001,6 +9160,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 +9853,198 @@ 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" + + +# ============================================================================ +# 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_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 + + +