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..86346db 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 not enabled 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..4398804 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,213 @@ 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 + 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: + # 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 (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( + 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 {} + + # 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": 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 + + 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: + # 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) + + # 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, + # 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, + ) + 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. + + 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 (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 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) + # ============================================================================ # 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..6a7954c 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,359 @@ 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_base_client(): + """A bare ServerClient for STT tests (no network/session mocking).""" + return ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + 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"}, + ) + 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 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 = _stt_base_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 = _stt_base_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"] + + +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 # ============================================================================= @@ -5009,8 +5363,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 +5586,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() 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: """