From afa2097bba977069113bba86955add1486ce8dbf Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Tue, 21 Jul 2026 14:57:08 +0530 Subject: [PATCH 1/3] feat: add Session Transfer Token (STT) support for CTE impersonation via session transfer Adds the initiator surface for Impersonation via Session Transfer on top of Custom Token Exchange: - request_session_transfer_token: mints an STT against the urn:{domain}:session_transfer audience. The actor is auto-sourced from the agent session's ID token (verified via JWKS, refreshed when expired) and overridable via an explicit actor_token; a blank actor_token is rejected, and no usable actor fails client-side with ACTOR_UNAVAILABLE before any network call. - build_session_transfer_redirect: builds the URL-encoded redirect that hands the STT to the target app's login URL (organization forwarded when present). - SessionTransferTokenResult model and the ACTOR_UNAVAILABLE / SETACTOR_REQUIRED / SESSION_TRANSFER_DISABLED error codes. - Developer guide (examples/CustomTokenExchange.md) + README pointer, and a Session Transfer Token test section covering actor resolution, the refresh path, stateless minting, error cases, and the redirect builder. Target-side redemption needs no new SDK code: start_interactive_login already forwards session_transfer_token to /authorize. --- README.md | 5 + examples/CustomTokenExchange.md | 86 +++++++ .../auth_server/server_client.py | 172 +++++++++++++ .../auth_types/__init__.py | 18 ++ src/auth0_server_python/error/__init__.py | 3 + .../tests/test_server_client.py | 235 +++++++++++++++++- 6 files changed, 516 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 55cce7e..93df38f 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,11 @@ response = await auth0.custom_token_exchange( print(response.access_token) ``` +Building on token exchange, the SDK also supports: + +- **[Delegation and Impersonation](examples/CustomTokenExchange.md#3-actor-tokens-delegation)** - exchange with an `actor_token` so the issued tokens record who is acting on whose behalf (the `act` claim). +- **[Impersonation via Session Transfer (STT)](examples/CustomTokenExchange.md#8-impersonation-via-session-transfer-stt)** - mint a Session Transfer Token to log an agent into a target app as a customer, via `request_session_transfer_token()` and `build_session_transfer_redirect()`. + For more details and examples, see [examples/CustomTokenExchange.md](examples/CustomTokenExchange.md). ### 5. Multiple Custom Domains (MCD) diff --git a/examples/CustomTokenExchange.md b/examples/CustomTokenExchange.md index 62cb817..d0589bd 100644 --- a/examples/CustomTokenExchange.md +++ b/examples/CustomTokenExchange.md @@ -179,6 +179,92 @@ Use standard URNs when possible: "urn:company:legacy-token" ``` +## 8. Impersonation via Session Transfer (STT) + +Custom Token Exchange can also mint a **Session Transfer Token (STT)** instead of an API access token. An STT lets an initiator app (for example a support console) log an agent into a target web app **as** a customer, with the agent recorded in the `act` claim - so a support engineer can reproduce a customer's exact experience without their password. + +This is a two-role, two-hop flow: + +- **Initiator** (the agent's app) mints the STT and redirects with it. This is where the new SDK methods live. +- **Target** (the customer's app) forwards the STT to `/authorize` on a normal interactive login, which establishes the impersonated session. + +The STT is opaque, single-use, and short-lived (~60s). The SDK requests it and helps build the redirect - it never decodes or stores it. + +### Initiator: request an STT and build the redirect + +```python +from auth0_server_python.auth_server.server_client import ServerClient +from auth0_server_python.error import CustomTokenExchangeError + +# Mint the STT. The audience (urn:{domain}:session_transfer), grant type, and the actor are +# set by the SDK - the actor is sourced from the logged-in agent's session. +result = await auth0.request_session_transfer_token( + subject_token=subject_token, # your proof of which customer to impersonate + subject_token_type="urn:acme:customer-subject", + organization=None, # optional; forwarded to the redirect + store_options={"request": request, "response": None}, +) + +# result.session_transfer_token is the opaque, one-shot STT (~60s). Never store it. +redirect_url = auth0.build_session_transfer_redirect( + "https://customer-app.example.com/auth/login", result, organization=None +) +return RedirectResponse(redirect_url) # your framework performs the redirect +``` + +`SessionTransferTokenResult` carries `session_transfer_token`, `issued_token_type` (the session-transfer URN - the field to branch on), `expires_in`, and an informational `token_type` (`N_A`). There is no `act` on this result; `act` appears later, on the target session. + +> **NOTE**: An actor is mandatory - an STT is only issued when the Action set one. By default the SDK sources the actor from the logged-in agent's session ID token, refreshing it when expired. If the agent is not logged in (no usable session ID token and none can be refreshed), the call fails client-side with `ACTOR_UNAVAILABLE` before any network request. + +> **NOTE**: To use your own actor token instead of the session, pass `actor_token` (and optionally `actor_token_type`, which defaults to the ID token URN). An explicit `actor_token` takes precedence and the session is not read at all. It must be an **unexpired, asymmetrically-signed JWT** (RS256 or PS256) - an Auth0 session ID token satisfies this; an HS256 or expired token is rejected by the server. +> +> ```python +> result = await auth0.request_session_transfer_token( +> subject_token=subject_token, +> subject_token_type="urn:acme:customer-subject", +> actor_token=agent_id_token, # explicit override - session is not used +> store_options={"request": request, "response": None}, +> ) +> ``` + +### Target: forward the STT to `/authorize` + +On the target, the STT rides through your normal login. `start_interactive_login` forwards arbitrary authorization parameters to `/authorize`, so your login route just passes `session_transfer_token` (and `organization`, when the STT was issued in an org context) straight through: + +```python +from auth0_server_python.auth_types import StartInteractiveLoginOptions + +url = await auth0.start_interactive_login( + StartInteractiveLoginOptions(authorization_params={ + "session_transfer_token": request.query_params["session_transfer_token"], + # "organization": org, # when the STT was issued in an org context + }), + store_options={"request": request, "response": None}, +) +return RedirectResponse(url) +``` + +After the callback completes, read the acting party off the session user - the same way as the [Actor Tokens (Delegation)](#3-actor-tokens-delegation) section above: + +```python +session = await auth0.get_session(store_options={"request": request, "response": None}) +act = (session or {}).get("user", {}).get("act") +if act: + print(f"Impersonated by: {act['sub']}") # drive an impersonation banner, etc. +``` + +> **NOTE**: Both clients need one-time configuration through the Auth0 Dashboard or Management API. The issuing (initiator) client must be allowed to create session transfer tokens. The redeeming (target) client must be allowed to accept delegated-access sessions and to receive the token as a query parameter. See the [Auth0 documentation](https://auth0.com/docs/authenticate/custom-token-exchange) for the exact client settings. + +> **NOTE**: `build_session_transfer_redirect` attaches a single-use credential to `target_login_url`, so that URL must be a trusted, app-controlled value - never one derived from untrusted input (such as a user-supplied `returnTo`), which could leak the token to an attacker host. + +> **NOTE**: The impersonation session is hard-capped at 2 hours and cannot mint a refresh token (`offline_access` is dropped when an actor is present). To continue past that, re-run the flow. + +### STT error codes + +- `ACTOR_UNAVAILABLE`: no usable actor token (client-side; raised before any network call) +- `SETACTOR_REQUIRED`: an STT was requested but the Action did not call `setActor` (server 400) +- `SESSION_TRANSFER_DISABLED`: the session-transfer feature is off for the tenant/client (server 400) + ## Additional Resources - [Auth0 Custom Token Exchange Documentation](https://auth0.com/docs/authenticate/custom-token-exchange) diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index 13b10b2..bea7bfb 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -41,6 +41,7 @@ PasskeySignupChallengeResponse, PasskeyTokenResponse, PasskeyUserProfile, + SessionTransferTokenResult, StartInteractiveLoginOptions, StateData, TokenExchangeResponse, @@ -86,6 +87,12 @@ INTERNAL_AUTHORIZE_PARAMS = ["client_id", "response_type", "code_challenge", "code_challenge_method", "state", "nonce", "scope"] +# issued_token_type URN for a Session Transfer Token (STT). +SESSION_TRANSFER_TOKEN_TYPE = "urn:auth0:params:oauth:token-type:session_transfer_token" + +# actor_token_type URN when the actor is sourced from the agent session's ID token. +ID_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:id_token" + class ServerClient(Generic[TStoreOptions]): """ @@ -2635,6 +2642,171 @@ async def login_with_custom_token_exchange( e ) + # ============================================================================ + # SESSION TRANSFER TOKEN (STT) + # Impersonation via Session Transfer, built on Custom Token Exchange. + # ============================================================================ + + async def _is_id_token_usable(self, token: str, store_options: Optional[dict[str, Any]]) -> bool: + """ + Verifies the agent session's ID token (signature + expiry) before using it as an actor. + + Full verification against JWKS - the same path the login callback uses - so an expired + or tampered token is rejected client-side rather than sent to the server as a dud. + """ + if not token: + return False + try: + domain = await self._resolve_current_domain(store_options) + metadata = await self._get_oidc_metadata_cached(domain) + jwks = await self._get_jwks_cached(domain, metadata) + await self._verify_and_decode_jwt(token, jwks, audience=self._client_id) + return True + except Exception: + return False + + async def _resolve_actor_token( + self, + actor_token: Optional[str], + actor_token_type: Optional[str], + store_options: Optional[dict[str, Any]] + ) -> tuple[str, str]: + """ + Resolves the (actor_token, actor_token_type) pair for a session transfer request. + + Raises: + CustomTokenExchangeError(ACTOR_UNAVAILABLE): if no usable actor can be resolved. + """ + # Explicit actor wins; a passed-but-blank value is a bug, not a fallback signal. + if actor_token is not None: + if not actor_token.strip(): + raise CustomTokenExchangeError( + CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT, + "actor_token cannot be empty or whitespace-only" + ) + return actor_token, (actor_token_type or ID_TOKEN_TYPE) + + # Otherwise source it from the agent's session ID token. + state_data = await self._state_store.get(self._state_identifier, store_options) + if state_data and hasattr(state_data, "dict") and callable(state_data.dict): + state_data = state_data.dict() + state_data = state_data or {} + + session_id_token = state_data.get("id_token") + + # Refresh a stale (or missing) ID token when the agent session has a refresh token. + if not await self._is_id_token_usable(session_id_token, store_options) and state_data.get("refresh_token"): + try: + refreshed = await self.get_token_by_refresh_token({ + "refresh_token": state_data["refresh_token"], + "domain": state_data.get("domain") or self._domain, + }) + updated_state_data = State.update_state_data( + self.DEFAULT_AUDIENCE_STATE_KEY, state_data, refreshed) + await self._state_store.set(self._state_identifier, updated_state_data, options=store_options) + session_id_token = refreshed.get("id_token") or session_id_token + except Exception: + session_id_token = None + + if await self._is_id_token_usable(session_id_token, store_options): + return session_id_token, ID_TOKEN_TYPE + + raise CustomTokenExchangeError( + CustomTokenExchangeErrorCode.ACTOR_UNAVAILABLE, + "No usable actor token: pass actor_token or ensure the agent has a valid session." + ) + + async def request_session_transfer_token( + self, + subject_token: str, + subject_token_type: str, + actor_token: Optional[str] = None, + actor_token_type: Optional[str] = None, + scope: Optional[str] = None, + organization: Optional[str] = None, + store_options: Optional[dict[str, Any]] = None + ) -> SessionTransferTokenResult: + """ + Requests a Session Transfer Token (STT) for impersonation via session transfer. + + Performs a custom token exchange against the session_transfer audience. The returned + STT is opaque and single-use; hand it to build_session_transfer_redirect and do not + decode or store it. The act claim is not on this result. + + Args: + subject_token: Your proof of which customer to impersonate (validated by your Action) + subject_token_type: The subject token type URI routing to your CTE Profile + actor_token: The acting party's token; optional. Defaults to the agent session's ID token + actor_token_type: Type URI of the actor token; defaults to the ID token URN + scope: Space-delimited list of scopes (optional) + organization: Organization identifier (optional) + store_options: Optional options used to read the agent session and resolve the domain + + Returns: + SessionTransferTokenResult containing the STT and its metadata + + Raises: + CustomTokenExchangeError: If no actor can be resolved or the exchange fails + """ + try: + actor_token, actor_token_type = await self._resolve_actor_token( + actor_token, actor_token_type, store_options) + + # Build the session_transfer audience from the resolved request domain. + domain = await self._resolve_current_domain(store_options) + audience = f"urn:{domain}:session_transfer" + + options = CustomTokenExchangeOptions( + subject_token=subject_token, + subject_token_type=subject_token_type, + audience=audience, + scope=scope, + actor_token=actor_token, + actor_token_type=actor_token_type, + organization=organization, + ) + + response = await self.custom_token_exchange(options, store_options) + + return SessionTransferTokenResult( + session_transfer_token=response.access_token, + issued_token_type=response.issued_token_type or SESSION_TRANSFER_TOKEN_TYPE, + expires_in=response.expires_in, + token_type=response.token_type, + scope=response.scope, + ) + except (CustomTokenExchangeError, ApiError): + raise + except Exception as e: + raise CustomTokenExchangeError( + CustomTokenExchangeErrorCode.TOKEN_EXCHANGE_FAILED, + f"Session transfer token request failed: {str(e)}", + e + ) + + def build_session_transfer_redirect( + self, + target_login_url: str, + result: SessionTransferTokenResult, + organization: Optional[str] = None + ) -> str: + """ + Builds the redirect URL that hands the STT to the target app's login URL. + + Args: + target_login_url: The target app's login URL + result: The SessionTransferTokenResult from request_session_transfer_token + organization: Organization identifier to forward (optional) + + Returns: + A URL string with session_transfer_token (and organization) as query parameters + """ + params = {"session_transfer_token": result.session_transfer_token} + if organization: + params["organization"] = organization + + return URL.build_url(target_login_url, params) + # ============================================================================ # MFA (Multi-Factor Authentication) # ============================================================================ diff --git a/src/auth0_server_python/auth_types/__init__.py b/src/auth0_server_python/auth_types/__init__.py index da2a228..e886938 100644 --- a/src/auth0_server_python/auth_types/__init__.py +++ b/src/auth0_server_python/auth_types/__init__.py @@ -397,6 +397,24 @@ class LoginWithCustomTokenExchangeResult(BaseModel): authorization_details: Optional[list[AuthorizationDetails]] = None +class SessionTransferTokenResult(BaseModel): + """ + Response from a session transfer token (STT) request. + + Attributes: + session_transfer_token: The opaque, single-use session transfer token + issued_token_type: Format of issued token (the session-transfer URN) + expires_in: Token lifetime in seconds + token_type: Token type as returned by the server (typically "N_A") + scope: Granted scopes (if returned) + """ + session_transfer_token: str + issued_token_type: str + expires_in: int + token_type: Optional[str] = None + scope: Optional[str] = None + + # ============================================================================= # Connected Accounts Types # ============================================================================= diff --git a/src/auth0_server_python/error/__init__.py b/src/auth0_server_python/error/__init__.py index a961c30..ee9279f 100644 --- a/src/auth0_server_python/error/__init__.py +++ b/src/auth0_server_python/error/__init__.py @@ -254,6 +254,9 @@ class CustomTokenExchangeErrorCode: MISSING_ACTOR_TOKEN = "missing_actor_token" TOKEN_EXCHANGE_FAILED = "token_exchange_failed" INVALID_RESPONSE = "invalid_response" + ACTOR_UNAVAILABLE = "actor_unavailable" + SETACTOR_REQUIRED = "setactor_required" + SESSION_TRANSFER_DISABLED = "session_transfer_disabled" # ============================================================================= diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index e9f17be..22714d0 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -33,6 +33,7 @@ PasskeyLoginResult, PasskeySignupChallengeResponse, PasskeyUserProfile, + SessionTransferTokenResult, StartInteractiveLoginOptions, StateData, TransactionData, @@ -3953,6 +3954,232 @@ def test_state_merge_preserves_user_act_claim(): assert updated["user"]["act"] == {"sub": "agent|abc"} +# ============================================================================= +# Session Transfer Token (STT) Tests +# ============================================================================= + +# Fixed wire-protocol URNs, pinned as literals so a bad edit to the SDK constants is caught. +STT_URN = "urn:auth0:params:oauth:token-type:session_transfer_token" +ID_TOKEN_URN = "urn:ietf:params:oauth:token-type:id_token" + + +def _stt_client(mocker, *, exchange_response=None): + """A ServerClient with the token endpoint mocked, returning (client, post_mock).""" + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret", + ) + mocker.patch.object( + client, "_fetch_oidc_metadata", + return_value={"token_endpoint": "https://auth0.local/oauth/token"}, + ) + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = exchange_response or { + "access_token": "the-stt", + "token_type": "N_A", + "expires_in": 60, + "scope": "openid profile", + "issued_token_type": STT_URN, + } + mock_response.headers.get.return_value = "application/json" + + post_mock = AsyncMock() + post_mock.__aenter__.return_value = post_mock + post_mock.__aexit__.return_value = None + post_mock.post.return_value = mock_response + mocker.patch("httpx.AsyncClient", return_value=post_mock) + return client, post_mock + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_success(mocker): + """Returns a SessionTransferTokenResult with the STT and session_transfer audience.""" + client, post_mock = _stt_client(mocker) + + result = await client.request_session_transfer_token( + subject_token="subj", + subject_token_type="urn:acme:customer-subject", + actor_token="agent-id-token", + ) + + assert isinstance(result, SessionTransferTokenResult) + assert result.session_transfer_token == "the-stt" + assert result.issued_token_type == STT_URN + assert result.expires_in == 60 + + data = post_mock.post.call_args[1]["data"] + assert data["audience"] == "urn:auth0.local:session_transfer" + assert data["grant_type"] == "urn:ietf:params:oauth:grant-type:token-exchange" + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_sources_actor_from_session(mocker): + """When actor_token is omitted, the agent session's ID token is used as the actor.""" + client, post_mock = _stt_client(mocker) + client._state_store.get.return_value = {"id_token": "session-id-token"} + mocker.patch.object(client, "_is_id_token_usable", return_value=True) + + await client.request_session_transfer_token( + subject_token="subj", + subject_token_type="urn:acme:customer-subject", + ) + + data = post_mock.post.call_args[1]["data"] + assert data["actor_token"] == "session-id-token" + assert data["actor_token_type"] == ID_TOKEN_URN + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_explicit_actor_overrides_session(mocker): + """An explicit actor_token wins and the session is never read.""" + client, post_mock = _stt_client(mocker) + + await client.request_session_transfer_token( + subject_token="subj", + subject_token_type="urn:acme:customer-subject", + actor_token="explicit-actor", + ) + + client._state_store.get.assert_not_called() + assert post_mock.post.call_args[1]["data"]["actor_token"] == "explicit-actor" + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_refreshes_expired_session_token(mocker): + """A stale session ID token is refreshed and the fresh token is used as the actor.""" + client, post_mock = _stt_client(mocker) + client._state_store.get.return_value = {"id_token": "stale", "refresh_token": "rt"} + # First check (stale) fails, second (refreshed) passes. + mocker.patch.object(client, "_is_id_token_usable", side_effect=[False, True]) + mocker.patch.object( + client, "get_token_by_refresh_token", + return_value={"id_token": "fresh-id-token", "access_token": "a", "expires_in": 3600}, + ) + + await client.request_session_transfer_token( + subject_token="subj", + subject_token_type="urn:acme:customer-subject", + ) + + assert post_mock.post.call_args[1]["data"]["actor_token"] == "fresh-id-token" + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_never_persists_stt(mocker): + """The mint is stateless: the one-shot STT must never be written to the session store.""" + client, _ = _stt_client(mocker) + + await client.request_session_transfer_token( + subject_token="subj", + subject_token_type="urn:acme:customer-subject", + actor_token="explicit-actor", + ) + + client._state_store.set.assert_not_called() + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_blank_actor_rejected(mocker): + """A passed-but-blank actor_token is a bug, not a fallback signal.""" + client, _ = _stt_client(mocker) + + with pytest.raises(CustomTokenExchangeError) as exc: + await client.request_session_transfer_token( + subject_token="subj", + subject_token_type="urn:acme:customer-subject", + actor_token=" ", + ) + assert exc.value.code == CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_no_actor_raises_before_network(mocker): + """No explicit actor and no usable session token fails client-side with ACTOR_UNAVAILABLE.""" + client, post_mock = _stt_client(mocker) + client._state_store.get.return_value = None + + with pytest.raises(CustomTokenExchangeError) as exc: + await client.request_session_transfer_token( + subject_token="subj", + subject_token_type="urn:acme:customer-subject", + ) + assert exc.value.code == CustomTokenExchangeErrorCode.ACTOR_UNAVAILABLE + post_mock.post.assert_not_called() + + +def test_build_session_transfer_redirect_encodes_and_includes_organization(): + """The redirect URL carries a URL-encoded STT and the organization when present.""" + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret", + ) + result = SessionTransferTokenResult( + session_transfer_token="a b/c+d", + issued_token_type=STT_URN, + expires_in=60, + ) + + url = client.build_session_transfer_redirect( + "https://app.example.com/auth/login", result, organization="org_globex" + ) + + query = parse_qs(urlparse(url).query) + assert query["session_transfer_token"] == ["a b/c+d"] # parse_qs decodes; encoding happened on the wire + assert query["organization"] == ["org_globex"] + assert "a+b" in url or "a%20b" in url # the raw URL is encoded + + +def _redirect_client(): + return ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + secret="some-secret", + ) + + +def test_build_session_transfer_redirect_omits_organization_when_absent(): + """No organization is passed → the parameter is absent, not empty.""" + result = SessionTransferTokenResult( + session_transfer_token="stt", issued_token_type=STT_URN, expires_in=60 + ) + + url = _redirect_client().build_session_transfer_redirect( + "https://app.example.com/auth/login", result + ) + + query = parse_qs(urlparse(url).query) + assert query["session_transfer_token"] == ["stt"] + assert "organization" not in query + + +def test_build_session_transfer_redirect_appends_to_existing_query(): + """A target login URL that already has a query string gets the STT appended with '&'.""" + result = SessionTransferTokenResult( + session_transfer_token="stt", issued_token_type=STT_URN, expires_in=60 + ) + + url = _redirect_client().build_session_transfer_redirect( + "https://app.example.com/auth/login?returnTo=/home", result + ) + + assert url.count("?") == 1 + query = parse_qs(urlparse(url).query) + assert query["returnTo"] == ["/home"] + assert query["session_transfer_token"] == ["stt"] + + # ============================================================================= # OIDC Metadata and JWKS Fetching Tests # ============================================================================= @@ -5009,8 +5236,9 @@ async def domain_resolver(context): user = await client.get_user(store_options={"request": {}}) assert user is None - -# ── MFA Integration Tests ──────────────────────────────────────────────────── +# ============================================================================= +# MFA Tests +# ============================================================================= @pytest.mark.asyncio @@ -5231,9 +5459,10 @@ async def _fake_fetch(self, domain): # ============================================================================= -# Organization and Invitation Tests +# ORGANIZATIONS AND INVITATION TESTS # ============================================================================= + def _make_org_client(mocker, transaction_data: TransactionData, **extra): """Helper: build a ServerClient with mocked stores and standard JWT mocks.""" mock_tx_store = AsyncMock() From a070d419debb9ec137a70de2751aeb9ded1b247b Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Tue, 21 Jul 2026 17:30:36 +0530 Subject: [PATCH 2/3] fix: address STT review feedback on PR #139 - Validate the redirect target URL in build_session_transfer_redirect: require an absolute https URL (http allowed only for localhost/loopback), via a reusable URL.validate_https_redirect_target helper. Prevents leaking the single-use STT to an untrusted host. - Surface issued_token_type exactly as the server returned it instead of defaulting to the STT URN, so a non-STT response is not mislabelled. - Validate subject_token/subject_token_type up front, before any session read or network. - Narrow the swallowed exceptions in actor resolution: the refresh catches only ApiError/AccessTokenError and _is_id_token_usable only token-verification errors, so infrastructure failures propagate instead of being masked as ACTOR_UNAVAILABLE. - Source the refresh domain from the session (resolver-safe) and reject an actor sourced from a session on a different domain in resolver mode. - Note the aud==client_id assumption in _is_id_token_usable. - Add tests for redirect validation, issued_token_type passthrough, up-front subject validation, and the actor-resolution paths. --- examples/CustomTokenExchange.md | 2 +- .../auth_server/server_client.py | 62 +++++-- .../tests/test_server_client.py | 159 ++++++++++++++++-- src/auth0_server_python/utils/helpers.py | 25 ++- 4 files changed, 220 insertions(+), 28 deletions(-) diff --git a/examples/CustomTokenExchange.md b/examples/CustomTokenExchange.md index d0589bd..86346db 100644 --- a/examples/CustomTokenExchange.md +++ b/examples/CustomTokenExchange.md @@ -263,7 +263,7 @@ if act: - `ACTOR_UNAVAILABLE`: no usable actor token (client-side; raised before any network call) - `SETACTOR_REQUIRED`: an STT was requested but the Action did not call `setActor` (server 400) -- `SESSION_TRANSFER_DISABLED`: the session-transfer feature is off for the tenant/client (server 400) +- `SESSION_TRANSFER_DISABLED`: the session-transfer feature is not enabled for the tenant/client (server 400) ## Additional Resources diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index bea7bfb..4398804 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -2656,13 +2656,16 @@ async def _is_id_token_usable(self, token: str, store_options: Optional[dict[str """ if not token: return False + domain = await self._resolve_current_domain(store_options) + metadata = await self._get_oidc_metadata_cached(domain) + jwks = await self._get_jwks_cached(domain, metadata) try: - domain = await self._resolve_current_domain(store_options) - metadata = await self._get_oidc_metadata_cached(domain) - jwks = await self._get_jwks_cached(domain, metadata) + # aud is the client_id for a standard Auth0 ID token. await self._verify_and_decode_jwt(token, jwks, audience=self._client_id) return True - except Exception: + except (jwt.PyJWTError, ValueError): + # Only token-level failures mean "unusable"; infra errors (JWKS fetch, domain + # resolution) propagate above rather than being masked as ACTOR_UNAVAILABLE. return False async def _resolve_actor_token( @@ -2692,21 +2695,34 @@ async def _resolve_actor_token( state_data = state_data.dict() state_data = state_data or {} + # In resolver mode, don't source the actor from a session on a different domain. + if self._domain_resolver: + session_domain = self._get_session_domain(state_data) + current_domain = await self._resolve_current_domain(store_options) + if not session_domain or self._normalize_url(session_domain) != self._normalize_url(current_domain): + raise CustomTokenExchangeError( + CustomTokenExchangeErrorCode.ACTOR_UNAVAILABLE, + "No usable actor token: the agent session is on a different domain." + ) + session_id_token = state_data.get("id_token") # Refresh a stale (or missing) ID token when the agent session has a refresh token. if not await self._is_id_token_usable(session_id_token, store_options) and state_data.get("refresh_token"): + refresh_domain = self._get_session_domain(state_data) or await self._resolve_current_domain(store_options) try: refreshed = await self.get_token_by_refresh_token({ "refresh_token": state_data["refresh_token"], - "domain": state_data.get("domain") or self._domain, + "domain": refresh_domain, }) + except (ApiError, AccessTokenError): + # A genuine refresh failure means no usable actor; unexpected errors propagate. + refreshed = None + if refreshed: updated_state_data = State.update_state_data( self.DEFAULT_AUDIENCE_STATE_KEY, state_data, refreshed) await self._state_store.set(self._state_identifier, updated_state_data, options=store_options) session_id_token = refreshed.get("id_token") or session_id_token - except Exception: - session_id_token = None if await self._is_id_token_usable(session_id_token, store_options): return session_id_token, ID_TOKEN_TYPE @@ -2749,6 +2765,18 @@ async def request_session_transfer_token( CustomTokenExchangeError: If no actor can be resolved or the exchange fails """ try: + # Validate the subject up front - before any session read/refresh/network. + if not subject_token or not subject_token.strip(): + raise CustomTokenExchangeError( + CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT, + "subject_token cannot be empty or whitespace-only" + ) + if not subject_token_type or not subject_token_type.strip(): + raise CustomTokenExchangeError( + CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT, + "subject_token_type cannot be empty or whitespace-only" + ) + actor_token, actor_token_type = await self._resolve_actor_token( actor_token, actor_token_type, store_options) @@ -2770,7 +2798,9 @@ async def request_session_transfer_token( return SessionTransferTokenResult( session_transfer_token=response.access_token, - issued_token_type=response.issued_token_type or SESSION_TRANSFER_TOKEN_TYPE, + # Return the server's value as-is; don't default to the STT URN, or a non-STT + # response would be mislabelled as an STT. + issued_token_type=response.issued_token_type or "", expires_in=response.expires_in, token_type=response.token_type, scope=response.scope, @@ -2793,16 +2823,28 @@ def build_session_transfer_redirect( """ Builds the redirect URL that hands the STT to the target app's login URL. + target_login_url must be a trusted, app-controlled absolute https URL (http is allowed + only for localhost/loopback) - the STT is a single-use credential and must not leak to an + untrusted host. + Args: - target_login_url: The target app's login URL + target_login_url: The target app's login URL (absolute, https) result: The SessionTransferTokenResult from request_session_transfer_token organization: Organization identifier to forward (optional) Returns: A URL string with session_transfer_token (and organization) as query parameters + + Raises: + MissingRequiredArgumentError: If target_login_url is missing or blank + InvalidArgumentError: If target_login_url is not an absolute https URL, or organization is blank """ + URL.validate_https_redirect_target(target_login_url, "target_login_url") + params = {"session_transfer_token": result.session_transfer_token} - if organization: + if organization is not None: + if not organization.strip(): + raise InvalidArgumentError("organization", "organization must not be blank") params["organization"] = organization return URL.build_url(target_login_url, params) diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 22714d0..6a7954c 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -3963,9 +3963,9 @@ def test_state_merge_preserves_user_act_claim(): ID_TOKEN_URN = "urn:ietf:params:oauth:token-type:id_token" -def _stt_client(mocker, *, exchange_response=None): - """A ServerClient with the token endpoint mocked, returning (client, post_mock).""" - client = ServerClient( +def _stt_base_client(): + """A bare ServerClient for STT tests (no network/session mocking).""" + return ServerClient( domain="auth0.local", client_id="", client_secret="", @@ -3973,6 +3973,11 @@ def _stt_client(mocker, *, exchange_response=None): transaction_store=AsyncMock(), secret="some-secret", ) + + +def _stt_client(mocker, *, exchange_response=None): + """A ServerClient with the token endpoint mocked, returning (client, post_mock).""" + client = _stt_base_client() mocker.patch.object( client, "_fetch_oidc_metadata", return_value={"token_endpoint": "https://auth0.local/oauth/token"}, @@ -4138,24 +4143,13 @@ def test_build_session_transfer_redirect_encodes_and_includes_organization(): assert "a+b" in url or "a%20b" in url # the raw URL is encoded -def _redirect_client(): - return ServerClient( - domain="auth0.local", - client_id="", - client_secret="", - state_store=AsyncMock(), - transaction_store=AsyncMock(), - secret="some-secret", - ) - - def test_build_session_transfer_redirect_omits_organization_when_absent(): """No organization is passed → the parameter is absent, not empty.""" result = SessionTransferTokenResult( session_transfer_token="stt", issued_token_type=STT_URN, expires_in=60 ) - url = _redirect_client().build_session_transfer_redirect( + url = _stt_base_client().build_session_transfer_redirect( "https://app.example.com/auth/login", result ) @@ -4170,7 +4164,7 @@ def test_build_session_transfer_redirect_appends_to_existing_query(): session_transfer_token="stt", issued_token_type=STT_URN, expires_in=60 ) - url = _redirect_client().build_session_transfer_redirect( + url = _stt_base_client().build_session_transfer_redirect( "https://app.example.com/auth/login?returnTo=/home", result ) @@ -4180,6 +4174,139 @@ def test_build_session_transfer_redirect_appends_to_existing_query(): assert query["session_transfer_token"] == ["stt"] +def _stt_result(): + return SessionTransferTokenResult( + session_transfer_token="stt", issued_token_type=STT_URN, expires_in=60 + ) + + +def test_build_session_transfer_redirect_allows_http_for_localhost(): + """http is allowed for localhost (local dev).""" + url = _stt_base_client().build_session_transfer_redirect( + "http://localhost:3000/auth/login", _stt_result()) + assert "session_transfer_token=stt" in url + + +def test_build_session_transfer_redirect_allows_http_for_127_0_0_1(): + """http is allowed for the 127.0.0.1 loopback address.""" + url = _stt_base_client().build_session_transfer_redirect( + "http://127.0.0.1:3000/auth/login", _stt_result()) + assert "session_transfer_token=stt" in url + + +def test_build_session_transfer_redirect_rejects_non_https(): + """A non-loopback http target is rejected - the STT must not ride an insecure URL.""" + with pytest.raises(InvalidArgumentError): + _stt_base_client().build_session_transfer_redirect( + "http://app.example.com/auth/login", _stt_result()) + + +def test_build_session_transfer_redirect_rejects_non_absolute(): + """A non-absolute target URL is rejected.""" + with pytest.raises(InvalidArgumentError): + _stt_base_client().build_session_transfer_redirect("app.example.com/auth/login", _stt_result()) + + +def test_build_session_transfer_redirect_rejects_blank_target(): + """A blank target URL is rejected.""" + with pytest.raises(MissingRequiredArgumentError): + _stt_base_client().build_session_transfer_redirect(" ", _stt_result()) + + +def test_build_session_transfer_redirect_rejects_blank_organization(): + """A blank organization fails fast rather than being forwarded as an empty param.""" + with pytest.raises(InvalidArgumentError): + _stt_base_client().build_session_transfer_redirect( + "https://app.example.com/auth/login", _stt_result(), organization=" ") + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_surfaces_server_issued_token_type(mocker): + """A non-STT issued_token_type is surfaced verbatim, never fabricated as the STT URN.""" + client, _ = _stt_client(mocker, exchange_response={ + "access_token": "tok", "token_type": "Bearer", "expires_in": 300, + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + }) + + result = await client.request_session_transfer_token( + subject_token="subj", subject_token_type="urn:acme:sub", actor_token="a", + ) + + assert result.issued_token_type == "urn:ietf:params:oauth:token-type:access_token" + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_empty_issued_token_type_when_absent(mocker): + """When the server omits issued_token_type, the result carries an empty string, not the URN.""" + client, _ = _stt_client(mocker, exchange_response={ + "access_token": "tok", "token_type": "N_A", "expires_in": 60, + }) + + result = await client.request_session_transfer_token( + subject_token="subj", subject_token_type="urn:acme:sub", actor_token="a", + ) + + assert result.issued_token_type == "" + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_blank_subject_token_rejected_before_network(mocker): + """A blank subject_token fails with INVALID_TOKEN_FORMAT before any network call.""" + client, post_mock = _stt_client(mocker) + with pytest.raises(CustomTokenExchangeError) as exc: + await client.request_session_transfer_token( + subject_token=" ", subject_token_type="urn:acme:sub", actor_token="a") + assert exc.value.code == CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT + post_mock.post.assert_not_called() + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_blank_subject_token_type_rejected(mocker): + """A blank subject_token_type fails with INVALID_TOKEN_FORMAT before any network call.""" + client, post_mock = _stt_client(mocker) + with pytest.raises(CustomTokenExchangeError) as exc: + await client.request_session_transfer_token( + subject_token="subj", subject_token_type=" ", actor_token="a") + assert exc.value.code == CustomTokenExchangeErrorCode.INVALID_TOKEN_FORMAT + post_mock.post.assert_not_called() + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_refresh_without_id_token_is_unavailable(mocker): + """A refresh that returns no ID token leaves no usable actor -> ACTOR_UNAVAILABLE.""" + client, _ = _stt_client(mocker) + client._state_store.get.return_value = {"id_token": "stale", "refresh_token": "rt"} + mocker.patch.object(client, "_is_id_token_usable", return_value=False) + mocker.patch.object(client, "get_token_by_refresh_token", + return_value={"access_token": "a", "expires_in": 3600}) # no id_token + + with pytest.raises(CustomTokenExchangeError) as exc: + await client.request_session_transfer_token( + subject_token="subj", subject_token_type="urn:acme:sub", + ) + assert exc.value.code == CustomTokenExchangeErrorCode.ACTOR_UNAVAILABLE + + +@pytest.mark.asyncio +async def test_request_session_transfer_token_rejects_cross_domain_session_in_resolver_mode(mocker): + """In resolver mode, an actor is not sourced from a session on a different domain.""" + async def resolver(context): + return "tenant-a.auth0.com" + + client = ServerClient( + domain=resolver, client_id="", client_secret="", + state_store=AsyncMock(), transaction_store=AsyncMock(), secret="some-secret", + ) + client._state_store.get.return_value = {"id_token": "t", "domain": "tenant-b.auth0.com"} + + with pytest.raises(CustomTokenExchangeError) as exc: + await client.request_session_transfer_token( + subject_token="subj", subject_token_type="urn:acme:sub", + store_options={"request": None, "response": None}, + ) + assert exc.value.code == CustomTokenExchangeErrorCode.ACTOR_UNAVAILABLE + + # ============================================================================= # OIDC Metadata and JWKS Fetching Tests # ============================================================================= diff --git a/src/auth0_server_python/utils/helpers.py b/src/auth0_server_python/utils/helpers.py index b875289..3ef0e53 100644 --- a/src/auth0_server_python/utils/helpers.py +++ b/src/auth0_server_python/utils/helpers.py @@ -8,7 +8,12 @@ from urllib.parse import parse_qs, urlencode, urlparse from auth0_server_python.auth_types import DomainResolverContext -from auth0_server_python.error import DomainResolverError, OrganizationTokenValidationError +from auth0_server_python.error import ( + DomainResolverError, + InvalidArgumentError, + MissingRequiredArgumentError, + OrganizationTokenValidationError, +) class PKCE: @@ -218,6 +223,24 @@ def is_session_ceiling_in_past( class URL: + @staticmethod + def validate_https_redirect_target(url: str, name: str) -> None: + """ + Require url to be an absolute https URL (http allowed only for localhost/loopback). + + Raises MissingRequiredArgumentError if blank, InvalidArgumentError otherwise. + """ + if not url or not url.strip(): + raise MissingRequiredArgumentError(name) + parsed = urlparse(url) + if not parsed.scheme or not parsed.netloc: + raise InvalidArgumentError(name, "must be an absolute URL") + is_loopback = parsed.hostname in ("localhost", "127.0.0.1", "::1") + if parsed.scheme != "https" and not (parsed.scheme == "http" and is_loopback): + raise InvalidArgumentError( + name, "must use https (http is allowed only for localhost/loopback)" + ) + @staticmethod def build_url(base_url: str, params: dict[str, Any]) -> str: """ From f741a1e6776eaea53c5c13d39f00a590961ef5f3 Mon Sep 17 00:00:00 2001 From: Snehil Kishore Date: Wed, 22 Jul 2026 17:34:36 +0530 Subject: [PATCH 3/3] fix: reject fragment in STT redirect target and cover the server-error path - validate_https_redirect_target now rejects a target_login_url with a fragment: appending the query after a fragment would place the STT inside the fragment, silently dropping the single-use token before it reaches the target's login handler. - Add a test asserting a server 400 on the STT exchange surfaces as CustomTokenExchangeError carrying the server's error/error_description. --- .../tests/test_server_client.py | 25 +++++++++++++++++++ src/auth0_server_python/utils/helpers.py | 3 +++ 2 files changed, 28 insertions(+) diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 6a7954c..16ce89e 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -4213,6 +4213,13 @@ def test_build_session_transfer_redirect_rejects_blank_target(): _stt_base_client().build_session_transfer_redirect(" ", _stt_result()) +def test_build_session_transfer_redirect_rejects_fragment(): + """A fragment is rejected - it would swallow the STT query param, dropping the token.""" + with pytest.raises(InvalidArgumentError): + _stt_base_client().build_session_transfer_redirect( + "https://app.example.com/auth/login#section", _stt_result()) + + def test_build_session_transfer_redirect_rejects_blank_organization(): """A blank organization fails fast rather than being forwarded as an empty param.""" with pytest.raises(InvalidArgumentError): @@ -4287,6 +4294,24 @@ async def test_request_session_transfer_token_refresh_without_id_token_is_unavai assert exc.value.code == CustomTokenExchangeErrorCode.ACTOR_UNAVAILABLE +@pytest.mark.asyncio +async def test_request_session_transfer_token_surfaces_server_error(mocker): + """A server 400 surfaces as CustomTokenExchangeError carrying the server's error/description.""" + client, post_mock = _stt_client(mocker) + post_mock.post.return_value.status_code = 400 + post_mock.post.return_value.json.return_value = { + "error": "invalid_request", + "error_description": "setActor is required when requesting a session transfer token via token exchange.", + } + + with pytest.raises(CustomTokenExchangeError) as exc: + await client.request_session_transfer_token( + subject_token="subj", subject_token_type="urn:acme:sub", actor_token="a", + ) + assert exc.value.code == "invalid_request" + assert "setActor is required" in str(exc.value) + + @pytest.mark.asyncio async def test_request_session_transfer_token_rejects_cross_domain_session_in_resolver_mode(mocker): """In resolver mode, an actor is not sourced from a session on a different domain.""" diff --git a/src/auth0_server_python/utils/helpers.py b/src/auth0_server_python/utils/helpers.py index 3ef0e53..e7d51cc 100644 --- a/src/auth0_server_python/utils/helpers.py +++ b/src/auth0_server_python/utils/helpers.py @@ -240,6 +240,9 @@ def validate_https_redirect_target(url: str, name: str) -> None: raise InvalidArgumentError( name, "must use https (http is allowed only for localhost/loopback)" ) + # A fragment would swallow the appended query params, dropping the token silently. + if parsed.fragment: + raise InvalidArgumentError(name, "must not contain a fragment") @staticmethod def build_url(base_url: str, params: dict[str, Any]) -> str: