Fixed an issue where the SDK would fail to validate webhook signature verification if the signature key was not the one returned in the /public/v3/public-key endpoint (#104)

Author: lordkyzr

CHANGES.rst

@@ -4,6 +4,7 @@ CHANGELOG for LaunchKey Python SDK
 -----
 
 * Fixed an issue where the SDK would improperly report which key was used when the encryption and signature keys differed
+* Fixed an issue where the SDK would fail to validate webhook signature verification if the signature key was not the one returned in the public-key endpoint
 
 3.8.0
 -----

launchkey/transports/jose_auth.py

@@ -572,7 +572,10 @@ def verify_jwt_request(self, compact_jwt, subject, method, path, body):
         :return: The claims of the JWT
         """
         try:
-            self._set_current_kid()
+            # Ensure key exists in cache, fetch from API by ID otherwise
+            kid = JWT().unpack(compact_jwt).headers["kid"]
+            public_key = self._find_key_by_kid(kid)
+            self._cache_public_key(kid, public_key)
             payload = self._get_jwt_payload(compact_jwt)
         except JWKESTException as reason:
             raise JWTValidationFailure("Unable to parse JWT", reason=reason)

tests/test_jose_auth_transport.py

@@ -453,7 +453,13 @@ def setUp(self):
         self._transport.content_hash_function().hexdigest.return_value = self._content_hash
 
         self._jwt_patch = patch("launchkey.transports.jose_auth.JWT", return_value=MagicMock(spec=JWT)).start()
-        self._jwt_patch.return_value.unpack.return_value.headers = faux_jwt_headers
+        self._faux_jwt_headers = {
+            "alg": "RS512",
+            "typ": "JWT",
+            # This should be different than the kid from the transport headers
+            "kid": "ff:ff:ff:ff:ff:ff:ff:ff:ff:ff:ff:ff:ff:ff:ff:ff"
+        }
+        self._jwt_patch.return_value.unpack.return_value.headers = self._faux_jwt_headers
 
         patcher = patch('launchkey.transports.jose_auth.sha256')
         patched = patcher.start()
@@ -474,6 +480,11 @@ def test_none_path_returns_payload(self):
         actual = self._transport.verify_jwt_request(MagicMock(), self.jwt_payload['sub'], 'POST', None, MagicMock())
         self.assertEqual(actual, self.jwt_payload)
 
+    def test_jwt_signed_with_key_not_in_cache_fetches_key(self):
+        self._verify_jwt_request()
+        self._transport.get.assert_called_once_with(
+            "/public/v3/public-key/%s" % self._faux_jwt_headers["kid"])
+
     def test_none_path_still_requires_jwt_request_path(self):
         del self.jwt_payload['request']['path']
         with self.assertRaises(JWTValidationFailure):