diff --git a/README.md b/README.md index e32a1d9..55cce7e 100644 --- a/README.md +++ b/README.md @@ -187,6 +187,14 @@ For more details and examples, see [examples/RetrievingData.md](examples/Retriev Sign users up or in with [WebAuthn](https://www.w3.org/TR/webauthn-2/) passkeys (Touch ID, Face ID, Windows Hello, or a security key) instead of a password, via [Auth0 passkeys](https://auth0.com/docs/authenticate/database-connections/passkeys). The ceremony is two steps — request a challenge, sign it in the browser, then complete sign-in — and establishes a server-side session like every other login path. For the signup and login flows, organizations, step-up MFA, and error handling, see [examples/Passkeys.md](examples/Passkeys.md). +### 8. My Account API — Authentication Methods + +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 new file mode 100644 index 0000000..42662b5 --- /dev/null +++ b/examples/MyAccountAuthenticationMethods.md @@ -0,0 +1,241 @@ +# My Account API — Authentication Methods & Factors + +The [My Account API](https://auth0.com/docs/manage-users/my-account-api) lets a **logged-in user manage their own account**. This guide covers the **authentication-methods** and **factors** surface: enrolling a new passkey (or other factor), and listing, reading, renaming, and deleting a user's enrolled methods. + +> [!NOTE] +> 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 **bind these calls to a held key**, pass an optional `dpop_key` — see [DPoP](#dpop) below. + +## Table of Contents + +- [Prerequisites](#prerequisites) +- [Obtaining a scoped token](#obtaining-a-scoped-token) +- [1. List factors available for enrollment](#1-list-factors-available-for-enrollment) +- [2. Enroll an authentication method (passkey)](#2-enroll-an-authentication-method-passkey) +- [3. List authentication methods](#3-list-authentication-methods) +- [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 + +1. [Activate the My Account API](https://auth0.com/docs/manage-users/my-account-api#activate-the-my-account-api) on your tenant and enable access for your application. +2. [Configure MRRT](https://auth0.com/docs/secure/tokens/refresh-tokens/multi-resource-refresh-token) so your refresh-token policy can mint tokens for the My Account audience (`https://{yourDomain}/me/`) with the authentication-methods scopes. +3. Passkey enrollment additionally requires a [Custom Domain](https://auth0.com/docs/customize/custom-domains) and the native passkey feature on your tenant. + +See [Auth0 Docs → Scope](https://auth0.com/docs/manage-users/my-account-api#scope) for the full scope reference. The scopes for this surface (note the **hyphens**): + +| Operation | Scope | +|-----------|-------| +| List factors | `read:me:factors` | +| List / get methods | `read:me:authentication-methods` | +| Enroll / verify | `create:me:authentication-methods` | +| Update | `update:me:authentication-methods` | +| Delete | `delete:me:authentication-methods` | + +> [!TIP] +> As with Connected Accounts, set the default `scope` for the My Account audience when constructing `ServerClient` to avoid a fresh token request per scope. See [ConnectedAccounts.md → A note about scopes](ConnectedAccounts.md#a-note-about-scopes). + +## Obtaining a scoped token + +`MyAccountClient` is **stateless** — it takes a correctly-scoped user access token on every call. Obtain that token from your `ServerClient` session via MRRT, then construct the client. See [Auth0 Docs → Get an access token](https://auth0.com/docs/manage-users/my-account-api#get-an-access-token) for the underlying token requirements. + +```python +from auth0_server_python.auth_server.my_account_client import MyAccountClient + +# Fresh My Account-scoped token for the current session (MRRT exchange) +access_token = await server_client.get_access_token( + store_options={"request": request, "response": response}, + audience=f"https://{YOUR_CUSTOM_DOMAIN}/me/", + scope="create:me:authentication-methods read:me:authentication-methods read:me:factors", +) + +my_account = MyAccountClient(domain=YOUR_CUSTOM_DOMAIN) +``` + +## 1. List factors available for enrollment + +```python +factors = await my_account.get_factors(access_token=access_token) +for factor in factors.factors: + print(factor.type, factor.usage) +``` + +## 2. Enroll an authentication method (passkey) + +Enrollment is a **two-step** ceremony, mirroring sign-in: request a challenge, sign it in the browser, then verify. See [Auth0 Docs → Enrollment flow](https://auth0.com/docs/manage-users/my-account-api#enrollment-flow) (and, for passkeys specifically, [Embedded login with native passkeys](https://auth0.com/docs/manage-users/my-account-api#embedded-login-with-native-passkeys)) for the platform-level description of this ceremony. + +### Step 1 — Start enrollment + +```python +from auth0_server_python.auth_types import EnrollAuthenticationMethodRequest + +challenge = await my_account.enroll_authentication_method( + access_token=access_token, + request=EnrollAuthenticationMethodRequest(type="passkey"), +) + +# challenge.authentication_method_id -> id of the new (unverified) method +# challenge.auth_session -> Tier 1 session credential (do not log) +# challenge.authn_params_public_key -> pass to navigator.credentials.create() +``` + +`EnrollAuthenticationMethodRequest.type` is a closed set: `passkey`, `email`, `phone`, `totp`, `push-notification`, `recovery-code`, `password`. For non-passkey types, supply the relevant fields (`email`, `phone_number`, `preferred_authentication_method`). An invalid type fails at construction with a clear `ValidationError`. + +### Step 2 — Create the credential in the browser + +Pass `challenge.authn_params_public_key` to `navigator.credentials.create()` and collect the resulting credential. + +### Step 3 — Verify enrollment + +```python +from auth0_server_python.auth_types import ( + VerifyAuthenticationMethodRequest, + PasskeyAuthResponse, +) + +method = await my_account.verify_authentication_method( + access_token=access_token, + authentication_method_id=challenge.authentication_method_id, + request=VerifyAuthenticationMethodRequest( + auth_session=challenge.auth_session, + authn_response=PasskeyAuthResponse( + id=credential["id"], + raw_id=credential["rawId"], + type="public-key", + response={ + "clientDataJSON": credential["response"]["clientDataJSON"], + "attestationObject": credential["response"]["attestationObject"], + }, + ), + ), +) +print(f"Enrolled: {method.id} ({method.type})") +``` + +> [!NOTE] +> For non-passkey types, set the matching field on `VerifyAuthenticationMethodRequest` instead of `authn_response`: `otp_code` (email/phone/totp), `recovery_code`, or `password`. A push enrollment needs only `auth_session`. + +## 3. List authentication methods + +See [Auth0 Docs → List authentication methods](https://auth0.com/docs/manage-users/my-account-api#list-authentication-methods). + +```python +all_methods = await my_account.list_authentication_methods(access_token=access_token) + +# Filter by type +passkeys = await my_account.list_authentication_methods( + access_token=access_token, + type_filter="passkey", +) +for m in passkeys.authentication_methods: + print(m.id, m.type, m.created_at) +``` + +> [!NOTE] +> `AuthenticationMethod` and `Factor` are forward-tolerant (`extra="allow"`): fields or method/factor types Auth0 adds later still deserialize. Don't switch exhaustively on `type` — handle unknown types gracefully. + +## 4. Get a single authentication method + +```python +method = await my_account.get_authentication_method( + access_token=access_token, + authentication_method_id="passkey|abc123", +) +``` + +> [!NOTE] +> Method IDs (e.g. `passkey|abc123`) can contain characters like `|`. The SDK URL-encodes every ID it places in a path, so pass the raw ID exactly as returned — do not pre-encode it. + +## 5. Update (rename) an authentication method + +```python +from auth0_server_python.auth_types import UpdateAuthenticationMethodRequest + +method = await my_account.update_authentication_method( + access_token=access_token, + authentication_method_id="totp|123", + request=UpdateAuthenticationMethodRequest(name="My Authenticator App"), +) +``` + +## 6. Delete an authentication method + +See [Auth0 Docs → Delete an authentication method](https://auth0.com/docs/manage-users/my-account-api#delete-an-authentication-method). + +```python +await my_account.delete_authentication_method( + access_token=access_token, + authentication_method_id="passkey|abc123", +) +# 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`. + +### Basic handling (recommended) + +```python +from auth0_server_python.error import Auth0Error + +try: + methods = await my_account.list_authentication_methods(access_token=access_token) +except Auth0Error as e: + return {"error": str(e)} +``` + +### Advanced handling (when actions differ by case) + +```python +from auth0_server_python.error import Auth0Error, MyAccountApiError + +try: + await my_account.enroll_authentication_method( + access_token=access_token, + request=EnrollAuthenticationMethodRequest(type="passkey"), + ) +except MyAccountApiError as e: + if e.status == 401: + return redirect_to_login() # token expired + if e.status == 403: + return {"error": "Missing required scope"} # e.g. create:me:authentication-methods + if e.status == 400 and e.validation_errors: + return {"error": "Validation failed", "details": e.validation_errors} + raise +except Auth0Error as e: + return {"error": str(e)} +``` + +> [!NOTE] +> Enrollment raises `MyAccountApiError`/`ApiError`, whereas passkey **sign-in** (`ServerClient`) raises `PasskeyError`. They are two distinct API surfaces — an auth grant versus a My Account resource — so write the `except` that matches the call you made. + +### Common error types + +- **`Auth0Error`** (base): catch for general handling +- **`MyAccountApiError`**: My Account API errors with `status`, `detail`, optional `validation_errors` +- **`MissingRequiredArgumentError`**: a required parameter (`access_token`, `authentication_method_id`, `request`) was not provided +- **`ApiError`**: transport failure or a non-JSON error body diff --git a/examples/Passkeys.md b/examples/Passkeys.md index ed50d59..0ad39bb 100644 --- a/examples/Passkeys.md +++ b/examples/Passkeys.md @@ -5,12 +5,16 @@ Passkeys let users sign up and log in with [WebAuthn](https://www.w3.org/TR/weba > [!NOTE] > Passkeys require a [Custom Domain](https://auth0.com/docs/customize/custom-domains) (WebAuthn binds the credential to the relying-party domain) and the native passkey feature enabled on your tenant. See the [Auth0 passkey documentation](https://auth0.com/docs/authenticate/database-connections/passkeys). +> [!NOTE] +> Managing a logged-in user's enrolled passkeys (enroll a new passkey, list, rename, delete) is a **separate** surface on the My Account API. See [examples/MyAccountAuthenticationMethods.md](MyAccountAuthenticationMethods.md). + ## Table of Contents - [How the flow works](#how-the-flow-works) - [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) @@ -132,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): @@ -143,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. @@ -155,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}, ) @@ -171,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.) @@ -217,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 499b981..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,16 +1,25 @@ - -from typing import Optional +import json +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, CompleteConnectAccountResponse, ConnectAccountRequest, ConnectAccountResponse, + EnrollAuthenticationMethodRequest, + EnrollmentChallengeResponse, + GetFactorsResponse, + ListAuthenticationMethodsResponse, ListConnectedAccountConnectionsResponse, ListConnectedAccountsResponse, + UpdateAuthenticationMethodRequest, + VerifyAuthenticationMethodRequest, ) from auth0_server_python.error import ( ApiError, @@ -19,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: """ @@ -340,3 +361,462 @@ async def list_connected_account_connections( f"Connected Accounts list connections request failed: {str(e) or 'Unknown error'}", e ) + + # ============================================================================ + # AUTHENTICATION METHODS & FACTORS (Passkey / MyAccount API) + # ============================================================================ + + 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. + + Raises: + MissingRequiredArgumentError: If access_token is not provided. + MyAccountApiError: If the API returns an error response. + ApiError: If the request fails due to network or other issues. + """ + if not access_token: + raise MissingRequiredArgumentError("access_token") + + try: + async with self._get_http_client() as client: + response = await client.get( + url=f"{self.audience}v1/factors", + auth=_make_auth(access_token, dpop_key), + ) + + if response.status_code != 200: + try: + error_data = response.json() + except (json.JSONDecodeError, ValueError): + raise ApiError( + "get_factors_error", + f"Get factors failed with status {response.status_code}", + ) + raise MyAccountApiError( + title=error_data.get("title", None), + type=error_data.get("type", None), + detail=error_data.get("detail", None), + status=error_data.get("status", None), + validation_errors=error_data.get("validation_errors", None), + ) + + # Auth0 /me/v1/factors returns a plain array, not {"factors":[...]} + raw = response.json() + payload = raw if isinstance(raw, dict) else {"factors": raw} + return GetFactorsResponse.model_validate(payload) + + except Exception as e: + if isinstance(e, (MyAccountApiError, ApiError)): + raise + raise ApiError( + "get_factors_error", + "Get factors request failed", + e, + ) + + 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. + + 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. + + Raises: + MissingRequiredArgumentError: If access_token is not provided. + MyAccountApiError: If the API returns an error response. + ApiError: If the request fails due to network or other issues. + """ + if not access_token: + raise MissingRequiredArgumentError("access_token") + + try: + async with self._get_http_client() as client: + params = {} + if type_filter: + params["type"] = type_filter + + response = await client.get( + url=f"{self.audience}v1/authentication-methods", + params=params, + auth=_make_auth(access_token, dpop_key), + ) + + if response.status_code != 200: + try: + error_data = response.json() + except (json.JSONDecodeError, ValueError): + raise ApiError( + "list_authentication_methods_error", + f"List authentication methods failed with status {response.status_code}", + ) + raise MyAccountApiError( + title=error_data.get("title", None), + type=error_data.get("type", None), + detail=error_data.get("detail", None), + status=error_data.get("status", None), + validation_errors=error_data.get("validation_errors", None), + ) + + return ListAuthenticationMethodsResponse.model_validate(response.json()) + + except Exception as e: + if isinstance(e, (MyAccountApiError, ApiError)): + raise + raise ApiError( + "list_authentication_methods_error", + "List authentication methods request failed", + e, + ) + + 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. + + 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. + + Raises: + MissingRequiredArgumentError: If access_token or authentication_method_id is not provided. + MyAccountApiError: If the API returns an error response. + ApiError: If the request fails due to network or other issues. + """ + if not access_token: + raise MissingRequiredArgumentError("access_token") + if not authentication_method_id: + raise MissingRequiredArgumentError("authentication_method_id") + + try: + 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=_make_auth(access_token, dpop_key), + ) + + if response.status_code != 200: + try: + error_data = response.json() + except (json.JSONDecodeError, ValueError): + raise ApiError( + "get_authentication_method_error", + f"Get authentication method failed with status {response.status_code}", + ) + raise MyAccountApiError( + title=error_data.get("title", None), + type=error_data.get("type", None), + detail=error_data.get("detail", None), + status=error_data.get("status", None), + validation_errors=error_data.get("validation_errors", None), + ) + + return AuthenticationMethod.model_validate(response.json()) + + except Exception as e: + if isinstance(e, (MyAccountApiError, ApiError)): + raise + raise ApiError( + "get_authentication_method_error", + "Get authentication method request failed", + e, + ) + + 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. + + 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. + MyAccountApiError: If the API returns an error response. + ApiError: If the request fails due to network or other issues. + """ + if not access_token: + raise MissingRequiredArgumentError("access_token") + if not authentication_method_id: + raise MissingRequiredArgumentError("authentication_method_id") + + try: + 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=_make_auth(access_token, dpop_key), + ) + + if response.status_code != 204: + try: + error_data = response.json() + except (json.JSONDecodeError, ValueError): + raise ApiError( + "delete_authentication_method_error", + f"Delete authentication method failed with status {response.status_code}", + ) + raise MyAccountApiError( + title=error_data.get("title", None), + type=error_data.get("type", None), + detail=error_data.get("detail", None), + status=error_data.get("status", None), + validation_errors=error_data.get("validation_errors", None), + ) + + except Exception as e: + if isinstance(e, (MyAccountApiError, ApiError)): + raise + raise ApiError( + "delete_authentication_method_error", + "Delete authentication method request failed", + e, + ) + + async def update_authentication_method( + self, + access_token: str, + authentication_method_id: str, + request: UpdateAuthenticationMethodRequest, + dpop_key: Optional["jwk.JWK"] = None, + ) -> AuthenticationMethod: + """ + Update an authentication method by ID. + + Args: + 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. + + Raises: + MissingRequiredArgumentError: If access_token, authentication_method_id, or request is not provided. + MyAccountApiError: If the API returns an error response. + ApiError: If the request fails due to network or other issues. + """ + if not access_token: + raise MissingRequiredArgumentError("access_token") + if not authentication_method_id: + raise MissingRequiredArgumentError("authentication_method_id") + if request is None: + raise MissingRequiredArgumentError("request") + + try: + async with self._get_http_client() as client: + response = await client.patch( + url=f"{self.audience}v1/authentication-methods/{quote(authentication_method_id, safe='')}", + json=request.model_dump(exclude_none=True), + auth=_make_auth(access_token, dpop_key), + ) + + if response.status_code != 200: + try: + error_data = response.json() + except (json.JSONDecodeError, ValueError): + raise ApiError( + "update_authentication_method_error", + f"Update authentication method failed with status {response.status_code}", + ) + raise MyAccountApiError( + title=error_data.get("title", None), + type=error_data.get("type", None), + detail=error_data.get("detail", None), + status=error_data.get("status", None), + validation_errors=error_data.get("validation_errors", None), + ) + + return AuthenticationMethod.model_validate(response.json()) + + except Exception as e: + if isinstance(e, (MyAccountApiError, ApiError)): + raise + raise ApiError( + "update_authentication_method_error", + "Update authentication method request failed", + e, + ) + + 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). + + For passkey enrollment, pass the returned authn_params_public_key to + navigator.credentials.create(), then call verify_authentication_method() + with the auth_session and credential result. + + Requires scope: create:me:authentication-methods + """ + if not access_token: + raise MissingRequiredArgumentError("access_token") + if request is None: + raise MissingRequiredArgumentError("request") + + try: + async with self._get_http_client() as client: + response = await client.post( + url=f"{self.audience}v1/authentication-methods", + json=request.model_dump(exclude_none=True), + auth=_make_auth(access_token, dpop_key), + ) + + if response.status_code != 202: + try: + error_data = response.json() + except (json.JSONDecodeError, ValueError): + raise ApiError( + "enroll_authentication_method_error", + f"Enroll authentication method failed with status {response.status_code}", + ) + raise MyAccountApiError( + title=error_data.get("title", None), + type=error_data.get("type", None), + detail=error_data.get("detail", None), + status=error_data.get("status", None), + validation_errors=error_data.get("validation_errors", None), + ) + + location = response.headers.get("location") + if not location: + raise ApiError( + "enroll_authentication_method_error", + "Enrollment succeeded (202) but Location header is missing", + ) + + parsed_path = urlparse(location).path.rstrip("/") + raw_id = parsed_path.rsplit("/", 1)[-1] if "/" in parsed_path else "" + authentication_method_id = unquote(raw_id) + if not authentication_method_id or authentication_method_id in ( + "authentication-methods", + "v1", + "me", + ): + raise ApiError( + "enroll_authentication_method_error", + "Enrollment succeeded (202) but could not extract ID from Location header", + ) + + try: + data = response.json() + except (json.JSONDecodeError, ValueError): + raise ApiError( + "enroll_authentication_method_error", + "Enrollment succeeded (202) but response body is not valid JSON", + ) + + auth_session = data.get("auth_session") + if not auth_session: + raise ApiError( + "enroll_authentication_method_error", + "Enrollment succeeded (202) but auth_session is missing from response", + ) + + return EnrollmentChallengeResponse.model_validate( + { + **data, + "authentication_method_id": authentication_method_id, + "auth_session": auth_session, + } + ) + + except Exception as e: + if isinstance(e, (MyAccountApiError, ApiError)): + raise + raise ApiError( + "enroll_authentication_method_error", + "Enroll authentication method request failed", + e, + ) + + async def verify_authentication_method( + self, + 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). + + Requires scope: create:me:authentication-methods + """ + if not access_token: + raise MissingRequiredArgumentError("access_token") + if not authentication_method_id: + raise MissingRequiredArgumentError("authentication_method_id") + if request is None: + raise MissingRequiredArgumentError("request") + + try: + async with self._get_http_client() as client: + 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=_make_auth(access_token, dpop_key), + ) + + if response.status_code != 201: + try: + error_data = response.json() + except (json.JSONDecodeError, ValueError): + raise ApiError( + "verify_authentication_method_error", + f"Verify authentication method failed with status {response.status_code}", + ) + raise MyAccountApiError( + title=error_data.get("title", None), + type=error_data.get("type", None), + detail=error_data.get("detail", None), + status=error_data.get("status", None), + validation_errors=error_data.get("validation_errors", None), + ) + + return AuthenticationMethod.model_validate(response.json()) + + except Exception as e: + if isinstance(e, (MyAccountApiError, ApiError)): + raise + raise ApiError( + "verify_authentication_method_error", + "Verify authentication method request failed", + e, + ) 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/auth_types/__init__.py b/src/auth0_server_python/auth_types/__init__.py index 7ae8a58..da2a228 100644 --- a/src/auth0_server_python/auth_types/__init__.py +++ b/src/auth0_server_python/auth_types/__init__.py @@ -16,6 +16,8 @@ # challenge type (e.g. a future webauthn second factor) does not fail closed. OobChannel = Literal["sms", "voice", "auth0", "email"] ChallengeType = Literal["otp", "oob"] +EnrollmentType = Literal["passkey", "email", "phone", "totp", "push-notification", "recovery-code", "password"] +PreferredAuthMethod = Literal["sms", "voice"] # Deprecated public aliases resolved lazily (PEP 562) so access emits a warning # while imports keep working. Remove in a future release. @@ -720,6 +722,22 @@ class PasskeyPublicKeyOptions(BaseModel): user_verification: Optional[str] = Field(None, alias="userVerification") +class EnrollAuthenticationMethodRequest(BaseModel): + type: EnrollmentType + email: Optional[str] = None + phone_number: Optional[str] = None + preferred_authentication_method: Optional[PreferredAuthMethod] = None + identity_user_id: Optional[str] = None # OAS: IdentityAuthenticationMethodBase.identity_user_id + connection: Optional[str] = None + + +class EnrollmentChallengeResponse(BaseModel): + model_config = ConfigDict(extra="allow") + authentication_method_id: str + auth_session: str + authn_params_public_key: Optional[PasskeyPublicKeyOptions] = None + + class PasskeyAuthResponse(BaseModel): model_config = ConfigDict(populate_by_name=True) id: str @@ -730,6 +748,58 @@ class PasskeyAuthResponse(BaseModel): client_extension_results: Optional[dict[str, Any]] = Field(None, alias="clientExtensionResults") +class VerifyAuthenticationMethodRequest(BaseModel): + auth_session: str + authn_response: Optional[PasskeyAuthResponse] = None + otp_code: Optional[str] = None + recovery_code: Optional[str] = None + password: Optional[str] = None + + +class AuthenticationMethod(BaseModel): + model_config = ConfigDict(extra="allow") + + id: str + type: str + created_at: str + confirmed: Optional[bool] = None + usage: Optional[list[str]] = None + identity_user_id: Optional[str] = None + credential_device_type: Optional[str] = None + credential_backed_up: Optional[bool] = None + key_id: Optional[str] = None + public_key: Optional[str] = None + transports: Optional[list[str]] = None + user_agent: Optional[str] = None + user_handle: Optional[str] = None + aaguid: Optional[str] = None + relying_party_id: Optional[str] = None + phone_number: Optional[str] = None + preferred_authentication_method: Optional[str] = None + email: Optional[str] = None + name: Optional[str] = None + last_password_reset: Optional[str] = None + + +class UpdateAuthenticationMethodRequest(BaseModel): + name: Optional[str] = None + preferred_authentication_method: Optional[str] = None + + +class ListAuthenticationMethodsResponse(BaseModel): + authentication_methods: list[AuthenticationMethod] + + +class Factor(BaseModel): + model_config = ConfigDict(extra="allow") + type: str + usage: Optional[list[str]] = None + + +class GetFactorsResponse(BaseModel): + factors: list[Factor] + + class PasskeyUserProfile(BaseModel): model_config = ConfigDict(extra="allow") email: Optional[str] = None 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 e4ff74c..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,9 +1,15 @@ +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, CompleteConnectAccountRequest, CompleteConnectAccountResponse, ConnectAccountRequest, @@ -11,10 +17,18 @@ ConnectedAccount, ConnectedAccountConnection, ConnectParams, + EnrollAuthenticationMethodRequest, + EnrollmentChallengeResponse, + GetFactorsResponse, + ListAuthenticationMethodsResponse, ListConnectedAccountConnectionsResponse, ListConnectedAccountsResponse, + PasskeyAuthResponse, + UpdateAuthenticationMethodRequest, + VerifyAuthenticationMethodRequest, ) from auth0_server_python.error import ( + ApiError, InvalidArgumentError, MissingRequiredArgumentError, MyAccountApiError, @@ -502,3 +516,909 @@ async def test_list_connected_account_connections_api_response_failure(mocker): mock_get.assert_awaited_once() assert "Invalid Token" in str(exc.value) + +# ============================================================================= +# AUTHENTICATION METHODS & FACTORS (Passkey / MyAccount API) +# ============================================================================= + + +@pytest.mark.asyncio +async def test_get_factors_success(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock( + return_value={"factors": [{"type": "phone", "usage": ["primary"]}]} + ) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + result = await client.get_factors(access_token="token123") + + assert isinstance(result, GetFactorsResponse) + assert len(result.factors) == 1 + assert result.factors[0].type == "phone" + assert result.factors[0].usage == ["primary"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("access_token", [None, ""]) +async def test_get_factors_missing_access_token(mocker, access_token): + client = MyAccountClient(domain="auth0.local") + mock_get = mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock) + + with pytest.raises(MissingRequiredArgumentError): + await client.get_factors(access_token=access_token) + + mock_get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_get_factors_api_error(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 403 + response.json = MagicMock(return_value={ + "title": "Forbidden", + "type": "forbidden", + "detail": "Insufficient scope", + "status": 403, + }) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + with pytest.raises(MyAccountApiError) as exc: + await client.get_factors(access_token="token123") + + assert exc.value.status == 403 + + +@pytest.mark.asyncio +async def test_get_factors_network_error(mocker): + client = MyAccountClient(domain="auth0.local") + mocker.patch( + "httpx.AsyncClient.get", new_callable=AsyncMock, side_effect=Exception("Connection refused") + ) + + with pytest.raises(ApiError): + await client.get_factors(access_token="token123") + + +@pytest.mark.asyncio +async def test_get_factors_empty_list(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={"factors": []}) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + result = await client.get_factors(access_token="token123") + assert result.factors == [] + + +@pytest.mark.asyncio +async def test_get_factors_extra_fields(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={ + "factors": [{"type": "webauthn-roaming", "usage": ["secondary"], "future_field": "value"}] + }) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + result = await client.get_factors(access_token="token123") + assert result.factors[0].type == "webauthn-roaming" + assert result.factors[0].model_extra["future_field"] == "value" + + +@pytest.mark.asyncio +async def test_list_authentication_methods_success(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={ + "authentication_methods": [ + {"id": "am_1", "type": "passkey", "created_at": "2026-01-01T00:00:00Z", "key_id": "kid1"} + ] + }) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + result = await client.list_authentication_methods(access_token="token123") + assert isinstance(result, ListAuthenticationMethodsResponse) + assert len(result.authentication_methods) == 1 + assert result.authentication_methods[0].type == "passkey" + + +@pytest.mark.asyncio +async def test_list_authentication_methods_with_type_filter(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) + + await client.list_authentication_methods(access_token="token123", type_filter="passkey") + mock_get.assert_awaited_once() + call_kwargs = mock_get.call_args[1] + assert call_kwargs["params"] == {"type": "passkey"} + + +@pytest.mark.asyncio +async def test_list_authentication_methods_empty(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={"authentication_methods": []}) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + result = await client.list_authentication_methods(access_token="token123") + assert result.authentication_methods == [] + + +@pytest.mark.asyncio +async def test_get_authentication_method_success(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" + }) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + result = await client.get_authentication_method( + access_token="token123", authentication_method_id="am_1" + ) + assert isinstance(result, AuthenticationMethod) + assert result.id == "am_1" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method_id", [None, ""]) +async def test_get_authentication_method_missing_id(mocker, method_id): + client = MyAccountClient(domain="auth0.local") + mock_get = mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock) + + with pytest.raises(MissingRequiredArgumentError): + await client.get_authentication_method( + access_token="token123", authentication_method_id=method_id + ) + + mock_get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_get_authentication_method_path_traversal(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={ + "id": "id/slash", "type": "passkey", "created_at": "2026-01-01T00:00:00Z" + }) + mock_get = mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + await client.get_authentication_method( + access_token="token123", authentication_method_id="id/slash" + ) + call_url = mock_get.call_args[1]["url"] + assert "id%2Fslash" in call_url + assert "id/slash" not in call_url.replace("https://auth0.local/me/", "") + + +@pytest.mark.asyncio +async def test_get_authentication_method_pipe_encoding(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 200 + response.json = MagicMock(return_value={ + "id": "passkey|new", "type": "passkey", "created_at": "2026-01-01T00:00:00Z" + }) + mock_get = mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + await client.get_authentication_method( + access_token="token123", authentication_method_id="passkey|new" + ) + call_url = mock_get.call_args[1]["url"] + assert "passkey%7Cnew" in call_url + + +@pytest.mark.asyncio +async def test_delete_authentication_method_success(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 204 + mocker.patch("httpx.AsyncClient.delete", new_callable=AsyncMock, return_value=response) + + result = await client.delete_authentication_method( + access_token="token123", authentication_method_id="am_1" + ) + assert result is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method_id", [None, ""]) +async def test_delete_authentication_method_missing_id(mocker, method_id): + client = MyAccountClient(domain="auth0.local") + mock_delete = mocker.patch("httpx.AsyncClient.delete", new_callable=AsyncMock) + + with pytest.raises(MissingRequiredArgumentError): + await client.delete_authentication_method( + access_token="token123", authentication_method_id=method_id + ) + + mock_delete.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_update_authentication_method_success(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", "name": "My Key", + }) + mock_patch = mocker.patch( + "httpx.AsyncClient.patch", new_callable=AsyncMock, return_value=response + ) + + req = UpdateAuthenticationMethodRequest(name="My Key") + result = await client.update_authentication_method( + access_token="token123", authentication_method_id="am_1", request=req + ) + assert result.name == "My Key" + call_kwargs = mock_patch.call_args[1] + assert call_kwargs["json"] == {"name": "My Key"} + + +@pytest.mark.asyncio +async def test_update_authentication_method_missing_request(mocker): + client = MyAccountClient(domain="auth0.local") + mocker.patch("httpx.AsyncClient.patch", new_callable=AsyncMock) + + with pytest.raises(MissingRequiredArgumentError): + await client.update_authentication_method( + access_token="token123", authentication_method_id="am_1", request=None + ) + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_success(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", + "authn_params_public_key": { + "challenge": "dGVzdA", + "rp": {"id": "auth0.local", "name": "My App"}, + "user": {"id": "dXNlcl8x", "name": "user@test.com", "displayName": "Test User"}, + "pubKeyCredParams": [{"type": "public-key", "alg": -7}], + "authenticatorSelection": {"residentKey": "required", "userVerification": "preferred"}, + "timeout": 60000, + }, + }) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="passkey") + result = await client.enroll_authentication_method(access_token="token123", request=req) + + assert isinstance(result, EnrollmentChallengeResponse) + assert result.authentication_method_id == "passkey|new" + assert result.auth_session == "session_abc" + assert result.authn_params_public_key is not None + assert result.authn_params_public_key.pub_key_cred_params[0].alg == -7 + assert result.authn_params_public_key.authenticator_selection.resident_key == "required" + assert result.authn_params_public_key.user.display_name == "Test User" + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_sends_identity_user_id(mocker): + """identity_user_id must serialize to the request body under its OAS wire key.""" + 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 + ) + + req = EnrollAuthenticationMethodRequest(type="passkey", identity_user_id="auth0|abc123") + await client.enroll_authentication_method(access_token="token123", request=req) + + sent_body = mock_post.call_args[1]["json"] + assert sent_body["identity_user_id"] == "auth0|abc123" + assert "user_identity_id" not in sent_body + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_public_key_extra_fields_preserved(mocker): + """Unknown WebAuthn fields (excludeCredentials, attestation, extensions) must not be dropped.""" + 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", + "authn_params_public_key": { + "challenge": "dGVzdA", + "rp": {"id": "auth0.local", "name": "My App"}, + "user": {"id": "dXNlcl8x", "name": "user@test.com"}, + "pubKeyCredParams": [{"type": "public-key", "alg": -7}], + "excludeCredentials": [{"type": "public-key", "id": "Y3JlZA"}], + "attestation": "direct", + "extensions": {"appid": "https://auth0.local"}, + }, + }) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="passkey") + result = await client.enroll_authentication_method(access_token="token123", request=req) + + pk = result.authn_params_public_key + assert pk.model_extra["excludeCredentials"] == [{"type": "public-key", "id": "Y3JlZA"}] + assert pk.model_extra["attestation"] == "direct" + assert pk.model_extra["extensions"] == {"appid": "https://auth0.local"} + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_missing_location(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 202 + response.headers = {} + response.json = MagicMock(return_value={"auth_session": "session_abc"}) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="passkey") + with pytest.raises(ApiError) as exc: + await client.enroll_authentication_method(access_token="token123", request=req) + + assert "Location header" in str(exc.value) + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_location_with_query(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 202 + response.headers = {"location": "/me/v1/authentication-methods/abc123?tracking=1"} + response.json = MagicMock(return_value={"auth_session": "session_abc"}) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="passkey") + result = await client.enroll_authentication_method(access_token="token123", request=req) + assert result.authentication_method_id == "abc123" + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_location_absolute_url(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 202 + response.headers = {"location": "https://tenant.auth0.com/me/v1/authentication-methods/am_xyz"} + response.json = MagicMock(return_value={"auth_session": "session_abc"}) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="passkey") + result = await client.enroll_authentication_method(access_token="token123", request=req) + assert result.authentication_method_id == "am_xyz" + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_totp_preserves_secret(mocker): + """TOTP enrollment response includes totp_secret and barcode_uri — must not be dropped.""" + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 202 + response.headers = {"location": "/me/v1/authentication-methods/totp|new"} + response.json = MagicMock(return_value={ + "auth_session": "session_totp", + "totp_secret": "JBSWY3DPEHPK3PXP", + "barcode_uri": "otpauth://totp/Example:alice@example.com?secret=JBSWY3DPEHPK3PXP", + }) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="totp") + result = await client.enroll_authentication_method(access_token="token123", request=req) + + assert result.authentication_method_id == "totp|new" + assert result.auth_session == "session_totp" + assert result.model_extra["totp_secret"] == "JBSWY3DPEHPK3PXP" + assert result.model_extra["barcode_uri"].startswith("otpauth://") + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_oob_preserves_oob_code(mocker): + """OOB (email/phone) enrollment response includes oob_code — must not be dropped.""" + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 202 + response.headers = {"location": "/me/v1/authentication-methods/email|new"} + response.json = MagicMock(return_value={ + "auth_session": "session_oob", + "oob_code": "oob_abc123", + }) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="email") + result = await client.enroll_authentication_method(access_token="token123", request=req) + + assert result.authentication_method_id == "email|new" + assert result.model_extra["oob_code"] == "oob_abc123" + + +@pytest.mark.asyncio +async def test_verify_authentication_method_success(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", "confirmed": True, + }) + mock_post = mocker.patch( + "httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response + ) + + authn_response = PasskeyAuthResponse( + id="cred1", + raw_id="cmF3MQ", + type="public-key", + authenticator_attachment="platform", + response={"clientDataJSON": "abc", "attestationObject": "def"}, + ) + req = VerifyAuthenticationMethodRequest( + auth_session="session_abc", authn_response=authn_response + ) + result = await client.verify_authentication_method( + access_token="token123", authentication_method_id="passkey|new", request=req + ) + + assert isinstance(result, AuthenticationMethod) + assert result.confirmed is True + + call_kwargs = mock_post.call_args[1] + body = call_kwargs["json"] + assert "rawId" in body["authn_response"] + assert "raw_id" not in body["authn_response"] + assert "authenticatorAttachment" in body["authn_response"] + assert body["auth_session"] == "session_abc" + assert "passkey%7Cnew" in call_kwargs["url"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method_id", [None, ""]) +async def test_verify_authentication_method_missing_id(mocker, method_id): + client = MyAccountClient(domain="auth0.local") + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock) + + req = VerifyAuthenticationMethodRequest(auth_session="session_abc", otp_code="123456") + with pytest.raises(MissingRequiredArgumentError): + await client.verify_authentication_method( + access_token="token123", authentication_method_id=method_id, request=req + ) + + +def test_verify_request_auth_session_only_is_valid(): + req = VerifyAuthenticationMethodRequest(auth_session="session_abc") + assert req.auth_session == "session_abc" + assert req.otp_code is None + assert req.authn_response is None + + +def test_verify_request_accepts_otp_code(): + req = VerifyAuthenticationMethodRequest(auth_session="session_abc", otp_code="123456") + assert req.otp_code == "123456" + + +def test_verify_request_accepts_authn_response(): + authn_resp = PasskeyAuthResponse( + id="cred1", + raw_id="cmF3MQ", + type="public-key", + response={"clientDataJSON": "abc", "attestationObject": "def"}, + ) + req = VerifyAuthenticationMethodRequest(auth_session="session_abc", authn_response=authn_resp) + 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") + response = AsyncMock() + response.status_code = 403 + response.json = MagicMock(return_value={ + "title": "Forbidden", "type": "forbidden", "detail": "Insufficient scope", "status": 403, + }) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + with pytest.raises(MyAccountApiError) as exc: + await client.list_authentication_methods(access_token="token123") + assert exc.value.status == 403 + + +@pytest.mark.asyncio +async def test_list_authentication_methods_network_error(mocker): + client = MyAccountClient(domain="auth0.local") + mocker.patch( + "httpx.AsyncClient.get", new_callable=AsyncMock, side_effect=Exception("Connection refused") + ) + + with pytest.raises(ApiError): + await client.list_authentication_methods(access_token="token123") + + +@pytest.mark.asyncio +async def test_get_authentication_method_api_error(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 404 + response.json = MagicMock(return_value={ + "title": "Not Found", "type": "not_found", "detail": "Not found", "status": 404, + }) + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, return_value=response) + + with pytest.raises(MyAccountApiError) as exc: + await client.get_authentication_method( + access_token="token123", authentication_method_id="am_1" + ) + assert exc.value.status == 404 + + +@pytest.mark.asyncio +async def test_get_authentication_method_network_error(mocker): + client = MyAccountClient(domain="auth0.local") + mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock, side_effect=Exception("timeout")) + + with pytest.raises(ApiError): + await client.get_authentication_method( + access_token="token123", authentication_method_id="am_1" + ) + + +@pytest.mark.asyncio +async def test_delete_authentication_method_api_error(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 404 + response.json = MagicMock(return_value={ + "title": "Not Found", "type": "not_found", "detail": "Not found", "status": 404, + }) + mocker.patch("httpx.AsyncClient.delete", new_callable=AsyncMock, return_value=response) + + with pytest.raises(MyAccountApiError) as exc: + await client.delete_authentication_method( + access_token="token123", authentication_method_id="am_1" + ) + assert exc.value.status == 404 + + +@pytest.mark.asyncio +async def test_delete_authentication_method_network_error(mocker): + client = MyAccountClient(domain="auth0.local") + mocker.patch( + "httpx.AsyncClient.delete", + new_callable=AsyncMock, + side_effect=Exception("Connection reset"), + ) + + with pytest.raises(ApiError): + await client.delete_authentication_method( + access_token="token123", authentication_method_id="am_1" + ) + + +@pytest.mark.asyncio +async def test_update_authentication_method_api_error(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 422 + response.json = MagicMock(return_value={ + "title": "Unprocessable", "type": "validation_error", "detail": "Invalid", "status": 422, + }) + mocker.patch("httpx.AsyncClient.patch", new_callable=AsyncMock, return_value=response) + + req = UpdateAuthenticationMethodRequest(name="x") + with pytest.raises(MyAccountApiError) as exc: + await client.update_authentication_method( + access_token="token123", authentication_method_id="am_1", request=req + ) + assert exc.value.status == 422 + + +@pytest.mark.asyncio +async def test_update_authentication_method_network_error(mocker): + client = MyAccountClient(domain="auth0.local") + mocker.patch( + "httpx.AsyncClient.patch", new_callable=AsyncMock, side_effect=Exception("timeout") + ) + + req = UpdateAuthenticationMethodRequest(name="x") + with pytest.raises(ApiError): + await client.update_authentication_method( + access_token="token123", authentication_method_id="am_1", request=req + ) + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_api_error(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 403 + response.json = MagicMock(return_value={ + "title": "Forbidden", "type": "forbidden", "detail": "Scope missing", "status": 403, + }) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="passkey") + with pytest.raises(MyAccountApiError) as exc: + await client.enroll_authentication_method(access_token="token123", request=req) + assert exc.value.status == 403 + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_network_error(mocker): + client = MyAccountClient(domain="auth0.local") + mocker.patch( + "httpx.AsyncClient.post", + new_callable=AsyncMock, + side_effect=Exception("Connection refused"), + ) + + req = EnrollAuthenticationMethodRequest(type="passkey") + with pytest.raises(ApiError): + await client.enroll_authentication_method(access_token="token123", request=req) + + +@pytest.mark.asyncio +async def test_verify_authentication_method_api_error(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 400 + response.json = MagicMock(return_value={ + "title": "Bad Request", "type": "invalid_request", "detail": "Invalid OTP", "status": 400, + }) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = VerifyAuthenticationMethodRequest(auth_session="session_abc", otp_code="000000") + with pytest.raises(MyAccountApiError) as exc: + await client.verify_authentication_method( + access_token="token123", authentication_method_id="am_1", request=req + ) + assert exc.value.status == 400 + + +@pytest.mark.asyncio +async def test_verify_authentication_method_network_error(mocker): + client = MyAccountClient(domain="auth0.local") + mocker.patch( + "httpx.AsyncClient.post", + new_callable=AsyncMock, + side_effect=Exception("Connection refused"), + ) + + req = VerifyAuthenticationMethodRequest(auth_session="session_abc", otp_code="123456") + with pytest.raises(ApiError): + await client.verify_authentication_method( + access_token="token123", authentication_method_id="am_1", request=req + ) + + +@pytest.mark.asyncio +async def test_enroll_authentication_method_location_collection_url(mocker): + client = MyAccountClient(domain="auth0.local") + response = AsyncMock() + response.status_code = 202 + response.headers = {"location": "/me/v1/authentication-methods/"} + response.json = MagicMock(return_value={"auth_session": "session_abc"}) + mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response) + + req = EnrollAuthenticationMethodRequest(type="passkey") + 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."""