diff --git a/EXAMPLES.md b/EXAMPLES.md index c4955dc..99d98eb 100644 --- a/EXAMPLES.md +++ b/EXAMPLES.md @@ -17,6 +17,11 @@ - [DPoP with MCD](#dpop-with-mcd) - [Discovery Cache Configuration](#discovery-cache-configuration) - [Protecting API Routes](#protecting-api-routes) +- [On-Behalf-Of Token Exchange](#on-behalf-of-token-exchange) + - [Performing the Exchange](#performing-the-exchange) + - [`get_token_on_behalf_of()` Return Value](#get_token_on_behalf_of-return-value) + - [Error Handling](#error-handling) + - [Inspecting Delegation After Token Verification](#inspecting-delegation-after-token-verification) ## Configuration @@ -356,3 +361,140 @@ async def protected_route(claims=Depends(auth0.require_auth())): > [!IMPORTANT] > The above is to protect API routes by the means of a bearer token, and not server-side rendering routes using a session. + +## On-Behalf-Of Token Exchange + +Use `get_token_on_behalf_of()` on the underlying `api_client` 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 flow has three steps: + +1. **Verify** the incoming access token so your API rejects invalid or mis-targeted tokens before exchanging. `require_auth()` does this for the route. +2. **Exchange** the verified token for a new access token scoped to the downstream API. +3. **Call** the downstream API using the exchanged token. + +`get_token_on_behalf_of()` requires a confidential client. Configure the plugin with `client_id` and `client_secret`. Calling it without client credentials raises `GetTokenByExchangeProfileError`. + +### Performing the Exchange + +Inside a protected route, extract the raw incoming token, exchange it for a downstream audience, then call the downstream API: + +```python +import httpx +from fastapi import Depends, FastAPI, Request +from fastapi_plugin import Auth0FastAPI + +app = FastAPI() + +auth0 = Auth0FastAPI( + domain="", # your MCP server's Auth0 tenant domain + audience="", # your MCP server's API audience + client_id="", # required for OBO + client_secret="", # required for OBO +) + +@app.post("/schedule-meeting") +async def schedule_meeting(request: Request, claims=Depends(auth0.require_auth())): + # require_auth() already verified the incoming token. Pass the raw token + # to the exchange, without the "Bearer " prefix. + incoming_access_token = request.headers["authorization"].split(" ", 1)[1] + + obo = await auth0.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: + response = await client.post( + "https://calendar-api.example.com/meetings", + headers={"Authorization": f"Bearer {obo['access_token']}"}, + json=await request.json(), + ) + response.raise_for_status() + + return {"user": claims["sub"], "meeting": response.json()} +``` + +> [!TIP] +> **Production notes:** +> - `require_auth()` verifies the incoming token before your handler runs. Always protect routes with `require_auth()` before calling `get_token_on_behalf_of()`. +> - Pass the raw JWT to `get_token_on_behalf_of()`. Do not include the `Bearer ` prefix or the full `Authorization` header. +> - The downstream `audience` must match an API identifier configured in your Auth0 tenant, and your client must be authorized to access it. +> - `get_token_on_behalf_of()` only returns access-token-oriented fields. It does not expose `id_token` or `refresh_token`. +> - OBO requires a **confidential client**. Calling it without client credentials raises `GetTokenByExchangeProfileError`. + +> [!NOTE] +> **DPoP:** `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 does not carry a DPoP proof or request a DPoP-bound result. The exchanged token comes back as a plain bearer token, so the downstream call uses `Authorization: Bearer`. If the incoming token was DPoP-bound, that binding is not preserved on the exchanged token. + +### `get_token_on_behalf_of()` Return Value + +On success, the method returns a dict containing: + +- `access_token`: The exchanged access token issued for the downstream API. +- `expires_in`: Token lifetime in seconds. +- `expires_at`: The access token expiration time, in seconds since the Unix epoch. +- `scope`: The scope granted for the exchanged token, if returned. +- `token_type`: The returned token type, if returned. +- `issued_token_type`: The returned RFC 8693 issued token type, if returned. + +### Error Handling + +These error types cover the failure scenarios you will encounter. All are re-exported from `fastapi_plugin`, and all subclass `BaseAuthError`, so you can catch that single type if you want one handler for everything: + +- `MissingRequiredArgumentError`: Raised when `audience` or `access_token` is empty. +- `GetTokenByExchangeProfileError`: Raised for a missing confidential client (no `client_id` or `client_secret`), and also at request time when the incoming token is malformed. This includes a token carrying the `Bearer ` prefix, a blank or whitespace-only token, a token with leading or trailing whitespace, and a `token_endpoint` missing from OIDC discovery. Pass the raw JWT with no `Bearer ` prefix to avoid the token-format cases. +- `ApiError`: Raised when Auth0 rejects the exchange. The error preserves the OAuth error code and description from Auth0 (for example, `invalid_target` when the client is not authorized to access the downstream API). +- `VerifyAccessTokenError`: Raised by `get_current_actor()` and `get_delegation_chain()` when the token's `act` claim is present but malformed. + +```python +from fastapi_plugin import ApiError, GetTokenByExchangeProfileError, MissingRequiredArgumentError + +try: + obo = await auth0.api_client.get_token_on_behalf_of( + access_token=incoming_access_token, + audience="https://calendar-api.example.com", + ) +except MissingRequiredArgumentError: + # audience or access_token was empty. + raise +except GetTokenByExchangeProfileError: + # Missing client credentials, or a malformed incoming token (for example the "Bearer " prefix). + raise +except ApiError as err: + # Auth0 rejected the exchange. err.get_error_code() carries the OAuth error code. + raise +``` + +### Inspecting Delegation After Token Verification + +When a downstream API receives an exchanged token, 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. The plugin re-exports `get_current_actor` and `get_delegation_chain` for this. + +```python +from fastapi import Depends, FastAPI, HTTPException +from fastapi_plugin import Auth0FastAPI, get_current_actor, get_delegation_chain + +app = FastAPI() + +auth0 = Auth0FastAPI( + domain="", + audience="https://calendar-api.example.com", +) + +ALLOWED_ACTORS = [""] + +@app.get("/meetings") +async def list_meetings(claims=Depends(auth0.require_auth())): + current_actor = get_current_actor(claims) + delegation_chain = get_delegation_chain(claims) + + if current_actor not in ALLOWED_ACTORS: + raise HTTPException(status_code=403, detail={"error": "insufficient_permissions"}) + + return { + "user_sub": claims["sub"], + "current_actor": current_actor, + "delegation_chain": delegation_chain, + } +``` + +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. See [RFC 8693, section 4.1](https://datatracker.ietf.org/doc/html/rfc8693#section-4.1) for details. diff --git a/README.md b/README.md index f92ce5d..1c5a282 100644 --- a/README.md +++ b/README.md @@ -327,6 +327,38 @@ When both `domain` and `domains` are configured, the SDK uses `domains` exclusiv For detailed examples including dynamic resolvers, cache configuration, security requirements, and DPoP integration, see the [Multiple Custom Domains section in EXAMPLES.md](EXAMPLES.md#multiple-custom-domains-mcd). +### 8. On-Behalf-Of Token Exchange + +If your API receives an Auth0 access token for itself and needs to call a downstream API on behalf of the same user (for example, an MCP server), use `get_token_on_behalf_of` on the underlying `api_client` to exchange the incoming token for one scoped to the downstream API. This requires a confidential client (`client_id` and `client_secret`). + +```python +import asyncio + +from fastapi_plugin import Auth0FastAPI + +async def main(): + auth0 = Auth0FastAPI( + domain="", + audience="", + client_id="", + client_secret="", + ) + incoming_access_token = "..." # the verified Auth0 access token to exchange + + result = await auth0.api_client.get_token_on_behalf_of( + access_token=incoming_access_token, + audience="https://calendar-api.example.com", + scope="calendar:read calendar:write", + ) + print(result["access_token"]) # short-lived token for the downstream API + +asyncio.run(main()) +``` + +A downstream API can inspect the delegation on a verified token with the re-exported `get_current_actor` and `get_delegation_chain` helpers. Only the outermost `act.sub` should be used for authorization decisions. + +For the full flow, production notes, and delegation inspection, see the [On-Behalf-Of Token Exchange section in EXAMPLES.md](EXAMPLES.md#on-behalf-of-token-exchange). + ## Feedback ### Contributing diff --git a/fastapi_plugin/__init__.py b/fastapi_plugin/__init__.py index 0a1b85c..315413b 100644 --- a/fastapi_plugin/__init__.py +++ b/fastapi_plugin/__init__.py @@ -1,20 +1,38 @@ from auth0_api_python import ( + ApiError, CacheAdapter, ConfigurationError, DomainsResolver, DomainsResolverContext, DomainsResolverError, + GetTokenByExchangeProfileError, InMemoryCache, + OnBehalfOfTokenResult, + get_current_actor, + get_delegation_chain, +) +from auth0_api_python.errors import ( + BaseAuthError, + MissingRequiredArgumentError, + VerifyAccessTokenError, ) from .fast_api_client import Auth0FastAPI __all__ = [ + "ApiError", "Auth0FastAPI", + "BaseAuthError", "CacheAdapter", "ConfigurationError", "DomainsResolver", "DomainsResolverContext", "DomainsResolverError", + "GetTokenByExchangeProfileError", "InMemoryCache", + "MissingRequiredArgumentError", + "OnBehalfOfTokenResult", + "VerifyAccessTokenError", + "get_current_actor", + "get_delegation_chain", ] diff --git a/tests/test_on_behalf_of.py b/tests/test_on_behalf_of.py new file mode 100644 index 0000000..0b7fb11 --- /dev/null +++ b/tests/test_on_behalf_of.py @@ -0,0 +1,139 @@ +""" +Tests for the On-Behalf-Of (OBO) surface this plugin owns: the require_auth() -> +pull the verified token -> exchange flow through a real FastAPI route, and the +re-exports the plugin adds on top of auth0-api-python. + +The exchange, the form well-formedness, the Basic auth encoding, and the act-claim +parsing are exercised in auth0-api-python's own suite (test_api_client.py, test_act.py), +so they are not repeated here. +""" +import base64 +import urllib.parse + +import pytest +from fastapi import Depends, FastAPI, Request +from fastapi.testclient import TestClient +from pytest_httpx import HTTPXMock + +from fastapi_plugin import ( + ApiError, + Auth0FastAPI, + BaseAuthError, + GetTokenByExchangeProfileError, + MissingRequiredArgumentError, + OnBehalfOfTokenResult, + VerifyAccessTokenError, + get_current_actor, + get_delegation_chain, +) + +from .test_utils import generate_token + +TOKEN_ENDPOINT = "https://auth0.local/oauth/token" + + +def _setup_obo_mocks(httpx_mock: HTTPXMock): + """OIDC discovery (with a token_endpoint), JWKS, and the token-exchange response.""" + httpx_mock.add_response( + method="GET", + url="https://auth0.local/.well-known/openid-configuration", + json={ + "issuer": "https://auth0.local/", + "jwks_uri": "https://auth0.local/.well-known/jwks.json", + "token_endpoint": TOKEN_ENDPOINT, + }, + ) + from .conftest import PUBLIC_DPOP_JWK, RSA_PUBLIC_KEY + httpx_mock.add_response( + method="GET", + url="https://auth0.local/.well-known/jwks.json", + json={"keys": [RSA_PUBLIC_KEY, PUBLIC_DPOP_JWK]}, + ) + httpx_mock.add_response( + method="POST", + url=TOKEN_ENDPOINT, + json={ + "access_token": "obo-access-token", + "expires_in": 3600, + "scope": "calendar:read", + "token_type": "Bearer", + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + }, + ) + + +# ============================================================================= +# The owned flow: require_auth() -> pull verified token -> exchange, via a route +# ============================================================================= + +@pytest.mark.asyncio +async def test_documented_route_verifies_then_exchanges(httpx_mock: HTTPXMock): + """A protected route pulls the verified token and exchanges it on behalf of the user.""" + _setup_obo_mocks(httpx_mock) + + access_token = await generate_token( + domain="auth0.local", + user_id="user_123", + audience="my-audience", + issuer="https://auth0.local/", + iat=True, + exp=True, + ) + + app = FastAPI() + auth0 = Auth0FastAPI( + domain="auth0.local", + audience="my-audience", + client_id="cid", + client_secret="csecret", + ) + + @app.post("/schedule-meeting") + async def schedule_meeting(request: Request, claims=Depends(auth0.require_auth())): + incoming = request.headers["authorization"].split(" ", 1)[1] + obo = await auth0.api_client.get_token_on_behalf_of( + access_token=incoming, + audience="https://calendar-api.example.com", + scope="calendar:read", + ) + return {"user": claims["sub"], "downstream_token": obo["access_token"]} + + client = TestClient(app) + response = client.post( + "/schedule-meeting", + headers={"Authorization": f"Bearer {access_token}"}, + ) + + assert response.status_code == 200 + assert response.json() == {"user": "user_123", "downstream_token": "obo-access-token"} + + exchange_req = httpx_mock.get_requests(method="POST", url=TOKEN_ENDPOINT)[-1] + form = urllib.parse.parse_qs(exchange_req.content.decode()) + assert form["subject_token"] == [access_token] + assert form["audience"] == ["https://calendar-api.example.com"] + assert "client_secret" not in form + auth_header = exchange_req.headers.get("authorization") + assert auth_header.startswith("Basic ") + assert base64.b64decode(auth_header.split(" ")[1]).decode() == "cid:csecret" + + +# ============================================================================= +# Re-exports the plugin adds on top of auth0-api-python +# ============================================================================= + +def test_obo_surface_reexported(): + """The OBO method's helpers, result type, and error types are re-exported from the plugin.""" + import auth0_api_python as dep + import auth0_api_python.errors as errors + + assert get_current_actor is dep.get_current_actor + assert get_delegation_chain is dep.get_delegation_chain + assert OnBehalfOfTokenResult is dep.OnBehalfOfTokenResult + assert GetTokenByExchangeProfileError is dep.GetTokenByExchangeProfileError + assert ApiError is dep.ApiError + # These three are only under auth0_api_python.errors upstream, not at its top level. + assert MissingRequiredArgumentError is errors.MissingRequiredArgumentError + assert VerifyAccessTokenError is errors.VerifyAccessTokenError + assert BaseAuthError is errors.BaseAuthError + for err in (GetTokenByExchangeProfileError, ApiError, MissingRequiredArgumentError, VerifyAccessTokenError): + assert issubclass(err, BaseAuthError)