feat: add Custom Token Exchange support (RFC 8693)

btiernay · claude · btiernay · commit 337a4454819a · 2025-10-25T13:25:22.000-04:00
Implements get_token_by_exchange_profile() method for exchanging subject tokens via Auth0 Token Exchange Profiles, following RFC 8693. Features: - Subject token exchange with automatic whitespace normalization - Support for optional audience, scope, and requested_token_type parameters - Extra parameters for Actions with reserved param validation (fail-fast) - Returns access_token, expires_in, expires_at, and optional fields - HTTP Basic authentication with confidential client credentials - Comprehensive error handling with upstream status preservation Implementation details: - Validates subject tokens (strips whitespace, rejects Bearer prefix) - Fails fast when reserved OAuth parameters supplied in extras - Preserves upstream 4xx error status codes - Adds GetTokenByExchangeProfileError for validation failures 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 Tests: 24 new tests covering success paths, validation, error handling, and edge cases. All 84 tests passing. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
diff --git a/README.md b/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.
diff --git a/src/auth0_api_python/__init__.py b/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"
 ]
diff --git a/src/auth0_api_python/api_client.py b/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,
@@ -501,6 +502,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
+        """
+        # Constants
+        TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"  # noqa: S105
+
+        # OAuth parameter denylist - fail fast if these are supplied in extra
+        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",
+        ])
+
+        # Validate required parameters
+        if not subject_token:
+            raise MissingRequiredArgumentError("subject_token")
+        if not subject_token_type:
+            raise MissingRequiredArgumentError("subject_token_type")
+
+        # Validate and normalize subject token
+        if not isinstance(subject_token, str):
+            raise GetTokenByExchangeProfileError("subject_token must be a string")
+
+        subject_token = subject_token.strip()
+
+        if not subject_token:
+            raise GetTokenByExchangeProfileError("subject_token cannot be blank or whitespace")
+        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):
+                    params[parameter_key] = parameter_value
+                else:
+                    params[parameter_key] = str(parameter_value)
+
+        # Make token exchange request
+        try:
+            async with httpx.AsyncClient() 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 Exception:  # noqa: S110
+                        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 Exception:
+                    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]:
diff --git a/src/auth0_api_python/errors.py b/src/auth0_api_python/errors.py
@@ -106,6 +106,16 @@ def get_error_code(self) -> str:
         return "get_access_token_for_connection_error"
 
 
+class GetTokenByExchangeProfileError(BaseAuthError):
+    """Error raised when getting a token via exchange profile fails."""
+
+    def get_status_code(self) -> int:
+        return 400
+
+    def get_error_code(self) -> str:
+        return "get_token_by_exchange_profile_error"
+
+
 class ApiError(BaseAuthError):
     """
     Error raised when an API request to Auth0 fails.
diff --git a/tests/test_api_client.py b/tests/test_api_client.py
