feat: reject already-expired sessions at login and generalize epoch claim extraction

Add a login-time lockout guard to complete_interactive_login: when the upstream
IdP asserts a session_expiry ceiling already in the past at login (compared
against the ID token iat with the same 30s leeway as read-time enforcement),
raise the new flow-agnostic SessionExpiredError instead of persisting an
already-expired session. A missing claim stays a no-op, preserving existing
behavior.

Generalize extract_session_expiry into extract_epoch_claim(claims, name),
reused for both session_expiry and iat, and rename the ceiling predicates to
is_session_ceiling_reached (read-time) and is_session_ceiling_in_past (login).

Document the login rejection in the README, RetrievingData guide, and the
ipsie-webapp example.

Author: kishore7snehil

README.md

@@ -175,7 +175,7 @@ For more details and examples, see [examples/MultipleCustomDomains.md](examples/
 
 ### 6. Session Expiry from the Upstream IdP
 
-For enterprise connections, the upstream identity provider can cap how long a user's session lives. When the connection is configured to honor it, Auth0 includes a `session_expiry` claim in the ID token, and the SDK enforces this ceiling on every session read. Once it is reached, `get_user()` and `get_session()` return `None`, and `get_access_token()` raises an `AccessTokenError` with code `session_expired`.
+For enterprise connections, the upstream identity provider can cap how long a user's session lives. When the connection is configured to honor it, Auth0 includes a `session_expiry` claim in the ID token, and the SDK enforces this ceiling on every session read. Once it is reached, `get_user()` and `get_session()` return `None`, and `get_access_token()` raises an `AccessTokenError` with code `session_expired`. If the asserted ceiling is already in the past at login, `complete_interactive_login()` raises a `SessionExpiredError` instead of persisting an already-expired session.
 
 For more details and examples, see [examples/RetrievingData.md](examples/RetrievingData.md#session-expiry-from-the-upstream-idp).
 

examples/RetrievingData.md

@@ -93,6 +93,21 @@ except AccessTokenError as error:
 
 When the ceiling is reached, the SDK deletes the stored session before returning, so the next request starts clean.
 
+If the upstream IdP asserts a ceiling that is already in the past at login time, `complete_interactive_login()` raises a `SessionExpiredError` rather than persisting an already-expired session:
+
+```python
+from auth0_server_python.error import SessionExpiredError
+
+try:
+    await server_client.complete_interactive_login(url, store_options=store_options)
+except SessionExpiredError:
+    # The session was already past its ceiling on arrival; start a new login.
+    ...
+```
+
+> [!NOTE]
+> **Upgrading:** with this feature enabled, `get_user()` and `get_session()` can return `None` for a user who was previously logged in, once the upstream ceiling passes. Applications that assumed these always return a value after login should add a null check and route the user back through login.
+
 The `session_expiry` value is also surfaced through the user claims, so you can read it without triggering enforcement:
 
 ```python

src/auth0_server_python/auth_server/server_client.py

@@ -55,6 +55,7 @@
     MissingRequiredArgumentError,
     MissingTransactionError,
     PollingApiError,
+    SessionExpiredError,
     StartLinkUserError,
 )
 from auth0_server_python.telemetry import Telemetry
@@ -640,12 +641,15 @@ async def complete_interactive_login(
         id_token = token_response.get("id_token")
         # IPSIE session_expiry ceiling, read from the verified ID token claims.
         session_expires_at = None
+        # ID token `iat`, used to detect a ceiling that is already past at login.
+        issued_at = None
 
         if user_info:
             user_claims = UserClaims.parse_obj(user_info)
             # authlib populates `userinfo` from parsed ID token claims, so the
             # IPSIE session_expiry claim may surface here.
-            session_expires_at = State.extract_session_expiry(user_info)
+            session_expires_at = State.extract_epoch_claim(user_info, "session_expiry")
+            issued_at = State.extract_epoch_claim(user_info, "iat")
         elif id_token:
             # Fetch JWKS for signature verification
             jwks = await self._get_jwks_cached(origin_domain, metadata)
@@ -663,7 +667,8 @@ async def complete_interactive_login(
 
                 user_claims = UserClaims.parse_obj(claims)
                 # IPSIE session_expiry ceiling from the verified ID token.
-                session_expires_at = State.extract_session_expiry(claims)
+                session_expires_at = State.extract_epoch_claim(claims, "session_expiry")
+                issued_at = State.extract_epoch_claim(claims, "iat")
             except ValueError as e:
                 raise ApiError("jwks_key_not_found", str(e))
             except jwt.InvalidSignatureError as e:
@@ -692,6 +697,10 @@ async def complete_interactive_login(
                 )
 
 
+        # Refuse to persist a session whose ceiling is already in the past.
+        if State.is_session_ceiling_in_past(session_expires_at, issued_at):
+            raise SessionExpiredError()
+
         # Build a token set using the token response data
         token_set = TokenSet(
             audience=transaction_data.audience or self.DEFAULT_AUDIENCE_STATE_KEY,
@@ -754,7 +763,7 @@ async def _is_session_expired_by_ceiling(
         """
         internal = state_data_dict.get("internal") or {}
         session_expires_at = internal.get("session_expires_at")
-        if State.is_session_expiry_reached(session_expires_at):
+        if State.is_session_ceiling_reached(session_expires_at):
             await self._state_store.delete(self._state_identifier, options=store_options)
             return True
         return False
@@ -1005,17 +1014,13 @@ async def get_access_token(
 
         merged_scope = self._merge_scope_with_defaults(scope, audience)
 
-        # IPSIE: once the upstream IdP session ceiling has passed, the session
-        # is expired. Surface "session expired" and do NOT serve a cached token
-        # or attempt a refresh-token exchange (which would race the platform's
-        # session revocation).
+        # Once the session ceiling has passed, fail instead of serving or refreshing a token.
         internal = (state_data_dict or {}).get("internal") or {}
-        if State.is_session_expiry_reached(internal.get("session_expires_at")):
+        if State.is_session_ceiling_reached(internal.get("session_expires_at")):
             await self._state_store.delete(self._state_identifier, options=store_options)
             raise AccessTokenError(
                 AccessTokenErrorCode.SESSION_EXPIRED,
-                "The session has expired because the upstream identity provider's "
-                "session_expiry was reached. The user needs to re-authenticate."
+                "The session has expired and the user must re-authenticate."
             )
 
         # Find matching token set

src/auth0_server_python/error/__init__.py

@@ -211,6 +211,19 @@ class AccessTokenForConnectionErrorCode:
     DOMAIN_MISMATCH = "domain_mismatch"
 
 
+class SessionExpiredError(Auth0Error):
+    """
+    Error raised when a session is rejected at login because its
+    session_expiry ceiling is already in the past.
+    """
+    code = AccessTokenErrorCode.SESSION_EXPIRED
+
+    def __init__(self, message: Optional[str] = None, cause=None):
+        super().__init__(message or "The session has expired and the user must re-authenticate.")
+        self.name = "SessionExpiredError"
+        self.cause = cause
+
+
 class CustomTokenExchangeError(Auth0Error):
     """
     Error raised during custom token exchange operations.