feat: Add optional jwt_issuer option for session token validation

README.md

@@ -58,6 +58,7 @@ The client reads credentials from the environment when not passed explicitly:
 | `WORKOS_CLIENT_ID` | WorkOS client ID |
 | `WORKOS_BASE_URL` | Override the API base URL (defaults to `https://api.workos.com/`) |
 | `WORKOS_REQUEST_TIMEOUT` | HTTP timeout in seconds (defaults to `60`) |
+| `WORKOS_ISSUER` | Expected `iss` claim of session access tokens, comma-separated to accept several (not validated when unset; also settable via `jwt_issuer=`) |
 
 ## Available Resources
 

src/workos/_base_client.py

@@ -9,7 +9,7 @@
 import random
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Any, Dict, Optional, Sequence, Type, cast, overload
+from typing import Any, Dict, List, Optional, Sequence, Type, Union, cast, overload
 from urllib.parse import quote
 
 import httpx
@@ -42,6 +42,14 @@
 RETRY_MULTIPLIER = 2
 
 
+def _parse_issuer_env(value: str) -> Optional[Union[str, List[str]]]:
+    issuers = [issuer.strip() for issuer in value.split(",")]
+    issuers = [issuer for issuer in issuers if issuer]
+    if not issuers:
+        return None
+    return issuers[0] if len(issuers) == 1 else issuers
+
+
 class _BaseWorkOSClient:
     """Shared WorkOS client implementation."""
 
@@ -53,6 +61,7 @@ def __init__(
         base_url: Optional[str] = None,
         request_timeout: Optional[int] = None,
         jwt_leeway: float = 0.0,
+        jwt_issuer: Optional[Union[str, Sequence[str]]] = None,
         max_retries: int = MAX_RETRIES,
         is_public: bool = False,
     ) -> None:
@@ -82,6 +91,13 @@ def __init__(
         )
         self._max_retries = max_retries
         self._jwt_leeway = jwt_leeway
+        if jwt_issuer is None:
+            env_issuer = os.environ.get("WORKOS_ISSUER")
+            self._jwt_issuer: Optional[Union[str, Sequence[str]]] = (
+                _parse_issuer_env(env_issuer) if env_issuer else None
+            )
+        else:
+            self._jwt_issuer = jwt_issuer
 
     @property
     def base_url(self) -> str:
@@ -362,6 +378,7 @@ def __init__(
         base_url: Optional[str] = None,
         request_timeout: Optional[int] = None,
         jwt_leeway: float = 0.0,
+        jwt_issuer: Optional[Union[str, Sequence[str]]] = None,
         max_retries: int = MAX_RETRIES,
         is_public: bool = False,
     ) -> None:
@@ -373,6 +390,9 @@ def __init__(
             base_url: Base URL for API requests. Falls back to WORKOS_BASE_URL or "https://api.workos.com".
             request_timeout: HTTP request timeout in seconds. Falls back to WORKOS_REQUEST_TIMEOUT or 60.
             jwt_leeway: JWT clock skew leeway in seconds.
+            jwt_issuer: Expected ``iss`` claim of session access tokens, or a list of
+                accepted issuers. Falls back to the WORKOS_ISSUER environment variable
+                (comma-separated for a list). When unset, the issuer is not validated.
             max_retries: Maximum number of retries for failed requests. Defaults to 3.
             is_public: When True, mark this client as public (PKCE / browser
                 / mobile / CLI). The API key is forced to None and the
@@ -388,6 +408,7 @@ def __init__(
             base_url=base_url,
             request_timeout=request_timeout,
             jwt_leeway=jwt_leeway,
+            jwt_issuer=jwt_issuer,
             max_retries=max_retries,
             is_public=is_public,
         )
@@ -593,6 +614,7 @@ def __init__(
         base_url: Optional[str] = None,
         request_timeout: Optional[int] = None,
         jwt_leeway: float = 0.0,
+        jwt_issuer: Optional[Union[str, Sequence[str]]] = None,
         max_retries: int = MAX_RETRIES,
         is_public: bool = False,
     ) -> None:
@@ -604,6 +626,9 @@ def __init__(
             base_url: Base URL for API requests. Falls back to WORKOS_BASE_URL or "https://api.workos.com".
             request_timeout: HTTP request timeout in seconds. Falls back to WORKOS_REQUEST_TIMEOUT or 60.
             jwt_leeway: JWT clock skew leeway in seconds.
+            jwt_issuer: Expected ``iss`` claim of session access tokens, or a list of
+                accepted issuers. Falls back to the WORKOS_ISSUER environment variable
+                (comma-separated for a list). When unset, the issuer is not validated.
             max_retries: Maximum number of retries for failed requests. Defaults to 3.
 
         Raises:
@@ -615,6 +640,7 @@ def __init__(
             base_url=base_url,
             request_timeout=request_timeout,
             jwt_leeway=jwt_leeway,
+            jwt_issuer=jwt_issuer,
             max_retries=max_retries,
             is_public=is_public,
         )

src/workos/session.py

@@ -289,6 +289,7 @@ def authenticate(
                 algorithms=self._JWK_ALGORITHMS,
                 options={"verify_aud": False},
                 leeway=self._client._jwt_leeway,
+                issuer=self._client._jwt_issuer,
             )
         except jwt.exceptions.InvalidTokenError:
             return AuthenticateWithSessionCookieErrorResponse(
@@ -370,6 +371,7 @@ def refresh(
                     algorithms=self._JWK_ALGORITHMS,
                     options={"verify_aud": False},
                     leeway=self._client._jwt_leeway,
+                    issuer=self._client._jwt_issuer,
                 )
             except (
                 jwt.exceptions.InvalidTokenError,
@@ -497,6 +499,7 @@ def authenticate(
                 algorithms=self._JWK_ALGORITHMS,
                 options={"verify_aud": False},
                 leeway=self._client._jwt_leeway,
+                issuer=self._client._jwt_issuer,
             )
         except jwt.exceptions.InvalidTokenError:
             return AuthenticateWithSessionCookieErrorResponse(
@@ -578,6 +581,7 @@ async def refresh(
                     algorithms=self._JWK_ALGORITHMS,
                     options={"verify_aud": False},
                     leeway=self._client._jwt_leeway,
+                    issuer=self._client._jwt_issuer,
                 )
             except (
                 jwt.exceptions.InvalidTokenError,

tests/test_session.py

@@ -6,7 +6,7 @@
 from cryptography.fernet import Fernet, InvalidToken
 from cryptography.hazmat.primitives.asymmetric import rsa
 
-from workos import WorkOSClient
+from workos import AsyncWorkOSClient, WorkOSClient
 from workos._errors import (
     AuthenticationError,
     AuthenticationMethodNotAllowedError,
@@ -210,6 +210,165 @@ def test_session_authenticate_expired_jwt(self):
         assert isinstance(result, AuthenticateWithSessionCookieErrorResponse)
         assert result.reason == AuthenticateWithSessionCookieFailureReason.INVALID_JWT
 
+    def _authenticate_with_client(self, client, claims=None):
+        token = _make_jwt(self.private_key, claims=claims)
+        sealed = self._make_sealed_session(access_token=token)
+        session = Session(
+            client=client, session_data=sealed, cookie_password=COOKIE_PASSWORD
+        )
+        session.jwks = self._mock_jwks()
+        return session.authenticate()
+
+    def test_session_authenticate_ignores_issuer_when_not_configured(self):
+        result = self._authenticate_with_client(
+            self.workos, claims={"iss": "https://other.example.com"}
+        )
+        assert isinstance(result, AuthenticateWithSessionCookieSuccessResponse)
+
+    def test_session_authenticate_accepts_configured_issuer(self):
+        client = WorkOSClient(
+            api_key="sk_test_123",
+            client_id="client_test_123",
+            jwt_issuer="https://api.workos.com/user_management/client_test_123",
+            max_retries=0,
+        )
+        try:
+            result = self._authenticate_with_client(
+                client,
+                claims={
+                    "iss": "https://api.workos.com/user_management/client_test_123"
+                },
+            )
+        finally:
+            client.close()
+        assert isinstance(result, AuthenticateWithSessionCookieSuccessResponse)
+
+    def test_session_authenticate_rejects_mismatched_issuer(self):
+        client = WorkOSClient(
+            api_key="sk_test_123",
+            client_id="client_test_123",
+            jwt_issuer="https://api.workos.com",
+            max_retries=0,
+        )
+        try:
+            result = self._authenticate_with_client(
+                client, claims={"iss": "https://other.example.com"}
+            )
+        finally:
+            client.close()
+        assert isinstance(result, AuthenticateWithSessionCookieErrorResponse)
+        assert result.reason == AuthenticateWithSessionCookieFailureReason.INVALID_JWT
+
+    def test_session_authenticate_accepts_any_listed_issuer(self):
+        client = WorkOSClient(
+            api_key="sk_test_123",
+            client_id="client_test_123",
+            jwt_issuer=["https://api.workos.com", "https://auth.example.com"],
+            max_retries=0,
+        )
+        try:
+            result = self._authenticate_with_client(
+                client, claims={"iss": "https://auth.example.com"}
+            )
+        finally:
+            client.close()
+        assert isinstance(result, AuthenticateWithSessionCookieSuccessResponse)
+
+    def test_session_authenticate_reads_issuer_list_from_env(self, monkeypatch):
+        monkeypatch.setenv(
+            "WORKOS_ISSUER", "https://api.workos.com, https://auth.example.com,"
+        )
+        client = WorkOSClient(
+            api_key="sk_test_123", client_id="client_test_123", max_retries=0
+        )
+        try:
+            assert client._jwt_issuer == [
+                "https://api.workos.com",
+                "https://auth.example.com",
+            ]
+            result = self._authenticate_with_client(
+                client, claims={"iss": "https://auth.example.com"}
+            )
+        finally:
+            client.close()
+        assert isinstance(result, AuthenticateWithSessionCookieSuccessResponse)
+
+    def test_session_empty_issuer_list_rejects_all_tokens(self):
+        client = WorkOSClient(
+            api_key="sk_test_123",
+            client_id="client_test_123",
+            jwt_issuer=[],
+            max_retries=0,
+        )
+        try:
+            assert client._jwt_issuer == []
+            result = self._authenticate_with_client(
+                client, claims={"iss": "https://api.workos.com"}
+            )
+        finally:
+            client.close()
+        assert isinstance(result, AuthenticateWithSessionCookieErrorResponse)
+        assert result.reason == AuthenticateWithSessionCookieFailureReason.INVALID_JWT
+
+    def _refresh_with_client(self, client, claims=None):
+        new_token = _make_jwt(self.private_key, claims=claims)
+        sealed = seal_data(
+            {"refresh_token": "rt_old", "user": {"id": "user_01"}}, COOKIE_PASSWORD
+        )
+        session = Session(
+            client=client, session_data=sealed, cookie_password=COOKIE_PASSWORD
+        )
+        session.jwks = self._mock_jwks()
+        session._client.request_raw = MagicMock(
+            return_value={
+                "access_token": new_token,
+                "refresh_token": "rt_new",
+                "user": {"id": "user_01"},
+            }
+        )
+        return session.refresh()
+
+    def test_session_refresh_accepts_configured_issuer(self):
+        client = WorkOSClient(
+            api_key="sk_test_123",
+            client_id="client_test_123",
+            jwt_issuer="https://api.workos.com",
+            max_retries=0,
+        )
+        try:
+            result = self._refresh_with_client(
+                client, claims={"iss": "https://api.workos.com"}
+            )
+        finally:
+            client.close()
+        assert isinstance(result, RefreshWithSessionCookieSuccessResponse)
+
+    def test_session_refresh_rejects_mismatched_issuer(self):
+        client = WorkOSClient(
+            api_key="sk_test_123",
+            client_id="client_test_123",
+            jwt_issuer="https://api.workos.com",
+            max_retries=0,
+        )
+        try:
+            result = self._refresh_with_client(
+                client, claims={"iss": "https://other.example.com"}
+            )
+        finally:
+            client.close()
+        assert isinstance(result, RefreshWithSessionCookieErrorResponse)
+        assert result.reason == AuthenticateWithSessionCookieFailureReason.INVALID_JWT
+
+    def test_session_single_env_issuer_stays_string(self, monkeypatch):
+        monkeypatch.setenv("WORKOS_ISSUER", "https://api.workos.com")
+        client = WorkOSClient(
+            api_key="sk_test_123", client_id="client_test_123", max_retries=0
+        )
+        try:
+            assert client._jwt_issuer == "https://api.workos.com"
+        finally:
+            client.close()
+
     def test_session_refresh_invalid_session(self):
         session = Session(
             client=self.workos, session_data="garbage", cookie_password=COOKIE_PASSWORD
@@ -581,6 +740,70 @@ async def test_async_session_authenticate_success(self, async_workos):
         assert isinstance(result, AuthenticateWithSessionCookieSuccessResponse)
         assert result.session_id == "session_01"
 
+    async def test_async_session_authenticate_rejects_mismatched_issuer(self):
+        private_key, public_key = _generate_rsa_key_pair()
+        token = _make_jwt(private_key, claims={"iss": "https://other.example.com"})
+        sealed = seal_data(
+            {"access_token": token, "refresh_token": "rt", "user": {"id": "u1"}},
+            COOKIE_PASSWORD,
+        )
+        client = AsyncWorkOSClient(
+            api_key="sk_test_123",
+            client_id="client_test_123",
+            jwt_issuer=["https://api.workos.com", "https://auth.example.com"],
+        )
+        try:
+            session = AsyncSession(
+                client=client, session_data=sealed, cookie_password=COOKIE_PASSWORD
+            )
+            session.jwks = self._mock_jwks(public_key)
+            result = session.authenticate()
+        finally:
+            await client.close()
+        assert isinstance(result, AuthenticateWithSessionCookieErrorResponse)
+        assert result.reason == AuthenticateWithSessionCookieFailureReason.INVALID_JWT
+
+    async def _refresh_with_issuer(self, jwt_issuer, iss):
+        from unittest.mock import AsyncMock
+
+        private_key, public_key = _generate_rsa_key_pair()
+        new_token = _make_jwt(private_key, claims={"iss": iss})
+        sealed = seal_data(
+            {"refresh_token": "rt_old", "user": {"id": "user_01"}}, COOKIE_PASSWORD
+        )
+        client = AsyncWorkOSClient(
+            api_key="sk_test_123", client_id="client_test_123", jwt_issuer=jwt_issuer
+        )
+        try:
+            session = AsyncSession(
+                client=client, session_data=sealed, cookie_password=COOKIE_PASSWORD
+            )
+            session.jwks = self._mock_jwks(public_key)
+            session._client.request_raw = AsyncMock(
+                return_value={
+                    "access_token": new_token,
+                    "refresh_token": "rt_new",
+                    "user": {"id": "user_01"},
+                }
+            )
+            return await session.refresh()
+        finally:
+            await client.close()
+
+    async def test_async_session_refresh_accepts_configured_issuer(self):
+        result = await self._refresh_with_issuer(
+            ["https://api.workos.com", "https://auth.example.com"],
+            "https://auth.example.com",
+        )
+        assert isinstance(result, RefreshWithSessionCookieSuccessResponse)
+
+    async def test_async_session_refresh_rejects_mismatched_issuer(self):
+        result = await self._refresh_with_issuer(
+            "https://api.workos.com", "https://other.example.com"
+        )
+        assert isinstance(result, RefreshWithSessionCookieErrorResponse)
+        assert result.reason == AuthenticateWithSessionCookieFailureReason.INVALID_JWT
+
     async def test_async_session_refresh_success(self, async_workos):
         from unittest.mock import AsyncMock