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.

Author: kishore7snehil

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
 

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)

src/auth0_server_python/tests/test_server_client.py

@@ -3963,16 +3963,21 @@ 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_id>",
         client_secret="<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"},
@@ -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_id>",
-        client_secret="<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_id>", client_secret="<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
 # =============================================================================

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:
         """