Merge pull request #13 from Authress/improve-caching-verify-token

Improve caching in verify_token.

Author: wparad

CHANGELOG.md

@@ -14,6 +14,7 @@ This is the changelog for [Authress SDK](readme.md).
 * [Breaking] Renamed `AccessRecordStatement` model to `Statement` in `models.statement.py`.
 * [Breaking] Renamed `AccessRecordResource` model to `Resource` in `models.resource.py`.
 * Add missing `If-Unmodified-Since` support to the `update_group` in the `Groups` API.
+* Improve caching in `verify_token`
 
 ## 2.0 ##
 * Add support for users and groups at the statement level of access records.

authress/api/token_verifier.py

@@ -51,7 +51,7 @@ def verify_token(self, authressCustomDomain, token, options=None):
     if (clientIdMatcher is not None and clientIdMatcher.group(1) != unverifiedPayload['sub']):
       raise Exception("Unauthorized", "Service ID does not match token sub claim")
 
-    jwk = options['expectedPublicKey'] if options is not None and 'expectedPublicKey' in options else self.get_public_key(f"{issuer}/.well-known/openid-configuration/jwks", kid)
+    jwk = self.get_public_key(f"{issuer}/.well-known/openid-configuration/jwks", kid)
 
     try:
       return jwt.decode(authenticationToken, jwt.api_jwk.PyJWK.from_dict(jwk).key, algorithms=['EdDSA'], options = { 'verify_aud': False })
@@ -61,7 +61,7 @@ def verify_token(self, authressCustomDomain, token, options=None):
   def get_public_key(self, jwkKeyListUrl, kid):
     hashKey = f"{jwkKeyListUrl}|{kid}"
 
-    if hashKey in self.keyMap is not None:
+    if hashKey in self.keyMap is None:
       self.keyMap[hashKey] = self.get_key_uncached(jwkKeyListUrl, kid)
 
     try:

test/test_token_verifier.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import
 
 import unittest
+import unittest.mock
 
 from authress.api import token_verifier
 
@@ -22,5 +23,27 @@ def test_get_token_for_eddsa(self):
     access_key = 'CLIENT.KEY.ACCOUNT.MC4CAQAwBQYDK2VwBCIEIDVjjrIVCH3dVRq4ixRzBwjVHSoB2QzZ2iJuHq1Wshwp'
     publicKey = { 'alg': 'EdDSA', 'kty': 'OKP', 'crv': 'Ed25519', 'x': 'JxtSC5tZZJuaW7Aeu5Kh_3tgCpPZRkHaaFyTj5sQ3KU' }
 
-    identity = token_verifier.TokenVerifier().verify_token(authressCustomDomain=f"https://{customDomain}", token=access_key, options={ 'expectedPublicKey': publicKey })
-    assert identity['iss'] == f'https://{customDomain}/v1/clients/CLIENT'
+    token_verifier_instance = token_verifier.TokenVerifier()
+
+    mock_get_key_uncached = unittest.mock.MagicMock(return_value=publicKey)
+    token_verifier_instance.get_key_uncached = mock_get_key_uncached
+    identity = token_verifier_instance.verify_token(authressCustomDomain=f"https://{customDomain}", token=access_key)
+
+    mock_get_key_uncached.assert_called_once_with(f"https://{customDomain}/v1/clients/CLIENT/.well-known/openid-configuration/jwks", "KEY")
+    assert identity['iss'] == f'https://{customDomain}/v1/clients/CLIENT'
+    assert identity['sub'] == "CLIENT"
+
+  def test_get_public_key(self):
+    token_verifier_instance = token_verifier.TokenVerifier()
+
+    test_key_value = "TestKeyValue"
+    mock_get_key_uncached = unittest.mock.MagicMock(return_value=test_key_value)
+    token_verifier_instance.get_key_uncached = mock_get_key_uncached
+
+    public_key_1 = token_verifier_instance.get_public_key(f'https://{customDomain}/v1/clients/CLIENT', "Test KID")
+    public_key_2 = token_verifier_instance.get_public_key(f'https://{customDomain}/v1/clients/CLIENT', "Test KID")
+
+    mock_get_key_uncached.assert_called_once_with(f'https://{customDomain}/v1/clients/CLIENT', "Test KID")
+
+    assert public_key_1 == public_key_2
+    assert public_key_1 == test_key_value