From a16963a60f0c9f4cedeafeddc91e4b9127e09201 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Fri, 14 Aug 2026 14:29:59 +0530 Subject: [PATCH 1/2] feat: add On-Behalf-Of token exchange support --- EXAMPLES.md | 137 ++++++++++++++++++ README.md | 32 +++++ fastapi_plugin/__init__.py | 10 ++ tests/test_on_behalf_of.py | 286 +++++++++++++++++++++++++++++++++++++ 4 files changed, 465 insertions(+) create mode 100644 tests/test_on_behalf_of.py diff --git a/EXAMPLES.md b/EXAMPLES.md index c4955dc..645e5a1 100644 --- a/EXAMPLES.md +++ b/EXAMPLES.md @@ -16,6 +16,11 @@ - [Security Requirements](#security-requirements) - [DPoP with MCD](#dpop-with-mcd) - [Discovery Cache Configuration](#discovery-cache-configuration) +- [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) - [Protecting API Routes](#protecting-api-routes) ## Configuration @@ -344,6 +349,138 @@ auth0 = Auth0FastAPI( ) ``` +## 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 relies on Auth0 to handle any DPoP-specific behavior for that 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 + +Two error types cover the failure scenarios you will encounter, both re-exported from `fastapi_plugin`: + +- `GetTokenByExchangeProfileError`: Raised when `client_id` or `client_secret` is not configured on the plugin. This is a configuration error and will not be resolved at request time. +- `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). + +```python +from fastapi_plugin import ApiError, GetTokenByExchangeProfileError + +try: + obo = await auth0.api_client.get_token_on_behalf_of( + access_token=incoming_access_token, + audience="https://calendar-api.example.com", + ) +except GetTokenByExchangeProfileError: + # The plugin is not configured with client credentials. Fix the configuration. + 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 +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 PermissionError("unexpected actor") + + 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. + ## Protecting API Routes To protect a FastAPI route, use the `require_auth()` dependency. The SDK automatically detects and validates both Bearer and DPoP authentication schemes. 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..46909ea 100644 --- a/fastapi_plugin/__init__.py +++ b/fastapi_plugin/__init__.py @@ -1,20 +1,30 @@ from auth0_api_python import ( + ApiError, CacheAdapter, ConfigurationError, DomainsResolver, DomainsResolverContext, DomainsResolverError, + GetTokenByExchangeProfileError, InMemoryCache, + OnBehalfOfTokenResult, + get_current_actor, + get_delegation_chain, ) from .fast_api_client import Auth0FastAPI __all__ = [ + "ApiError", "Auth0FastAPI", "CacheAdapter", "ConfigurationError", "DomainsResolver", "DomainsResolverContext", "DomainsResolverError", + "GetTokenByExchangeProfileError", "InMemoryCache", + "OnBehalfOfTokenResult", + "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..5a4221b --- /dev/null +++ b/tests/test_on_behalf_of.py @@ -0,0 +1,286 @@ +""" +Tests for On-Behalf-Of (OBO) token exchange and the act-claim helpers exposed +through the FastAPI plugin. + +The exchange itself is performed by the underlying auth0-api-python ApiClient, +reached via auth0.api_client. These tests confirm the wrapper surfaces it and +re-exports the act helpers and result type. +""" +import base64 +import urllib.parse + +import pytest +from pytest_httpx import HTTPXMock + +from fastapi_plugin import ( + ApiError, + Auth0FastAPI, + GetTokenByExchangeProfileError, + OnBehalfOfTokenResult, + get_current_actor, + get_delegation_chain, +) + +DISCOVERY_URL = "https://auth0.local/.well-known/openid-configuration" +TOKEN_ENDPOINT = "https://auth0.local/oauth/token" + + +def _mock_discovery(httpx_mock: HTTPXMock): + httpx_mock.add_response( + method="GET", + url=DISCOVERY_URL, + json={"token_endpoint": TOKEN_ENDPOINT}, + ) + + +def _last_form(httpx_mock: HTTPXMock) -> dict[str, list[str]]: + req = httpx_mock.get_requests()[-1] + return urllib.parse.parse_qs(req.content.decode()) + + +def _confidential_client() -> Auth0FastAPI: + return Auth0FastAPI( + domain="auth0.local", + audience="my-audience", + client_id="cid", + client_secret="csecret", + ) + + +# ============================================================================= +# Exchange - configuration guards +# ============================================================================= + +@pytest.mark.asyncio +async def test_obo_requires_client_credentials(): + """OBO requires a confidential client configured on the plugin.""" + auth0 = Auth0FastAPI(domain="auth0.local", audience="my-audience") + + with pytest.raises(GetTokenByExchangeProfileError) as err: + await auth0.api_client.get_token_on_behalf_of( + access_token="incoming-access-token", + audience="https://api.backend.com", + ) + + assert "client credentials are required" in str(err.value).lower() + + +@pytest.mark.asyncio +async def test_obo_requires_client_secret(): + """OBO requires client_secret when only client_id is configured.""" + auth0 = Auth0FastAPI( + domain="auth0.local", + audience="my-audience", + client_id="cid", + ) + + with pytest.raises(GetTokenByExchangeProfileError) as err: + await auth0.api_client.get_token_on_behalf_of( + access_token="incoming-access-token", + audience="https://api.backend.com", + ) + + assert "client credentials are required" in str(err.value).lower() + + +@pytest.mark.asyncio +async def test_obo_requires_audience(): + """OBO requires an explicit downstream audience.""" + from auth0_api_python.errors import MissingRequiredArgumentError + + auth0 = _confidential_client() + + with pytest.raises(MissingRequiredArgumentError): + await auth0.api_client.get_token_on_behalf_of( + access_token="incoming-access-token", + audience="", + ) + + +# ============================================================================= +# Exchange - success path and request well-formedness +# ============================================================================= + +@pytest.mark.asyncio +async def test_obo_success_sends_fixed_token_types(httpx_mock: HTTPXMock): + """Successful OBO exchange sends the fixed RFC 8693 access-token types.""" + _mock_discovery(httpx_mock) + 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", + }, + ) + + auth0 = _confidential_client() + result = await auth0.api_client.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" + + form = _last_form(httpx_mock) + assert form["grant_type"] == ["urn:ietf:params:oauth:grant-type:token-exchange"] + assert form["subject_token"] == ["incoming-access-token"] + assert form["subject_token_type"] == ["urn:ietf:params:oauth:token-type:access_token"] + assert form["requested_token_type"] == ["urn:ietf:params:oauth:token-type:access_token"] + assert form["audience"] == ["https://api.backend.com"] + assert form["scope"] == ["read:data write:data"] + # Client credentials go via HTTP Basic auth, not the form body. + assert "client_id" not in form + assert "client_secret" not in form + + auth_header = httpx_mock.get_requests()[-1].headers.get("authorization") + assert auth_header is not None and auth_header.startswith("Basic ") + decoded = base64.b64decode(auth_header.split(" ")[1]).decode() + assert decoded == "cid:csecret" + + +@pytest.mark.asyncio +async def test_obo_omits_scope_when_not_provided(httpx_mock: HTTPXMock): + """OBO omits the scope field when no scope is requested.""" + _mock_discovery(httpx_mock) + 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", + }, + ) + + auth0 = _confidential_client() + result = await auth0.api_client.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_obo_does_not_expose_id_or_refresh_token(httpx_mock: HTTPXMock): + """OBO result only exposes access-token-oriented fields.""" + _mock_discovery(httpx_mock) + 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", + }, + ) + + auth0 = _confidential_client() + result = await auth0.api_client.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_obo_propagates_exchange_error(httpx_mock: HTTPXMock): + """OBO surfaces the underlying exchange error when Auth0 rejects it.""" + _mock_discovery(httpx_mock) + httpx_mock.add_response( + method="POST", + url=TOKEN_ENDPOINT, + status_code=400, + json={ + "error": "invalid_target", + "error_description": "The target API is not allowed", + }, + ) + + auth0 = _confidential_client() + with pytest.raises(ApiError) as err: + await auth0.api_client.get_token_on_behalf_of( + access_token="incoming-access-token", + audience="https://api.backend.com", + ) + + assert err.value.get_status_code() == 400 + + +# ============================================================================= +# Re-exports +# ============================================================================= + +def test_act_helpers_and_types_reexported(): + """The act helpers, OBO result type, and error types are re-exported from fastapi_plugin.""" + import auth0_api_python as dep + + 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 + + +# ============================================================================= +# act-claim helpers +# ============================================================================= + +def test_get_current_actor_and_chain_none_when_act_missing(): + """No act claim means no current actor and an empty delegation chain.""" + claims = {"sub": "auth0|user123"} + + assert get_current_actor(claims) is None + assert get_delegation_chain(claims) == [] + + +def test_get_current_actor_and_chain_from_nested_act(): + """Current actor is the outermost act.sub; chain runs newest to oldest.""" + claims = { + "sub": "auth0|user123", + "act": { + "sub": "mcp_server_2_client_id", + "act": { + "sub": "mcp_server_1_client_id", + "act": {"sub": "spa_client_id"}, + }, + }, + } + + assert get_current_actor(claims) == "mcp_server_2_client_id" + assert get_delegation_chain(claims) == [ + "mcp_server_2_client_id", + "mcp_server_1_client_id", + "spa_client_id", + ] + + +def test_act_helpers_reject_malformed_act_claim(): + """A present but malformed act claim raises VerifyAccessTokenError.""" + from auth0_api_python.errors import VerifyAccessTokenError + + with pytest.raises(VerifyAccessTokenError): + get_current_actor({"sub": "auth0|user123", "act": "not-an-object"}) + + with pytest.raises(VerifyAccessTokenError): + get_delegation_chain( + {"act": {"sub": "mcp_server_client_id", "act": "spa_client_id"}} + ) From 3ea1975d76116b8a94dc56ee68231d60aaf7d317 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Fri, 14 Aug 2026 15:40:35 +0530 Subject: [PATCH 2/2] fix: address OBO review feedback on docs, exports, and tests - Re-export MissingRequiredArgumentError, VerifyAccessTokenError, and BaseAuthError from fastapi_plugin so the documented OBO surface is catchable without reaching into auth0_api_python.errors - Correct the Error Handling docs: GetTokenByExchangeProfileError also fires at request time for a malformed token, and document the request-time cases - Fix the authorization example to raise HTTPException(403) instead of PermissionError, which escaped as a 500 - Clarify the DPoP note: the binding is not preserved on the exchanged token - Move the OBO section after Protecting API Routes so require_auth() is introduced first - Add a TestClient test for the require_auth -> exchange flow and drop the duplicates of upstream auth0-api-python tests --- EXAMPLES.md | 47 +++--- fastapi_plugin/__init__.py | 8 + tests/test_on_behalf_of.py | 303 ++++++++++--------------------------- 3 files changed, 112 insertions(+), 246 deletions(-) diff --git a/EXAMPLES.md b/EXAMPLES.md index 645e5a1..99d98eb 100644 --- a/EXAMPLES.md +++ b/EXAMPLES.md @@ -16,12 +16,12 @@ - [Security Requirements](#security-requirements) - [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) -- [Protecting API Routes](#protecting-api-routes) ## Configuration @@ -349,6 +349,19 @@ auth0 = Auth0FastAPI( ) ``` +## Protecting API Routes + +To protect a FastAPI route, use the `require_auth()` dependency. The SDK automatically detects and validates both Bearer and DPoP authentication schemes. + +```python +@app.get("/api/protected") +async def protected_route(claims=Depends(auth0.require_auth())): + return {"user_id": claims["sub"]} +``` + +> [!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. @@ -411,7 +424,7 @@ async def schedule_meeting(request: Request, claims=Depends(auth0.require_auth() > - 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 relies on Auth0 to handle any DPoP-specific behavior for that token. +> **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 @@ -426,21 +439,26 @@ On success, the method returns a dict containing: ### Error Handling -Two error types cover the failure scenarios you will encounter, both re-exported from `fastapi_plugin`: +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: -- `GetTokenByExchangeProfileError`: Raised when `client_id` or `client_secret` is not configured on the plugin. This is a configuration error and will not be resolved at request time. +- `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 +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: - # The plugin is not configured with client credentials. Fix the configuration. + # 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. @@ -452,7 +470,7 @@ except ApiError as err: 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 +from fastapi import Depends, FastAPI, HTTPException from fastapi_plugin import Auth0FastAPI, get_current_actor, get_delegation_chain app = FastAPI() @@ -470,7 +488,7 @@ async def list_meetings(claims=Depends(auth0.require_auth())): delegation_chain = get_delegation_chain(claims) if current_actor not in ALLOWED_ACTORS: - raise PermissionError("unexpected actor") + raise HTTPException(status_code=403, detail={"error": "insufficient_permissions"}) return { "user_sub": claims["sub"], @@ -480,16 +498,3 @@ async def list_meetings(claims=Depends(auth0.require_auth())): ``` 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. - -## Protecting API Routes - -To protect a FastAPI route, use the `require_auth()` dependency. The SDK automatically detects and validates both Bearer and DPoP authentication schemes. - -```python -@app.get("/api/protected") -async def protected_route(claims=Depends(auth0.require_auth())): - return {"user_id": claims["sub"]} -``` - -> [!IMPORTANT] -> The above is to protect API routes by the means of a bearer token, and not server-side rendering routes using a session. diff --git a/fastapi_plugin/__init__.py b/fastapi_plugin/__init__.py index 46909ea..315413b 100644 --- a/fastapi_plugin/__init__.py +++ b/fastapi_plugin/__init__.py @@ -11,12 +11,18 @@ 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", @@ -24,7 +30,9 @@ "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 index 5a4221b..0b7fb11 100644 --- a/tests/test_on_behalf_of.py +++ b/tests/test_on_behalf_of.py @@ -1,286 +1,139 @@ """ -Tests for On-Behalf-Of (OBO) token exchange and the act-claim helpers exposed -through the FastAPI plugin. +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 itself is performed by the underlying auth0-api-python ApiClient, -reached via auth0.api_client. These tests confirm the wrapper surfaces it and -re-exports the act helpers and result type. +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, ) -DISCOVERY_URL = "https://auth0.local/.well-known/openid-configuration" +from .test_utils import generate_token + TOKEN_ENDPOINT = "https://auth0.local/oauth/token" -def _mock_discovery(httpx_mock: HTTPXMock): +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=DISCOVERY_URL, - json={"token_endpoint": TOKEN_ENDPOINT}, - ) - - -def _last_form(httpx_mock: HTTPXMock) -> dict[str, list[str]]: - req = httpx_mock.get_requests()[-1] - return urllib.parse.parse_qs(req.content.decode()) - - -def _confidential_client() -> Auth0FastAPI: - return Auth0FastAPI( - domain="auth0.local", - audience="my-audience", - client_id="cid", - client_secret="csecret", - ) - - -# ============================================================================= -# Exchange - configuration guards -# ============================================================================= - -@pytest.mark.asyncio -async def test_obo_requires_client_credentials(): - """OBO requires a confidential client configured on the plugin.""" - auth0 = Auth0FastAPI(domain="auth0.local", audience="my-audience") - - with pytest.raises(GetTokenByExchangeProfileError) as err: - await auth0.api_client.get_token_on_behalf_of( - access_token="incoming-access-token", - audience="https://api.backend.com", - ) - - assert "client credentials are required" in str(err.value).lower() - - -@pytest.mark.asyncio -async def test_obo_requires_client_secret(): - """OBO requires client_secret when only client_id is configured.""" - auth0 = Auth0FastAPI( - domain="auth0.local", - audience="my-audience", - client_id="cid", - ) - - with pytest.raises(GetTokenByExchangeProfileError) as err: - await auth0.api_client.get_token_on_behalf_of( - access_token="incoming-access-token", - audience="https://api.backend.com", - ) - - assert "client credentials are required" in str(err.value).lower() - - -@pytest.mark.asyncio -async def test_obo_requires_audience(): - """OBO requires an explicit downstream audience.""" - from auth0_api_python.errors import MissingRequiredArgumentError - - auth0 = _confidential_client() - - with pytest.raises(MissingRequiredArgumentError): - await auth0.api_client.get_token_on_behalf_of( - access_token="incoming-access-token", - audience="", - ) - - -# ============================================================================= -# Exchange - success path and request well-formedness -# ============================================================================= - -@pytest.mark.asyncio -async def test_obo_success_sends_fixed_token_types(httpx_mock: HTTPXMock): - """Successful OBO exchange sends the fixed RFC 8693 access-token types.""" - _mock_discovery(httpx_mock) - httpx_mock.add_response( - method="POST", - url=TOKEN_ENDPOINT, + url="https://auth0.local/.well-known/openid-configuration", 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", + "issuer": "https://auth0.local/", + "jwks_uri": "https://auth0.local/.well-known/jwks.json", + "token_endpoint": TOKEN_ENDPOINT, }, ) - - auth0 = _confidential_client() - result = await auth0.api_client.get_token_on_behalf_of( - access_token="incoming-access-token", - audience="https://api.backend.com", - scope="read:data write:data", + 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]}, ) - - 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" - - form = _last_form(httpx_mock) - assert form["grant_type"] == ["urn:ietf:params:oauth:grant-type:token-exchange"] - assert form["subject_token"] == ["incoming-access-token"] - assert form["subject_token_type"] == ["urn:ietf:params:oauth:token-type:access_token"] - assert form["requested_token_type"] == ["urn:ietf:params:oauth:token-type:access_token"] - assert form["audience"] == ["https://api.backend.com"] - assert form["scope"] == ["read:data write:data"] - # Client credentials go via HTTP Basic auth, not the form body. - assert "client_id" not in form - assert "client_secret" not in form - - auth_header = httpx_mock.get_requests()[-1].headers.get("authorization") - assert auth_header is not None and auth_header.startswith("Basic ") - decoded = base64.b64decode(auth_header.split(" ")[1]).decode() - assert decoded == "cid:csecret" - - -@pytest.mark.asyncio -async def test_obo_omits_scope_when_not_provided(httpx_mock: HTTPXMock): - """OBO omits the scope field when no scope is requested.""" - _mock_discovery(httpx_mock) 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", }, ) - auth0 = _confidential_client() - result = await auth0.api_client.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) +# ============================================================================= +# The owned flow: require_auth() -> pull verified token -> exchange, via a route +# ============================================================================= @pytest.mark.asyncio -async def test_obo_does_not_expose_id_or_refresh_token(httpx_mock: HTTPXMock): - """OBO result only exposes access-token-oriented fields.""" - _mock_discovery(httpx_mock) - 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", - }, - ) +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) - auth0 = _confidential_client() - result = await auth0.api_client.get_token_on_behalf_of( - access_token="incoming-access-token", - audience="https://api.backend.com", + access_token = await generate_token( + domain="auth0.local", + user_id="user_123", + audience="my-audience", + issuer="https://auth0.local/", + iat=True, + exp=True, ) - assert result["access_token"] == "obo-access-token" - assert "id_token" not in result - assert "refresh_token" not in result + 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"]} -@pytest.mark.asyncio -async def test_obo_propagates_exchange_error(httpx_mock: HTTPXMock): - """OBO surfaces the underlying exchange error when Auth0 rejects it.""" - _mock_discovery(httpx_mock) - httpx_mock.add_response( - method="POST", - url=TOKEN_ENDPOINT, - status_code=400, - json={ - "error": "invalid_target", - "error_description": "The target API is not allowed", - }, + client = TestClient(app) + response = client.post( + "/schedule-meeting", + headers={"Authorization": f"Bearer {access_token}"}, ) - auth0 = _confidential_client() - with pytest.raises(ApiError) as err: - await auth0.api_client.get_token_on_behalf_of( - access_token="incoming-access-token", - audience="https://api.backend.com", - ) + assert response.status_code == 200 + assert response.json() == {"user": "user_123", "downstream_token": "obo-access-token"} - assert err.value.get_status_code() == 400 + 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 +# Re-exports the plugin adds on top of auth0-api-python # ============================================================================= -def test_act_helpers_and_types_reexported(): - """The act helpers, OBO result type, and error types are re-exported from fastapi_plugin.""" +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 - - -# ============================================================================= -# act-claim helpers -# ============================================================================= - -def test_get_current_actor_and_chain_none_when_act_missing(): - """No act claim means no current actor and an empty delegation chain.""" - claims = {"sub": "auth0|user123"} - - assert get_current_actor(claims) is None - assert get_delegation_chain(claims) == [] - - -def test_get_current_actor_and_chain_from_nested_act(): - """Current actor is the outermost act.sub; chain runs newest to oldest.""" - claims = { - "sub": "auth0|user123", - "act": { - "sub": "mcp_server_2_client_id", - "act": { - "sub": "mcp_server_1_client_id", - "act": {"sub": "spa_client_id"}, - }, - }, - } - - assert get_current_actor(claims) == "mcp_server_2_client_id" - assert get_delegation_chain(claims) == [ - "mcp_server_2_client_id", - "mcp_server_1_client_id", - "spa_client_id", - ] - - -def test_act_helpers_reject_malformed_act_claim(): - """A present but malformed act claim raises VerifyAccessTokenError.""" - from auth0_api_python.errors import VerifyAccessTokenError - - with pytest.raises(VerifyAccessTokenError): - get_current_actor({"sub": "auth0|user123", "act": "not-an-object"}) - - with pytest.raises(VerifyAccessTokenError): - get_delegation_chain( - {"act": {"sub": "mcp_server_client_id", "act": "spa_client_id"}} - ) + # 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)