Doc changes and minor code optimizations

Author: rmad17

README.md

@@ -185,29 +185,7 @@ For more details and examples, see [examples/RetrievingData.md](examples/Retriev
 
 ### 7. Passkey Authentication
 
-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. 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:
-
-```python
-from auth0_server_python.auth_types import PasskeyUserProfile, PasskeyAuthResponse
-
-# Step 1 — request a challenge
-challenge = await auth0.passkey_login_challenge(
-    store_options={"request": request, "response": response}
-)
-
-# Step 2 — browser signs: navigator.credentials.get(challenge.authn_params_public_key)
-
-# Step 3 — complete sign-in and establish the session
-result = await auth0.signin_with_passkey(
-    auth_session=challenge.auth_session,
-    authn_response=PasskeyAuthResponse(**credential),
-    store_options={"request": request, "response": response}
-)
-
-user = result.state_data["user"]
-```
-
-For signup, organizations, step-up MFA, and error handling, see [examples/Passkeys.md](examples/Passkeys.md).
+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
 

src/auth0_server_python/auth_schemes/dpop_auth.py

@@ -21,15 +21,17 @@ def _validate_dpop_key(key: "jwk.JWK") -> dict:
     return public_jwk
 
 
-def make_dpop_proof_for_token_endpoint(
-    key: "jwk.JWK", method: str, url: str, nonce: Optional[str] = None
+def _build_dpop_proof(
+    key: "jwk.JWK",
+    public_jwk: dict,
+    method: str,
+    url: str,
+    *,
+    ath: Optional[str] = None,
+    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)
+    """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 = {
@@ -38,13 +40,27 @@ def make_dpop_proof_for_token_endpoint(
         "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
@@ -59,12 +75,6 @@ def __init__(self, token: str, key: "jwk.JWK") -> None:
         self._key = key
         self._public_jwk = public_jwk
 
-    def __repr__(self) -> str:
-        return "DPoPAuth(token=[REDACTED], key=[REDACTED])"
-
-    def __str__(self) -> str:
-        return "DPoPAuth(token=[REDACTED], key=[REDACTED])"
-
     def auth_flow(self, request: httpx.Request):
         proof = self._make_proof(request.method, str(request.url))
         request.headers["Authorization"] = f"DPoP {self._token}"
@@ -80,20 +90,5 @@ def auth_flow(self, request: httpx.Request):
             yield request
 
     def _make_proof(self, method: str, url: str, nonce: Optional[str] = None) -> str:
-        htu = url.split("?")[0].split("#")[0]
         ath = _base64url(hashlib.sha256(self._token.encode("ascii")).digest())
-
-        header = {"typ": "dpop+jwt", "alg": "ES256", "jwk": self._public_jwk}
-        payload = {
-            "jti": str(uuid.uuid4()),
-            "htm": method.upper(),
-            "htu": htu,
-            "iat": int(time.time()),
-            "ath": ath,
-        }
-        if nonce is not None:
-            payload["nonce"] = nonce
-
-        token = jwcrypto_jwt.JWT(header=header, claims=payload)
-        token.make_signed_token(self._key)
-        return token.serialize()
+        return _build_dpop_proof(self._key, self._public_jwk, method, url, ath=ath, nonce=nonce)

src/auth0_server_python/auth_server/mfa_client.py

@@ -142,39 +142,6 @@ def decrypt_mfa_token(self, encrypted_token: str) -> MfaTokenContext:
     # MFA STATE
     # ============================================================================
 
-    async def store_pending_mfa(
-        self,
-        encrypted_token: str,
-        store_options: Optional[dict[str, Any]] = None,
-    ) -> None:
-        """Save an in-progress MFA token so challenge and verify can proceed without the client carrying the token."""
-        if self._state_store:
-            await self._state_store.set(
-                MFA_PENDING_IDENTIFIER,
-                {"mfa_token": encrypted_token},
-                options=store_options,
-            )
-
-    async def get_pending_mfa(
-        self,
-        store_options: Optional[dict[str, Any]] = None,
-    ) -> Optional[str]:
-        """Retrieve the in-progress MFA token if a challenge is pending for the current session, or None."""
-        if not self._state_store:
-            return None
-        data = await self._state_store.get(MFA_PENDING_IDENTIFIER, store_options)
-        if data and isinstance(data, dict):
-            return data.get("mfa_token")
-        return None
-
-    async def _clear_pending_mfa(
-        self,
-        store_options: Optional[dict[str, Any]] = None,
-    ) -> None:
-        """Clear the in-progress MFA state after successful verification."""
-        if self._state_store:
-            await self._state_store.delete(MFA_PENDING_IDENTIFIER, store_options)
-
     def _resolve_encrypted_token(
         self,
         options: dict[str, Any],
@@ -192,12 +159,18 @@ def _resolve_encrypted_token(
     @staticmethod
     def _parse_error_body(response: httpx.Response) -> dict[str, Any]:
         """
-        Parse an error response body as JSON, falling back to a status-coded
-        stub when the body is not JSON (e.g. a gateway 502/504 HTML page).
+        Parse an error response body as JSON.
+
+        Falls back to a status-coded stub when the body is not JSON (e.g. a
+        gateway 502/504 HTML page), so the caller always gets a readable dict
+        rather than a JSON-parser exception folded into the message.
 
-        Never raises — the caller always gets a dict it can read error fields
-        from, so a non-JSON error surfaces the real HTTP status rather than a
-        JSON-parser exception folded into the message.
+        Args:
+            response: The HTTP error response to parse.
+
+        Returns:
+            The parsed JSON object, or a stub dict whose 'error_description'
+            names the HTTP status when the body is not a JSON object.
         """
         try:
             data = response.json()
@@ -223,14 +196,25 @@ async def _raise_mfa_required(
         Encrypt the server-issued mfa_token and raise MfaRequiredError.
 
         Shared by every site that handles an `mfa_required` response so the
-        encrypt-then-raise behaviour cannot drift between entry points. Returns
-        only when the response carries no mfa_token (caller then falls through
-        to its own typed error).
-
-        store_pending controls whether the encrypted token is persisted to the
-        state store before raising. It is an explicit argument so the difference
-        between entry points is visible: the passkey grant persists it here,
-        while the refresh-token path relies on its get_access_token caller.
+        encrypt-then-raise behaviour cannot drift between entry points.
+
+        Args:
+            error_data: The parsed `mfa_required` error body from Auth0.
+            audience: Audience to bind into the encrypted token context.
+            scope: Scope to bind into the encrypted token context.
+            default_description: Message used when the response omits
+                'error_description'.
+            store_pending: When True, persist the encrypted token to the state
+                store before raising (the passkey grant does; the refresh-token
+                path relies on its get_access_token caller instead).
+            store_options: Optional options passed to the State Store.
+
+        Returns:
+            None. Returns without raising only when the response carries no
+            mfa_token, so the caller can fall through to its own typed error.
+
+        Raises:
+            MfaRequiredError: When the response carries an mfa_token.
         """
         raw_mfa_token = error_data.get("mfa_token")
         if not raw_mfa_token:
@@ -245,8 +229,14 @@ async def _raise_mfa_required(
             scope=scope,
             mfa_requirements=mfa_requirements,
         )
-        if store_pending:
-            await self.store_pending_mfa(encrypted_token, store_options)
+        if store_pending and self._state_store:
+            # Persist the in-progress MFA token so challenge and verify can
+            # proceed without the client carrying the token.
+            await self._state_store.set(
+                MFA_PENDING_IDENTIFIER,
+                {"mfa_token": encrypted_token},
+                options=store_options,
+            )
         raise MfaRequiredError(
             error_data.get("error_description", default_description),
             mfa_token=encrypted_token,
@@ -545,7 +535,9 @@ async def verify(
                 token_response = response.json()
                 verify_response = MfaVerifyResponse(**token_response)
 
-                await self._clear_pending_mfa(store_options)
+                # Clear the in-progress MFA state after successful verification.
+                if self._state_store:
+                    await self._state_store.delete(MFA_PENDING_IDENTIFIER, store_options)
 
                 if options.get("persist") and self._state_store:
                     await self._persist_mfa_tokens(
@@ -635,5 +627,5 @@ async def _persist_mfa_tokens(
             raise
         except Exception as e:
             raise MfaVerifyError(
-                f"Failed to persist MFA tokens to state store: {str(e)}"
-            )
+                "Failed to persist MFA tokens to state store"
+            ) from e

src/auth0_server_python/auth_server/server_client.py

@@ -2925,14 +2925,28 @@ async def signin_with_passkey(
                         PasskeyErrorCode.INVALID_RESPONSE, "Failed to parse passkey token response as JSON"
                     )
 
+                # Add required fields if they are missing
+                if "expires_in" in token_data and "expires_at" not in token_data:
+                    token_data["expires_at"] = int(time.time()) + token_data["expires_in"]
+
                 token_response = PasskeyTokenResponse.model_validate(token_data)
 
-                if dpop_key is not None and token_response.token_type.lower() != "dpop":
+                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(

src/auth0_server_python/auth_types/__init__.py

@@ -3,14 +3,21 @@
 These Pydantic models provide type safety and validation for all SDK data structures.
 """
 
-import time
 from typing import Any, Literal, Optional, Union
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 # Upper bound (Unix seconds) for a plausible session_expiry
 SESSION_EXPIRY_MAX_PLAUSIBLE = 10_000_000_000
 
+# Type aliases using Literal types. Used to validate caller-supplied input.
+# Server-controlled response fields use plain str instead, so a new factor or
+# 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"]
+
 
 class UserClaims(BaseModel):
     """
@@ -471,12 +478,6 @@ class ListConnectedAccountConnectionsResponse(BaseModel):
 # MFA Types
 # =============================================================================
 
-# Type aliases using Literal types. Used to validate caller-supplied input.
-# Server-controlled response fields use plain str instead, so a new factor or
-# challenge type (e.g. a future webauthn second factor) does not fail closed.
-OobChannel = Literal["sms", "voice", "auth0", "email"]
-ChallengeType = Literal["otp", "oob"]
-
 
 class AuthenticatorResponse(BaseModel):
     """Represents an MFA authenticator enrolled by a user."""
@@ -701,10 +702,6 @@ class PasskeyPublicKeyOptions(BaseModel):
     user_verification: Optional[str] = Field(None, alias="userVerification")
 
 
-EnrollmentType = Literal["passkey", "email", "phone", "totp", "push-notification", "recovery-code", "password"]
-PreferredAuthMethod = Literal["sms", "voice"]
-
-
 class EnrollAuthenticationMethodRequest(BaseModel):
     type: EnrollmentType
     email: Optional[str] = None
@@ -720,14 +717,6 @@ class EnrollmentChallengeResponse(BaseModel):
     auth_session: str
     authn_params_public_key: Optional[PasskeyPublicKeyOptions] = None
 
-    def __repr__(self) -> str:
-        return (
-            f"EnrollmentChallengeResponse("
-            f"authentication_method_id={self.authentication_method_id!r}, "
-            f"auth_session=[REDACTED], "
-            f"authn_params_public_key={self.authn_params_public_key!r})"
-        )
-
 
 class PasskeyAuthResponse(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
@@ -807,13 +796,6 @@ class _PasskeyChallengeResponseBase(BaseModel):
     auth_session: str
     authn_params_public_key: PasskeyPublicKeyOptions
 
-    def __repr__(self) -> str:
-        return (
-            f"{self.__class__.__name__}("
-            f"auth_session=[REDACTED], "
-            f"authn_params_public_key={self.authn_params_public_key!r})"
-        )
-
 
 class PasskeySignupChallengeResponse(_PasskeyChallengeResponseBase):
     pass
@@ -832,22 +814,3 @@ class PasskeyTokenResponse(BaseModel):
     scope: Optional[str] = None
     id_token: Optional[str] = None
     refresh_token: Optional[str] = None
-
-    @model_validator(mode="before")
-    @classmethod
-    def _backfill_expires_at(cls, data: Any) -> Any:
-        if isinstance(data, dict) and "expires_at" not in data and "expires_in" in data:
-            data["expires_at"] = int(time.time()) + int(data["expires_in"])
-        return data
-
-    def __repr__(self) -> str:
-        return (
-            f"PasskeyTokenResponse("
-            f"token_type={self.token_type!r}, "
-            f"expires_in={self.expires_in!r}, "
-            f"expires_at={self.expires_at!r}, "
-            f"scope={self.scope!r}, "
-            f"access_token=[REDACTED], "
-            f"id_token=[REDACTED], "
-            f"refresh_token=[REDACTED])"
-        )

src/auth0_server_python/tests/test_dpop_auth.py

@@ -157,12 +157,10 @@ def test_token_endpoint_proof_rejects_non_ec_key():
         make_dpop_proof_for_token_endpoint(rsa_key, "POST", "https://example.com/oauth/token")
 
 
-def test_dpop_repr_redacts_credentials(ec_key):
+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)
-    assert "[REDACTED]" in repr(auth)
-    assert "[REDACTED]" in str(auth)
 
 
 def test_dpop_rejects_non_ec_key():