feat: validate CTE actor token pairing and surface act claim for delegation

Author: kishore7snehil

examples/CustomTokenExchange.md

@@ -64,7 +64,7 @@ print(f"User logged in: {user['sub']}")
 
 ## 3. Actor Tokens (Delegation)
 
-Enable delegation scenarios where one service acts on behalf of a user.
+Enable delegation scenarios where one party acts on behalf of a user. The acting party is supplied via `actor_token`, and Auth0 records it in the [`act` claim](https://datatracker.ietf.org/doc/html/rfc8693#section-4.1) on the issued tokens.
 
 ```python
 # Service acting on behalf of a user
@@ -77,8 +77,32 @@ response = await auth0.custom_token_exchange(
         audience="https://api.example.com"
     )
 )
+
+# The actor claim is surfaced on the response. It may nest for delegation chains.
+if response.act:
+    print(f"Acting party: {response.act['sub']}")
 ```
 
+When you establish a session with `login_with_custom_token_exchange()`, the `act` claim is persisted on the session user and can be read back later via `get_user()`:
+
+```python
+result = await auth0.login_with_custom_token_exchange(
+    LoginWithCustomTokenExchangeOptions(
+        subject_token="user-access-token",
+        subject_token_type="urn:ietf:params:oauth:token-type:access_token",
+        actor_token="service-access-token",
+        actor_token_type="urn:ietf:params:oauth:token-type:access_token",
+    ),
+    store_options={"request": request, "response": response}
+)
+
+user = result.state_data["user"]
+if user.get("act"):
+    print(f"Acting party: {user['act']['sub']}")
+```
+
+> **NOTE**: When an `actor_token` is present, Auth0 does not issue a refresh token (the `offline_access` scope is dropped). A subsequent refresh-token grant therefore cannot re-emit the `act` claim, so the acting party is fixed at exchange time.
+
 ## 4. Custom Authorization Parameters
 
 Pass additional parameters to the token endpoint.
@@ -133,6 +157,7 @@ except CustomTokenExchangeError as e:
 
 - `INVALID_TOKEN_FORMAT`: Token is empty, whitespace-only, or has "Bearer " prefix
 - `MISSING_ACTOR_TOKEN_TYPE`: `actor_token` provided without `actor_token_type`
+- `MISSING_ACTOR_TOKEN`: `actor_token_type` provided without `actor_token`
 - `TOKEN_EXCHANGE_FAILED`: General token exchange failure
 - `INVALID_RESPONSE`: Auth0 returned a non-JSON response
 

src/auth0_server_python/auth_server/server_client.py

@@ -2243,10 +2243,35 @@ async def custom_token_exchange(
             https://datatracker.ietf.org/doc/html/rfc8693
         """
         try:
-            # Validate options (Pydantic handles this automatically)
             if not isinstance(options, CustomTokenExchangeOptions):
                 options = CustomTokenExchangeOptions(**options)
 
+            if not options.subject_token.strip():
+                raise CustomTokenExchangeError(
+                    CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
+                    "subject_token cannot be empty or whitespace-only"
+                )
+            if options.subject_token.strip().startswith("Bearer "):
+                raise CustomTokenExchangeError(
+                    CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
+                    "subject_token should not include 'Bearer ' prefix"
+                )
+            if options.actor_token and options.actor_token.strip().startswith("Bearer "):
+                raise CustomTokenExchangeError(
+                    CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT,
+                    "actor_token should not include 'Bearer ' prefix"
+                )
+            if options.actor_token and not options.actor_token_type:
+                raise CustomTokenExchangeError(
+                    CustomTokenExchangeErrorCode.MISSING_ACTOR_TOKEN_TYPE,
+                    "actor_token_type is required when actor_token is provided"
+                )
+            if options.actor_token_type and not options.actor_token:
+                raise CustomTokenExchangeError(
+                    CustomTokenExchangeErrorCode.MISSING_ACTOR_TOKEN,
+                    "actor_token is required when actor_token_type is provided"
+                )
+
             # Resolve domain
             domain = await self._resolve_current_domain(store_options)
             metadata = await self._get_oidc_metadata_cached(domain)
@@ -2309,8 +2334,17 @@ async def custom_token_exchange(
                         "Failed to parse token response as JSON"
                     )
 
-                # Validate and return response
-                return TokenExchangeResponse(**token_data)
+                token_response = TokenExchangeResponse(**token_data)
+
+                # Surface the actor claim for delegation exchanges
+                if options.actor_token and token_response.id_token:
+                    jwks = await self._get_jwks_cached(domain, metadata)
+                    claims = await self._verify_and_decode_jwt(
+                        token_response.id_token, jwks, audience=self._client_id
+                    )
+                    token_response.act = claims.get("act")
+
+                return token_response
 
         except ValidationError as e:
             raise CustomTokenExchangeError(

src/auth0_server_python/auth_types/__init__.py

@@ -5,7 +5,7 @@
 
 from typing import Any, Literal, Optional, Union
 
-from pydantic import BaseModel, Field, field_validator, model_validator
+from pydantic import BaseModel, Field
 
 
 class UserClaims(BaseModel):
@@ -257,33 +257,15 @@ class CustomTokenExchangeOptions(BaseModel):
         organization: Organization identifier for the token exchange (optional)
         authorization_params: Additional OAuth parameters (optional)
     """
-    subject_token: str = Field(..., min_length=1)
-    subject_token_type: str = Field(..., min_length=1)
+    subject_token: str
+    subject_token_type: str
     audience: Optional[str] = None
     scope: Optional[str] = None
     actor_token: Optional[str] = None
     actor_token_type: Optional[str] = None
     organization: Optional[str] = None
     authorization_params: Optional[dict[str, Any]] = None
 
-    @field_validator('subject_token', 'actor_token')
-    @classmethod
-    def validate_token_format(cls, v: Optional[str]) -> Optional[str]:
-        """Validate token doesn't have Bearer prefix and isn't whitespace-only."""
-        if v is not None:
-            if not v.strip():
-                raise ValueError("Token cannot be empty or whitespace-only")
-            if v.strip().startswith("Bearer "):
-                raise ValueError("Token should not include 'Bearer ' prefix")
-        return v
-
-    @model_validator(mode='after')
-    def validate_actor_token_type(self) -> 'CustomTokenExchangeOptions':
-        """Ensure actor_token_type is provided if actor_token is present."""
-        if self.actor_token and not self.actor_token_type:
-            raise ValueError("actor_token_type is required when actor_token is provided")
-        return self
-
 
 class TokenExchangeResponse(BaseModel):
     """
@@ -297,6 +279,7 @@ class TokenExchangeResponse(BaseModel):
         issued_token_type: Format of issued token
         id_token: OpenID Connect ID token (optional)
         refresh_token: Refresh token (optional)
+        act: Actor claim for delegation/impersonation exchanges (optional)
     """
     access_token: str
     token_type: str = "Bearer"
@@ -305,6 +288,7 @@ class TokenExchangeResponse(BaseModel):
     issued_token_type: Optional[str] = None
     id_token: Optional[str] = None
     refresh_token: Optional[str] = None
+    act: Optional[dict[str, Any]] = None
 
 
 class LoginWithCustomTokenExchangeOptions(BaseModel):
@@ -313,33 +297,15 @@ class LoginWithCustomTokenExchangeOptions(BaseModel):
 
     Combines token exchange parameters with session management.
     """
-    subject_token: str = Field(..., min_length=1)
-    subject_token_type: str = Field(..., min_length=1)
+    subject_token: str
+    subject_token_type: str
     audience: Optional[str] = None
     scope: Optional[str] = None
     actor_token: Optional[str] = None
     actor_token_type: Optional[str] = None
     organization: Optional[str] = None
     authorization_params: Optional[dict[str, Any]] = None
 
-    @field_validator('subject_token', 'actor_token')
-    @classmethod
-    def validate_token_format(cls, v: Optional[str]) -> Optional[str]:
-        """Validate token doesn't have Bearer prefix and isn't whitespace-only."""
-        if v is not None:
-            if not v.strip():
-                raise ValueError("Token cannot be empty or whitespace-only")
-            if v.strip().startswith("Bearer "):
-                raise ValueError("Token should not include 'Bearer ' prefix")
-        return v
-
-    @model_validator(mode='after')
-    def validate_actor_token_type(self) -> 'LoginWithCustomTokenExchangeOptions':
-        """Ensure actor_token_type is provided if actor_token is present."""
-        if self.actor_token and not self.actor_token_type:
-            raise ValueError("actor_token_type is required when actor_token is provided")
-        return self
-
 
 class LoginWithCustomTokenExchangeResult(BaseModel):
     """

src/auth0_server_python/error/__init__.py

@@ -225,6 +225,7 @@ 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"
     TOKEN_EXCHANGE_FAILED = "token_exchange_failed"
     INVALID_RESPONSE = "invalid_response"