CG-1904 improve expired session ux (#45)

*    Improve the user experience around old stale sessions that appear to be
    initialized, but are actually expired. This is done by providing the new
    utility method Auth.ensure_request_authenticator_is_ready().

*    Save computed expiration time and issued time in token files. This allows
    for the persistence of this information when dealing with opaque tokens.

*    Support non-expiring tokens.

Author: carl-alexander-adams

docs/changelog.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## 2.3.0 - TBD
+- Improve the user experience around old stale sessions that appear to be
+  initialized, but are actually expired.  This is done by providing the new
+  utility method `Auth.ensure_request_authenticator_is_ready()`.
+- Save computed expiration time and issued time in token files. This allows
+  for the persistence of this information when dealing with opaque tokens.
+  - **Note**: Previously saved OAuth access tokens that are not JWTs with
+    an `exp` claim that can be inspected will be considered to expire in
+    `expires_in` seconds from the time they are loaded, since the time
+    they were issued was not saved in the past.
+- Support non-expiring tokens.
+
 ## 2.2.0 - 2025-10-02
 - Update supported python versions.
   Support for 3.9 dropped.  Support through 3.14 added.

docs/examples/auth-client/oauth/make-oauth-authenticated-httpx-request.py

@@ -6,6 +6,7 @@
 def main():
     logging.basicConfig(level=logging.DEBUG)
     auth_ctx = planet_auth_utils.PlanetAuthFactory.initialize_auth_client_context(auth_profile_opt="my-custom-profile")
+    auth_ctx.ensure_request_authenticator_is_ready(allow_open_browser=True, allow_tty_prompt=True)
     result = httpx.get(
         url="https://api.planet.com/basemaps/v1/mosaics",
         auth=auth_ctx.request_authenticator(),

docs/examples/auth-client/oauth/make-oauth-authenticated-requests-request.py

@@ -6,6 +6,7 @@
 def main():
     logging.basicConfig(level=logging.DEBUG)
     auth_ctx = planet_auth_utils.PlanetAuthFactory.initialize_auth_client_context(auth_profile_opt="my-custom-profile")
+    auth_ctx.ensure_request_authenticator_is_ready(allow_open_browser=True, allow_tty_prompt=True)
     result = requests.get(
         url="https://api.planet.com/basemaps/v1/mosaics",
         auth=auth_ctx.request_authenticator(),

docs/examples/auth-client/oauth/perform-oauth-initial-login-2.py

@@ -0,0 +1,13 @@
+import logging
+import planet_auth_utils
+
+
+def main():
+    logging.basicConfig(level=logging.DEBUG)
+    auth_ctx = planet_auth_utils.PlanetAuthFactory.initialize_auth_client_context(auth_profile_opt="my-custom-profile")
+
+    auth_ctx.ensure_request_authenticator_is_ready(allow_open_browser=True, allow_tty_prompt=True)
+
+
+if __name__ == "__main__":
+    main()

src/planet_auth/__init__.py

@@ -56,7 +56,7 @@
       class exists.
 """
 
-from .auth import Auth
+from .auth import Auth, AuthClientContextException
 from .auth_exception import AuthException
 from .auth_client import AuthClientConfig, AuthClient
 from .credential import Credential
@@ -145,6 +145,7 @@ class exists.
     "Auth",
     "AuthClient",
     "AuthClientConfig",
+    "AuthClientContextException",
     "AuthCodeAuthClient",
     "AuthCodeClientConfig",
     "AuthCodeWithClientSecretAuthClient",

src/planet_auth/auth.py

@@ -18,16 +18,18 @@
 from typing import Optional, Union
 
 from planet_auth.auth_client import AuthClient, AuthClientConfig
+from planet_auth.auth_exception import AuthException
 from planet_auth.credential import Credential
 from planet_auth.request_authenticator import CredentialRequestAuthenticator
 from planet_auth.storage_utils import ObjectStorageProvider
 from planet_auth.logging.auth_logger import getAuthLogger
 
 auth_logger = getAuthLogger()
 
-# class AuthClientContextException(AuthException):
-#     def __init__(self, **kwargs):
-#         super().__init__(**kwargs)
+
+class AuthClientContextException(AuthException):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
 
 
 class Auth:
@@ -113,19 +115,129 @@ def request_authenticator_is_ready(self) -> bool:
         For example, simple API key clients only need an API key in their
         configuration.  OAuth2 user clients need to have performed
         a user login and obtained access or refresh tokens.
+
+        Note: This will not detect when a credential is expired or
+        otherwise invalid.
         """
         return self._request_authenticator.is_initialized() or self._auth_client.can_login_unattended()
 
-    def login(self, **kwargs) -> Credential:
+    def ensure_request_authenticator_is_ready(
+        self, allow_open_browser: Optional[bool] = False, allow_tty_prompt: Optional[bool] = False
+    ) -> None:
+        """
+        Do everything necessary to ensure the request authenticator is ready for use,
+        while still biasing towards just-in-time operations and not making
+        unnecessary network requests or prompts for user interaction.
+
+        This can be more complex than it sounds given the variations in the
+        capabilities of authentication clients and possible session states.
+        Clients may be initialized with active sessions, initialized with stale
+        but still valid sessions, initialized with invalid or expired
+        sessions, or completely uninitialized. The process taken to ensure
+        client readiness with as little user disruption as possible
+        is as follows:
+
+        1. If the client has been logged in and has a non-expired
+           short-term access token, the client will be considered
+           ready without prompting the user or probing the network.
+           This will not require user interaction.
+        2. If the client has not been logged in and is a type that
+           can do so without prompting the user, the client will be
+           considered ready without prompting the user or probing
+           the network. This will not require user interaction.
+           Login will be delayed until it is required.
+        3. If the client has been logged in and has an expired
+           short-term access token, the network will be probed to attempt
+           a refresh of the session. This should not require user interaction.
+           If refresh fails, the user will be prompted to perform a fresh
+           login, requiring user interaction.
+        4. If the client has never been logged in and is a type that
+           requires a user interactive login, a user interactive
+           login will be initiated.
+
+        There still may be conditions where we believe we are
+        ready, but requests still ultimately fail.  For example, if
+        the auth context holds a static API key or username/password, it is
+        assumed to be ready but the credentials could be bad. Even when ready
+        with valid credentia, requests could fail if the service
+        rejects the request due to its own policy configuration.
+
+        Parameters:
+            allow_open_browser: specify whether login is permitted to open
+                a browser window.
+            allow_tty_prompt: specify whether login is permitted to request
+                input from the terminal.
+        """
+
+        def _has_credential() -> bool:
+            # Does not do any JIT checks
+            return self._request_authenticator.is_initialized()
+
+        def _can_obtain_credentials_unattended() -> bool:
+            # Does not do any JIT checks
+            return self._auth_client.can_login_unattended()
+
+        def _is_expired() -> bool:
+            # Does not do any JIT check
+            new_cred = self._request_authenticator.credential(refresh_if_needed=False)
+            if new_cred:
+                return new_cred.is_expired()
+            return True
+
+        # Case #1 above.
+        if _has_credential() and not _is_expired():
+            return
+
+        # Case #2 above.
+        if _can_obtain_credentials_unattended():
+            # Should we fetch one?  We do not by default because the bias is towards
+            #  JIT operations and silent operations.  This so programs can initialize and
+            #  not fail for auth reasons unless the credential is actually needed.
+            return
+
+        # Case #3 above.
+        if _has_credential() and _is_expired():
+            try:
+                # This takes care of making sure the authenticator's credential is
+                # current with the update. No further action needed on our part.
+                new_cred = self._request_authenticator.credential(refresh_if_needed=True)
+                if not new_cred:
+                    raise RuntimeError("Unable to refresh credentials - Unknown error")
+                if new_cred.is_expired():
+                    raise RuntimeError("Unable to refresh credentials - Refreshed credentials are still expired.")
+                return
+            except Exception as e:
+                auth_logger.warning(
+                    msg=f"Failed to refresh expired credentials (Error: {str(e)}).  Attempting interactive login."
+                )
+
+        # Case #4 above.
+        self.login(allow_open_browser=allow_open_browser, allow_tty_prompt=allow_tty_prompt)
+
+    def login(
+        self, allow_open_browser: Optional[bool] = False, allow_tty_prompt: Optional[bool] = False, **kwargs
+    ) -> Credential:
         """
         Perform a login with the configured auth client.
         This higher level function will ensure that the token is saved to
         storage if Auth context has been configured with a suitable token
         storage path.  Otherwise, the token will be held only in memory.
         In all cases, the request authenticator will also be updated with
         the new credentials.
+
+        Parameters:
+            allow_open_browser: specify whether login is permitted to open
+                a browser window.
+            allow_tty_prompt: specify whether login is permitted to request
+                input from the terminal.
         """
-        new_credential = self._auth_client.login(**kwargs)
+        new_credential = self._auth_client.login(
+            allow_open_browser=allow_open_browser, allow_tty_prompt=allow_tty_prompt, **kwargs
+        )
+        if not new_credential:
+            # AuthClient.login() is supposed to raise on failure.
+            raise AuthClientContextException(message="Unknown login failure. No credentials and no error returned.")
+
         new_credential.set_path(self._token_file_path)
         new_credential.set_storage_provider(self._storage_provider)
         new_credential.save()

src/planet_auth/credential.py

@@ -12,6 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+from typing import Optional
+
 from planet_auth.storage_utils import FileBackedJsonObject
 
 
@@ -22,7 +25,69 @@ class Credential(FileBackedJsonObject):
     storage provider implementation, clear-text .json files or .sops.json files
     with field level encryption are supported.  Custom storage providers may
     offer different functionality.
+
+    Subclass Implementor Notes:
+        The `Credential` class reserves the data fields `_iat` and `_exp` for
+        internal use.  These are used to record the time the credential was
+        issued and the time that it expires respectively, expressed as seconds
+        since the epoch.  A None or NULL value for `_exp` indicates that
+        the credential never expires.
+
+        This base class does not set these values.  It is the responsibility
+        of subclasses to set these values as appropriate.  If left unset,
+        the credential will be treated as a non-expiring credential with an
+        indeterminate issued time.
+
+        Subclasses that wish to provide values should do so in their
+        constructor and in their `set_data()` methods.
     """
 
     def __init__(self, data=None, file_path=None, storage_provider=None):
         super().__init__(data=data, file_path=file_path, storage_provider=storage_provider)
+
+    def expiry_time(self) -> Optional[int]:
+        """
+        The time that the credential expires, expressed as seconds since the epoch.
+        """
+        return self.lazy_get("_exp")
+
+    def issued_time(self) -> Optional[int]:
+        """
+        The time that the credential was issued, expressed as seconds since the epoch.
+        """
+        return self.lazy_get("_iat")
+
+    def is_expiring(self) -> bool:
+        """
+        Return true if the credential has an expiry time.
+        """
+        return self.expiry_time() is not None
+
+    def is_non_expiring(self) -> bool:
+        """
+        Return true if the credential never expires.
+        """
+        return not self.is_expiring()
+
+    def is_expired(self, at_time: Optional[int] = None) -> bool:
+        """
+        Return true if the credential is expired at the specified time.
+        If no time is specified, the current time is used.
+        Non-expiring credentials will always return false.
+        """
+        if self.is_non_expiring():
+            return False
+
+        if at_time is None:
+            at_time = int(time.time())
+
+        exp = self.expiry_time()
+        return bool(at_time >= exp)  # type: ignore[operator]
+
+    def is_not_expired(self, at_time: Optional[int] = None) -> bool:
+        """
+        Return true if the credential is not expired at the specified time.
+        If no time is specified, the current time is used.
+        Non-expiring credentials will always return true.
+        """
+        return not self.is_expired(at_time)

src/planet_auth/oidc/oidc_credential.py

@@ -12,19 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
 from typing import Optional
 
 from planet_auth.credential import Credential
 from planet_auth.storage_utils import InvalidDataException, ObjectStorageProvider
+from planet_auth.oidc.token_validator import TokenValidator, InvalidArgumentException
 
 
 class FileBackedOidcCredential(Credential):
     """
     Credential object for storing OAuth/OIDC tokens.
+    Credential should conform to the "token" response
+    defined in RFC 6749 for OAuth access tokens with OIDC extensions
+    for ID tokens.
     """
 
     def __init__(self, data=None, credential_file=None, storage_provider: Optional[ObjectStorageProvider] = None):
         super().__init__(data=data, file_path=credential_file, storage_provider=storage_provider)
+        self._augment_rfc6749_data()
 
     def check_data(self, data):
         """
@@ -36,6 +42,65 @@ def check_data(self, data):
                 message="'access_token', 'id_token', or 'refresh_token' not found in file {}".format(self._file_path)
             )
 
+    def _augment_rfc6749_data(self):
+        # RFC 6749 includes an optional "expires_in" expressing the lifespan of
+        # the token.  But without knowing when a token was issued it tells us
+        # nothing about whether a token is actually valid.
+        #
+        # This function lest us augment our representation of this data to
+        # make this credential useful when reconstructed from saved data
+        # at a time that is distant from when the token was obtained from the
+        # authorization server.
+        #
+        # Edge case - It's possible that a JWT ID token has an expiration time
+        # that is different from the access token. It's also possible that
+        # we have a refresh token and not any other tokens (this state could
+        # be used for bootstrapping). We are really only tracking
+        # access token expiration at this time.
+        if not self._data:
+            return
+
+        try:
+            access_token_str = self.access_token()
+            if access_token_str:
+                (_, jwt_hazmat_body, _) = TokenValidator.hazmat_unverified_decode(access_token_str)
+            else:
+                jwt_hazmat_body = None
+        except InvalidArgumentException:
+            # Proceed as if it's not a JWT.
+            jwt_hazmat_body = None
+
+        # It's possible for the combination of a transparent bearer token,
+        # saved iat and exp values, and a expires_in value to be
+        # over-constrained.  We apply the following priority, from highest
+        # to lowest:
+        #     - Bearer token claims
+        #     - Saved values in the credential file
+        #     - Newly calculated values
+        # If a reasonable expiration time cannot be derived,
+        # tokens are assumed to never expire.
+        rfc6749_lifespan = self._data.get("expires_in", 0)
+        if jwt_hazmat_body:
+            _iat = jwt_hazmat_body.get("iat", self._data.get("_iat", int(time.time())))
+            _exp = jwt_hazmat_body.get("exp", self._data.get("_exp", None))
+        else:
+            _iat = self._data.get("_iat", int(time.time()))
+            _exp = self._data.get("_exp", None)
+
+        if _exp is None and rfc6749_lifespan > 0:
+            _exp = _iat + rfc6749_lifespan
+
+        self._data["_iat"] = _iat
+        self._data["_exp"] = _exp
+
+    def set_data(self, data, copy_data: bool = True):
+        """
+        Set credential data for an OAuth/OIDC credential.  The data structure is expected
+        to be an RFC 6749 /token response structure.
+        """
+        super().set_data(data, copy_data)
+        self._augment_rfc6749_data()
+
     def access_token(self):
         """
         Get the current access token.