feat: add Custom Token Exchange support (RFC 8693)

Implements get_token_by_exchange_profile() method for exchanging subject tokens
via Auth0 Token Exchange Profiles, following RFC 8693.

Features:
- Subject token exchange with strict validation (fail-fast on whitespace)
- Support for optional audience, scope, and requested_token_type parameters
- Extra parameters for Actions with reserved param validation (fail-fast)
- DoS protection with 20-item array size limit for extra parameters
- Returns access_token, expires_in, expires_at, and optional fields
- HTTP Basic authentication with confidential client credentials
- Comprehensive error handling with upstream status preservation
- 10-second HTTP timeout to prevent hanging requests

Implementation details:
- Validates subject tokens strictly (rejects whitespace, Bearer prefix)
- Fails fast when reserved OAuth parameters supplied in extras
- Enforces MAX_ARRAY_VALUES_PER_KEY=20 for DoS protection
- Preserves upstream 4xx error status codes
- Module-level constants (TOKEN_EXCHANGE_GRANT_TYPE, RESERVED_PARAMS)
- Case-insensitive header handling (normalizes to lowercase)
- Precise error handling (ValueError for JSON parse errors)
- Adds GetTokenByExchangeProfileError for validation failures

Validation against auth0-auth-js:
- Whitespace handling matches strict JS SDK behavior (fail-fast)
- Array size limit matches JS SDK (20 items for DoS protection)
- Reserved params list aligned with JS SDK implementation
- Improves on JS SDK: fails fast on reserved params vs silent ignore
- Enhances return value: includes both expires_in and expires_at

Documentation:
- Added Early Access feature note with confidential client requirement
- Documented Token Exchange Profile URI matching and namespace rules
- Security warning for extra parameters (form fields, no secrets)
- Clarified audience optionality with inline example
- Linked to auth0-auth-js companion SDK and Auth0 docs

Tests: 24 new tests covering success paths, validation, error handling,
and edge cases. All 84 tests passing with 85% coverage.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: btiernay

.gitignore

@@ -22,4 +22,7 @@ setup.py
 test.py
 test-script.py
 .coverage
-coverage.xml
+coverage.xml
+
+# IDE
+.idea/

README.md

@@ -113,6 +113,69 @@ asyncio.run(main())
 
 More info https://auth0.com/docs/secure/tokens/token-vault
 
+### 5. Custom Token Exchange (Early Access)
+
+> [!NOTE]
+> This feature is currently available in [Early Access](https://auth0.com/docs/troubleshoot/product-lifecycle/product-release-stages#early-access) for Enterprise customers. Please reach out to Auth0 support to get it enabled for your tenant.
+
+This feature requires a [confidential client](https://auth0.com/docs/get-started/applications/confidential-and-public-applications#confidential-applications) (both `client_id` and `client_secret` must be configured).
+
+Custom Token Exchange allows you to exchange a subject token for Auth0 tokens using RFC 8693. This is useful for:
+- Getting Auth0 tokens for another audience
+- Integrating external identity providers
+- Migrating to Auth0
+
+```python
+import asyncio
+
+from auth0_api_python import ApiClient, ApiClientOptions
+
+async def main():
+    api_client = ApiClient(ApiClientOptions(
+        domain="<AUTH0_DOMAIN>",
+        audience="<AUTH0_AUDIENCE>",
+        client_id="<AUTH0_CLIENT_ID>",
+        client_secret="<AUTH0_CLIENT_SECRET>",
+    ))
+
+    subject_token = "..."  # Token from your legacy system or external source
+
+    result = await api_client.get_token_by_exchange_profile(
+        subject_token=subject_token,
+        subject_token_type="urn:example:subject-token",
+        audience="https://api.example.com"  # Optional - omit to use the client's default audience
+    )
+
+    # Result contains access_token, expires_in, expires_at, and optionally id_token, refresh_token
+
+asyncio.run(main())
+```
+
+The `subject_token_type` must match a Token Exchange Profile configured in Auth0. This URI identifies which profile will process the exchange and cannot use reserved OAuth namespaces (e.g., `urn:ietf:params:oauth:*`).
+
+#### Additional Parameters
+
+You can pass additional parameters for your Token Exchange Profile or Actions via the `extra` parameter. These are sent as form fields to Auth0 and may be inspected by Actions:
+
+```python
+result = await api_client.get_token_by_exchange_profile(
+    subject_token=subject_token,
+    subject_token_type="urn:example:subject-token",
+    audience="https://api.example.com",
+    extra={
+        "device_id": "device-12345",
+        "session_id": "sess-abc"
+    }
+)
+```
+
+> [!WARNING]
+> Extra parameters are sent as form fields and may appear in logs. Do not include secrets or sensitive data. Reserved OAuth parameter names (like `grant_type`, `client_id`, `scope`) cannot be used and will raise an error.
+
+**Related SDKs:** [auth0-auth-js](https://github.com/auth0/auth0-auth-js) (JavaScript/TypeScript)
+
+More info: https://auth0.com/docs/authenticate/custom-token-exchange
+
 #### Requiring Additional Claims
 
 If your application demands extra claims, specify them with `required_claims`:
@@ -126,7 +189,7 @@ decoded_and_verified_token = await api_client.verify_access_token(
 
 If the token lacks `my_custom_claim` or fails any standard check (issuer mismatch, expired token, invalid signature), the method raises a `VerifyAccessTokenError`.
 
-### 5. DPoP Authentication
+### 6. DPoP Authentication
 
 > [!NOTE]  
 > This feature is currently available in [Early Access](https://auth0.com/docs/troubleshoot/product-lifecycle/product-release-stages#early-access). Please reach out to Auth0 support to get it enabled for your tenant.

src/auth0_api_python/__init__.py

@@ -7,8 +7,10 @@
 
 from .api_client import ApiClient
 from .config import ApiClientOptions
+from .errors import GetTokenByExchangeProfileError
 
 __all__ = [
     "ApiClient",
-    "ApiClientOptions"
+    "ApiClientOptions",
+    "GetTokenByExchangeProfileError"
 ]

src/auth0_api_python/api_client.py

@@ -1,5 +1,5 @@
 import time
-from typing import Any, Optional
+from typing import Any, Optional, Union
 
 import httpx
 from authlib.jose import JsonWebKey, JsonWebToken
@@ -9,6 +9,7 @@
     ApiError,
     BaseAuthError,
     GetAccessTokenForConnectionError,
+    GetTokenByExchangeProfileError,
     InvalidAuthSchemeError,
     InvalidDpopProofError,
     MissingAuthorizationError,
@@ -24,6 +25,19 @@
     sha256_base64url,
 )
 
+# Token Exchange constants
+TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"  # noqa: S105
+MAX_ARRAY_VALUES_PER_KEY = 20  # DoS protection for extra parameter arrays
+
+# OAuth parameter denylist - parameters that cannot be overridden via extras
+RESERVED_PARAMS = frozenset([
+    "grant_type", "client_id", "client_secret", "client_assertion",
+    "client_assertion_type", "subject_token", "subject_token_type",
+    "requested_token_type", "actor_token", "actor_token_type",
+    "audience", "aud", "resource", "resources", "resource_indicator",
+    "scope", "connection", "login_hint", "organization", "assertion",
+])
+
 
 class ApiClient:
     """
@@ -78,6 +92,9 @@ async def verify_request(
             InvalidDpopProofError: If DPoP verification fails
             VerifyAccessTokenError: If access token verification fails
         """
+        # Normalize header keys to lowercase for robust access
+        headers = {k.lower(): v for k, v in headers.items()}
+
         authorization_header = headers.get("authorization", "")
         dpop_proof = headers.get("dpop")
 
@@ -86,22 +103,20 @@ async def verify_request(
                 raise self._prepare_error(
                         InvalidAuthSchemeError("")
                     )
-            else :
+            else:
                 raise self._prepare_error(MissingAuthorizationError())
 
-
-        parts = authorization_header.split(" ")
+        # Split authorization header on first whitespace
+        parts = authorization_header.split(None, 1)
         if len(parts) != 2:
-            if len(parts) < 2:
-                raise self._prepare_error(MissingAuthorizationError())
-            elif len(parts) > 2:
-                raise self._prepare_error(
-                    InvalidAuthSchemeError("")
-                )
+            raise self._prepare_error(MissingAuthorizationError())
 
         scheme, token = parts
+        scheme = scheme.lower()
 
-        scheme = scheme.strip().lower()
+        # Tokens should not contain spaces (indicates malformed header)
+        if " " in token:
+            raise self._prepare_error(InvalidAuthSchemeError(""))
 
         if self.is_dpop_required() and scheme != "dpop":
             raise self._prepare_error(
@@ -501,6 +516,211 @@ async def get_access_token_for_connection(self, options: dict[str, Any]) -> dict
                 exc
             )
 
+    async def get_token_by_exchange_profile(
+        self,
+        subject_token: str,
+        subject_token_type: str,
+        audience: Optional[str] = None,
+        scope: Optional[str] = None,
+        requested_token_type: Optional[str] = None,
+        extra: Optional[dict[str, Union[str, list[str]]]] = None
+    ) -> dict[str, Any]:
+        """
+        Exchange a subject token for an Auth0 token using RFC 8693.
+
+        The matching Token Exchange Profile is selected by subject_token_type.
+        This method requires a confidential client (client_id and client_secret must be configured).
+
+        Args:
+            subject_token: The token to be exchanged
+            subject_token_type: URI identifying the token type (must match a Token Exchange Profile)
+            audience: Optional target API identifier for the exchanged tokens
+            scope: Optional space-separated OAuth 2.0 scopes to request
+            requested_token_type: Optional type of token to issue (defaults to access token)
+            extra: Optional additional parameters sent as form fields to Auth0.
+                   Cannot override reserved OAuth parameters.
+
+        Returns:
+            Dictionary containing:
+            - access_token (str): The Auth0 access token
+            - expires_in (int): Token lifetime in seconds
+            - expires_at (int): Unix timestamp when token expires
+            - id_token (str, optional): OpenID Connect ID token
+            - refresh_token (str, optional): Refresh token
+            - scope (str, optional): Granted scopes
+            - token_type (str, optional): Token type (typically "Bearer")
+            - issued_token_type (str, optional): RFC 8693 issued token type identifier
+
+        Raises:
+            MissingRequiredArgumentError: If required parameters are missing
+            GetTokenByExchangeProfileError: If client credentials not configured, validation fails,
+                                           or reserved parameters are supplied in extra
+            ApiError: If the token endpoint returns an error
+
+        Example:
+            >>> result = await api_client.get_token_by_exchange_profile(
+            ...     subject_token=token,
+            ...     subject_token_type="urn:example:subject-token",
+            ...     audience="https://api.backend.com"
+            ... )
+
+        References:
+            - Custom Token Exchange: https://auth0.com/docs/authenticate/custom-token-exchange
+            - RFC 8693: https://datatracker.ietf.org/doc/html/rfc8693
+            - Related SDK: https://github.com/auth0/auth0-auth-js
+        """
+        # Validate required parameters
+        if not subject_token:
+            raise MissingRequiredArgumentError("subject_token")
+        if not subject_token_type:
+            raise MissingRequiredArgumentError("subject_token_type")
+
+        # Validate subject token format (fail fast to ensure token integrity)
+        if not isinstance(subject_token, str):
+            raise GetTokenByExchangeProfileError("subject_token must be a string")
+
+        # Fail fast on blank or whitespace-only
+        if not subject_token.strip():
+            raise GetTokenByExchangeProfileError("subject_token cannot be blank or whitespace")
+
+        # Be explicit about surrounding spaces (prevents token ambiguity)
+        if subject_token != subject_token.strip():
+            raise GetTokenByExchangeProfileError(
+                "subject_token must not include leading or trailing whitespace"
+            )
+
+        # Very common copy-paste mistake (case-insensitive check)
+        if subject_token.lower().startswith("bearer "):
+            raise GetTokenByExchangeProfileError(
+                "subject_token must not include the 'Bearer ' prefix"
+            )
+
+        # Require client credentials
+        client_id = self.options.client_id
+        client_secret = self.options.client_secret
+        if not client_id or not client_secret:
+            raise GetTokenByExchangeProfileError(
+                "Client credentials are required to use get_token_by_exchange_profile"
+            )
+
+        # Discover token endpoint
+        metadata = await self._discover()
+        token_endpoint = metadata.get("token_endpoint")
+        if not token_endpoint:
+            raise GetTokenByExchangeProfileError("Token endpoint missing in OIDC metadata")
+
+        # Build request parameters (client_id sent via HTTP Basic auth only)
+        params = {
+            "grant_type": TOKEN_EXCHANGE_GRANT_TYPE,
+            "subject_token": subject_token,
+            "subject_token_type": subject_token_type,
+        }
+
+        # Add optional parameters
+        if audience:
+            params["audience"] = audience
+        if scope:
+            params["scope"] = scope
+        if requested_token_type:
+            params["requested_token_type"] = requested_token_type
+
+        # Append extra parameters with validation
+        if extra:
+            for parameter_key, parameter_value in extra.items():
+                # Fail fast if reserved parameter is supplied
+                if parameter_key in RESERVED_PARAMS:
+                    raise GetTokenByExchangeProfileError(
+                        f"Parameter '{parameter_key}' is reserved and cannot be overridden"
+                    )
+
+                # Store value (httpx handles list encoding as multiple key=value pairs)
+                if isinstance(parameter_value, list):
+                    # DoS protection: enforce array size limit
+                    if len(parameter_value) > MAX_ARRAY_VALUES_PER_KEY:
+                        raise GetTokenByExchangeProfileError(
+                            f"Parameter '{parameter_key}' exceeds maximum array size of {MAX_ARRAY_VALUES_PER_KEY}"
+                        )
+                    params[parameter_key] = parameter_value
+                else:
+                    params[parameter_key] = str(parameter_value)
+
+        # Make token exchange request
+        try:
+            async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
+                response = await client.post(
+                    token_endpoint,
+                    data=params,
+                    auth=(client_id, client_secret)
+                )
+
+                if response.status_code != 200:
+                    error_data = {}
+                    try:
+                        if "json" in response.headers.get("content-type", "").lower():
+                            error_data = response.json()
+                    except ValueError:
+                        pass  # Ignore JSON parse errors, use generic error message below
+
+                    raise ApiError(
+                        error_data.get("error", "token_exchange_error"),
+                        error_data.get(
+                            "error_description",
+                            f"Failed to exchange token of type '{subject_token_type}'"
+                            + (f" for audience '{audience}'" if audience else "")
+                        ),
+                        response.status_code
+                    )
+
+                try:
+                    token_response = response.json()
+                except ValueError:
+                    raise ApiError("invalid_json", "Token endpoint returned invalid JSON.", 502)
+
+                # Validate required fields
+                access_token = token_response.get("access_token")
+                if not isinstance(access_token, str) or not access_token:
+                    raise ApiError(
+                        "invalid_response",
+                        "Missing or invalid access_token in response.",
+                        502
+                    )
+
+                expires_in_raw = token_response.get("expires_in", 3600)
+                try:
+                    expires_in = int(expires_in_raw)
+                except (TypeError, ValueError):
+                    raise ApiError("invalid_response", "expires_in is not an integer.", 502)
+
+                # Build response with required fields
+                result = {
+                    "access_token": access_token,
+                    "expires_in": expires_in,
+                    "expires_at": int(time.time()) + expires_in,
+                }
+
+                # Add optional fields if present
+                optional_fields = ["scope", "id_token", "refresh_token", "token_type", "issued_token_type"]
+                for field in optional_fields:
+                    if value := token_response.get(field):
+                        result[field] = value
+
+                return result
+
+        except httpx.TimeoutException as exc:
+            raise ApiError(
+                "timeout_error",
+                f"Request to token endpoint timed out: {str(exc)}",
+                504,
+                exc
+            )
+        except httpx.HTTPError as exc:
+            raise ApiError(
+                "network_error",
+                f"Network error occurred: {str(exc)}",
+                502,
+                exc
+            )
+
     # ===== Private Methods =====
 
     async def _discover(self) -> dict[str, Any]: