diff --git a/examples/Passwordless.md b/examples/Passwordless.md new file mode 100644 index 0000000..099c057 --- /dev/null +++ b/examples/Passwordless.md @@ -0,0 +1,359 @@ +# Passwordless Authentication + +Passwordless lets users sign in with a one-time code sent by email or SMS, or with a magic link sent by email. This guide covers the **embedded login** flow on `ServerClient.passwordless` and how each path establishes a server-side session. + +> [!NOTE] +> Passwordless API flows use Auth0 Legacy Passwordless connections (`email` and `sms`). Enable the **Passwordless OTP** grant for your application under **Applications -> Your App -> Advanced Settings -> Grant Types**. See the [Auth0 Passwordless API documentation](https://auth0.com/docs/authenticate/passwordless/implement-login/embedded-login/relevant-api-endpoints). + +> [!IMPORTANT] +> These flows are for confidential server-side applications. Tokens stay on the server; the browser should only receive your application's session cookie or opaque session reference. + +## Table of Contents + +- [How the flow works](#how-the-flow-works) +- [Prerequisites](#prerequisites) +- [1. Email OTP](#1-email-otp) +- [2. SMS OTP](#2-sms-otp) +- [3. Email magic link](#3-email-magic-link) +- [4. Custom scopes and audiences](#4-custom-scopes-and-audiences) +- [5. Forwarding the end-user IP](#5-forwarding-the-end-user-ip) +- [6. Organizations](#6-organizations) +- [Completing MFA during passwordless login](#completing-mfa-during-passwordless-login) +- [Error Handling](#error-handling) + +## How the flow works + +Passwordless has two shapes: + +1. **OTP code** - `start()` sends a code by email or SMS. Your app collects that code, then `verify()` exchanges it at `/oauth/token` with the passwordless OTP grant and **creates a server-side session**. +2. **Magic link** - `start(send="link")` sends a one-click email link. Auth0 redirects the user back to your callback URL, and your app completes the flow with `complete_interactive_login()`. The callback creates the server-side session. + +OTP start does **not** create a session. The session exists only after `verify()` succeeds. Magic-link start writes a transaction so the callback can validate the returned `state`; the session exists only after the callback completes. + +## Prerequisites + +```python +from auth0_server_python.auth_server.server_client import ServerClient + +server_client = ServerClient( + domain="YOUR_AUTH0_DOMAIN", + client_id="YOUR_CLIENT_ID", + client_secret="YOUR_CLIENT_SECRET", + secret="YOUR_SECRET", + redirect_uri="https://app.example.com/auth/callback", +) +``` + +For apps using request/response-backed stores or multiple custom domains, pass `store_options={"request": request, "response": response}` to each method that reads or writes transaction/session state. + +## 1. Email OTP + +### Step 1 - Send the code + +```python +from auth0_server_python.auth_types import StartPasswordlessEmailOptions + +start_result = await server_client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="code", + language="en-US", # optional x-request-language header + ), + store_options={"request": request, "response": response}, +) + +# start_result.id is Auth0's request identifier when returned by the API. +``` + +### Step 2 - Verify the code and establish the session + +```python +from auth0_server_python.auth_types import VerifyPasswordlessOtpOptions + +result = await server_client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", + email="user@example.com", + verification_code=user_entered_code, + ), + store_options={"request": request, "response": response}, +) + +user = result["state_data"]["user"] +print(f"Signed in: {user['sub']}") +``` + +The SDK verifies the returned ID token, validates the issuer and audience, persists the tokens in the configured state store, and sources the session `sid` from the verified ID token when available. + +## 2. SMS OTP + +SMS has the same two-step shape. Phone numbers must be in E.164 format. + +```python +from auth0_server_python.auth_types import ( + StartPasswordlessSmsOptions, + VerifyPasswordlessOtpOptions, +) + +await server_client.passwordless.start( + StartPasswordlessSmsOptions( + phone_number="+14155550100", + ), + store_options={"request": request, "response": response}, +) + +result = await server_client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="sms", + phone_number="+14155550100", + verification_code=user_entered_code, + ), + store_options={"request": request, "response": response}, +) +``` + +By default, email OTP requests `openid profile email`; SMS OTP requests `openid profile` because SMS identities do not have an email claim to satisfy. + +## 3. Email magic link + +Magic links are email-only. `start(send="link")` persists a transaction and includes SDK-owned `redirect_uri`, `response_type`, and `state` in Auth0's `authParams`. + +```python +from auth0_server_python.auth_types import StartPasswordlessEmailOptions + +await server_client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="link", + auth_params={ + "scope": "openid profile email", + "login_hint": "user@example.com", + }, + ), + store_options={"request": request, "response": response}, +) +``` + +> [!IMPORTANT] +> Magic-link start requires `store_options` whenever your transaction store needs the framework request/response to write state. Without that transaction, the callback cannot validate the returned `state`. + +When the user clicks the emailed link, Auth0 redirects back to your configured callback URL. Complete it with the standard interactive-login callback: + +```python +callback_url = str(request.url) + +result = await server_client.complete_interactive_login( + callback_url, + store_options={"request": request, "response": response}, +) + +user = result["state_data"]["user"] +``` + +> [!WARNING] +> Do not let callers override `redirect_uri`, `state`, `response_type`, `nonce`, or PKCE fields in magic-link `auth_params`. The SDK owns these values so the emailed authorization code and state cannot be redirected to an attacker-controlled URL. + +## 4. Custom scopes and audiences + +For OTP flows, pass `scope` and `audience` to `verify()`. These become the `/oauth/token` request parameters. + +```python +result = await server_client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", + email="user@example.com", + verification_code=user_entered_code, + audience="https://api.example.com", + scope="openid profile email offline_access read:orders", + ), + store_options={"request": request, "response": response}, +) +``` + +For magic links, pass allowed authorization parameters through `auth_params` at `start()` time: + +```python +await server_client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="link", + auth_params={ + "audience": "https://api.example.com", + "scope": "openid profile email offline_access read:orders", + "login_hint": "user@example.com", + }, + ), + store_options={"request": request, "response": response}, +) +``` + +> [!NOTE] +> `state` is intentionally not a caller-supplied auth parameter in this SDK. If you need app-specific return data, store it server-side against your own transaction/session context instead of putting it into the Auth0 magic-link `state`. + +## 5. Forwarding the end-user IP + +Auth0 attack protection and rate limiting normally see the IP address of the server making the API call. For confidential clients, Auth0 can use the `auth0-forwarded-for` header when the **Trust Token Endpoint IP Header** setting is enabled. + +```python +await server_client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="code", + client_ip=request.client.host, + ), + store_options={"request": request, "response": response}, +) + +result = await server_client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", + email="user@example.com", + verification_code=user_entered_code, + client_ip=request.client.host, + ), + store_options={"request": request, "response": response}, +) +``` + +> [!WARNING] +> Only forward a trusted, normalized end-user IP from your edge/proxy layer. Do not blindly copy arbitrary client-supplied headers into `client_ip`. + +## 6. Organizations + +Magic links can carry an organization through `authParams`; the SDK stores the expected organization in the transaction and validates the callback token claims. + +```python +await server_client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="link", + organization="org_abc123", + ), + store_options={"request": request, "response": response}, +) +``` + +For OTP verification, pass `organization` only when your Auth0 passwordless OTP configuration returns organization claims for that flow. The SDK validates the ID token's `org_id` or `org_name` claim against the supplied value. + +```python +result = await server_client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", + email="user@example.com", + verification_code=user_entered_code, + organization="org_abc123", + ), + store_options={"request": request, "response": response}, +) +``` + +If the ID token does not include a matching organization claim, verification fails before a session is persisted. + +## Completing MFA during passwordless login + +Auth0 can require MFA during passwordless OTP verification. In that case, the SDK raises `MfaRequiredError` before it creates a session. Complete the MFA challenge with `server_client.mfa`, then persist the returned tokens according to your framework's session integration. + +```python +from auth0_server_python.error import MfaRequiredError + +try: + result = await server_client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", + email="user@example.com", + verification_code=user_entered_code, + ), + store_options={"request": request, "response": response}, + ) + user = result["state_data"]["user"] + +except MfaRequiredError as e: + await server_client.mfa.challenge_authenticator( + {"mfa_token": e.mfa_token, "factor_type": "otp"}, + store_options={"request": request, "response": response}, + ) + + verify_response = await server_client.mfa.verify( + {"mfa_token": e.mfa_token, "otp": mfa_code}, + store_options={"request": request, "response": response}, + ) + + save_session_for_user( + access_token=verify_response.access_token, + id_token=verify_response.id_token, + refresh_token=verify_response.refresh_token, + ) +``` + +> [!NOTE] +> Passwordless OTP MFA is like passkey-first MFA: there is no existing application session yet. Use the returned MFA tokens to create the session in your framework layer rather than trying to update a session that does not exist. + +## Error Handling + +Passwordless methods raise typed SDK errors: + +- `PasswordlessStartError` - `POST /passwordless/start` failed +- `PasswordlessVerifyError` - OTP token exchange or ID-token verification failed +- `MfaRequiredError` - Auth0 requires MFA before completing login +- `MissingRequiredArgumentError` - required SDK input is missing, such as magic-link `store_options` +- `InvalidArgumentError` - caller input is rejected before a network call +- `OrganizationTokenValidationError` - requested organization does not match returned token claims + +### Basic handling + +```python +from auth0_server_python.error import Auth0Error + +try: + result = await server_client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", + email="user@example.com", + verification_code=user_entered_code, + ), + store_options={"request": request, "response": response}, + ) +except Auth0Error as e: + return {"error": str(e)} +``` + +### Advanced handling + +```python +from auth0_server_python.error import ( + Auth0Error, + MfaRequiredError, + PasswordlessStartError, + PasswordlessVerifyError, +) + +try: + result = await server_client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", + email="user@example.com", + verification_code=user_entered_code, + ), + store_options={"request": request, "response": response}, + ) +except MfaRequiredError as e: + return start_mfa(e.mfa_token) +except PasswordlessVerifyError as e: + return {"error": e.code, "detail": e.message} +except PasswordlessStartError as e: + return {"error": e.code, "detail": e.message} +except Auth0Error as e: + return {"error": str(e)} +``` + +### Common error codes (`PasswordlessErrorCode`) + +- `bad.connection` - the passwordless connection is disabled or invalid +- `bad.email` - the email address is invalid or rejected by Auth0 +- `sms_provider_error` - Auth0 could not send the SMS +- `too_many_requests` - rate limiting or attack protection blocked the request +- `invalid_grant` - the OTP is invalid, expired, or already used +- `invalid_audience` - returned ID token audience does not match the SDK client +- `discovery_error` - the SDK could not load authorization server metadata +- `passwordless_start_failed` - SDK-side start failure +- `passwordless_verify_failed` - SDK-side verify failure diff --git a/src/auth0_server_python/auth_server/passwordless_client.py b/src/auth0_server_python/auth_server/passwordless_client.py new file mode 100644 index 0000000..119c1da --- /dev/null +++ b/src/auth0_server_python/auth_server/passwordless_client.py @@ -0,0 +1,399 @@ +""" +Passwordless Client for auth0-server-python SDK. + +Implements embedded passwordless login (Legacy Passwordless connections) for a +confidential (Regular Web App) client: + +* Email OTP / SMS OTP — ``start()`` sends a code, ``verify()`` exchanges it for + tokens via the passwordless-OTP grant and establishes a server-side session. +* Magic link — ``start(send="link")`` emails a one-click link; completion is + handled by the standard callback (``ServerClient.complete_interactive_login``), + not by ``verify()``. + +Tokens never leave the server; the browser holds only the opaque session +reference (RWA / BFF posture). +""" + +from typing import TYPE_CHECKING, Any, Optional + +import jwt + +from auth0_server_python.auth_types import ( + PASSWORDLESS_ALLOWED_AUTH_PARAMS, + PASSWORDLESS_RESERVED_AUTH_PARAMS, + PasswordlessStartResult, + StartPasswordlessEmailOptions, + StartPasswordlessOptions, + StartPasswordlessSmsOptions, + TransactionData, + UserClaims, + VerifyPasswordlessOtpOptions, +) +from auth0_server_python.error import ( + InvalidArgumentError, + IssuerValidationError, + MissingRequiredArgumentError, + PasswordlessErrorCode, + PasswordlessStartError, + PasswordlessVerifyError, +) +from auth0_server_python.utils import PKCE +from auth0_server_python.utils.helpers import validate_org_claims + +if TYPE_CHECKING: # avoid a circular import at runtime + from auth0_server_python.auth_server.server_client import ServerClient + +PASSWORDLESS_OTP_GRANT_TYPE = "http://auth0.com/oauth/grant-type/passwordless/otp" +# Email flows request the `email` scope; SMS has no email claim to satisfy. +DEFAULT_PASSWORDLESS_EMAIL_SCOPE = "openid profile email" +DEFAULT_PASSWORDLESS_SMS_SCOPE = "openid profile" +# Header Auth0 reads for the real end-user IP (confidential clients with +# "Trust Token Endpoint IP Header" enabled). +FORWARDED_FOR_HEADER = "auth0-forwarded-for" + + +class PasswordlessClient: + """ + Client for Auth0 embedded passwordless operations. + + Composes the parent :class:`ServerClient` to reuse domain resolution, OIDC + discovery, JWKS/ID-token verification, and session persistence rather than + duplicating that security-critical logic. + """ + + def __init__(self, server_client: "ServerClient"): + self._client = server_client + + # ------------------------------------------------------------------ start + + async def start( + self, + options: StartPasswordlessOptions, + store_options: Optional[dict[str, Any]] = None, + ) -> PasswordlessStartResult: + """ + Start a passwordless flow by sending an OTP code or a magic link. + + Args: + options: ``StartPasswordlessEmailOptions`` or + ``StartPasswordlessSmsOptions``. + store_options: Options passed to the transaction store (e.g. + request/response) — required for the magic-link flow so the + transaction cookie can be written. + + Returns: + PasswordlessStartResult with Auth0's start response payload. + + Raises: + PasswordlessStartError: When ``POST /passwordless/start`` fails. + InvalidArgumentError: When caller ``auth_params`` attempts to + override an SDK-owned parameter. + MissingRequiredArgumentError: When a magic link is requested but no + ``redirect_uri`` is configured on the client. + """ + client = self._client + origin_domain = await client._resolve_current_domain(store_options) + + body: dict[str, Any] = { + "client_id": client._client_id, + "client_secret": client._client_secret, + "connection": options.connection, + } + + if isinstance(options, StartPasswordlessEmailOptions): + body["email"] = options.email + body["send"] = options.send + elif isinstance(options, StartPasswordlessSmsOptions): + body["phone_number"] = options.phone_number + else: + raise InvalidArgumentError( + "options", + "options must be StartPasswordlessEmailOptions or StartPasswordlessSmsOptions", + ) + + if options.captcha: + body["captcha"] = options.captcha + + is_magic_link = ( + isinstance(options, StartPasswordlessEmailOptions) and options.send == "link" + ) + + if is_magic_link: + body["authParams"] = await self._build_magic_link_auth_params( + options, origin_domain, store_options + ) + elif options.auth_params: + # OTP flows: forward safe passthrough params only. + body["authParams"] = self._sanitize_caller_auth_params(options.auth_params) + + headers = {"Content-Type": "application/json"} + if options.language: + headers["x-request-language"] = options.language + if options.client_ip: + headers[FORWARDED_FOR_HEADER] = options.client_ip + + base_url = client._normalize_url(origin_domain) + url = f"{base_url}/passwordless/start" + + try: + async with client._get_http_client() as http: + response = await http.post(url, json=body, headers=headers) + except Exception as e: + raise PasswordlessStartError( + PasswordlessErrorCode.START_FAILED, + f"Unexpected error during passwordless start: {str(e)}", + e, + ) + + if response.status_code not in (200, 201): + error_body = self._safe_json(response) + raise PasswordlessStartError( + error_body.get("error", PasswordlessErrorCode.START_FAILED), + error_body.get("error_description", "Failed to start passwordless flow"), + error_body, + ) + + return PasswordlessStartResult(**self._safe_json(response)) + + # ----------------------------------------------------------------- verify + + async def verify( + self, + options: VerifyPasswordlessOtpOptions, + store_options: Optional[dict[str, Any]] = None, + ) -> dict[str, Any]: + """ + Verify a passwordless OTP and establish a server-side session. + + Only for the OTP flows (email/SMS code). Magic link completes via the + standard callback handler, not here. + + Args: + options: VerifyPasswordlessOtpOptions. + store_options: Options passed to the state store (e.g. + request/response) so the session can be written. + + Returns: + Dict containing ``state_data`` for the established session. + + Raises: + PasswordlessVerifyError: When token exchange or ID-token + verification fails. + """ + client = self._client + origin_domain = await client._resolve_current_domain(store_options) + + try: + metadata = await client._get_oidc_metadata_cached(origin_domain) + except Exception as e: + raise PasswordlessVerifyError( + PasswordlessErrorCode.DISCOVERY_ERROR, + "Failed to fetch authorization server metadata", + e, + ) + + token_endpoint = metadata["token_endpoint"] + origin_issuer = metadata.get("issuer") + + default_scope = ( + DEFAULT_PASSWORDLESS_EMAIL_SCOPE + if options.connection == "email" + else DEFAULT_PASSWORDLESS_SMS_SCOPE + ) + body: dict[str, Any] = { + "grant_type": PASSWORDLESS_OTP_GRANT_TYPE, + "client_id": client._client_id, + "client_secret": client._client_secret, + "realm": options.connection, + "username": options.username, + "otp": options.verification_code, + "scope": options.scope or default_scope, + } + if options.audience: + body["audience"] = options.audience + + headers = {"Content-Type": "application/x-www-form-urlencoded"} + if options.client_ip: + headers[FORWARDED_FOR_HEADER] = options.client_ip + + try: + async with client._get_http_client() as http: + response = await http.post( + token_endpoint, + data=body, + headers=headers, + ) + except Exception as e: + raise PasswordlessVerifyError( + PasswordlessErrorCode.VERIFY_FAILED, + f"Unexpected error during passwordless verify: {str(e)}", + e, + ) + + if response.status_code != 200: + error_body = self._safe_json(response) + raise PasswordlessVerifyError( + error_body.get("error", PasswordlessErrorCode.INVALID_GRANT), + error_body.get("error_description", "Passwordless verification failed"), + error_body, + ) + + token_response = response.json() + + user_claims, id_token_claims = await self._verify_id_token( + token_response, origin_domain, origin_issuer, metadata, options.organization + ) + + state_data = await client._persist_session_from_token_response( + token_response=token_response, + user_claims=user_claims, + origin_domain=origin_domain, + audience=options.audience, + session_expires_at=user_claims.session_expiry, + issued_at=id_token_claims.get("iat"), + id_token_claims=id_token_claims, + store_options=store_options, + ) + + return {"state_data": state_data.model_dump()} + + # ------------------------------------------------------------- internals + + async def _build_magic_link_auth_params( + self, + options: StartPasswordlessEmailOptions, + origin_domain: str, + store_options: Optional[dict[str, Any]], + ) -> dict[str, Any]: + """ + Build the magic-link ``authParams`` and persist the transaction. + + The SDK owns ``redirect_uri`` / ``response_type`` / ``state``; caller + ``auth_params`` may only contribute non-reserved passthrough keys. + """ + client = self._client + + redirect_uri = client._redirect_uri + if not redirect_uri: + raise MissingRequiredArgumentError("redirect_uri") + + auth_params = self._sanitize_caller_auth_params(options.auth_params) + + # Required to persist the transaction cookie; checked after input + # validation so bad auth_params / missing redirect_uri surface first. + if store_options is None: + raise MissingRequiredArgumentError("store_options") + + state = PKCE.generate_random_string(32) + auth_params["redirect_uri"] = redirect_uri + auth_params["response_type"] = "code" + auth_params["state"] = state + # Magic link is email-only, so the email scope is always appropriate. + auth_params.setdefault("scope", DEFAULT_PASSWORDLESS_EMAIL_SCOPE) + if options.organization: + auth_params["organization"] = options.organization + + # Magic link uses a plain authorization-code exchange (no PKCE), so the + # transaction stores no code_verifier. Single-use is enforced by + # transaction deletion on the callback; remove_if_expires signals the + # store to drop the transaction once expired. Its effective lifetime is + # the store's configured duration, not a fixed value set here. + transaction_data = TransactionData( + code_verifier=None, + audience=auth_params.get("audience"), + redirect_uri=redirect_uri, + domain=origin_domain, + organization=options.organization, + ) + await client._transaction_store.set( + f"{client._transaction_identifier}:{state}", + transaction_data, + remove_if_expires=True, + options=store_options, + ) + + return auth_params + + def _sanitize_caller_auth_params(self, auth_params: Optional[dict[str, Any]]) -> dict[str, Any]: + """ + Copy caller-supplied auth params, forwarding only allowlisted keys. + + Enforced as an allowlist (Global §3): a key outside + ``PASSWORDLESS_ALLOWED_AUTH_PARAMS`` is rejected. SDK-owned keys get a + precise "set by the SDK" message; anything else is reported as + unsupported so a new authorize param cannot pass through silently. + + Raises: + InvalidArgumentError: When a reserved or unrecognized param is present. + """ + if not auth_params: + return {} + for key in auth_params: + if key in PASSWORDLESS_RESERVED_AUTH_PARAMS: + raise InvalidArgumentError( + "auth_params", + f"'{key}' is set by the SDK and cannot be overridden", + ) + if key not in PASSWORDLESS_ALLOWED_AUTH_PARAMS: + raise InvalidArgumentError( + "auth_params", + f"'{key}' is not an allowed passthrough auth parameter", + ) + return dict(auth_params) + + async def _verify_id_token( + self, + token_response: dict[str, Any], + origin_domain: str, + origin_issuer: Optional[str], + metadata: dict[str, Any], + expected_org: Optional[str], + ) -> tuple[UserClaims, dict[str, Any]]: + """Verify the ID token from the OTP exchange and return its claims.""" + client = self._client + id_token = token_response.get("id_token") + if not id_token: + raise PasswordlessVerifyError( + PasswordlessErrorCode.VERIFY_FAILED, + "Token response did not include an ID token; ensure 'openid' scope is requested", + ) + + jwks = await client._get_jwks_cached(origin_domain, metadata) + + try: + claims = await client._verify_and_decode_jwt(id_token, jwks, audience=client._client_id) + except ValueError as e: + raise PasswordlessVerifyError(PasswordlessErrorCode.VERIFY_FAILED, str(e), e) + except jwt.InvalidAudienceError as e: + raise PasswordlessVerifyError( + PasswordlessErrorCode.INVALID_AUDIENCE, + "ID token audience mismatch. Ensure your client_id is configured correctly.", + e, + ) + except jwt.InvalidTokenError as e: + # Covers expired signature, bad signature, and other token defects. + raise PasswordlessVerifyError( + PasswordlessErrorCode.VERIFY_FAILED, + f"ID token verification failed: {str(e)}", + e, + ) + + token_issuer = claims.get("iss", "") + if client._normalize_url(token_issuer) != client._normalize_url(origin_issuer): + raise IssuerValidationError( + "ID token issuer mismatch. Ensure your Auth0 domain is configured correctly." + ) + + if expected_org: + validate_org_claims(claims, expected_org) + + return UserClaims.model_validate(claims), claims + + @staticmethod + def _safe_json(response) -> dict[str, Any]: + """Parse a response body as JSON, returning {} on failure.""" + try: + data = response.json() + return data if isinstance(data, dict) else {} + except Exception: + return {} diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index 4398804..6580858 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -22,6 +22,7 @@ from auth0_server_python.auth_schemes.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_server.passwordless_client import PasswordlessClient from auth0_server_python.auth_types import ( CompleteConnectAccountRequest, CompleteConnectAccountResponse, @@ -214,6 +215,9 @@ def __init__( headers=self._telemetry_headers, ) + # Initialize Passwordless client (composes this client) + self._passwordless_client = PasswordlessClient(self) + def _get_http_client(self, **kwargs) -> httpx.AsyncClient: """Return an httpx.AsyncClient with telemetry headers injected.""" headers = {**kwargs.pop("headers", {}), **self._telemetry_headers} @@ -610,6 +614,72 @@ async def start_interactive_login( return auth_url + async def _persist_session_from_token_response( + self, + token_response: dict[str, Any], + user_claims: "UserClaims", + origin_domain: str, + audience: Optional[str], + session_expires_at: Optional[int], + issued_at: Optional[int], + id_token_claims: Optional[dict[str, Any]] = None, + user_info: Optional[dict[str, Any]] = None, + store_options: Optional[dict[str, Any]] = None, + ) -> StateData: + """ + Build and persist a session ``StateData`` from a token endpoint response. + + Shared by the interactive-login callback and the passwordless OTP verify + flow so both derive the session id, enforce the IPSIE ceiling, and write + the state store identically. + + The session ``sid`` is taken from the verified ID token claims (falling + back to userinfo, then a random value) so OIDC back-channel logout — which + matches sessions by ``sid`` — can target sessions created here. + + Raises: + SessionExpiredError: If the session ceiling is already in the past. + """ + # Refuse to persist a session whose ceiling is already in the past. + if State.is_session_ceiling_in_past(session_expires_at, issued_at): + raise SessionExpiredError() + + now = int(time.time()) + token_set = TokenSet( + audience=audience or self.DEFAULT_AUDIENCE_STATE_KEY, + access_token=token_response.get("access_token", ""), + scope=token_response.get("scope", ""), + expires_at=now + token_response.get("expires_in", 3600), + ) + + # Prefer the ID token's `sid` claim, then userinfo, then a random value. + # A random sid would make the session untargetable by back-channel logout. + sid = None + if id_token_claims and id_token_claims.get("sid"): + sid = id_token_claims["sid"] + elif user_info and user_info.get("sid"): + sid = user_info["sid"] + if not sid: + sid = PKCE.generate_random_string(32) + + state_data = StateData( + user=user_claims, + id_token=token_response.get("id_token"), + refresh_token=token_response.get("refresh_token"), + token_sets=[token_set], + domain=origin_domain, + internal={ + "sid": sid, + "created_at": now, + "session_expires_at": session_expires_at, + }, + ) + + await self._state_store.set( + self._state_identifier, state_data, options=store_options + ) + return state_data + async def complete_interactive_login( self, url: str, @@ -684,6 +754,9 @@ async def complete_interactive_login( # ID token `iat`, used to detect a ceiling that is already past at login. issued_at = None id_token = token_response.get("id_token") + # Verified ID token claims, retained so the session `sid` can be sourced + # from them (back-channel logout matches on `sid`). + id_token_claims = None expected_org = transaction_data.organization @@ -726,6 +799,7 @@ async def complete_interactive_login( validate_org_claims(claims, expected_org) user_claims = UserClaims.parse_obj(claims) + id_token_claims = claims session_expires_at = user_claims.session_expiry issued_at = claims.get("iat") except ValueError as e: @@ -756,41 +830,24 @@ async def complete_interactive_login( ) - # Refuse to persist a session whose ceiling is already in the past. - if State.is_session_ceiling_in_past(session_expires_at, issued_at): + # Build + persist the session via the shared helper (enforces the IPSIE + # ceiling and sources `sid` from the ID token claims). On a past ceiling, + # clean up the transaction before surfacing the error. + try: + state_data = await self._persist_session_from_token_response( + token_response=token_response, + user_claims=user_claims, + origin_domain=origin_domain, + audience=transaction_data.audience, + session_expires_at=session_expires_at, + issued_at=issued_at, + id_token_claims=id_token_claims, + user_info=user_info if isinstance(user_info, dict) else None, + store_options=store_options, + ) + except SessionExpiredError: await self._transaction_store.delete(transaction_identifier, options=store_options) - raise SessionExpiredError() - - # Build a token set using the token response data - token_set = TokenSet( - audience=transaction_data.audience or self.DEFAULT_AUDIENCE_STATE_KEY, - access_token=token_response.get("access_token", ""), - scope=token_response.get("scope", ""), - expires_at=int(time.time()) + - token_response.get("expires_in", 3600) - ) - - # Generate a session id (sid) from token_response or transaction data, or create a new one - sid = user_info.get( - "sid") if user_info and "sid" in user_info else PKCE.generate_random_string(32) - - # Construct state data to represent the session - state_data = StateData( - user=user_claims, - id_token=token_response.get("id_token"), - # might be None if not provided - refresh_token=token_response.get("refresh_token"), - token_sets=[token_set], - domain=origin_domain, - internal={ - "sid": sid, - "created_at": int(time.time()), - "session_expires_at": session_expires_at - } - ) - - # Store the state data in the state store using store_options (Response required) - await self._state_store.set(self._state_identifier, state_data, options=store_options) + raise # Clean up transaction data after successful login await self._transaction_store.delete(transaction_identifier, options=store_options) @@ -2858,6 +2915,15 @@ def mfa(self) -> MfaClient: """Access the MFA client for multi-factor authentication operations.""" return self._mfa_client + # ============================================================================ + # Passwordless (embedded login) + # ============================================================================ + + @property + def passwordless(self) -> PasswordlessClient: + """Access the passwordless client for embedded passwordless operations.""" + return self._passwordless_client + # ============================================================================ # PASSKEY AUTHENTICATION # ============================================================================ diff --git a/src/auth0_server_python/auth_types/__init__.py b/src/auth0_server_python/auth_types/__init__.py index e886938..32941c2 100644 --- a/src/auth0_server_python/auth_types/__init__.py +++ b/src/auth0_server_python/auth_types/__init__.py @@ -3,10 +3,11 @@ These Pydantic models provide type safety and validation for all SDK data structures. """ +import re import warnings from typing import Any, Literal, Optional, Union -from pydantic import BaseModel, ConfigDict, Field, field_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator # Upper bound (Unix seconds) for a plausible session_expiry SESSION_EXPIRY_MAX_PLAUSIBLE = 10_000_000_000 @@ -61,7 +62,7 @@ class UserClaims(BaseModel): class Config: extra = "allow" # Allow additional fields not defined in the model - @field_validator('session_expiry', mode='before') + @field_validator("session_expiry", mode="before") @classmethod def _sanitize_session_expiry(cls, value: Any) -> Optional[int]: if isinstance(value, bool) or not isinstance(value, int): @@ -140,7 +141,9 @@ class TransactionData(BaseModel): """ audience: Optional[str] = None - code_verifier: str + # Optional: interactive login sets this for PKCE; the passwordless magic-link + # transaction has no verifier (plain auth-code exchange), so it stays None. + code_verifier: Optional[str] = None app_state: Optional[Any] = None auth_session: Optional[str] = None redirect_uri: Optional[str] = None @@ -684,6 +687,172 @@ class MfaTokenContext(BaseModel): created_at: int +# ============================================================================= +# Passwordless Types +# ============================================================================= + +# Passwordless connection strategies (Legacy Passwordless connections). +PasswordlessConnection = Literal["email", "sms"] + +# authParams keys the SDK owns and MUST NOT let a caller override for the +# magic-link flow. A caller-controlled redirect_uri/state would allow the +# emailed code+state to be redirected to an attacker (authorization-code +# interception); the PKCE/nonce/response_type keys are protocol-controlled. +# Mirrors nextjs-auth0's MAGIC_LINK_EXCLUDED_PARAMS / INTERNAL_AUTHORIZE_PARAMS. +# Kept explicit so a rejected override gets a precise "set by the SDK" message. +PASSWORDLESS_RESERVED_AUTH_PARAMS = frozenset( + { + "client_id", + "client_secret", + "redirect_uri", + "response_type", + "state", + "nonce", + "code_challenge", + "code_challenge_method", + } +) + +# Caller-supplied authParams keys the SDK will forward. This is an allowlist +# (Global §3: allowlists, not denylists) — any key outside it is rejected, so a +# future security-relevant authorize parameter cannot pass through silently on +# an SDK upgrade. Extend deliberately as new safe passthrough params are needed. +PASSWORDLESS_ALLOWED_AUTH_PARAMS = frozenset( + { + "audience", + "login_hint", + "ui_locales", + "screen_hint", + "prompt", + "max_age", + "acr_values", + } +) + +# Minimal BCP 47 language tag: primary subtag plus optional subtags. +_BCP47_LANGUAGE_RE = r"^[A-Za-z]{2,3}(-[A-Za-z0-9]{1,8})*$" +# E.164 phone number: '+' followed by up to 15 digits, first digit non-zero. +_E164_RE = re.compile(r"^\+[1-9]\d{1,14}$") + + +class _StartPasswordlessBase(BaseModel): + """Shared options for starting a passwordless flow.""" + + # BCP 47 tag (e.g. "fr", "en-US"). Forwarded as x-request-language to + # localise the email/SMS template. + language: Optional[str] = None + # Extra params forwarded to /passwordless/start. SDK-owned keys + # (PASSWORDLESS_RESERVED_AUTH_PARAMS) are stripped in the client. + auth_params: Optional[dict[str, Any]] = None + # Organization id or name; validated against ID token claims on verify. + organization: Optional[str] = None + # Attempted solution to a captcha challenge, when the tenant requires one. + captcha: Optional[str] = None + # End-user client IP, relayed to Auth0 as `auth0-forwarded-for` so brute-force + # and suspicious-IP protection key on the real user, not the app server. Only + # honored for confidential clients with "Trust Token Endpoint IP Header" on. + client_ip: Optional[str] = None + + @field_validator("language") + @classmethod + def _validate_language(cls, value: Optional[str]) -> Optional[str]: + if value is None: + return None + if not re.match(_BCP47_LANGUAGE_RE, value): + raise ValueError("language must be a valid BCP 47 tag (e.g. 'fr', 'en-US')") + return value + + +class StartPasswordlessEmailOptions(_StartPasswordlessBase): + """Options for starting an email passwordless flow (OTP code or magic link).""" + + connection: Literal["email"] = "email" + email: str + # "code" -> email OTP; "link" -> magic link. + send: Literal["code", "link"] = "code" + + +class StartPasswordlessSmsOptions(_StartPasswordlessBase): + """Options for starting an SMS passwordless (OTP) flow.""" + + connection: Literal["sms"] = "sms" + # E.164 format, e.g. "+14155550100". + phone_number: str + + @field_validator("phone_number") + @classmethod + def _validate_phone_number(cls, value: str) -> str: + if not _E164_RE.match(value): + raise ValueError("phone_number must be in E.164 format (e.g. '+14155550100')") + return value + + +StartPasswordlessOptions = Union[StartPasswordlessEmailOptions, StartPasswordlessSmsOptions] + + +class VerifyPasswordlessOtpOptions(BaseModel): + """ + Options for verifying a passwordless OTP and establishing a session. + + Exactly one of ``email`` / ``phone_number`` must be provided and must + match ``connection`` (email -> email, sms -> phone_number). + """ + + connection: PasswordlessConnection + # Public field name mirrors nextjs-auth0's `verificationCode`; sent to + # Auth0 as the `otp` form parameter. + verification_code: str + email: Optional[str] = None + phone_number: Optional[str] = None + scope: Optional[str] = None + audience: Optional[str] = None + organization: Optional[str] = None + # End-user client IP, relayed to Auth0 as `auth0-forwarded-for` on the OTP + # token exchange so brute-force protection keys on the real user, not the + # app server. Honored only for confidential clients with "Trust Token + # Endpoint IP Header" enabled. + client_ip: Optional[str] = None + + @field_validator("phone_number") + @classmethod + def _validate_phone_number(cls, value: Optional[str]) -> Optional[str]: + if value is None: + return None + if not _E164_RE.match(value): + raise ValueError("phone_number must be in E.164 format (e.g. '+14155550100')") + return value + + @model_validator(mode="after") + def _validate_identifier(self) -> "VerifyPasswordlessOtpOptions": + if self.connection == "email": + if not self.email: + raise ValueError("email is required when connection='email'") + if self.phone_number: + raise ValueError("phone_number must not be set when connection='email'") + else: # sms + if not self.phone_number: + raise ValueError("phone_number is required when connection='sms'") + if self.email: + raise ValueError("email must not be set when connection='sms'") + return self + + @property + def username(self) -> str: + """The Auth0 `username` value for the OTP grant (email or phone).""" + return self.email if self.connection == "email" else self.phone_number + + +class PasswordlessStartResult(BaseModel): + """Success payload from POST /passwordless/start.""" + + # Auth0 returns the request id as `_id`; alias so `.id` is populated. + id: Optional[str] = Field(default=None, alias="_id") + + class Config: + extra = "allow" # Allow additional fields returned by Auth0 + populate_by_name = True # accept both `_id` (alias) and `id` + + # ============================================================================= # Passkey & MyAccount Authentication Methods Types # ============================================================================= diff --git a/src/auth0_server_python/error/__init__.py b/src/auth0_server_python/error/__init__.py index ee9279f..f482a49 100644 --- a/src/auth0_server_python/error/__init__.py +++ b/src/auth0_server_python/error/__init__.py @@ -2,6 +2,7 @@ Error classes for the auth0-server-python SDK. These exceptions provide specific error types for different failure scenarios. """ + from typing import Any, Optional @@ -19,6 +20,7 @@ class MissingTransactionError(Auth0Error): This typically happens during the callback phase when the transaction from the initial authorization request cannot be found. """ + code = "missing_transaction_error" def __init__(self, message=None): @@ -56,6 +58,7 @@ def __init__(self, code: str, message: str, interval: Optional[int], cause=None) super().__init__(code, message, cause) self.interval = interval + class MyAccountApiError(Auth0Error): """ Error raised when an API request to My Account API fails. @@ -63,13 +66,13 @@ class MyAccountApiError(Auth0Error): """ def __init__( - self, - title: Optional[str], - type: Optional[str], - detail: Optional[str], - status: Optional[int], - validation_errors: Optional[list[dict[str, str]]] = None - ): + self, + title: Optional[str], + type: Optional[str], + detail: Optional[str], + status: Optional[int], + validation_errors: Optional[list[dict[str, str]]] = None, + ): super().__init__(detail) self.title = title self.type = type @@ -77,6 +80,7 @@ def __init__( self.status = status self.validation_errors = validation_errors + class AccessTokenError(Auth0Error): """Error raised when there's an issue with access tokens.""" @@ -92,6 +96,7 @@ class MissingRequiredArgumentError(Auth0Error): Error raised when a required argument is missing. Includes the name of the missing argument in the error message. """ + code = "missing_required_argument_error" def __init__(self, argument: str): @@ -106,16 +111,19 @@ class ConfigurationError(Auth0Error): Error raised when SDK configuration is invalid. This includes invalid combinations of parameters or incorrect configuration values. """ + code = "configuration_error" def __init__(self, message: str): super().__init__(message) self.name = "ConfigurationError" + class InvalidArgumentError(Auth0Error): """ Error raised when a given argument is an invalid value. """ + code = "invalid_argument" def __init__(self, argument: str, message: str): @@ -130,6 +138,7 @@ class IssuerValidationError(Auth0Error): This can happen when the issuer claim in a token does not match the expected issuer for the configured domain. """ + code = "issuer_validation_error" def __init__(self, message: str): @@ -142,6 +151,7 @@ class BackchannelLogoutError(Auth0Error): Error raised during backchannel logout processing. This can happen when validating or processing logout tokens. """ + code = "backchannel_logout_error" def __init__(self, message: str): @@ -156,6 +166,7 @@ class DomainResolverError(Auth0Error): This error indicates an issue with the custom domain resolver function provided for MCD (Multiple Custom Domains) support. """ + code = "domain_resolver_error" def __init__(self, message: str, original_error: Exception = None): @@ -172,12 +183,14 @@ def __init__(self, code: str, message: str): self.code = code self.name = "AccessTokenForConnectionError" + class StartLinkUserError(Auth0Error): """ Error raised when user linking process fails to start. This typically happens when trying to link accounts without having an authenticated user first. """ + code = "start_link_user_error" def __init__(self, message: str): @@ -187,8 +200,10 @@ def __init__(self, message: str): # Error code enumerations - these can be used to identify specific error scenarios + class AccessTokenErrorCode: """Error codes for access token operations.""" + MISSING_SESSION = "missing_session" MISSING_REFRESH_TOKEN = "missing_refresh_token" FAILED_TO_REFRESH_TOKEN = "failed_to_refresh_token" @@ -206,6 +221,7 @@ class OrganizationTokenValidationError(Auth0Error): Raised when org_id or org_name claim in the ID token fails validation against the organization value that was requested at login. """ + code = "organization_token_validation_error" def __init__(self, message: str): @@ -215,6 +231,7 @@ def __init__(self, message: str): class AccessTokenForConnectionErrorCode: """Error codes for connection-specific token operations.""" + MISSING_REFRESH_TOKEN = "missing_refresh_token" FAILED_TO_RETRIEVE = "failed_to_retrieve" API_ERROR = "api_error" @@ -228,6 +245,7 @@ class SessionExpiredError(Auth0Error): Error raised when a session is rejected at login because its session_expiry ceiling is already in the past. """ + code = AccessTokenErrorCode.SESSION_EXPIRED def __init__(self, message: Optional[str] = None, cause=None): @@ -240,6 +258,7 @@ class CustomTokenExchangeError(Auth0Error): """ Error raised during custom token exchange operations. """ + def __init__(self, code: str, message: str, cause=None): super().__init__(message) self.code = code @@ -249,6 +268,7 @@ def __init__(self, code: str, message: str, cause=None): class CustomTokenExchangeErrorCode: """Error codes for custom token exchange operations.""" + INVALID_TOKEN_FORMAT = "invalid_token_format" MISSING_ACTOR_TOKEN_TYPE = "missing_actor_token_type" MISSING_ACTOR_TOKEN = "missing_actor_token" @@ -263,15 +283,11 @@ class CustomTokenExchangeErrorCode: # MFA Error Classes # ============================================================================= + class MfaApiError(Auth0Error): """Base class for MFA API errors.""" - def __init__( - self, - code: str, - message: str, - cause: Optional[dict[str, Any]] = None - ): + def __init__(self, code: str, message: str, cause: Optional[dict[str, Any]] = None): super().__init__(message) self.code = code self.cause = cause @@ -315,11 +331,7 @@ class MfaRequiredError(AccessTokenError): """ def __init__( - self, - message: str, - mfa_token: str, - mfa_requirements=None, - cause: Optional[Exception] = None + self, message: str, mfa_token: str, mfa_requirements=None, cause: Optional[Exception] = None ): super().__init__("mfa_required", message, cause) self.mfa_token = mfa_token @@ -342,6 +354,64 @@ def __init__(self): self.code = "mfa_token_invalid" +# ============================================================================= +# Passwordless Error Classes +# ============================================================================= + + +class PasswordlessError(ApiError): + """ + Base class for passwordless (embedded login) errors. + + Carries the Auth0 ``error`` / ``error_description`` from the API response + body so integrators can branch on a typed exception rather than parsing + strings. + """ + + def __init__(self, code: str, message: str, cause=None): + super().__init__(code, message, cause) + self.name = "PasswordlessError" + # When cause is the raw error dict from Auth0, surface its fields even + # though ApiError only reads attributes off exception-like causes. + if isinstance(cause, dict): + self.error = cause.get("error") + self.error_description = cause.get("error_description") + + +class PasswordlessStartError(PasswordlessError): + """Error raised when POST /passwordless/start fails.""" + + def __init__(self, code: str, message: str, cause=None): + super().__init__(code, message, cause) + self.name = "PasswordlessStartError" + + +class PasswordlessVerifyError(PasswordlessError): + """Error raised when the passwordless OTP token exchange fails.""" + + def __init__(self, code: str, message: str, cause=None): + super().__init__(code, message, cause) + self.name = "PasswordlessVerifyError" + + +class PasswordlessErrorCode: + """Error codes for passwordless operations.""" + + # Start errors + BAD_CONNECTION = "bad.connection" + BAD_EMAIL = "bad.email" + SMS_PROVIDER_ERROR = "sms_provider_error" + TOO_MANY_REQUESTS = "too_many_requests" + # Verify errors + INVALID_GRANT = "invalid_grant" + INVALID_ISSUER = "invalid_issuer" + INVALID_AUDIENCE = "invalid_audience" + DISCOVERY_ERROR = "discovery_error" + # SDK-side + START_FAILED = "passwordless_start_failed" + VERIFY_FAILED = "passwordless_verify_failed" + + # ============================================================================= # Passkey Error Classes # ============================================================================= diff --git a/src/auth0_server_python/tests/test_passwordless_client.py b/src/auth0_server_python/tests/test_passwordless_client.py new file mode 100644 index 0000000..1e8a49b --- /dev/null +++ b/src/auth0_server_python/tests/test_passwordless_client.py @@ -0,0 +1,512 @@ +""" +Tests for PasswordlessClient — embedded passwordless (OTP + magic link). +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest +from pydantic import ValidationError + +from auth0_server_python.auth_server.passwordless_client import ( + PASSWORDLESS_OTP_GRANT_TYPE, + PasswordlessClient, +) +from auth0_server_python.auth_server.server_client import ServerClient +from auth0_server_python.auth_types import ( + StartPasswordlessEmailOptions, + StartPasswordlessSmsOptions, + TransactionData, + VerifyPasswordlessOtpOptions, +) +from auth0_server_python.error import ( + InvalidArgumentError, + IssuerValidationError, + MissingRequiredArgumentError, + PasswordlessStartError, + PasswordlessVerifyError, + SessionExpiredError, +) + +DOMAIN = "tenant.auth0.com" +CLIENT_ID = "test_client" +CLIENT_SECRET = "test_secret" +SECRET = "test_secret_key_32_chars_long!!!" +REDIRECT_URI = "https://app.example.com/auth/callback" +ISSUER = "https://tenant.auth0.com/" +METADATA = {"token_endpoint": f"https://{DOMAIN}/oauth/token", "issuer": ISSUER} + + +def _make_client(**overrides) -> ServerClient: + kwargs = { + "domain": DOMAIN, + "client_id": CLIENT_ID, + "client_secret": CLIENT_SECRET, + "secret": SECRET, + "redirect_uri": REDIRECT_URI, + "transaction_store": AsyncMock(), + "state_store": AsyncMock(), + } + kwargs.update(overrides) + return ServerClient(**kwargs) + + +def _mock_http(client: ServerClient, status_code: int, json_body): + """Patch client._get_http_client so post() returns the given response.""" + response = MagicMock(status_code=status_code) + response.json = MagicMock(return_value=json_body) + http = AsyncMock() + http.post = AsyncMock(return_value=response) + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=http) + ctx.__aexit__ = AsyncMock(return_value=False) + client._get_http_client = MagicMock(return_value=ctx) + return http + + +# ── Wiring ─────────────────────────────────────────────────────────────────── + + +class TestWiring: + def test_passwordless_property(self): + client = _make_client() + assert isinstance(client.passwordless, PasswordlessClient) + assert client.passwordless._client is client + + +# ── start(): OTP ─────────────────────────────────────────────────────────── + + +class TestStartOtp: + @pytest.mark.asyncio + async def test_email_otp_start(self): + client = _make_client() + http = _mock_http(client, 200, {"_id": "req_123"}) + + result = await client.passwordless.start( + StartPasswordlessEmailOptions(email="user@example.com", send="code") + ) + + body = http.post.call_args.kwargs["json"] + assert body["connection"] == "email" + assert body["email"] == "user@example.com" + assert body["send"] == "code" + assert body["client_id"] == CLIENT_ID + assert body["client_secret"] == CLIENT_SECRET + # OTP flow does not create a transaction. + client._transaction_store.set.assert_not_awaited() + # Auth0 returns the request id as `_id`; the model aliases it to `.id`. + assert result.id == "req_123" + + @pytest.mark.asyncio + async def test_sms_otp_start(self): + client = _make_client() + http = _mock_http(client, 200, {}) + + await client.passwordless.start(StartPasswordlessSmsOptions(phone_number="+14155550100")) + + body = http.post.call_args.kwargs["json"] + assert body["connection"] == "sms" + assert body["phone_number"] == "+14155550100" + assert "send" not in body + + @pytest.mark.asyncio + async def test_language_sets_header(self): + client = _make_client() + http = _mock_http(client, 200, {}) + + await client.passwordless.start( + StartPasswordlessEmailOptions(email="user@example.com", send="code", language="fr") + ) + + headers = http.post.call_args.kwargs["headers"] + assert headers["x-request-language"] == "fr" + + @pytest.mark.asyncio + async def test_start_error_maps_to_typed_exception(self): + client = _make_client() + _mock_http( + client, + 400, + {"error": "bad.connection", "error_description": "Connection disabled"}, + ) + + with pytest.raises(PasswordlessStartError) as exc: + await client.passwordless.start( + StartPasswordlessEmailOptions(email="user@example.com", send="code") + ) + assert exc.value.code == "bad.connection" + assert exc.value.error == "bad.connection" + assert exc.value.error_description == "Connection disabled" + + @pytest.mark.asyncio + async def test_sms_e164_rejected_at_model(self): + # Validation happens at model construction, before any network call. + with pytest.raises(ValidationError): + StartPasswordlessSmsOptions(phone_number="4155550100") + + +# ── start(): Magic link ────────────────────────────────────────────────────── + + +class TestStartMagicLink: + @pytest.mark.asyncio + async def test_magic_link_sets_sdk_owned_params_and_persists_tx(self): + client = _make_client() + http = _mock_http(client, 200, {}) + + await client.passwordless.start( + StartPasswordlessEmailOptions(email="user@example.com", send="link"), + store_options={}, + ) + + body = http.post.call_args.kwargs["json"] + ap = body["authParams"] + assert ap["redirect_uri"] == REDIRECT_URI + assert ap["response_type"] == "code" + assert "state" in ap and len(ap["state"]) >= 16 + assert ap["scope"] # default scope applied + + # Transaction persisted, single-use + bounded TTL, no PKCE verifier. + client._transaction_store.set.assert_awaited_once() + set_call = client._transaction_store.set.await_args + tx_key = set_call.args[0] + tx_data = set_call.args[1] + assert tx_key == f"{client._transaction_identifier}:{ap['state']}" + assert set_call.kwargs["remove_if_expires"] is True + assert tx_data.code_verifier is None + assert tx_data.redirect_uri == REDIRECT_URI + + @pytest.mark.asyncio + async def test_magic_link_requires_store_options(self): + # No store_options -> transaction cookie can't be persisted -> fail loudly. + client = _make_client() + _mock_http(client, 200, {}) + + with pytest.raises(MissingRequiredArgumentError): + await client.passwordless.start( + StartPasswordlessEmailOptions(email="user@example.com", send="link"), + store_options=None, + ) + client._transaction_store.set.assert_not_awaited() + + @pytest.mark.asyncio + async def test_magic_link_requires_redirect_uri(self): + client = _make_client(redirect_uri=None) + _mock_http(client, 200, {}) + + with pytest.raises(MissingRequiredArgumentError): + await client.passwordless.start( + StartPasswordlessEmailOptions(email="user@example.com", send="link") + ) + + @pytest.mark.asyncio + async def test_caller_cannot_override_reserved_param(self): + client = _make_client() + _mock_http(client, 200, {}) + + with pytest.raises(InvalidArgumentError): + await client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="link", + auth_params={"redirect_uri": "https://evil.example.com/steal"}, + ) + ) + # No transaction written when the override is rejected. + client._transaction_store.set.assert_not_awaited() + + @pytest.mark.asyncio + async def test_caller_safe_param_passed_through(self): + client = _make_client() + http = _mock_http(client, 200, {}) + + await client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="link", + auth_params={"login_hint": "user@example.com"}, + ), + store_options={}, + ) + ap = http.post.call_args.kwargs["json"]["authParams"] + assert ap["login_hint"] == "user@example.com" + assert ap["redirect_uri"] == REDIRECT_URI # still SDK-owned + + @pytest.mark.asyncio + async def test_caller_unrecognized_param_rejected(self): + # Allowlist posture: a param outside the allowlist is rejected, not + # silently forwarded, so a future authorize param can't slip through. + client = _make_client() + _mock_http(client, 200, {}) + + with pytest.raises(InvalidArgumentError): + await client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="link", + auth_params={"response_mode": "fragment"}, + ) + ) + client._transaction_store.set.assert_not_awaited() + + @pytest.mark.asyncio + async def test_connection_scope_not_allowed(self): + # connection_scope is a federated-connection param with no meaning for + # email/SMS passwordless; it is not in the allowlist and is rejected. + client = _make_client() + _mock_http(client, 200, {}) + + with pytest.raises(InvalidArgumentError): + await client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", + send="link", + auth_params={"connection_scope": "read:foo"}, + ) + ) + client._transaction_store.set.assert_not_awaited() + + @pytest.mark.asyncio + async def test_client_ip_forwarded_on_start(self): + client = _make_client() + http = _mock_http(client, 200, {}) + + await client.passwordless.start( + StartPasswordlessEmailOptions( + email="user@example.com", send="code", client_ip="203.0.113.7" + ) + ) + headers = http.post.call_args.kwargs["headers"] + assert headers["auth0-forwarded-for"] == "203.0.113.7" + + +# ── Magic link callback completion (complete_interactive_login) ────────────── + + +class TestMagicLinkCallback: + @pytest.mark.asyncio + async def test_magic_link_callback_exchanges_code_without_pkce(self, mocker): + # Magic link is a plain auth-code exchange: lock that code_verifier=None + # reaches fetch_token (authlib drops the falsy field) so a forced verifier + # — which Auth0 would reject — is caught. + client = _make_client() + client._transaction_store.get.return_value = TransactionData( + code_verifier=None, + redirect_uri=REDIRECT_URI, + domain=DOMAIN, + ) + + mocker.patch.object(client, "_get_oidc_metadata_cached", return_value=METADATA) + mocker.patch.object(client._oauth, "metadata", METADATA) + mocker.patch.object( + client, + "_get_jwks_cached", + return_value={"keys": [{"kty": "RSA", "kid": "k1"}]}, + ) + mocker.patch.object( + client, + "_verify_and_decode_jwt", + return_value={"iss": ISSUER, "sub": "auth0|1", "sid": "SID-1", "iat": 1_000}, + ) + fetch_token = AsyncMock( + return_value={ + "access_token": "at", + "id_token": "idt", + "expires_in": 3600, + "scope": "openid", + } + ) + mocker.patch.object(client._oauth, "fetch_token", fetch_token) + + result = await client.complete_interactive_login( + f"{REDIRECT_URI}?code=AUTHCODE&state=STATE-1" + ) + + assert fetch_token.await_args.kwargs["code_verifier"] is None + assert fetch_token.await_args.kwargs["code"] == "AUTHCODE" + + # Session established; transaction consumed (single-use). + client._state_store.set.assert_awaited_once() + client._transaction_store.delete.assert_awaited_once() + assert result["state_data"]["internal"]["sid"] == "SID-1" + + +# ── verify() ───────────────────────────────────────────────────────────────── + + +class TestVerify: + def _patch_verify_deps(self, client, mocker, claims): + mocker.patch.object(client, "_get_oidc_metadata_cached", return_value=METADATA) + mocker.patch.object( + client, + "_get_jwks_cached", + return_value={"keys": [{"kty": "RSA", "kid": "k1"}]}, + ) + mocker.patch.object(client, "_verify_and_decode_jwt", return_value=claims) + + @pytest.mark.asyncio + async def test_email_otp_verify_creates_session(self, mocker): + client = _make_client() + claims = {"iss": ISSUER, "sub": "auth0|1", "sid": "SID-123", "iat": 1_000} + self._patch_verify_deps(client, mocker, claims) + http = _mock_http( + client, + 200, + {"access_token": "at", "id_token": "idt", "expires_in": 3600, "scope": "openid"}, + ) + + result = await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + + # Correct grant + params on the token call. + data = http.post.call_args.kwargs["data"] + assert data["grant_type"] == PASSWORDLESS_OTP_GRANT_TYPE + assert data["realm"] == "email" + assert data["username"] == "user@example.com" + assert data["otp"] == "123456" + + # Session persisted, sid taken from the ID token claim (not random). + client._state_store.set.assert_awaited_once() + assert "state_data" in result + assert result["state_data"]["internal"]["sid"] == "SID-123" + + @pytest.mark.asyncio + async def test_sms_otp_verify_username_is_phone(self, mocker): + client = _make_client() + claims = {"iss": ISSUER, "sub": "auth0|2", "sid": "SID-9", "iat": 1_000} + self._patch_verify_deps(client, mocker, claims) + http = _mock_http( + client, 200, {"access_token": "at", "id_token": "idt", "expires_in": 3600} + ) + + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="sms", phone_number="+14155550100", verification_code="000111" + ) + ) + data = http.post.call_args.kwargs["data"] + assert data["realm"] == "sms" + assert data["username"] == "+14155550100" + # SMS has no email claim to satisfy, so the default scope omits `email`. + assert data["scope"] == "openid profile" + + @pytest.mark.asyncio + async def test_email_verify_default_scope_includes_email(self, mocker): + client = _make_client() + claims = {"iss": ISSUER, "sub": "auth0|1", "sid": "s", "iat": 1_000} + self._patch_verify_deps(client, mocker, claims) + http = _mock_http( + client, 200, {"access_token": "at", "id_token": "idt", "expires_in": 3600} + ) + + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + assert http.post.call_args.kwargs["data"]["scope"] == "openid profile email" + + @pytest.mark.asyncio + async def test_client_ip_forwarded_on_verify(self, mocker): + client = _make_client() + claims = {"iss": ISSUER, "sub": "auth0|1", "sid": "s", "iat": 1_000} + self._patch_verify_deps(client, mocker, claims) + http = _mock_http( + client, 200, {"access_token": "at", "id_token": "idt", "expires_in": 3600} + ) + + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", + email="user@example.com", + verification_code="123456", + client_ip="203.0.113.7", + ) + ) + headers = http.post.call_args.kwargs["headers"] + assert headers["auth0-forwarded-for"] == "203.0.113.7" + + @pytest.mark.asyncio + async def test_verify_invalid_otp_maps_to_typed_error(self, mocker): + client = _make_client() + mocker.patch.object(client, "_get_oidc_metadata_cached", return_value=METADATA) + _mock_http( + client, + 403, + {"error": "invalid_grant", "error_description": "Wrong code"}, + ) + + with pytest.raises(PasswordlessVerifyError) as exc: + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="000000" + ) + ) + assert exc.value.code == "invalid_grant" + client._state_store.set.assert_not_awaited() + + @pytest.mark.asyncio + async def test_verify_issuer_mismatch_rejected(self, mocker): + client = _make_client() + claims = {"iss": "https://attacker.evil.com/", "sub": "x", "sid": "s", "iat": 1_000} + self._patch_verify_deps(client, mocker, claims) + _mock_http(client, 200, {"access_token": "at", "id_token": "idt", "expires_in": 3600}) + + with pytest.raises(IssuerValidationError): + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + client._state_store.set.assert_not_awaited() + + @pytest.mark.asyncio + async def test_verify_missing_id_token_rejected(self, mocker): + client = _make_client() + mocker.patch.object(client, "_get_oidc_metadata_cached", return_value=METADATA) + _mock_http(client, 200, {"access_token": "at", "expires_in": 3600}) + + with pytest.raises(PasswordlessVerifyError): + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + + @pytest.mark.asyncio + async def test_verify_discovery_failure_mapped(self, mocker): + client = _make_client() + mocker.patch.object(client, "_get_oidc_metadata_cached", side_effect=Exception("boom")) + + with pytest.raises(PasswordlessVerifyError) as exc: + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + assert exc.value.code == "discovery_error" + + @pytest.mark.asyncio + async def test_verify_ceiling_in_past_rejected(self, mocker): + client = _make_client() + # session_expiry well before iat -> ceiling already past. + claims = { + "iss": ISSUER, + "sub": "x", + "sid": "s", + "iat": 2_000_000_000, + "session_expiry": 1_000, + } + self._patch_verify_deps(client, mocker, claims) + _mock_http(client, 200, {"access_token": "at", "id_token": "idt", "expires_in": 3600}) + + with pytest.raises(SessionExpiredError): + await client.passwordless.verify( + VerifyPasswordlessOtpOptions( + connection="email", email="user@example.com", verification_code="123456" + ) + ) + client._state_store.set.assert_not_awaited() diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 16ce89e..c4bd309 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -4705,6 +4705,122 @@ async def test_complete_login_issuer_validation_success(mocker): assert "state_data" in result +@pytest.mark.asyncio +async def test_complete_login_sid_sourced_from_id_token_claim(mocker): + """ + ID-token-only login must persist the session sid from the ID token's `sid` + claim (not a random value) so OIDC back-channel logout, which matches on + sid, can target the session. + """ + mock_tx_store = AsyncMock() + mock_tx_store.get.return_value = TransactionData( + code_verifier="123", + domain="tenant.auth0.com", + ) + + mock_state_store = AsyncMock() + + client = ServerClient( + domain="tenant.auth0.com", + client_id="test_client", + client_secret="test_secret", + transaction_store=mock_tx_store, + state_store=mock_state_store, + secret="test_secret_key_32_chars_long!!", + ) + + # Mock OIDC metadata + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"issuer": "https://tenant.auth0.com/", "token_endpoint": "https://tenant.auth0.com/token"} + ) + + # Mock JWKS fetch + mocker.patch.object( + client, + "_get_jwks_cached", + return_value={"keys": [{"kty": "RSA", "kid": "test-key"}]} + ) + + # Mock OAuth fetch_token (ID-token-only response, no userinfo) + async_fetch_token = AsyncMock() + async_fetch_token.return_value = { + "access_token": "token123", + "id_token": "id_token_jwt" + } + mocker.patch.object(client._oauth, "fetch_token", async_fetch_token) + + # Verified claims carry a `sid` + mocker.patch.object( + client, + "_verify_and_decode_jwt", + return_value={"sub": "user123", "iss": "https://tenant.auth0.com/", "sid": "SID-from-claim"} + ) + + result = await client.complete_interactive_login("http://localhost/callback?code=abc&state=xyz") + + assert result["state_data"]["internal"]["sid"] == "SID-from-claim" + + +@pytest.mark.asyncio +async def test_complete_login_sid_falls_back_to_random_without_claim(mocker): + """ + When the ID token carries no `sid`, a random one is generated so the session + is still persisted (back-channel logout simply cannot target it). + """ + mock_tx_store = AsyncMock() + mock_tx_store.get.return_value = TransactionData( + code_verifier="123", + domain="tenant.auth0.com", + ) + + mock_state_store = AsyncMock() + + client = ServerClient( + domain="tenant.auth0.com", + client_id="test_client", + client_secret="test_secret", + transaction_store=mock_tx_store, + state_store=mock_state_store, + secret="test_secret_key_32_chars_long!!", + ) + + # Mock OIDC metadata + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"issuer": "https://tenant.auth0.com/", "token_endpoint": "https://tenant.auth0.com/token"} + ) + + # Mock JWKS fetch + mocker.patch.object( + client, + "_get_jwks_cached", + return_value={"keys": [{"kty": "RSA", "kid": "test-key"}]} + ) + + # Mock OAuth fetch_token (ID-token-only response, no userinfo) + async_fetch_token = AsyncMock() + async_fetch_token.return_value = { + "access_token": "token123", + "id_token": "id_token_jwt" + } + mocker.patch.object(client._oauth, "fetch_token", async_fetch_token) + + # Verified claims carry NO `sid` + mocker.patch.object( + client, + "_verify_and_decode_jwt", + return_value={"sub": "user123", "iss": "https://tenant.auth0.com/"} + ) + + result = await client.complete_interactive_login("http://localhost/callback?code=abc&state=xyz") + + sid = result["state_data"]["internal"]["sid"] + assert sid and sid != "user123" + + @pytest.mark.asyncio async def test_complete_login_issuer_mismatch_raises_error(mocker): """Test that issuer mismatch in ID token raises IssuerValidationError."""