Merge pull request #28 from auth0/tokenvault-at-sync

Add token vault subject_token_type access_token to api sdk

Author: adamjmcgrath

README.md

@@ -85,6 +85,34 @@ asyncio.run(main())
 
 In this example, the returned dictionary contains the decoded claims (like `sub`, `scope`, etc.) from the verified token.
 
+### 4. Get an access token for a connection
+
+If you need to get an access token for an upstream idp via a connection, you can use the `get_access_token_for_connection` method:
+
+```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>",
+    ))
+    connection = "my-connection" # The Auth0 connection to the upstream idp
+    access_token = "..." # The Auth0 access token to exchange
+
+    connection_access_token = await api_client.get_access_token_for_connection({"connection": connection, "access_token": access_token})
+    # The returned token is the access token for the upstream idp
+    print(connection_access_token)
+
+asyncio.run(main())
+```
+
+More info https://auth0.com/docs/secure/tokens/token-vault
+
 #### Requiring Additional Claims
 
 If your application demands extra claims, specify them with `required_claims`:
@@ -98,7 +126,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`.
 
-### 4. DPoP Authentication
+### 5. 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/api_client.py

@@ -1,11 +1,14 @@
 import time
 from typing import Any, Optional
 
+import httpx
 from authlib.jose import JsonWebKey, JsonWebToken
 
 from .config import ApiClientOptions
 from .errors import (
+    ApiError,
     BaseAuthError,
+    GetAccessTokenForConnectionError,
     InvalidAuthSchemeError,
     InvalidDpopProofError,
     MissingAuthorizationError,
@@ -390,6 +393,114 @@ async def verify_dpop_proof(
 
         return claims
 
+    async def get_access_token_for_connection(self, options: dict[str, Any]) -> dict[str, Any]:
+        """
+        Retrieves a token for a connection.
+
+        Args:
+            options: Options for retrieving an access token for a connection.
+                Must include 'connection' and 'access_token' keys.
+                May optionally include 'login_hint'.
+
+        Raises:
+            GetAccessTokenForConnectionError: If there was an issue requesting the access token.
+            ApiError: If the token exchange endpoint returns an error.
+
+        Returns:
+            Dictionary containing the token response with access_token, expires_in, and scope.
+        """
+        # Constants
+        SUBJECT_TYPE_ACCESS_TOKEN = "urn:ietf:params:oauth:token-type:access_token"  # noqa S105
+        REQUESTED_TOKEN_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN = "http://auth0.com/oauth/token-type/federated-connection-access-token"  # noqa S105
+        GRANT_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN = "urn:auth0:params:oauth:grant-type:token-exchange:federated-connection-access-token"  # noqa S105
+        connection = options.get("connection")
+        access_token = options.get("access_token")
+
+        if not connection:
+            raise MissingRequiredArgumentError("connection")
+
+        if not access_token:
+            raise MissingRequiredArgumentError("access_token")
+
+        client_id = self.options.client_id
+        client_secret = self.options.client_secret
+        if not client_id or not client_secret:
+            raise GetAccessTokenForConnectionError("You must configure the SDK with a client_id and client_secret to use get_access_token_for_connection.")
+
+        metadata = await self._discover()
+
+        token_endpoint = metadata.get("token_endpoint")
+        if not token_endpoint:
+            raise GetAccessTokenForConnectionError("Token endpoint missing in OIDC metadata")
+
+        # Prepare parameters
+        params = {
+            "connection": connection,
+            "requested_token_type": REQUESTED_TOKEN_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN,
+            "grant_type": GRANT_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN,
+            "client_id": client_id,
+            "subject_token": access_token,
+            "subject_token_type": SUBJECT_TYPE_ACCESS_TOKEN,
+        }
+
+        # Add login_hint if provided
+        if "login_hint" in options and options["login_hint"]:
+            params["login_hint"] = options["login_hint"]
+
+        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 = response.json() if "json" in response.headers.get(
+                        "content-type", "").lower() else {}
+                    raise ApiError(
+                        error_data.get("error", "connection_token_error"),
+                        error_data.get(
+                            "error_description", f"Failed to get token for connection: {response.status_code}"),
+                        response.status_code
+                    )
+
+                try:
+                    token_endpoint_response = response.json()
+                except Exception:
+                    raise ApiError("invalid_json", "Token endpoint returned invalid JSON.")
+
+                access_token = token_endpoint_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_endpoint_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)
+
+                return {
+                    "access_token": access_token,
+                    "expires_at": int(time.time()) + expires_in,
+                    "scope": token_endpoint_response.get("scope", "")
+                }
+
+        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]:

src/auth0_api_python/config.py

@@ -17,16 +17,20 @@ class ApiClientOptions:
         dpop_required: Whether DPoP is required (default: False, allows both Bearer and DPoP).
         dpop_iat_leeway: Leeway in seconds for DPoP proof iat claim (default: 30).
         dpop_iat_offset: Maximum age in seconds for DPoP proof iat claim (default: 300).
+        client_id: Optional required if you want to use get_access_token_for_connection.
+        client_secret: Optional required if you want to use get_access_token_for_connection.
     """
     def __init__(
-        self,
-        domain: str,
-        audience: str,
-        custom_fetch: Optional[Callable[..., object]] = None,
-        dpop_enabled: bool = True,
-        dpop_required: bool = False,
-        dpop_iat_leeway: int = 30,
-        dpop_iat_offset: int = 300,
+            self,
+            domain: str,
+            audience: str,
+            custom_fetch: Optional[Callable[..., object]] = None,
+            dpop_enabled: bool = True,
+            dpop_required: bool = False,
+            dpop_iat_leeway: int = 30,
+            dpop_iat_offset: int = 300,
+            client_id: Optional[str] = None,
+            client_secret: Optional[str] = None,
     ):
         self.domain = domain
         self.audience = audience
@@ -35,3 +39,5 @@ def __init__(
         self.dpop_required = dpop_required
         self.dpop_iat_leeway = dpop_iat_leeway
         self.dpop_iat_offset = dpop_iat_offset
+        self.client_id = client_id
+        self.client_secret = client_secret

src/auth0_api_python/errors.py

@@ -94,3 +94,39 @@ def get_status_code(self) -> int:
 
     def get_error_code(self) -> str:
         return "invalid_request"
+
+
+class GetAccessTokenForConnectionError(BaseAuthError):
+    """Error raised when getting a token for a connection fails."""
+
+    def get_status_code(self) -> int:
+        return 400
+
+    def get_error_code(self) -> str:
+        return "get_access_token_for_connection_error"
+
+
+class ApiError(BaseAuthError):
+    """
+    Error raised when an API request to Auth0 fails.
+    Contains details about the original error from Auth0.
+    """
+
+    def __init__(self, code: str, message: str, status_code=500, cause=None):
+        super().__init__(message)
+        self.code = code
+        self.status_code = status_code
+        self.cause = cause
+
+        if cause:
+            self.error = getattr(cause, "error", None)
+            self.error_description = getattr(cause, "error_description", None)
+        else:
+            self.error = None
+            self.error_description = None
+
+    def get_status_code(self) -> int:
+        return self.status_code
+
+    def get_error_code(self) -> str:
+        return self.code