Merge pull request #72 from richardkeit/expiry

feat(ApiAuth): reduce probability of token expiry moments after retrieval

Author: Anton0

README.md

@@ -21,6 +21,16 @@ export STAX_ACCESS_KEY=<your_access_key>
 export STAX_SECRET_KEY=<your_secret_key>
 ```
 
+*Optional configuration:*
+
+##### Authentication token expiry
+
+Allows configuration of the threshold to when the Auth library should re-cache the credentials
+*Suggested use when running within CI/CD tools to reduce overall auth calls*
+~~~bash
+export TOKEN_EXPIRY_THRESHOLD_IN_MINS=2 # Type: Integer representing minutes
+~~~
+
 ## Usage
 
 ### Read Accounts

staxapp/auth.py

@@ -1,5 +1,6 @@
 #!/usr/local/bin/python3
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
+from os import environ
 
 import boto3
 from aws_requests_auth.aws_auth import AWSRequestsAuth
@@ -139,7 +140,10 @@ def requests_auth(username, password, **kwargs):
 class ApiTokenAuth:
     @staticmethod
     def requests_auth(username, password, **kwargs):
-        if StaxConfig.expiration and StaxConfig.expiration > datetime.now(timezone.utc):
+        # Minimize the potentical for token to expire while still being used for auth (say within a lambda function)
+        if StaxConfig.expiration and StaxConfig.expiration - timedelta(
+            minutes=int(environ.get("TOKEN_EXPIRY_THRESHOLD_IN_MINS", 1))
+        ) > datetime.now(timezone.utc):
             return StaxConfig.auth
 
         return StaxAuth("ApiAuth").requests_auth(username, password, **kwargs)

tests/test_auth.py

@@ -6,6 +6,7 @@
 """
 
 import unittest
+from unittest.mock import patch
 import jwt
 import botocore
 import requests
@@ -15,6 +16,7 @@
 from botocore.client import Config as BotoConfig
 from botocore.stub import Stubber, ANY
 from datetime import datetime, timedelta, timezone
+from os import environ
 
 from staxapp.api import Api
 from staxapp.auth import StaxAuth, ApiTokenAuth, RootAuth
@@ -156,7 +158,7 @@ def testCredsClient(self):
                 expected_params=expected_parameters,
             )
         self.cognito_stub.activate()
-        
+
         with self.assertRaises(InvalidCredentialsException) as e:
             sa.sts_from_cognito_identity_pool(jwt_token.get("sub"), cognito_client=self.cognito_client)
 
@@ -305,6 +307,52 @@ def testApiTokenAuth(self):
         )
         self.assertIsNotNone(StaxConfig.auth)
 
+
+    @patch("test_auth.StaxAuth.requests_auth")
+    def testApiTokenAuthExpiring(self, requests_auth_mock):
+        """
+        Test credentials close to expiration get refreshed
+        """
+        sa = StaxAuth("ApiAuth")
+        StaxConfig = Config
+        ## expiration 20 minutes in the future, no need to refresh
+        StaxConfig.expiration = datetime.now(timezone.utc) + timedelta(minutes=20)
+
+        ApiTokenAuth.requests_auth(
+            "username",
+            "password",
+            srp_client=self.aws_srp_client,
+            cognito_client=self.cognito_client,
+        )
+        requests_auth_mock.assert_not_called()
+
+        requests_auth_mock.reset_mock()
+        ## expiration in 5 seconds from now, refresh to avoid token becoming stale used
+        StaxConfig.expiration = datetime.now(timezone.utc) + timedelta(seconds=5)
+
+        ApiTokenAuth.requests_auth(
+            "username",
+            "password",
+            srp_client=self.aws_srp_client,
+            cognito_client=self.cognito_client,
+        )
+        requests_auth_mock.assert_called_once()
+
+
+        requests_auth_mock.reset_mock()
+        ## expiration in 5 minutes from now, refresh to avoid token becoming stale used
+        ## override default triggering library to not refresh
+        environ["TOKEN_EXPIRY_THRESHOLD_IN_MINS"] = "10"
+        StaxConfig.expiration = datetime.now(timezone.utc) + timedelta(minutes=2)
+
+        ApiTokenAuth.requests_auth(
+            "username",
+            "password",
+            srp_client=self.aws_srp_client,
+            cognito_client=self.cognito_client,
+        )
+        requests_auth_mock.assert_called_once()
+
     def testRootAuthNotExpired(self):
         """
         Test credentials have not expired
@@ -359,5 +407,6 @@ def testApiAuth(self):
         self.assertIsNotNone(Api._requests_auth)
 
 
+
 if __name__ == "__main__":
     unittest.main()