diff --git a/EXAMPLES.md b/EXAMPLES.md index db6e8f6..3bd6e0c 100644 --- a/EXAMPLES.md +++ b/EXAMPLES.md @@ -2,6 +2,64 @@ This document provides examples for using the `auth0-api-python` package to validate Auth0 tokens in your API. +## On Behalf Of Token Exchange + +Use `get_token_on_behalf_of()` when your API receives an `Auth0` access token for itself and needs +to exchange it for another `Auth0` access token targeting a downstream API while preserving the same +user identity. This is especially useful for `MCP` servers and other intermediary APIs that need to +call downstream APIs on behalf of the user. + +The following example verifies the incoming access token for your API, exchanges it for a token for the downstream API, and then calls the downstream API with the exchanged token. + +```python +import asyncio +import httpx + +from auth0_api_python import ApiClient, ApiClientOptions + +async def exchange_on_behalf_of(): + api_client = ApiClient(ApiClientOptions( + domain="your-tenant.auth0.com", + audience="https://mcp-server.example.com", + client_id="", + client_secret="" + )) + + incoming_access_token = "incoming-auth0-access-token" + + claims = await api_client.verify_access_token(access_token=incoming_access_token) + + result = await api_client.get_token_on_behalf_of( + access_token=incoming_access_token, + audience="https://calendar-api.example.com", + scope="calendar:read calendar:write" + ) + + async with httpx.AsyncClient() as client: + downstream_response = await client.get( + "https://calendar-api.example.com/events", + headers={"Authorization": f"Bearer {result.access_token}"} + ) + + downstream_response.raise_for_status() + + return { + "user": claims["sub"], + "data": downstream_response.json(), + } + +asyncio.run(exchange_on_behalf_of()) +``` + +> [!TIP] Production notes: +> - Pass the raw access token to `get_token_on_behalf_of()`. Do not pass the full `Authorization` header or include the `Bearer ` prefix. +> - Verify the incoming token for your API before exchanging it so your application rejects invalid or mis-targeted tokens early. +> - The downstream `audience` must match an API identifier configured in your Auth0 tenant. +> - `get_token_on_behalf_of()` only returns access-token-oriented fields. It does not expose `id_token` or `refresh_token`. + +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. + ## Bearer Authentication Bearer authentication is the standard OAuth 2.0 token authentication method. @@ -157,4 +215,4 @@ async def verify_dpop_token(access_token, dpop_proof, http_method, http_url): "token_claims": token_claims, "proof_claims": proof_claims } -``` \ No newline at end of file +``` diff --git a/README.md b/README.md index 26332bf..c3609b4 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,44 @@ except ApiError as e: More info: https://auth0.com/docs/authenticate/custom-token-exchange +#### On Behalf Of Token Exchange + +Use `get_token_on_behalf_of()` when your API receives an `Auth0` access token for itself and needs +to exchange it for another `Auth0` access token targeting a downstream API while preserving the +same user identity. This is especially useful for `MCP` servers and other intermediary APIs that +need to call downstream APIs on behalf of the user. + +The following example verifies the incoming access token for your API, exchanges it for a token for the downstream API, and then calls the downstream API with the exchanged token. + +```python +import httpx + +async def handle_calendar_request(incoming_access_token: str): + await api_client.verify_access_token(access_token=incoming_access_token) + + result = await api_client.get_token_on_behalf_of( + access_token=incoming_access_token, + audience="https://calendar-api.example.com", + scope="calendar:read calendar:write" + ) + + async with httpx.AsyncClient() as client: + downstream_response = await client.get( + "https://calendar-api.example.com/events", + headers={"Authorization": f"Bearer {result.access_token}"} + ) + + downstream_response.raise_for_status() + + return downstream_response.json() +``` + +The OBO wrapper reuses the existing RFC 8693 exchange support and fixes both token-type parameters +to Auth0 access-token exchange. In the current implementation, the SDK forwards the incoming access +token as the `subject_token` and relies on Auth0 to handle any DPoP-specific behavior for that token. +The OBO result only includes access-token-oriented fields. It does not expose `id_token` or +`refresh_token`. + #### Requiring Additional Claims If your application demands extra claims, specify them with `required_claims`: @@ -353,4 +391,4 @@ Please do not report security vulnerabilities on the public GitHub issue tracker

This project is licensed under the MIT license. See the LICENSE file for more info. -

\ No newline at end of file +

diff --git a/src/auth0_api_python/__init__.py b/src/auth0_api_python/__init__.py index ef27ea2..e5a7919 100644 --- a/src/auth0_api_python/__init__.py +++ b/src/auth0_api_python/__init__.py @@ -14,7 +14,7 @@ DomainsResolverError, GetTokenByExchangeProfileError, ) -from .types import DomainsResolver, DomainsResolverContext +from .types import DomainsResolver, DomainsResolverContext, OnBehalfOfTokenResult __all__ = [ "ApiClient", @@ -27,4 +27,5 @@ "DomainsResolverError", "GetTokenByExchangeProfileError", "InMemoryCache", + "OnBehalfOfTokenResult", ] diff --git a/src/auth0_api_python/api_client.py b/src/auth0_api_python/api_client.py index 36f23cf..a4056d5 100644 --- a/src/auth0_api_python/api_client.py +++ b/src/auth0_api_python/api_client.py @@ -21,6 +21,7 @@ MissingRequiredArgumentError, VerifyAccessTokenError, ) +from .types import OnBehalfOfTokenResult from .utils import ( calculate_jwk_thumbprint, fetch_jwks, @@ -34,6 +35,7 @@ # Token Exchange constants TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange" # noqa: S105 +OBO_ACCESS_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" # noqa: S105 MAX_ARRAY_VALUES_PER_KEY = 20 # DoS protection for extra parameter arrays # OAuth parameter denylist - parameters that cannot be overridden via extras @@ -962,6 +964,64 @@ async def example(): exc ) + async def get_token_on_behalf_of( + self, + access_token: str, + audience: str, + scope: Optional[str] = None, + ) -> OnBehalfOfTokenResult: + """ + Exchange an Auth0 access token for another Auth0 access token targeting a downstream API + while acting on behalf of the same end user (OBO). + + This is a convenience wrapper around get_token_by_exchange_profile() that fixes the + RFC 8693 token types for Auth0 access-token-to-access-token exchange. + + Args: + access_token: The Auth0 access token to exchange + audience: Target API identifier for the exchanged access token + scope: Optional space-separated OAuth 2.0 scopes to request + + Returns: + 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 + - 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 are not configured or validation fails + ApiError: If the token endpoint returns an error + """ + if not audience: + raise MissingRequiredArgumentError("audience") + + result = await self.get_token_by_exchange_profile( + subject_token=access_token, + subject_token_type=OBO_ACCESS_TOKEN_TYPE, + audience=audience, + scope=scope, + requested_token_type=OBO_ACCESS_TOKEN_TYPE, + ) + + obo_result: OnBehalfOfTokenResult = { + "access_token": result["access_token"], + "expires_in": result["expires_in"], + "expires_at": result["expires_at"], + } + + if "scope" in result: + obo_result["scope"] = result["scope"] + if "token_type" in result: + obo_result["token_type"] = result["token_type"] + if "issued_token_type" in result: + obo_result["issued_token_type"] = result["issued_token_type"] + + return obo_result + # ===== Private Methods ===== def _apply_extra( diff --git a/src/auth0_api_python/config.py b/src/auth0_api_python/config.py index 6c5f1c7..1929d55 100644 --- a/src/auth0_api_python/config.py +++ b/src/auth0_api_python/config.py @@ -27,8 +27,10 @@ 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: Required for get_access_token_for_connection and get_token_by_exchange_profile. - client_secret: Required for get_access_token_for_connection and get_token_by_exchange_profile. + client_id: Required for get_access_token_for_connection, get_token_by_exchange_profile, + and get_token_on_behalf_of. + client_secret: Required for get_access_token_for_connection, get_token_by_exchange_profile, + and get_token_on_behalf_of. timeout: HTTP timeout in seconds for token endpoint requests (default: 10.0). """ def __init__( diff --git a/src/auth0_api_python/types.py b/src/auth0_api_python/types.py index 122f59e..f665bc8 100644 --- a/src/auth0_api_python/types.py +++ b/src/auth0_api_python/types.py @@ -19,6 +19,27 @@ class DomainsResolverContext(TypedDict, total=False): request_headers: Optional[dict] unverified_iss: str + +class OnBehalfOfTokenResult(TypedDict, total=False): + """ + Result returned from an On Behalf Of token exchange. + + 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. + 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. + """ + + access_token: str + expires_in: int + expires_at: int + scope: str + token_type: str + issued_token_type: str + DomainsResolver = Callable[ [DomainsResolverContext], Union[list[str], Awaitable[list[str]]] ] diff --git a/tests/test_api_client.py b/tests/test_api_client.py index 13ba329..eb59528 100644 --- a/tests/test_api_client.py +++ b/tests/test_api_client.py @@ -2779,6 +2779,159 @@ async def test_get_token_by_exchange_profile_custom_timeout_honored(httpx_mock: assert err.value.status_code == 504 +@pytest.mark.asyncio +async def test_get_token_on_behalf_of_missing_client_credentials(): + """Test that OBO requires confidential client credentials.""" + api_client = ApiClient(ApiClientOptions( + domain="auth0.local", + audience="my-audience", + )) + + with pytest.raises(GetTokenByExchangeProfileError) as err: + await api_client.get_token_on_behalf_of( + access_token="token", + audience="https://api.backend.com" + ) + + assert "client credentials are required" in str(err.value).lower() + + +@pytest.mark.asyncio +async def test_get_token_on_behalf_of_missing_audience(api_client_confidential): + """Test that OBO requires an explicit downstream audience.""" + with pytest.raises(MissingRequiredArgumentError): + await api_client_confidential.get_token_on_behalf_of( + access_token="token", + audience="" + ) + + +@pytest.mark.asyncio +async def test_get_token_on_behalf_of_success(mock_discovery, api_client_confidential, httpx_mock): + """Test successful OBO exchange with fixed access-token types.""" + httpx_mock.add_response( + method="POST", + url=TOKEN_ENDPOINT, + json={ + "access_token": "obo-access-token", + "expires_in": 3600, + "scope": "read:data write:data", + "token_type": "Bearer", + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + } + ) + + result = await api_client_confidential.get_token_on_behalf_of( + access_token="incoming-access-token", + audience="https://api.backend.com", + scope="read:data write:data" + ) + + assert result["access_token"] == "obo-access-token" + assert result["expires_in"] == 3600 + assert isinstance(result["expires_at"], int) + assert result["scope"] == "read:data write:data" + assert result["token_type"] == "Bearer" + assert result["issued_token_type"] == "urn:ietf:params:oauth:token-type:access_token" + + assert_form_post( + httpx_mock, + expect_fields={ + "grant_type": ["urn:ietf:params:oauth:grant-type:token-exchange"], + "subject_token": ["incoming-access-token"], + "subject_token_type": ["urn:ietf:params:oauth:token-type:access_token"], + "requested_token_type": ["urn:ietf:params:oauth:token-type:access_token"], + "audience": ["https://api.backend.com"], + "scope": ["read:data write:data"], + }, + forbid_fields=["client_id", "client_secret"], + expect_basic_auth=("cid", "csecret") + ) + + +@pytest.mark.asyncio +async def test_get_token_on_behalf_of_without_scope(mock_discovery, api_client_confidential, httpx_mock): + """Test OBO exchange omits scope when not provided.""" + httpx_mock.add_response( + method="POST", + url=TOKEN_ENDPOINT, + json=token_success( + access_token="obo-access-token", + issued_token_type="urn:ietf:params:oauth:token-type:access_token", + token_type="Bearer", + ) + ) + + result = await api_client_confidential.get_token_on_behalf_of( + access_token="incoming-access-token", + audience="https://api.backend.com", + ) + + assert result["access_token"] == "obo-access-token" + assert "scope" not in last_form(httpx_mock) + + +@pytest.mark.asyncio +async def test_get_token_on_behalf_of_does_not_expose_id_or_refresh_token( + mock_discovery, api_client_confidential, httpx_mock +): + """Test OBO result only exposes access-token-oriented fields.""" + httpx_mock.add_response( + method="POST", + url=TOKEN_ENDPOINT, + json={ + "access_token": "obo-access-token", + "expires_in": 3600, + "token_type": "Bearer", + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + "id_token": "id-token", + "refresh_token": "refresh-token", + } + ) + + result = await api_client_confidential.get_token_on_behalf_of( + access_token="incoming-access-token", + audience="https://api.backend.com", + ) + + assert result["access_token"] == "obo-access-token" + assert "id_token" not in result + assert "refresh_token" not in result + + +@pytest.mark.asyncio +async def test_get_token_on_behalf_of_api_error(mock_discovery, api_client_confidential, httpx_mock): + """Test that OBO reuses the existing exchange error semantics.""" + httpx_mock.add_response( + method="POST", + url=TOKEN_ENDPOINT, + status_code=400, + json={ + "error": "invalid_target", + "error_description": "The target API is not allowed" + } + ) + + with pytest.raises(ApiError) as err: + await api_client_confidential.get_token_on_behalf_of( + access_token="incoming-access-token", + audience="https://api.backend.com", + ) + + assert err.value.code == "invalid_target" + assert err.value.status_code == 400 + + +@pytest.mark.asyncio +async def test_get_token_on_behalf_of_empty_access_token(api_client_confidential): + """Test that OBO validates the incoming access token via the shared exchange path.""" + with pytest.raises(MissingRequiredArgumentError): + await api_client_confidential.get_token_on_behalf_of( + access_token="", + audience="https://api.backend.com", + ) + + # ===== MCD (Multi-Custom Domain) Tests ===== @pytest.mark.asyncio @@ -4135,4 +4288,3 @@ def capturing_resolver(context): assert ctx["request_headers"]["x-custom-header"] == "test-value" assert ctx["unverified_iss"] == "https://tenant1.auth0.com/" -