Expire and renew the access token when using OAuth 2.0

okta/client.py

@@ -116,7 +116,7 @@ def __init__(self, user_config: dict = {}):
         client_config_setter = ConfigSetter()
         client_config_setter._apply_config({'client': user_config})
         self._config = client_config_setter.get_config()
-        # Prune configuration to remove unnecesary fields
+        # Prune configuration to remove unnecessary fields
         self._config = client_config_setter._prune_config(self._config)
         # Validate configuration
         ConfigValidator(self._config)
@@ -128,6 +128,7 @@ def __init__(self, user_config: dict = {}):
         self._client_id = None
         self._scopes = None
         self._private_key = None
+        self._oauth_token_renewal_offset = None
 
         # Determine which cache to use
         cache = NoOpCache()
@@ -154,6 +155,7 @@ def __init__(self, user_config: dict = {}):
             self._client_id = self._config["client"]["clientId"]
             self._scopes = self._config["client"]["scopes"]
             self._private_key = self._config["client"]["privateKey"]
+            self._oauth_token_renewal_offset = self._config["client"]["oauthTokenRenewalOffset"]
 
         setup_logging(log_level=self._config["client"]["logging"]["logLevel"])
         # Check if logging should be enabled

okta/config/config_setter.py

@@ -37,7 +37,8 @@ class ConfigSetter():
             },
             "rateLimit": {
                 "maxRetries": ''
-            }
+            },
+            "oauthTokenRenewalOffset": ''
         },
         "testing": {
             "testingDisableHttpsCheck": ''
@@ -116,6 +117,7 @@ def _apply_default_values(self):
         self._config["client"]["rateLimit"] = {
             "maxRetries": 2
         }
+        self._config["client"]["oauthTokenRenewalOffset"] = 5
 
         self._config["testing"]["testingDisableHttpsCheck"] = False
 

okta/config/config_validator.py

@@ -45,9 +45,8 @@ def validate_config(self):
                 self._validate_token(
                     client.get('token', ""))
         elif client.get('authorizationMode') == "PrivateKey":
-            client_fields = ['clientId', 'scopes', 'privateKey']
-            client_fields_values = [self._config.get(
-                'client').get(field, "") for field in client_fields]
+            client_fields = ['clientId', 'scopes', 'privateKey', 'oauthTokenRenewalOffset']
+            client_fields_values = [client.get(field, "") for field in client_fields]
             errors += self._validate_client_fields(*client_fields_values)
         else:  # Not a valid authorization mode
             errors += [
@@ -61,7 +60,7 @@ def validate_config(self):
                              f"See {REPO_URL} for usage")
 
     def _validate_client_fields(self, client_id, client_scopes,
-                                client_private_key):
+                                client_private_key, oauth_token_renewal_offset):
         client_fields_errors = []
 
         # check client id
@@ -77,6 +76,14 @@ def _validate_client_fields(self, client_id, client_scopes,
         if not (client_scopes and client_private_key):
             client_fields_errors.append(ERROR_MESSAGE_SCOPES_PK_MISSING)
 
+        # Validate oauthTokenRenewalOffset
+        if not oauth_token_renewal_offset:
+            client_fields_errors.append("oauthTokenRenewalOffset must be provided")
+        if not isinstance(oauth_token_renewal_offset, int):
+            client_fields_errors.append("oauthTokenRenewalOffset must be a valid integer")
+        if isinstance(oauth_token_renewal_offset, int) and oauth_token_renewal_offset < 0:
+            client_fields_errors.append("oauthTokenRenewalOffset must be a non-negative integer")
+
         return client_fields_errors
 
     def _validate_token(self, token: str):

okta/oauth.py

@@ -1,3 +1,4 @@
+import time
 from urllib.parse import urlencode, quote
 from okta.jwt import JWT
 from okta.http_client import HTTPClient
@@ -37,6 +38,14 @@ async def get_access_token(self):
             str, Exception: Tuple of the access token, error that was raised
             (if any)
         """
+
+        # Check if access token has expired or will expire soon
+        current_time = int(time.time())
+        if self._access_token and hasattr(self, '_access_token_expiry_time'):
+            renewal_offset = self._config["client"]["oauthTokenRenewalOffset"] * 60  # Convert minutes to seconds
+            if current_time + renewal_offset >= self._access_token_expiry_time:
+                self.clear_access_token()
+
         # Return token if already generated
         if self._access_token:
             return (self._access_token, None)
@@ -83,6 +92,9 @@ async def get_access_token(self):
 
         # Otherwise set token and return it
         self._access_token = parsed_response["access_token"]
+
+        # Set token expiry time
+        self._access_token_expiry_time = int(time.time()) + parsed_response["expires_in"]
         return (self._access_token, None)
 
     def clear_access_token(self):
@@ -92,3 +104,4 @@ def clear_access_token(self):
         self._access_token = None
         self._request_executor._cache.delete("OKTA_ACCESS_TOKEN")
         self._request_executor._default_headers.pop("Authorization", None)
+        self._access_token_expiry_time = None