diff --git a/README.md b/README.md index 0ba9b43..55cce7e 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,10 @@ Sign users up or in with [WebAuthn](https://www.w3.org/TR/webauthn-2/) passkeys Let a logged-in user manage their own enrolled authentication methods — enroll a new passkey (or other factor), list, rename, and delete — via the [My Account API](https://auth0.com/docs/manage-users/my-account-api). For obtaining a scoped token, the enroll/verify ceremony, listing, updating, deleting, and error handling, see [examples/MyAccountAuthenticationMethods.md](examples/MyAccountAuthenticationMethods.md). +### 9. DPoP — Sender-Constrained Tokens (Passkeys & MyAccount) + +Bind tokens to a key your server holds ([RFC 9449](https://www.rfc-editor.org/rfc/rfc9449)) so a stolen token alone cannot be replayed. DPoP is supported for Passkey sign-in (`signin_with_passkey`) and the authentication-methods/factors methods on `MyAccountClient`. For key generation and usage, see [examples/Passkeys.md](examples/Passkeys.md#3-dpop-bound-passkey-tokens-optional) and [examples/MyAccountAuthenticationMethods.md](examples/MyAccountAuthenticationMethods.md#dpop). + ## Feedback ### Contributing diff --git a/examples/MFA.md b/examples/MFA.md index 095061c..a91ffe0 100644 --- a/examples/MFA.md +++ b/examples/MFA.md @@ -34,6 +34,7 @@ The Auth0 MFA API allows you to manage multi-factor authentication for users in - [Verify with OTP](#verify-with-otp) - [Verify with Recovery Code](#verify-with-recovery-code) - [Verify with Push Notification (Polling)](#verify-with-push-notification-polling) + - [Verify with DPoP (sender-constrained tokens)](#verify-with-dpop-sender-constrained-tokens) - [Session Persistence](#session-persistence) - [Automatic Session Update](#automatic-session-update) - [Manual Session Update](#manual-session-update) @@ -519,6 +520,27 @@ async def poll_push_verification(server_client, mfa_token, oob_code, timeout=60) > [!NOTE] > When polling for push notification approval, the API returns an `authorization_pending` error until the user approves or denies the request. A `slow_down` error indicates you should increase the polling interval. +### Verify with DPoP (sender-constrained tokens) + +When the login that triggered MFA was DPoP-bound (for example a `signin_with_passkey(dpop_key=...)` that returned `MfaRequiredError`), pass the **same** `dpop_key` to `verify()` so the token minted by the MFA step-up stays sender-constrained: + +```python +verify_response = await server_client.mfa.verify( + { + "mfa_token": mfa_token, + "otp": "123456", + }, + dpop_key=dpop_key, # the same EC P-256 key the login was bound to +) + +assert verify_response.token_type == "DPoP" +``` + +The SDK does not store your private key, so you must re-supply it on the `verify()` call. It attaches a token-endpoint DPoP proof, transparently handles the server-nonce challenge, and **rejects a Bearer downgrade** — if `dpop_key` was supplied but the server returned an unbound token (or vice versa), `verify()` raises `MfaVerifyError` instead of silently dropping the sender constraint. + +> [!WARNING] +> The `dpop_key` is a **Tier 0 secret**. Keep it in your secret store, never log it, use one key per user/session, and use **EC P-256 only**. + ## Session Persistence By default, `verify()` returns tokens without persisting them to the session store. However, you can automatically persist tokens by setting `persist=True`. diff --git a/examples/MyAccountAuthenticationMethods.md b/examples/MyAccountAuthenticationMethods.md index 94c434d..42662b5 100644 --- a/examples/MyAccountAuthenticationMethods.md +++ b/examples/MyAccountAuthenticationMethods.md @@ -6,7 +6,7 @@ The [My Account API](https://auth0.com/docs/manage-users/my-account-api) lets a > This is a different My Account resource from [Connected Accounts](ConnectedAccounts.md) (Token Vault). Connected-accounts management is exposed as convenience methods on `ServerClient`; **authentication-method management is on `MyAccountClient` directly**, because each call takes a user access token you obtain yourself. The two share the same My Account setup (activation, MRRT, scopes, `MyAccountApiError`) — see [ConnectedAccounts.md → Pre-requisites](ConnectedAccounts.md#pre-requisites) for that common setup. > [!NOTE] -> To **sign in** with a passkey (rather than manage one), see [examples/Passkeys.md](Passkeys.md). +> To **sign in** with a passkey (rather than manage one), see [examples/Passkeys.md](Passkeys.md). To **bind these calls to a held key**, pass an optional `dpop_key` — see [DPoP](#dpop) below. ## Table of Contents @@ -18,6 +18,7 @@ The [My Account API](https://auth0.com/docs/manage-users/my-account-api) lets a - [4. Get a single authentication method](#4-get-a-single-authentication-method) - [5. Update (rename) an authentication method](#5-update-rename-an-authentication-method) - [6. Delete an authentication method](#6-delete-an-authentication-method) +- [DPoP](#dpop) - [Error Handling](#error-handling) ## Prerequisites @@ -174,6 +175,24 @@ await my_account.delete_authentication_method( # Returns None on success (HTTP 204). ``` +## DPoP + +Every method above accepts an optional `dpop_key` to present a sender-constrained token (`Authorization: DPoP` + a per-request proof) instead of a Bearer token ([RFC 9449](https://www.rfc-editor.org/rfc/rfc9449)). Pass the **same key** the access token was bound to at sign-in: + +```python +from jwcrypto import jwk + +dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") # the key the token was bound to + +methods = await my_account.list_authentication_methods( + access_token=access_token, + dpop_key=dpop_key, +) +``` + +> [!WARNING] +> The `dpop_key` private key is a **Tier 0 secret**. Keep it in your secret store (KMS/HSM), never log it (`repr()` is redacted, but `key.export_private()` is not), use **one key per user/session** (never share across principals), and use **EC P-256 only** — any other key type fails closed with a `ValueError`. + ## Error Handling All errors inherit from `Auth0Error`. My Account API errors are `MyAccountApiError` (RFC 7807 problem-details, carrying `status`, `detail`, and optional `validation_errors`); missing arguments raise `MissingRequiredArgumentError`; transport or non-JSON responses surface as `ApiError`. diff --git a/examples/Passkeys.md b/examples/Passkeys.md index 835d85c..0ad39bb 100644 --- a/examples/Passkeys.md +++ b/examples/Passkeys.md @@ -14,6 +14,7 @@ Passkeys let users sign up and log in with [WebAuthn](https://www.w3.org/TR/weba - [Prerequisites](#prerequisites) - [1. Passkey Signup](#1-passkey-signup) - [2. Passkey Login](#2-passkey-login) +- [3. DPoP-bound passkey tokens (optional)](#3-dpop-bound-passkey-tokens-optional) - [Completing MFA on a passkey login (and where the session comes from)](#completing-mfa-on-a-passkey-login-and-where-the-session-comes-from) - [Error Handling](#error-handling) @@ -135,6 +136,31 @@ result = await server_client.signin_with_passkey( > [!NOTE] > The SDK is transparent to the signup-vs-login difference in the credential `response` — both flow through the same `PasskeyAuthResponse.response` dict. Send exactly the keys the browser produced. +## 3. DPoP-bound passkey tokens (optional) + +Pass an optional `dpop_key` to bind the issued tokens to a key your server holds ([RFC 9449](https://www.rfc-editor.org/rfc/rfc9449)), so a stolen token alone cannot be replayed. DPoP is **opt-in**: omit `dpop_key` and sign-in returns ordinary Bearer tokens with no behaviour change. + +```python +from jwcrypto import jwk + +dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") # you generate and keep this key (Tier 0) + +result = await server_client.signin_with_passkey( + auth_session=challenge.auth_session, + authn_response=authn_response, + dpop_key=dpop_key, + store_options={"request": request, "response": response}, +) +``` + +When `dpop_key` is supplied, the SDK attaches a token-endpoint proof so Auth0 issues a DPoP-bound token, transparently handles the server-nonce challenge, and **rejects a Bearer downgrade** — if the server returns an unbound token, `signin_with_passkey` raises `PasskeyError` rather than silently accepting a token bound to a key it never used. + +> [!TIP] +> Reuse the **same** `dpop_key` for any subsequent My Account API calls made with the resulting token — the token is bound to that one key. See [MyAccountAuthenticationMethods.md → DPoP](MyAccountAuthenticationMethods.md#dpop). + +> [!WARNING] +> The `dpop_key` private key is a **Tier 0 secret**. Keep it in your secret store (KMS/HSM), never log it (`repr()` is redacted, but `key.export_private()` is not), use **one key per user/session** (never share across principals), and use **EC P-256 only** — any other key type fails closed with a `ValueError` before any network call. + ### Completing MFA on a passkey login (and where the session comes from) When a passkey login needs a second factor, `signin_with_passkey` raises `MfaRequiredError` **before** it creates a session. You finish the login by challenging and verifying through `client.mfa`, then **store the returned tokens yourself** — on this path the SDK does not persist the session for you (`persist` defaults to `False`, and there is no existing session to update yet): @@ -146,6 +172,7 @@ try: result = await server_client.signin_with_passkey( auth_session=auth_session, authn_response=authn_response, + dpop_key=dpop_key, # optional; omit for Bearer tokens store_options={"request": request, "response": response}, ) # No MFA needed: signin_with_passkey already persisted the session for you. @@ -158,10 +185,12 @@ except MfaRequiredError as e: store_options={"request": request, "response": response}, ) - # 2. Verify the user's code. persist=False (the default) → the SDK - # returns the tokens instead of writing a session. + # 2. Verify the user's code. Re-supply the SAME dpop_key so the issued + # token stays DPoP-bound. persist=False (the default) → the SDK returns + # the tokens instead of writing a session. verify_response = await server_client.mfa.verify( {"mfa_token": e.mfa_token, "otp": otp_code}, + dpop_key=dpop_key, # same key given to signin_with_passkey store_options={"request": request, "response": response}, ) @@ -174,6 +203,9 @@ except MfaRequiredError as e: ) ``` +> [!IMPORTANT] +> Re-supply the **same** `dpop_key` to `verify`. Omitting it when the login was DPoP-bound would downgrade the result to a Bearer token; `verify` **rejects** that mismatch with `MfaVerifyError` rather than silently dropping the sender constraint. DPoP is preserved end to end — `persist=False` affects only *who writes the session*, never the token binding. + > [!NOTE] > Do **not** pass `persist=True` on this path. It updates an *existing* session, and a passkey-first login has none yet, so it raises `MfaVerifyError("No existing session found…")` — discarding the tokens `verify` just obtained. Use `persist=False` and store the returned tokens as shown above. (This is pre-existing MFA-client behavior, unrelated to passkeys.) @@ -220,7 +252,7 @@ except Auth0Error as e: ### Common error codes (`PasskeyErrorCode`) - `passkey_challenge_error` — the signup/login challenge request failed -- `passkey_token_error` — token exchange failed +- `passkey_token_error` — token exchange failed (also used for a rejected DPoP downgrade) - `invalid_response` — Auth0 returned a response that could not be parsed > [!NOTE] diff --git a/src/auth0_server_python/auth_schemes/__init__.py b/src/auth0_server_python/auth_schemes/__init__.py index 1c2c869..ef37613 100644 --- a/src/auth0_server_python/auth_schemes/__init__.py +++ b/src/auth0_server_python/auth_schemes/__init__.py @@ -1,3 +1,4 @@ from .bearer_auth import BearerAuth +from .dpop_auth import DPoPAuth -__all__ = ["BearerAuth"] +__all__ = ["BearerAuth", "DPoPAuth"] diff --git a/src/auth0_server_python/auth_schemes/dpop_auth.py b/src/auth0_server_python/auth_schemes/dpop_auth.py new file mode 100644 index 0000000..6c99548 --- /dev/null +++ b/src/auth0_server_python/auth_schemes/dpop_auth.py @@ -0,0 +1,94 @@ +import base64 +import hashlib +import time +import uuid +from typing import Optional + +import httpx +from jwcrypto import jwk +from jwcrypto import jwt as jwcrypto_jwt + + +def _base64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _validate_dpop_key(key: "jwk.JWK") -> dict: + """Return the public JWK after enforcing the EC P-256 requirement (ES256).""" + public_jwk = key.export_public(as_dict=True) + if public_jwk.get("kty") != "EC" or public_jwk.get("crv") != "P-256": + raise ValueError("DPoP key must be an EC P-256 key") + return public_jwk + + +def _build_dpop_proof( + key: "jwk.JWK", + public_jwk: dict, + method: str, + url: str, + *, + ath: Optional[str] = None, + nonce: Optional[str] = None, +) -> str: + """Sign a DPoP proof JWT (RFC 9449 §4.2). `ath` binds the proof to an + access token and is omitted for token-endpoint proofs.""" + htu = url.split("?")[0].split("#")[0] + header = {"typ": "dpop+jwt", "alg": "ES256", "jwk": public_jwk} + payload = { + "jti": str(uuid.uuid4()), + "htm": method.upper(), + "htu": htu, + "iat": int(time.time()), + } + if ath is not None: + payload["ath"] = ath + if nonce is not None: + payload["nonce"] = nonce + token = jwcrypto_jwt.JWT(header=header, claims=payload) + token.make_signed_token(key) + return token.serialize() + + +def make_dpop_proof_for_token_endpoint( + key: "jwk.JWK", method: str, url: str, nonce: Optional[str] = None +) -> str: + """ + Build a DPoP proof JWT for use at the token endpoint (RFC 9449 §4.2). + Unlike resource-server proofs, token-endpoint proofs do NOT include `ath` + because no access token exists yet at issuance time. + """ + public_jwk = _validate_dpop_key(key) + return _build_dpop_proof(key, public_jwk, method, url, nonce=nonce) + + +class DPoPAuth(httpx.Auth): + # Buffer the body (sync/async-aware) so the nonce retry can resend it. + requires_request_body = True + + def __init__(self, token: str, key: "jwk.JWK") -> None: + public_jwk = _validate_dpop_key(key) + try: + token.encode("ascii") + except UnicodeEncodeError: + raise ValueError("Access token must contain only ASCII characters") + self._token = token + self._key = key + self._public_jwk = public_jwk + + def auth_flow(self, request: httpx.Request): + proof = self._make_proof(request.method, str(request.url)) + request.headers["Authorization"] = f"DPoP {self._token}" + request.headers["DPoP"] = proof + response = yield request + + # RFC 9449 §8.2 — server-nonce retry + if response.status_code == 401 and response.headers.get("DPoP-Nonce"): + nonce = response.headers["DPoP-Nonce"] + request.headers["DPoP"] = self._make_proof( + request.method, str(request.url), nonce=nonce + ) + yield request + + def _make_proof(self, method: str, url: str, nonce: Optional[str] = None) -> str: + ath = _base64url(hashlib.sha256(self._token.encode("ascii")).digest()) + return _build_dpop_proof(self._key, self._public_jwk, method, url, ath=ath, nonce=nonce) diff --git a/src/auth0_server_python/auth_server/mfa_client.py b/src/auth0_server_python/auth_server/mfa_client.py index b7fb614..18e198f 100644 --- a/src/auth0_server_python/auth_server/mfa_client.py +++ b/src/auth0_server_python/auth_server/mfa_client.py @@ -5,11 +5,15 @@ import json import time -from typing import Any, Callable, Optional, Union +from typing import TYPE_CHECKING, Any, Callable, Optional, Union import httpx from auth0_server_python.auth_schemes.bearer_auth import BearerAuth +from auth0_server_python.auth_schemes.dpop_auth import make_dpop_proof_for_token_endpoint + +if TYPE_CHECKING: + from jwcrypto import jwk from auth0_server_python.auth_types import ( AuthenticatorResponse, ChallengeResponse, @@ -447,6 +451,7 @@ async def verify( self, options: dict[str, Any], store_options: Optional[dict[str, Any]] = None, + dpop_key: Optional["jwk.JWK"] = None, ) -> MfaVerifyResponse: """ Verifies an MFA code and completes authentication. @@ -466,12 +471,16 @@ async def verify( - 'audience': str (optional, required if persist=True) - Audience for token_set - 'scope': str (optional) - Scope for token_set store_options: Optional options passed to the State Store (e.g. request/response). + dpop_key: Optional EC P-256 JWK to DPoP-bind the token. Pass the same + key used at login (e.g. given to signin_with_passkey) to preserve + the sender constraint through step-up. Never stored by the SDK. Returns: MfaVerifyResponse with access_token, token_type, etc. Raises: - MfaVerifyError: When verification fails. + MfaVerifyError: When verification fails, or when dpop_key was supplied + but the server returned an unbound (Bearer) token. MfaRequiredError: When chained MFA is required. """ mfa_token = options.get("mfa_token") @@ -506,12 +515,33 @@ async def verify( token_endpoint = f"{base_url}/oauth/token" async with self._get_http_client() as client: + headers = {"Content-Type": "application/x-www-form-urlencoded"} + if dpop_key is not None: + headers["DPoP"] = make_dpop_proof_for_token_endpoint( + dpop_key, "POST", token_endpoint + ) response = await client.post( token_endpoint, data=body, - headers={"Content-Type": "application/x-www-form-urlencoded"} + headers=headers ) + # Rebuild the proof with the nonce and retry once. + if ( + dpop_key is not None + and response.status_code in (400, 401) + and response.headers.get("DPoP-Nonce") + ): + nonce = response.headers["DPoP-Nonce"] + headers["DPoP"] = make_dpop_proof_for_token_endpoint( + dpop_key, "POST", token_endpoint, nonce=nonce + ) + response = await client.post( + token_endpoint, + data=body, + headers=headers + ) + if response.status_code != 200: error_data = self._parse_error_body(response) @@ -533,6 +563,15 @@ async def verify( token_response = response.json() verify_response = MfaVerifyResponse(**token_response) + # Reject a Bearer downgrade when a DPoP token was requested + # (RFC 9449: a bound token has token_type "DPoP"). + token_is_dpop = verify_response.token_type.lower() == "dpop" + if dpop_key is not None and not token_is_dpop: + raise MfaVerifyError( + "DPoP token binding failed: expected token_type 'DPoP', " + f"got '{verify_response.token_type}'" + ) + # Clear the in-progress MFA state after successful verification. if self._state_store: await self._state_store.delete(MFA_PENDING_IDENTIFIER, store_options) 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 7406156..e4e10f2 100644 --- a/src/auth0_server_python/auth_server/my_account_client.py +++ b/src/auth0_server_python/auth_server/my_account_client.py @@ -1,10 +1,11 @@ import json -from typing import Optional +from typing import TYPE_CHECKING, Optional from urllib.parse import quote, unquote, urlparse import httpx from auth0_server_python.auth_schemes.bearer_auth import BearerAuth +from auth0_server_python.auth_schemes.dpop_auth import DPoPAuth from auth0_server_python.auth_types import ( AuthenticationMethod, CompleteConnectAccountRequest, @@ -27,6 +28,18 @@ MyAccountApiError, ) +if TYPE_CHECKING: + from jwcrypto import jwk + + +def _make_auth( + access_token: str, + dpop_key: Optional["jwk.JWK"] = None, +) -> httpx.Auth: + if dpop_key is not None: + return DPoPAuth(access_token, dpop_key) + return BearerAuth(access_token) + class MyAccountClient: """ @@ -356,12 +369,14 @@ async def list_connected_account_connections( async def get_factors( self, access_token: str, + dpop_key: Optional["jwk.JWK"] = None, ) -> GetFactorsResponse: """ Retrieve the list of factors available for enrollment. Args: access_token: User's access token (scope: read:me:factors). + dpop_key: Optional EC P-256 key for DPoP-bound token presentation. Returns: GetFactorsResponse containing the available factors. @@ -378,7 +393,7 @@ async def get_factors( async with self._get_http_client() as client: response = await client.get( url=f"{self.audience}v1/factors", - auth=BearerAuth(access_token), + auth=_make_auth(access_token, dpop_key), ) if response.status_code != 200: @@ -415,6 +430,7 @@ async def list_authentication_methods( self, access_token: str, type_filter: Optional[str] = None, + dpop_key: Optional["jwk.JWK"] = None, ) -> ListAuthenticationMethodsResponse: """ List the user's enrolled authentication methods. @@ -422,6 +438,7 @@ async def list_authentication_methods( Args: access_token: User's access token (scope: read:me:authentication-methods). type_filter: Optional authentication-method type to filter results by. + dpop_key: Optional EC P-256 key for DPoP-bound token presentation. Returns: ListAuthenticationMethodsResponse containing the enrolled methods. @@ -443,7 +460,7 @@ async def list_authentication_methods( response = await client.get( url=f"{self.audience}v1/authentication-methods", params=params, - auth=BearerAuth(access_token), + auth=_make_auth(access_token, dpop_key), ) if response.status_code != 200: @@ -477,6 +494,7 @@ async def get_authentication_method( self, access_token: str, authentication_method_id: str, + dpop_key: Optional["jwk.JWK"] = None, ) -> AuthenticationMethod: """ Retrieve a single authentication method by ID. @@ -484,6 +502,7 @@ async def get_authentication_method( Args: access_token: User's access token (scope: read:me:authentication-methods). authentication_method_id: ID of the authentication method to retrieve. + dpop_key: Optional EC P-256 key for DPoP-bound token presentation. Returns: AuthenticationMethod for the requested ID. @@ -502,7 +521,7 @@ async def get_authentication_method( async with self._get_http_client() as client: response = await client.get( url=f"{self.audience}v1/authentication-methods/{quote(authentication_method_id, safe='')}", - auth=BearerAuth(access_token), + auth=_make_auth(access_token, dpop_key), ) if response.status_code != 200: @@ -536,6 +555,7 @@ async def delete_authentication_method( self, access_token: str, authentication_method_id: str, + dpop_key: Optional["jwk.JWK"] = None, ) -> None: """ Delete an authentication method by ID. @@ -543,6 +563,7 @@ async def delete_authentication_method( Args: access_token: User's access token (scope: delete:me:authentication-methods). authentication_method_id: ID of the authentication method to delete. + dpop_key: Optional EC P-256 key for DPoP-bound token presentation. Raises: MissingRequiredArgumentError: If access_token or authentication_method_id is not provided. @@ -558,7 +579,7 @@ async def delete_authentication_method( async with self._get_http_client() as client: response = await client.delete( url=f"{self.audience}v1/authentication-methods/{quote(authentication_method_id, safe='')}", - auth=BearerAuth(access_token), + auth=_make_auth(access_token, dpop_key), ) if response.status_code != 204: @@ -591,6 +612,7 @@ async def update_authentication_method( access_token: str, authentication_method_id: str, request: UpdateAuthenticationMethodRequest, + dpop_key: Optional["jwk.JWK"] = None, ) -> AuthenticationMethod: """ Update an authentication method by ID. @@ -599,6 +621,7 @@ async def update_authentication_method( access_token: User's access token (scope: update:me:authentication-methods). authentication_method_id: ID of the authentication method to update. request: Fields to update on the authentication method. + dpop_key: Optional EC P-256 key for DPoP-bound token presentation. Returns: AuthenticationMethod reflecting the updated state. @@ -620,7 +643,7 @@ async def update_authentication_method( response = await client.patch( url=f"{self.audience}v1/authentication-methods/{quote(authentication_method_id, safe='')}", json=request.model_dump(exclude_none=True), - auth=BearerAuth(access_token), + auth=_make_auth(access_token, dpop_key), ) if response.status_code != 200: @@ -654,6 +677,7 @@ async def enroll_authentication_method( self, access_token: str, request: EnrollAuthenticationMethodRequest, + dpop_key: Optional["jwk.JWK"] = None, ) -> EnrollmentChallengeResponse: """Step 1 of 2: Start enrollment (POST /me/v1/authentication-methods). @@ -673,7 +697,7 @@ async def enroll_authentication_method( response = await client.post( url=f"{self.audience}v1/authentication-methods", json=request.model_dump(exclude_none=True), - auth=BearerAuth(access_token), + auth=_make_auth(access_token, dpop_key), ) if response.status_code != 202: @@ -749,6 +773,7 @@ async def verify_authentication_method( access_token: str, authentication_method_id: str, request: VerifyAuthenticationMethodRequest, + dpop_key: Optional["jwk.JWK"] = None, ) -> AuthenticationMethod: """Step 2 of 2: Verify enrollment (POST /me/v1/authentication-methods/{id}/verify). @@ -766,7 +791,7 @@ async def verify_authentication_method( response = await client.post( url=f"{self.audience}v1/authentication-methods/{quote(authentication_method_id, safe='')}/verify", json=request.model_dump(by_alias=True, exclude_none=True), - auth=BearerAuth(access_token), + auth=_make_auth(access_token, dpop_key), ) if response.status_code != 201: diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index 91c2b38..13b10b2 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -7,7 +7,10 @@ import json import time from collections import OrderedDict -from typing import Any, Callable, Generic, Optional, TypeVar, Union +from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union + +if TYPE_CHECKING: + from jwcrypto import jwk from urllib.parse import parse_qs, urlencode, urlparse, urlunparse import httpx @@ -16,6 +19,7 @@ from authlib.integrations.httpx_client import AsyncOAuth2Client from pydantic import ValidationError +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 from auth0_server_python.auth_types import ( @@ -2801,6 +2805,7 @@ async def signin_with_passkey( organization: Optional[str] = None, scope: Optional[str] = None, audience: Optional[str] = None, + dpop_key: Optional["jwk.JWK"] = None, ) -> PasskeyLoginResult: """ Completes passkey authentication by exchanging the WebAuthn assertion @@ -2818,6 +2823,9 @@ async def signin_with_passkey( organization: Auth0 organization ID or name. scope: OAuth2 scope string. audience: Target API audience. + dpop_key: Optional EC P-256 JWK for DPoP-bound token exchange. When provided, + attaches a DPoP proof header so Auth0 issues a DPoP-bound token + (token_type: DPoP). Required when the tenant mandates DPoP binding. Returns: PasskeyLoginResult containing state_data with user claims and token sets, @@ -2862,7 +2870,27 @@ async def signin_with_passkey( body["audience"] = audience async with self._get_http_client() as client: - response = await client.post(token_endpoint, json=body) + headers = {} + if dpop_key is not None: + headers["DPoP"] = make_dpop_proof_for_token_endpoint( + dpop_key, "POST", token_endpoint + ) + response = await client.post(token_endpoint, json=body, headers=headers) + + # RFC 9449 — the authorization server signals a required nonce + # with HTTP 400 + error="use_dpop_nonce" + DPoP-Nonce. Accept + # 401 as well so the retry holds against servers that mirror the + # resource-server status. + if ( + dpop_key is not None + and response.status_code in (400, 401) + and response.headers.get("DPoP-Nonce") + ): + nonce = response.headers["DPoP-Nonce"] + headers["DPoP"] = make_dpop_proof_for_token_endpoint( + dpop_key, "POST", token_endpoint, nonce=nonce + ) + response = await client.post(token_endpoint, json=body, headers=headers) if response.status_code != 200: try: @@ -2903,6 +2931,23 @@ async def signin_with_passkey( token_response = PasskeyTokenResponse.model_validate(token_data) + token_is_dpop = token_response.token_type.lower() == "dpop" + if dpop_key is not None and not token_is_dpop: + raise PasskeyError( + PasskeyErrorCode.TOKEN_EXCHANGE_FAILED, + f"DPoP token binding failed: expected token_type 'DPoP', " + f"got '{token_response.token_type}'", + ) + if dpop_key is None and token_is_dpop: + # Server issued a DPoP-bound token but no proof key was + # supplied — we cannot prove possession on later calls, so + # storing it as Bearer would silently fail open. Reject. + raise PasskeyError( + PasskeyErrorCode.TOKEN_EXCHANGE_FAILED, + "Server returned a DPoP-bound token but no dpop_key was " + "provided; pass dpop_key to signin_with_passkey to bind it", + ) + if resolved_org and not token_response.id_token: raise OrganizationTokenValidationError( "Organization was requested but the token response included no ID token; " diff --git a/src/auth0_server_python/tests/test_dpop_auth.py b/src/auth0_server_python/tests/test_dpop_auth.py new file mode 100644 index 0000000..6952b04 --- /dev/null +++ b/src/auth0_server_python/tests/test_dpop_auth.py @@ -0,0 +1,185 @@ +import base64 +import hashlib +import json + +import httpx +import pytest +from jwcrypto import jwk + +from auth0_server_python.auth_schemes.bearer_auth import BearerAuth +from auth0_server_python.auth_schemes.dpop_auth import ( + DPoPAuth, + _base64url, + make_dpop_proof_for_token_endpoint, +) +from auth0_server_python.auth_server.my_account_client import _make_auth + + +@pytest.fixture +def ec_key(): + return jwk.JWK.generate(kty="EC", crv="P-256") + + +def _decode_jwt_parts(token: str) -> tuple[dict, dict]: + parts = token.split(".") + header = json.loads(base64.urlsafe_b64decode(parts[0] + "==")) + payload = json.loads(base64.urlsafe_b64decode(parts[1] + "==")) + return header, payload + + +def test_dpop_headers_set(ec_key): + auth = DPoPAuth(token="test_token", key=ec_key) + request = httpx.Request("POST", "https://example.com/me/v1/authentication-methods") + flow = auth.auth_flow(request) + modified = next(flow) + + assert modified.headers["Authorization"] == "DPoP test_token" + assert "DPoP" in modified.headers + assert "Bearer" not in modified.headers["Authorization"] + + +def test_dpop_proof_structure(ec_key): + auth = DPoPAuth(token="test_token", key=ec_key) + request = httpx.Request("POST", "https://example.com/me/v1/authentication-methods") + flow = auth.auth_flow(request) + modified = next(flow) + + proof = modified.headers["DPoP"] + header, payload = _decode_jwt_parts(proof) + + assert header["typ"] == "dpop+jwt" + assert header["alg"] == "ES256" + assert "jwk" in header + assert header["jwk"]["kty"] == "EC" + assert header["jwk"]["crv"] == "P-256" + + assert "jti" in payload + assert payload["htm"] == "POST" + assert payload["htu"] == "https://example.com/me/v1/authentication-methods" + assert "iat" in payload + assert "ath" in payload + + +def test_dpop_htm_binding(ec_key): + auth = DPoPAuth(token="test_token", key=ec_key) + + get_request = httpx.Request("GET", "https://example.com/me/v1/factors") + flow = auth.auth_flow(get_request) + modified = next(flow) + _, payload = _decode_jwt_parts(modified.headers["DPoP"]) + assert payload["htm"] == "GET" + + post_request = httpx.Request("post", "https://example.com/me/v1/factors") + flow = auth.auth_flow(post_request) + modified = next(flow) + _, payload = _decode_jwt_parts(modified.headers["DPoP"]) + assert payload["htm"] == "POST" + + +def test_dpop_htu_strips_query_and_fragment(ec_key): + auth = DPoPAuth(token="test_token", key=ec_key) + request = httpx.Request("GET", "https://example.com/me/v1/factors?foo=bar#section") + flow = auth.auth_flow(request) + modified = next(flow) + _, payload = _decode_jwt_parts(modified.headers["DPoP"]) + assert payload["htu"] == "https://example.com/me/v1/factors" + + +def test_dpop_htu_preserves_port(ec_key): + auth = DPoPAuth(token="test_token", key=ec_key) + request = httpx.Request("GET", "https://example.com:8443/me/v1/factors") + flow = auth.auth_flow(request) + modified = next(flow) + _, payload = _decode_jwt_parts(modified.headers["DPoP"]) + assert payload["htu"] == "https://example.com:8443/me/v1/factors" + + +def test_dpop_ath_binding(ec_key): + token = "my_access_token_value" + auth = DPoPAuth(token=token, key=ec_key) + request = httpx.Request("GET", "https://example.com/me/v1/factors") + flow = auth.auth_flow(request) + modified = next(flow) + _, payload = _decode_jwt_parts(modified.headers["DPoP"]) + + expected_ath = _base64url(hashlib.sha256(token.encode("ascii")).digest()) + assert payload["ath"] == expected_ath + + +def test_dpop_proof_uniqueness(ec_key): + auth = DPoPAuth(token="test_token", key=ec_key) + jtis = set() + for _ in range(10): + request = httpx.Request("GET", "https://example.com/me/v1/factors") + flow = auth.auth_flow(request) + modified = next(flow) + _, payload = _decode_jwt_parts(modified.headers["DPoP"]) + jtis.add(payload["jti"]) + + assert len(jtis) == 10 + + +def test_dpop_nonce_retry_on_resource_server(ec_key): + auth = DPoPAuth(token="test_token", key=ec_key) + request = httpx.Request("GET", "https://example.com/me/v1/factors") + flow = auth.auth_flow(request) + + first = next(flow) + _, first_payload = _decode_jwt_parts(first.headers["DPoP"]) + assert "nonce" not in first_payload + + nonce_response = httpx.Response( + status_code=401, + headers={"DPoP-Nonce": "rs-nonce-xyz"}, + request=first, + ) + retried = flow.send(nonce_response) + _, retried_payload = _decode_jwt_parts(retried.headers["DPoP"]) + assert retried_payload["nonce"] == "rs-nonce-xyz" + assert retried_payload["ath"] == first_payload["ath"] + assert retried_payload["jti"] != first_payload["jti"] + + +def test_dpop_no_retry_without_nonce_header(ec_key): + auth = DPoPAuth(token="test_token", key=ec_key) + request = httpx.Request("GET", "https://example.com/me/v1/factors") + flow = auth.auth_flow(request) + next(flow) + + ok_response = httpx.Response(status_code=200, request=request) + with pytest.raises(StopIteration): + flow.send(ok_response) + + +def test_token_endpoint_proof_rejects_non_ec_key(): + rsa_key = jwk.JWK.generate(kty="RSA", size=2048) + with pytest.raises(ValueError, match="EC P-256"): + make_dpop_proof_for_token_endpoint(rsa_key, "POST", "https://example.com/oauth/token") + + +def test_dpop_repr_does_not_leak_credentials(ec_key): + auth = DPoPAuth(token="secret_access_token_value", key=ec_key) + assert "secret_access_token_value" not in repr(auth) + assert "secret_access_token_value" not in str(auth) + + +def test_dpop_rejects_non_ec_key(): + rsa_key = jwk.JWK.generate(kty="RSA", size=2048) + with pytest.raises(ValueError, match="EC P-256"): + DPoPAuth(token="token", key=rsa_key) + + +def test_dpop_rejects_wrong_curve(): + p384_key = jwk.JWK.generate(kty="EC", crv="P-384") + with pytest.raises(ValueError, match="EC P-256"): + DPoPAuth(token="token", key=p384_key) + + +def test_make_auth_bearer_fallback(): + auth = _make_auth("token123", dpop_key=None) + assert isinstance(auth, BearerAuth) + + +def test_make_auth_dpop_when_key_provided(ec_key): + auth = _make_auth("token123", dpop_key=ec_key) + assert isinstance(auth, DPoPAuth) diff --git a/src/auth0_server_python/tests/test_mfa_client.py b/src/auth0_server_python/tests/test_mfa_client.py index ac7e9a4..ac820df 100644 --- a/src/auth0_server_python/tests/test_mfa_client.py +++ b/src/auth0_server_python/tests/test_mfa_client.py @@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest +from jwcrypto import jwk from auth0_server_python.auth_server.mfa_client import DEFAULT_MFA_TOKEN_TTL, MfaClient from auth0_server_python.auth_types import ( @@ -892,3 +893,106 @@ async def test_verify_persist_store_failure_raises(self, mocker): {"mfa_token": _enc(), "otp": "123456", "persist": True, "audience": "https://api.example.com"} ) + + @pytest.mark.asyncio + async def test_verify_dpop_attaches_proof_header(self, mocker): + """When dpop_key is supplied, a DPoP proof header is sent and a bound token accepted.""" + client = _make_client() + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") + response = AsyncMock() + response.status_code = 200 + response.headers = {} + response.json = MagicMock(return_value={ + "access_token": "bound_at", "token_type": "DPoP", "expires_in": 3600 + }) + + captured_request = {} + + async def mock_post(self_client, url, **kwargs): + captured_request["kwargs"] = kwargs + return response + + mocker.patch("httpx.AsyncClient.post", new=mock_post) + + result = await client.verify( + {"mfa_token": _enc(), "otp": "123456"}, + dpop_key=dpop_key, + ) + assert result.token_type == "DPoP" + assert "DPoP" in captured_request["kwargs"]["headers"] + + @pytest.mark.asyncio + async def test_verify_dpop_nonce_retry(self, mocker): + """RFC 9449 §8.2: a DPoP-Nonce challenge triggers exactly one retry with the nonce.""" + client = _make_client() + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") + + challenge = AsyncMock() + challenge.status_code = 400 + challenge.headers = {"DPoP-Nonce": "server-nonce-123"} + challenge.json = MagicMock(return_value={"error": "use_dpop_nonce"}) + + success = AsyncMock() + success.status_code = 200 + success.headers = {} + success.json = MagicMock(return_value={ + "access_token": "bound_at", "token_type": "DPoP", "expires_in": 3600 + }) + + proofs = [] + + async def mock_post(self_client, url, **kwargs): + proofs.append(kwargs["headers"].get("DPoP")) + return challenge if len(proofs) == 1 else success + + mocker.patch("httpx.AsyncClient.post", new=mock_post) + + result = await client.verify( + {"mfa_token": _enc(), "otp": "123456"}, + dpop_key=dpop_key, + ) + assert result.token_type == "DPoP" + assert len(proofs) == 2 + assert proofs[0] != proofs[1] + + @pytest.mark.asyncio + async def test_verify_dpop_rejects_bearer_downgrade(self, mocker): + """dpop_key supplied but server returns Bearer: reject rather than downgrade.""" + client = _make_client() + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") + response = AsyncMock() + response.status_code = 200 + response.headers = {} + response.json = MagicMock(return_value={ + "access_token": "at", "token_type": "Bearer", "expires_in": 3600 + }) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + with pytest.raises(MfaVerifyError, match="DPoP token binding failed"): + await client.verify( + {"mfa_token": _enc(), "otp": "123456"}, + dpop_key=dpop_key, + ) + + @pytest.mark.asyncio + async def test_verify_without_dpop_no_dpop_header(self, mocker): + """Without dpop_key the request carries no DPoP header and Bearer is accepted.""" + client = _make_client() + response = AsyncMock() + response.status_code = 200 + response.headers = {} + response.json = MagicMock(return_value={ + "access_token": "at", "token_type": "Bearer", "expires_in": 3600 + }) + + captured_request = {} + + async def mock_post(self_client, url, **kwargs): + captured_request["kwargs"] = kwargs + return response + + mocker.patch("httpx.AsyncClient.post", new=mock_post) + + result = await client.verify({"mfa_token": _enc(), "otp": "123456"}) + assert result.token_type == "Bearer" + assert "DPoP" not in captured_request["kwargs"]["headers"] 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 f2b83c6..f917eef 100644 --- a/src/auth0_server_python/tests/test_my_account_client.py +++ b/src/auth0_server_python/tests/test_my_account_client.py @@ -1,7 +1,12 @@ +import base64 +import json from unittest.mock import ANY, AsyncMock, MagicMock +import httpx import pytest +from jwcrypto import jwk as jwk_module +from auth0_server_python.auth_schemes.dpop_auth import DPoPAuth from auth0_server_python.auth_server.my_account_client import MyAccountClient from auth0_server_python.auth_types import ( AuthenticationMethod, @@ -1015,6 +1020,138 @@ def test_verify_request_accepts_authn_response(): assert req.authn_response is not None +@pytest.mark.asyncio +async def test_get_factors_with_dpop_key(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock( + return_value={"factors": [{"type": "phone", "usage": ["primary"]}]} + ) + mock_get = mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + await client.get_factors(access_token="token123", dpop_key=dpop_key) + + mock_get.assert_awaited_once() + call_kwargs = mock_get.call_args[1] + assert isinstance(call_kwargs["auth"], DPoPAuth) + + +@pytest.mark.asyncio +async def test_list_authentication_methods_with_dpop_key(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={"authentication_methods": []}) + mock_get = mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + await client.list_authentication_methods(access_token="token123", dpop_key=dpop_key) + + assert isinstance(mock_get.call_args[1]["auth"], DPoPAuth) + + +@pytest.mark.asyncio +async def test_get_authentication_method_with_dpop_key(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={ + "id": "am_1", "type": "passkey", "created_at": "2026-01-01T00:00:00Z" + }) + mock_get = mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + await client.get_authentication_method( + access_token="token123", authentication_method_id="am_1", dpop_key=dpop_key + ) + + assert isinstance(mock_get.call_args[1]["auth"], DPoPAuth) + + +@pytest.mark.asyncio +async def test_delete_authentication_method_with_dpop_key(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 204 + mock_delete = mocker.patch( + "httpx.AsyncClient.delete", new_callable=AsyncMock, return_value=response + ) + + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + await client.delete_authentication_method( + access_token="token123", authentication_method_id="am_1", dpop_key=dpop_key + ) + + assert isinstance(mock_delete.call_args[1]["auth"], DPoPAuth) + + +@pytest.mark.asyncio +async def test_update_authentication_method_with_dpop_key(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={ + "id": "am_1", "type": "passkey", "created_at": "2026-01-01T00:00:00Z" + }) + mock_patch = mocker.patch( + "httpx.AsyncClient.patch", new_callable=AsyncMock, return_value=response + ) + + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + req = UpdateAuthenticationMethodRequest(name="New Name") + await client.update_authentication_method( + access_token="token123", authentication_method_id="am_1", request=req, dpop_key=dpop_key + ) + + assert isinstance(mock_patch.call_args[1]["auth"], DPoPAuth) + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_with_dpop_key(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 202 + response.headers = {"location": "/me/v1/authentication-methods/passkey|new"} + response.json = MagicMock(return_value={"auth_session": "session_abc"}) + mock_post = mocker.patch( + "httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response + ) + + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + req = EnrollAuthenticationMethodRequest(type="passkey") + await client.enroll_authentication_method( + access_token="token123", request=req, dpop_key=dpop_key + ) + + assert isinstance(mock_post.call_args[1]["auth"], DPoPAuth) + + +@pytest.mark.asyncio +async def test_verify_authentication_method_with_dpop_key(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 201 + response.json = MagicMock(return_value={ + "id": "am_1", "type": "passkey", "created_at": "2026-01-01T00:00:00Z" + }) + mock_post = mocker.patch( + "httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response + ) + + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + req = VerifyAuthenticationMethodRequest(auth_session="session_abc", otp_code="123456") + await client.verify_authentication_method( + access_token="token123", + authentication_method_id="am_1", + request=req, + dpop_key=dpop_key, + ) + + assert isinstance(mock_post.call_args[1]["auth"], DPoPAuth) + + @pytest.mark.asyncio async def test_list_authentication_methods_api_error(mocker): client = MyAccountClient(domain="auth0.local") @@ -1210,3 +1347,78 @@ async def test_enroll_authentication_method_location_collection_url(mocker): with pytest.raises(ApiError) as exc: await client.enroll_authentication_method(access_token="token123", request=req) assert "could not extract ID" in str(exc.value) + + +# ============================================================================= +# DPoP nonce retry (RFC 9449 §8.2) — tests DPoPAuth.auth_flow directly +# ============================================================================= + + +def test_dpop_auth_flow_retries_with_nonce_on_401(): + """ + DPoPAuth.auth_flow() must retry with DPoP-Nonce when server responds 401 + + DPoP-Nonce header (RFC 9449 §8.2). Tested by driving the generator directly. + """ + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + auth = DPoPAuth(token="test_access_token", key=dpop_key) + + request = httpx.Request("GET", "https://auth0.local/me/v1/factors") + flow = auth.auth_flow(request) + + # First yield — initial request + first_request = next(flow) + assert "DPoP" in first_request.headers + assert "Authorization" in first_request.headers + + # First proof must not have nonce + proof1 = first_request.headers["DPoP"] + payload1_b64 = proof1.split(".")[1] + padding = 4 - len(payload1_b64) % 4 + payload1 = json.loads(base64.urlsafe_b64decode(payload1_b64 + "=" * padding)) + assert "nonce" not in payload1 + + # Simulate 401 + DPoP-Nonce response + nonce_response = httpx.Response( + status_code=401, + headers={"DPoP-Nonce": "server-nonce-abc"}, + content=b'{"error":"use_dpop_nonce"}', + request=request, + ) + + # Second yield — retry request with nonce + try: + second_request = flow.send(nonce_response) + except StopIteration: + second_request = None + + assert second_request is not None + proof2 = second_request.headers["DPoP"] + payload2_b64 = proof2.split(".")[1] + padding = 4 - len(payload2_b64) % 4 + payload2 = json.loads(base64.urlsafe_b64decode(payload2_b64 + "=" * padding)) + assert payload2["nonce"] == "server-nonce-abc" + + +def test_dpop_auth_flow_no_retry_on_non_401(): + """DPoPAuth.auth_flow() must NOT retry when the response is not 401.""" + dpop_key = jwk_module.JWK.generate(kty="EC", crv="P-256") + auth = DPoPAuth(token="test_access_token", key=dpop_key) + + request = httpx.Request("GET", "https://auth0.local/me/v1/factors") + flow = auth.auth_flow(request) + next(flow) + + success_response = httpx.Response( + status_code=200, + content=b'{"factors":[]}', + request=request, + ) + + try: + flow.send(success_response) + retried = True + except StopIteration: + retried = False + + assert not retried + diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 9cc7951..e9f17be 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -1,11 +1,15 @@ +import base64 import json import time import unicodedata from unittest.mock import ANY, AsyncMock, MagicMock, patch from urllib.parse import parse_qs, urlparse +import httpx import pytest +from jwcrypto import jwk +from auth0_server_python.auth_schemes.dpop_auth import DPoPAuth from auth0_server_python.auth_server.mfa_client import MfaClient from auth0_server_python.auth_server.my_account_client import MyAccountClient from auth0_server_python.auth_server.server_client import ServerClient @@ -6207,6 +6211,15 @@ async def capture_set(key, value, options=None): "scope": "openid profile", } +_PASSKEY_TOKEN_RESPONSE_DPOP = { + "access_token": "at_passkey_dpop_123", + "id_token": "eyJ.test.jwt", + "token_type": "DPoP", + "expires_in": 86400, + "scope": "openid profile", +} + + def _make_passkey_authn_response(): return PasskeyAuthResponse( id="cred_abc123", @@ -6870,6 +6883,287 @@ async def test_signin_with_passkey_missing_expires_at_calculates(mocker): assert abs(result.state_data["token_sets"][0]["expires_at"] - (int(time.time()) + 60)) <= 2 +@pytest.mark.asyncio +async def test_signin_with_passkey_dpop_attaches_proof_header(mocker): + client = ServerClient( + domain="auth0.local", + client_id="test_client_id", + client_secret="test_client_secret", + 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_DPOP) + mock_post.return_value = mock_response + + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") + await client.signin_with_passkey( + auth_session="session_xyz", + authn_response=_make_passkey_authn_response(), + dpop_key=dpop_key, + ) + + args, kwargs = mock_post.call_args + assert "DPoP" in kwargs["headers"] + + # Decode proof and assert no ath claim (token endpoint proof — RFC 9449 §4.2) + proof = kwargs["headers"]["DPoP"] + payload_b64 = proof.split(".")[1] + padding = 4 - len(payload_b64) % 4 + payload = json.loads(base64.urlsafe_b64decode(payload_b64 + "=" * padding)) + assert "ath" not in payload + assert "jti" in payload + assert payload["htm"] == "POST" + assert payload["htu"] == "https://auth0.local/oauth/token" + + +@pytest.mark.asyncio +async def test_signin_with_passkey_dpop_nonce_retry(mocker): + client = ServerClient( + domain="auth0.local", + client_id="test_client_id", + client_secret="test_client_secret", + 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) + + # RFC 9449 §8.1 — the token endpoint signals a required nonce with HTTP 400. + nonce_response = AsyncMock() + nonce_response.status_code = 400 + nonce_response.headers = {"DPoP-Nonce": "server-nonce-abc"} + nonce_response.json = MagicMock(return_value={"error": "use_dpop_nonce"}) + + success_response = AsyncMock() + success_response.status_code = 200 + success_response.json = MagicMock(return_value=_PASSKEY_TOKEN_RESPONSE_DPOP) + + mock_post.side_effect = [nonce_response, success_response] + + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") + result = await client.signin_with_passkey( + auth_session="session_xyz", + authn_response=_make_passkey_authn_response(), + dpop_key=dpop_key, + ) + + assert mock_post.await_count == 2 + assert result.state_data["token_sets"][0]["access_token"] == "at_passkey_dpop_123" + + # Second call must include the nonce in the DPoP proof + second_call_kwargs = mock_post.call_args_list[1][1] + proof = second_call_kwargs["headers"]["DPoP"] + payload_b64 = proof.split(".")[1] + padding = 4 - len(payload_b64) % 4 + payload = json.loads(base64.urlsafe_b64decode(payload_b64 + "=" * padding)) + assert payload["nonce"] == "server-nonce-abc" + + +@pytest.mark.asyncio +async def test_signin_with_passkey_dpop_nonce_retry_on_401(mocker): + """Token endpoint nonce retry must also hold when the server returns 401 + DPoP-Nonce.""" + client = ServerClient( + domain="auth0.local", + client_id="test_client_id", + client_secret="test_client_secret", + 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) + + nonce_response = AsyncMock() + nonce_response.status_code = 401 + nonce_response.headers = {"DPoP-Nonce": "server-nonce-401"} + nonce_response.json = MagicMock(return_value={"error": "use_dpop_nonce"}) + + success_response = AsyncMock() + success_response.status_code = 200 + success_response.json = MagicMock(return_value=_PASSKEY_TOKEN_RESPONSE_DPOP) + + mock_post.side_effect = [nonce_response, success_response] + + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") + result = await client.signin_with_passkey( + auth_session="session_xyz", + authn_response=_make_passkey_authn_response(), + dpop_key=dpop_key, + ) + + assert mock_post.await_count == 2 + assert result.state_data["token_sets"][0]["access_token"] == "at_passkey_dpop_123" + second_call_kwargs = mock_post.call_args_list[1][1] + proof = second_call_kwargs["headers"]["DPoP"] + payload_b64 = proof.split(".")[1] + padding = 4 - len(payload_b64) % 4 + payload = json.loads(base64.urlsafe_b64decode(payload_b64 + "=" * padding)) + assert payload["nonce"] == "server-nonce-401" + + +@pytest.mark.asyncio +async def test_signin_with_passkey_dpop_rejects_bearer_downgrade(mocker): + """Server returning token_type=Bearer when DPoP was requested must raise PasskeyError.""" + + client = ServerClient( + domain="auth0.local", + client_id="test_client_id", + client_secret="test_client_secret", + 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/"}, + ) + 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 + + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") + with pytest.raises(PasskeyError) as exc: + await client.signin_with_passkey( + auth_session="session_xyz", + authn_response=_make_passkey_authn_response(), + dpop_key=dpop_key, + ) + assert exc.value.code == PasskeyErrorCode.TOKEN_EXCHANGE_FAILED + + +@pytest.mark.asyncio +async def test_signin_with_passkey_dpop_bound_token_without_key_rejected(mocker): + """Server returning token_type=DPoP when no dpop_key was passed must fail + closed, not store a DPoP-bound token the SDK can never prove possession of.""" + + client = ServerClient( + domain="auth0.local", + client_id="test_client_id", + client_secret="test_client_secret", + 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/"}, + ) + 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_DPOP) + mock_post.return_value = mock_response + + with pytest.raises(PasskeyError) as exc: + await client.signin_with_passkey( + auth_session="session_xyz", + authn_response=_make_passkey_authn_response(), + ) + assert exc.value.code == PasskeyErrorCode.TOKEN_EXCHANGE_FAILED + client._state_store.set.assert_not_called() + + +@pytest.mark.asyncio +async def test_signin_with_passkey_missing_issuer_in_metadata(mocker): + """Missing 'issuer' in OIDC metadata must raise IssuerValidationError, not silently pass.""" + client = ServerClient( + domain="auth0.local", + client_id="test_client_id", + client_secret="test_client_secret", + 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"}, + ) + 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 + + with pytest.raises(IssuerValidationError): + await client.signin_with_passkey( + auth_session="session_xyz", + authn_response=_make_passkey_authn_response(), + ) + + +@pytest.mark.asyncio +async def test_signin_with_passkey_without_dpop_no_dpop_header(mocker): + client = ServerClient( + domain="auth0.local", + client_id="test_client_id", + client_secret="test_client_secret", + 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(), + ) + + args, kwargs = mock_post.call_args + assert "DPoP" not in kwargs.get("headers", {}) + + @pytest.mark.asyncio async def test_signin_with_passkey_creates_session_in_state_store(mocker): """signin_with_passkey must persist a session — consistent with complete_interactive_login.""" @@ -7107,6 +7401,130 @@ async def test_signin_with_passkey_mfa_required_stores_pending_mfa(mocker): assert store_payload["mfa_token"] == exc.value.mfa_token +@pytest.mark.asyncio +async def test_dpop_passkey_mfa_verify_preserves_binding(mocker): + """End-to-end: a DPoP passkey login that steps up through MFA keeps the + sender constraint when the caller re-supplies dpop_key to mfa.verify.""" + dpop_key = jwk.JWK.generate(kty="EC", crv="P-256") + mock_store = AsyncMock() + mock_store.get = AsyncMock(return_value=None) + mock_store.set = AsyncMock() + client = ServerClient( + domain="auth0.local", + client_id="test_client_id", + client_secret="test_client_secret", + state_store=mock_store, + 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/"}, + ) + + # 1. Passkey grant returns mfa_required (DPoP proof was attached to this call). + mfa_required = AsyncMock() + mfa_required.status_code = 403 + mfa_required.headers = {} + mfa_required.json = MagicMock(return_value={ + "error": "mfa_required", + "error_description": "MFA required", + "mfa_token": "raw_mfa_token_xyz", + }) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mfa_required) + + with pytest.raises(MfaRequiredError) as exc: + await client.signin_with_passkey( + auth_session="session_abc", + authn_response=_make_passkey_authn_response(), + dpop_key=dpop_key, + ) + + encrypted_mfa_token = exc.value.mfa_token + + # 2. Caller re-supplies the same dpop_key to mfa.verify; token must be DPoP-bound. + captured = {} + + async def mock_verify_post(self_client, url, **kwargs): + captured["headers"] = kwargs.get("headers", {}) + bound = AsyncMock() + bound.status_code = 200 + bound.headers = {} + bound.json = MagicMock(return_value={ + "access_token": "bound_at", "token_type": "DPoP", "expires_in": 3600 + }) + return bound + + mocker.patch("httpx.AsyncClient.post", new=mock_verify_post) + + result = await client.mfa.verify( + {"mfa_token": encrypted_mfa_token, "otp": "123456"}, + dpop_key=dpop_key, + ) + + assert result.token_type == "DPoP" + assert "DPoP" in captured["headers"] + + +@pytest.mark.asyncio +async def test_dpop_auth_async_json_body_survives_nonce_retry(): + """A DPoP-bound POST over AsyncClient (the MyAccount path) must resend an + identical, non-empty body when the server answers 401 + DPoP-Nonce.""" + key = jwk.JWK.generate(kty="EC", crv="P-256") + seen_bodies = [] + + async def handler(request: httpx.Request) -> httpx.Response: + seen_bodies.append(await request.aread()) + if len(seen_bodies) == 1: + return httpx.Response(401, headers={"DPoP-Nonce": "srv-nonce-1"}) + return httpx.Response(200, json={"ok": True}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport) as client: + response = await client.post( + "https://example.com/me/v1/authentication-methods", + json={"type": "passkey", "nested": {"a": 1, "b": [1, 2, 3]}}, + auth=DPoPAuth("access_token_xyz", key), + ) + + assert response.status_code == 200 + assert len(seen_bodies) == 2 + assert seen_bodies[0] == seen_bodies[1] + assert len(seen_bodies[0]) > 0 + + +@pytest.mark.asyncio +async def test_dpop_auth_async_streaming_body_survives_nonce_retry(): + """A streaming (async-generator) body must not crash the DPoP nonce retry + and must arrive identical on both sends. Guards against a manual sync + request.read() inside auth_flow, which raises on async streaming bodies.""" + key = jwk.JWK.generate(kty="EC", crv="P-256") + seen_bodies = [] + + async def body_stream(): + yield b'{"type":"passkey"}' + + async def handler(request: httpx.Request) -> httpx.Response: + seen_bodies.append(await request.aread()) + if len(seen_bodies) == 1: + return httpx.Response(401, headers={"DPoP-Nonce": "srv-nonce-1"}) + return httpx.Response(200, json={"ok": True}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport) as client: + response = await client.post( + "https://example.com/me/v1/authentication-methods", + content=body_stream(), + headers={"content-type": "application/json"}, + auth=DPoPAuth("access_token_xyz", key), + ) + + assert response.status_code == 200 + assert len(seen_bodies) == 2 + assert seen_bodies[0] == seen_bodies[1] == b'{"type":"passkey"}' + + @pytest.mark.asyncio async def test_passkey_signup_challenge_uses_client_default_organization(mocker): """When organization is not passed, self._organization is used."""