Expire and renew the access token when using OAuth 2.0

[GraemeMeyerGT]

This PR fixes #363 (Access token expiry not handled when using OAuth 2.0) by adding logic to expire and renew the access token when using OAuth 2.0 to authenticate with the Okta API. This issue and the fix was also discussed in PR #402 (Feature: Autorenew OAuth tokens]), but I believe this fix is simpler and superior, as it does not require importing any new libraries (other than time) and is more in line with the existing codebase.

I have reviewed the CLA and believe that this falls under the remit of an Obvious Fix and doesn't not require singing the CLA. I would be happy to do so however if the Okta team feels it is necessary.

I have also followed the steps CONTRIBUTING guide when submitting this PR.

I don't have the ability to run the pytest suite on my work PC, but I've run this code manually by modifying the local okta-sdk-python and it works. I've also implemented a hotfix library that imports the Okta code, replaces the OAuth.py functions code with my code and runs it, and it works well.

[bryanapellanes-okta]

@GraemeMeyerGT thanks for your contribution! Please see my comment here: #402 (comment). We hope to release a preview of a new version of the SDK in the coming weeks (a firm date is not yet determined). We would be grateful to have your input once it is ready and perhaps we can work together to address this concern in the next version of the SDK.

Further, we welcome additional comment from @haggrip on this PR to continue to engage our Python developers so we can work together to provide the best SDK experience for all. We'll prioritize internally to determine how best to move forward with this and other outstanding PRs. Thanks again!

[GraemeMeyerGT]

Understood, thanks Bryan. I look forward to trying out the new version soon!

For anyone else dealing with this issue right now, you can use the same work around as me. Essentially I am patching the loaded methods from the Okta library with my fixed versions.

First, create a new file called okta_fix.py like so:

okta_fix.py

import time
from urllib.parse import urlencode, quote
from okta.jwt import JWT
from okta.http_client import HTTPClient
from okta.oauth import OAuth

async def patched_get_access_token(self):
        """
        Retrieves or generates the OAuth access token for the Okta Client

        Returns:
            str, Exception: Tuple of the access token, error that was raised
            (if any)
        """
        
        # Check if access token has expired or will expire in the next 5 minutes
        current_time = int(time.time())
        if self._access_token and hasattr(self, '_access_token_expiry_time'):
            if current_time + 300 >= self._access_token_expiry_time:
                self.clear_access_token()
        
        # Return token if already generated
        if self._access_token:
            return (self._access_token, None)

        # Otherwise create new one
        # Get JWT and create parameters for new Oauth token
        jwt = self.get_JWT()
        parameters = {
            'grant_type': 'client_credentials',
            'scope': ' '.join(self._config["client"]["scopes"]),
            'client_assertion_type':
                'urn:ietf:params:oauth:client-assertion-type:jwt-bearer',
            'client_assertion': jwt
        }

        encoded_parameters = urlencode(parameters, quote_via=quote)
        org_url = self._config["client"]["orgUrl"]
        url = f"{org_url}{OAuth.OAUTH_ENDPOINT}?" + \
            encoded_parameters

        # Craft request
        oauth_req, err = await self._request_executor.create_request(
            "POST", url, None, {
                'Accept': "application/json",
                'Content-Type': 'application/x-www-form-urlencoded'
            }, oauth=True)

        # TODO Make max 1 retry
        # Shoot request
        if err:
            return (None, err)
        _, res_details, res_json, err = \
            await self._request_executor.fire_request(oauth_req)
        # Return HTTP Client error if raised
        if err:
            return (None, err)

        # Check response body for error message
        parsed_response, err = HTTPClient.check_response_for_error(
            url, res_details, res_json)
        # Return specific error if found in response
        if err:
            return (None, err)

        # Otherwise set token and return it
        self._access_token = parsed_response["access_token"]
        
        # Set token expiry time as epoch time
        self._access_token_expiry_time = int(time.time()) + parsed_response["expires_in"]
        return (self._access_token, None)


def patched_clear_access_token(self):
    """
    Clear currently used OAuth access token, probably expired
    """
    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

# Apply the patches
OAuth.get_access_token = patched_get_access_token
OAuth.clear_access_token = patched_clear_access_token

This file is pretty much just the two modified methods, and then at the end they are patched on-top-of the Okta SDK's methods:

# Apply the patches
OAuth.get_access_token = patched_get_access_token
OAuth.clear_access_token = patched_clear_access_token

Now I just import okta_fix.py before any use of the Okta library, and it all works well. E.g. I am writing a Flask app, so I import it right at the top of my app's __init__.py file:

import okta_fix
from flask import Flask
from config import Config
from .extensions import login_manager

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)
...

[haggrip]

I don't want this to be a tough ask, but could this be also present in the config when creating the OktaClient? Have the renewal time also be configurable and default to minutes

[GraemeMeyerGT]

Happy to take on some feedback.

Just to clarify: You want the renewal time - the number of minutes before the token expires that the token is automatically renewed - to be configurable when creating the Okta Client. Are you happy with default of 5 minutes?

So configuring the client might look something like this:

  configuration = {
      'orgUrl': f'https://{current_app.config["OKTA_BASE_URL"]}',
      'authorizationMode': 'PrivateKey',
      'clientId': current_app.config['OKTA_API_CLIENT_ID'],
      'scopes': current_app.config['OKTA_API_SCOPES'],
      'privateKey': current_app.config['OKTA_API_PRIVATE_KEY'],
      'kid': current_app.config['OKTA_API_PRIVATE_KEY_ID'],
      'oauthTokenRenewalOffset': 10 # Example of changing the default of 5 minutes to 10 minutes before expiration
  }
  current_app.okta_client = OktaClient(configuration)

And the token renewal code would look something like:

# 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.get('oauthTokenRenewalOffset', 5) * 60  # Convert minutes to seconds
    if current_time + renewal_offset >= self._access_token_expiry_time:
        self.clear_access_token()

[haggrip]

LGTM. The 5 minute default is okay IMO

[GraemeMeyerGT]

Thanks @haggrip, I've made these changes, and I've integrated the changes more completely into the application's convention/structure - initialising the configuration variable with the Client, setting the default value with the ConfigSetter and applying the default value with _apply_default_values etc.

Hopefully this is what you had in mind. In my manual testing, it works well.

[bryanapellanes-okta]

@GraemeMeyerGT I'm really happy to see the collaboration here! May I request that you add a unit test that tests your change?

[GraemeMeyerGT]

Thanks @bryanapellanes-okta. Unfortunately I don't have any experience with Python tests. I've spent a couple of hours looking at it with an LLM, but TBH I'm not understanding most of what it's spitting out, or any of the existing tests. I'll give myself a crash course in Python testing over the next few days, but I'm not confident yet that I'll be able to come up to the level of this project any time soon.

If you have any pointers on the exact type of test(s) you'd like to see and which file you'd like to see them in, that would be helpful (as would a pointer to one or two of the existing tests that you think would be a good reference).