update: add 'act' claim helpers

Author: nandan-bhat

EXAMPLES.md

@@ -38,7 +38,7 @@ async def exchange_on_behalf_of():
     async with httpx.AsyncClient() as client:
         downstream_response = await client.get(
             "https://calendar-api.example.com/events",
-            headers={"Authorization": f"Bearer {result.access_token}"}
+            headers={"Authorization": f"Bearer {result['access_token']}"}
         )
 
     downstream_response.raise_for_status()
@@ -60,6 +60,63 @@ asyncio.run(exchange_on_behalf_of())
 In the current implementation, `get_token_on_behalf_of()` forwards the incoming access token as
 the [RFC 8693](https://datatracker.ietf.org/doc/html/rfc8693#section-2.1) `subject_token` and relies on Auth0 to handle any DPoP-specific behavior for that token.
 
+## Inspecting Delegation After Token Verification
+
+When a downstream API or `MCP` server receives an access token that may have been issued through
+delegation, it can verify the token first and then inspect the `act` claim to identify the current
+actor for authorization and the full delegation chain for audit or attribution.
+
+```python
+import asyncio
+import logging
+
+from auth0_api_python import (
+    ApiClient,
+    ApiClientOptions,
+    get_current_actor,
+    get_delegation_chain,
+)
+
+logger = logging.getLogger(__name__)
+
+async def inspect_delegated_token():
+    api_client = ApiClient(ApiClientOptions(
+        domain="your-tenant.auth0.com",
+        audience="https://calendar-api.example.com"
+    ))
+
+    access_token = "delegated-auth0-access-token"
+
+    claims = await api_client.verify_access_token(access_token=access_token)
+
+    current_actor = get_current_actor(claims)
+    delegation_chain = get_delegation_chain(claims)
+
+    if current_actor != "mcp_server_client_id":
+        raise PermissionError("unexpected actor")
+
+    logger.info(
+        "delegated request",
+        extra={
+            "user_sub": claims["sub"],
+            "current_actor": current_actor,
+            "delegation_chain": delegation_chain,
+        },
+    )
+
+    return {
+        "user_sub": claims["sub"],
+        "current_actor": current_actor,
+        "delegation_chain": delegation_chain,
+    }
+
+asyncio.run(inspect_delegated_token())
+```
+
+Only the outermost `act.sub` represents the current actor and should be used for authorization
+decisions. Nested `act` values represent prior actors and are better suited for logging, audit, or
+attribution.
+
 ## Bearer Authentication
 
 Bearer authentication is the standard OAuth 2.0 token authentication method.

README.md

@@ -236,7 +236,7 @@ async def handle_calendar_request(incoming_access_token: str):
     async with httpx.AsyncClient() as client:
         downstream_response = await client.get(
             "https://calendar-api.example.com/events",
-            headers={"Authorization": f"Bearer {result.access_token}"}
+            headers={"Authorization": f"Bearer {result['access_token']}"}
         )
 
     downstream_response.raise_for_status()
@@ -250,6 +250,54 @@ token as the `subject_token` and relies on Auth0 to handle any DPoP-specific beh
 The OBO result only includes access-token-oriented fields. It does not expose `id_token` or
 `refresh_token`.
 
+#### Inspecting Delegation After Token Verification
+
+When a downstream API or `MCP` server receives an access token that may have been issued through
+delegation, it can verify the token first and then inspect the `act` claim to identify the current
+actor for authorization and the full delegation chain for logging or audit.
+
+```python
+import logging
+
+from auth0_api_python import (
+    ApiClient,
+    ApiClientOptions,
+    get_current_actor,
+    get_delegation_chain,
+)
+
+logger = logging.getLogger(__name__)
+
+api_client = ApiClient(ApiClientOptions(
+    domain="<AUTH0_DOMAIN>",
+    audience="<AUTH0_AUDIENCE>",
+))
+
+async def authorize_delegated_request(access_token: str):
+    claims = await api_client.verify_access_token(access_token=access_token)
+
+    current_actor = get_current_actor(claims)
+    delegation_chain = get_delegation_chain(claims)
+
+    if current_actor != "mcp_server_client_id":
+        raise PermissionError("unexpected actor")
+
+    logger.info(
+        "delegated request",
+        extra={
+            "user_sub": claims["sub"],
+            "current_actor": current_actor,
+            "delegation_chain": delegation_chain,
+        },
+    )
+
+    return claims
+```
+
+Only the outermost `act.sub` represents the current actor and should be used for authorization
+decisions. Nested `act` values represent prior actors in the delegation chain and are better suited
+for logging, audit, or attribution.
+
 #### Requiring Additional Claims
 
 If your application demands extra claims, specify them with `required_claims`:

src/auth0_api_python/__init__.py

@@ -5,6 +5,7 @@
 in server-side APIs, using Authlib for OIDC discovery and JWKS fetching.
 """
 
+from .act import get_current_actor, get_delegation_chain
 from .api_client import ApiClient
 from .cache import CacheAdapter, InMemoryCache
 from .config import ApiClientOptions
@@ -14,7 +15,11 @@
     DomainsResolverError,
     GetTokenByExchangeProfileError,
 )
-from .types import DomainsResolver, DomainsResolverContext, OnBehalfOfTokenResult
+from .types import (
+    DomainsResolver,
+    DomainsResolverContext,
+    OnBehalfOfTokenResult,
+)
 
 __all__ = [
     "ApiClient",
@@ -26,6 +31,8 @@
     "DomainsResolverContext",
     "DomainsResolverError",
     "GetTokenByExchangeProfileError",
+    "get_current_actor",
+    "get_delegation_chain",
     "InMemoryCache",
     "OnBehalfOfTokenResult",
 ]

src/auth0_api_python/act.py

@@ -0,0 +1,87 @@
+"""
+Helpers for working with the `act` claim on verified access token claims.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from .errors import VerifyAccessTokenError
+from .types import ActClaim
+
+
+def get_current_actor(claims: Mapping[str, Any]) -> str | None:
+    """
+    Return the current actor from the outermost `act.sub`, if present.
+
+    Only the outermost `act.sub` should be used for authorization decisions.
+    Nested `act` values represent prior actors and are informational.
+    """
+
+    act_claim = _get_validated_act_claim(claims)
+    if act_claim is None:
+        return None
+
+    return act_claim["sub"]
+
+
+def get_delegation_chain(claims: Mapping[str, Any]) -> list[str]:
+    """
+    Return the delegation chain from newest actor to oldest actor.
+
+    The first entry is the current actor (outermost `act.sub`). Later entries are
+    prior actors from nested `act` values and are typically most useful for audit
+    and attribution.
+    """
+
+    act_claim = _get_validated_act_claim(claims)
+    if act_claim is None:
+        return []
+
+    chain: list[str] = []
+    current: ActClaim | None = act_claim
+    while current is not None:
+        chain.append(current["sub"])
+        current = current.get("act")
+
+    return chain
+
+
+def _get_validated_act_claim(claims: Mapping[str, Any]) -> ActClaim | None:
+    if not isinstance(claims, Mapping):
+        raise VerifyAccessTokenError("Verified access token claims must be an object")
+
+    act_claim = claims.get("act")
+    if act_claim is None:
+        return None
+
+    return _parse_act_claim(act_claim, path="act", seen=set())
+
+
+def _parse_act_claim(value: Any, *, path: str, seen: set[int]) -> ActClaim:
+    if not isinstance(value, Mapping):
+        raise VerifyAccessTokenError(f"{path} must be an object")
+
+    object_id = id(value)
+    if object_id in seen:
+        raise VerifyAccessTokenError(f"{path} contains a circular reference")
+
+    seen.add(object_id)
+    try:
+        sub = value.get("sub")
+        if not isinstance(sub, str) or not sub.strip():
+            raise VerifyAccessTokenError(f"{path}.sub must be a non-empty string")
+
+        parsed: ActClaim = {"sub": sub}
+        nested_act = value.get("act")
+        if nested_act is not None:
+            parsed["act"] = _parse_act_claim(
+                nested_act,
+                path=f"{path}.act",
+                seen=seen,
+            )
+
+        return parsed
+    finally:
+        seen.remove(object_id)

src/auth0_api_python/api_client.py

@@ -234,7 +234,7 @@ async def verify_request(
             http_url: The HTTP URL (required for DPoP, also used for MCD resolver context)
 
         Returns:
-            The decoded access token claims
+            The decoded access token claims, including `act` when present.
 
         Raises:
             MissingRequiredArgumentError: If required args are missing
@@ -414,7 +414,7 @@ async def verify_access_token(
             required_claims: Optional list of additional claim names that must be present
 
         Returns:
-            The decoded token claims if valid.
+            The decoded token claims if valid, including `act` when present.
 
         Raises:
             MissingRequiredArgumentError: If no token is provided.
@@ -796,7 +796,7 @@ async def get_token_by_exchange_profile(
             Dictionary containing:
             - access_token (str): The Auth0 access token
             - expires_in (int): Token lifetime in seconds
-            - expires_at (int): Unix timestamp when token expires
+            - expires_at (int): Absolute expiration time as a Unix timestamp in seconds, calculated by the SDK from expires_in
             - id_token (str, optional): OpenID Connect ID token
             - refresh_token (str, optional): Refresh token
             - scope (str, optional): Granted scopes
@@ -986,7 +986,7 @@ async def get_token_on_behalf_of(
             Dictionary containing:
             - access_token (str): The exchanged Auth0 access token
             - expires_in (int): Token lifetime in seconds
-            - expires_at (int): Unix timestamp when token expires
+            - expires_at (int): Absolute expiration time as a Unix timestamp in seconds, calculated by the SDK from expires_in
             - scope (str, optional): Granted scopes
             - token_type (str, optional): Token type (typically "Bearer")
             - issued_token_type (str, optional): RFC 8693 issued token type identifier

src/auth0_api_python/types.py

@@ -2,8 +2,23 @@
 Type definitions for auth0-api-python SDK
 """
 
+from __future__ import annotations
+
 from collections.abc import Awaitable, Callable
-from typing import Optional, TypedDict, Union
+from typing import TypedDict
+
+
+class ActClaim(TypedDict, total=False):
+    """
+    Actor claim carried by access tokens issued through delegation.
+
+    Attributes:
+        sub: The current actor for this step of the delegation chain.
+        act: The prior actor in the delegation chain, if present.
+    """
+
+    sub: str
+    act: ActClaim
 
 
 class DomainsResolverContext(TypedDict, total=False):
@@ -15,8 +30,8 @@ class DomainsResolverContext(TypedDict, total=False):
         request_headers: Request headers dict (e.g., Host, X-Forwarded-Host) (optional)
         unverified_iss: The issuer claim from the unverified token
     """
-    request_url: Optional[str]
-    request_headers: Optional[dict]
+    request_url: str | None
+    request_headers: dict | None
     unverified_iss: str
 
 
@@ -27,7 +42,7 @@ class OnBehalfOfTokenResult(TypedDict, total=False):
     Attributes:
         access_token: The access token issued for the downstream API.
         expires_in: Token lifetime in seconds.
-        expires_at: Unix timestamp when the token expires.
+        expires_at: Absolute expiration time as a Unix timestamp in seconds, calculated by the SDK from expires_in.
         scope: Granted scopes, if returned by Auth0.
         token_type: Token type, if returned by Auth0.
         issued_token_type: RFC 8693 issued token type, if returned by Auth0.
@@ -41,7 +56,7 @@ class OnBehalfOfTokenResult(TypedDict, total=False):
     issued_token_type: str
 
 DomainsResolver = Callable[
-    [DomainsResolverContext], Union[list[str], Awaitable[list[str]]]
+    [DomainsResolverContext], list[str] | Awaitable[list[str]]
 ]
 """
 Type alias for domains resolver function.