Merge pull request #229 from johngian/pr-196

Added RS256 verification through OP JWKS endpoint

HISTORY.rst

@@ -9,6 +9,8 @@ History
 * Add OIDC_AUTHENTICATION_CALLBACK_URL as a new configuration parameter
 * Fail earlier when JWS algorithm does not OIDC_RP_SIGN_ALGO.
   Thanks `@anlutro`_
+* RS256 verification through ``settings.OIDC_OP_JWKS_ENDPOINT``
+  Thanks `@GermanoGuerrini`_
 
 Backwards-incompatible changes:
 

docs/installation.rst

@@ -47,6 +47,24 @@ The OpenID Connect provider (OP) will then give you the following:
 
 You'll need these values for settings.
 
+Choose the appropriate algorithm
+--------------------------------
+
+Depending on your OpenID Connect provider (OP) you might need to change the
+default signing algorithm from ``HS256`` to ``RS256`` by settings the
+``OIDC_RP_SIGN_ALGO`` value accordingly.
+
+For ``RS256`` algorithm to work, you need to set either the OP signing key or
+the OP JWKS Endpoint.
+
+The corresponding settings values are:
+
+.. code-block:: python
+
+    OIDC_RP_IDP_SIGN_KEY = "<OP signing key in PEM or DER format>"
+    OIDC_OP_JWKS_ENDPOINT = "<URL of the OIDC OP jwks endpoint>"
+
+If both specified, the key takes precedence.
 
 Add settings to settings.py
 ---------------------------

mozilla_django_oidc/auth.py

@@ -17,7 +17,7 @@
 from django.utils import six
 
 from josepy.jwk import JWK
-from josepy.jws import JWS
+from josepy.jws import JWS, Header
 
 from mozilla_django_oidc.utils import absolutify, import_from_settings
 
@@ -51,13 +51,16 @@ def __init__(self, *args, **kwargs):
         """Initialize settings."""
         self.OIDC_OP_TOKEN_ENDPOINT = import_from_settings('OIDC_OP_TOKEN_ENDPOINT')
         self.OIDC_OP_USER_ENDPOINT = import_from_settings('OIDC_OP_USER_ENDPOINT')
+        self.OIDC_OP_JWKS_ENDPOINT = import_from_settings('OIDC_OP_JWKS_ENDPOINT', None)
         self.OIDC_RP_CLIENT_ID = import_from_settings('OIDC_RP_CLIENT_ID')
         self.OIDC_RP_CLIENT_SECRET = import_from_settings('OIDC_RP_CLIENT_SECRET')
         self.OIDC_RP_SIGN_ALGO = import_from_settings('OIDC_RP_SIGN_ALGO', 'HS256')
         self.OIDC_RP_IDP_SIGN_KEY = import_from_settings('OIDC_RP_IDP_SIGN_KEY', None)
 
-        if self.OIDC_RP_SIGN_ALGO.startswith('RS') and self.OIDC_RP_IDP_SIGN_KEY is None:
-            raise ImproperlyConfigured('IDP Signing key not provided with RS signing algorithm')
+        if (self.OIDC_RP_SIGN_ALGO.startswith('RS') and
+                (self.OIDC_RP_IDP_SIGN_KEY is None and self.OIDC_OP_JWKS_ENDPOINT is None)):
+            msg = '{} alg requires OIDC_RP_IDP_SIGN_KEY or OIDC_OP_JWKS_ENDPOINT to be configured.'
+            raise ImproperlyConfigured(msg.format(self.OIDC_RP_SIGN_ALGO))
 
         self.UserModel = get_user_model()
 
@@ -97,7 +100,6 @@ def update_user(self, user, claims):
 
     def _verify_jws(self, payload, key):
         """Verify the given JWS payload with the given key and return the payload"""
-
         jws = JWS.from_compact(payload)
 
         try:
@@ -111,28 +113,57 @@ def _verify_jws(self, payload, key):
                   "OIDC_RP_SIGN_ALGO.".format(alg)
             raise SuspiciousOperation(msg)
 
-        jwk = JWK.load(key)
+        if isinstance(key, six.string_types):
+            # Use smart_bytes here since the key string comes from settings.
+            jwk = JWK.load(smart_bytes(key))
+        else:
+            # The key is a json returned from the IDP JWKS endpoint.
+            jwk = JWK.from_json(key)
+
         if not jws.verify(jwk):
             msg = 'JWS token verification failed.'
             raise SuspiciousOperation(msg)
 
         return jws.payload
 
+    def retrieve_matching_jwk(self, token):
+        """Get the signing key by exploring the JWKS endpoint of the OP."""
+        response_jwks = requests.get(
+            self.OIDC_OP_JWKS_ENDPOINT,
+            verify=import_from_settings('OIDC_VERIFY_SSL', True)
+        )
+        response_jwks.raise_for_status()
+        jwks = response_jwks.json()
+
+        # Compute the current header from the given token to find a match
+        jws = JWS.from_compact(token)
+        json_header = jws.signature.protected
+        header = Header.json_loads(json_header)
+
+        key = None
+        for jwk in jwks['keys']:
+            if jwk['alg'] == smart_text(header.alg) and jwk['kid'] == smart_text(header.kid):
+                key = jwk
+        if key is None:
+            raise SuspiciousOperation('Could not find a valid JWKS.')
+        return key
+
     def verify_token(self, token, **kwargs):
         """Validate the token signature."""
         nonce = kwargs.get('nonce')
 
+        token = force_bytes(token)
         if self.OIDC_RP_SIGN_ALGO.startswith('RS'):
-            key = self.OIDC_RP_IDP_SIGN_KEY
+            if self.OIDC_RP_IDP_SIGN_KEY is not None:
+                key = self.OIDC_RP_IDP_SIGN_KEY
+            else:
+                key = self.retrieve_matching_jwk(token)
         else:
             key = self.OIDC_RP_CLIENT_SECRET
 
         # Verify the token
-        verified_token = self._verify_jws(
-            force_bytes(token),
-            # Use smart_bytes here since the key string comes from settings.
-            smart_bytes(key),
-        )
+        verified_token = self._verify_jws(token, key)
+
         # The 'verified_token' will always be a byte string since it's
         # the result of base64.urlsafe_b64decode().
         # The payload is always the result of base64.urlsafe_b64decode().

tests/test_auth.py

@@ -345,7 +345,7 @@ def test_jwt_decode_params(self, request_mock, jws_mock):
         request_mock.post.return_value = post_json_mock
         self.backend.authenticate(request=auth_request)
         calls = [
-            call(force_bytes('token'), force_bytes('client_secret'))
+            call(force_bytes('token'), 'client_secret')
         ]
         jws_mock.assert_has_calls(calls)
 
@@ -375,7 +375,7 @@ def test_jwt_decode_params_verify_false(self, request_mock, jws_mock):
         }
         request_mock.post.return_value = post_json_mock
         calls = [
-            call(force_bytes('token'), force_bytes('client_secret'))
+            call(force_bytes('token'), 'client_secret')
         ]
         self.backend.authenticate(request=auth_request)
         jws_mock.assert_has_calls(calls)
@@ -601,5 +601,102 @@ def update_user(user, claims):
         self.assertEqual(User.objects.get().first_name, 'a_username')
 
 
+class OIDCAuthenticationBackendRS256WithKeyTestCase(TestCase):
+    """Authentication tests with ALG RS256 and provided IdP Sign Key."""
+
+    @override_settings(OIDC_OP_TOKEN_ENDPOINT='https://server.example.com/token')
+    @override_settings(OIDC_OP_USER_ENDPOINT='https://server.example.com/user')
+    @override_settings(OIDC_RP_CLIENT_ID='example_id')
+    @override_settings(OIDC_RP_CLIENT_SECRET='client_secret')
+    @override_settings(OIDC_RP_SIGN_ALGO='RS256')
+    @override_settings(OIDC_RP_IDP_SIGN_KEY='sign_key')
+    def setUp(self):
+        self.backend = OIDCAuthenticationBackend()
+
+    @override_settings(OIDC_USE_NONCE=False)
+    @patch('mozilla_django_oidc.auth.OIDCAuthenticationBackend._verify_jws')
+    @patch('mozilla_django_oidc.auth.requests')
+    def test_jwt_verify_sign_key(self, request_mock, jws_mock):
+        """Test jwt verification signature."""
+        auth_request = RequestFactory().get('/foo', {'code': 'foo',
+                                                     'state': 'bar'})
+        auth_request.session = {}
+
+        jws_mock.return_value = json.dumps({
+            'aud': 'audience'
+        }).encode('utf-8')
+        get_json_mock = Mock()
+        get_json_mock.json.return_value = {
+            'nickname': 'username',
+            'email': 'email@example.com'
+        }
+        request_mock.get.return_value = get_json_mock
+        post_json_mock = Mock()
+        post_json_mock.json.return_value = {
+            'id_token': 'token',
+            'access_token': 'access_token'
+        }
+        request_mock.post.return_value = post_json_mock
+        self.backend.authenticate(request=auth_request)
+        calls = [
+            call(force_bytes('token'), 'sign_key')
+        ]
+        jws_mock.assert_has_calls(calls)
+
+
+class OIDCAuthenticationBackendRS256WithJwksEndpointTestCase(TestCase):
+    """Authentication tests with ALG RS256 and IpD JWKS Endpoint."""
+
+    @override_settings(OIDC_OP_TOKEN_ENDPOINT='https://server.example.com/token')
+    @override_settings(OIDC_OP_USER_ENDPOINT='https://server.example.com/user')
+    @override_settings(OIDC_RP_CLIENT_ID='example_id')
+    @override_settings(OIDC_RP_CLIENT_SECRET='client_secret')
+    @override_settings(OIDC_RP_SIGN_ALGO='RS256')
+    @override_settings(OIDC_OP_JWKS_ENDPOINT='https://server.example.com/jwks')
+    def setUp(self):
+        self.backend = OIDCAuthenticationBackend()
+
+    @override_settings(OIDC_USE_NONCE=False)
+    @patch('mozilla_django_oidc.auth.OIDCAuthenticationBackend._verify_jws')
+    @patch('mozilla_django_oidc.auth.OIDCAuthenticationBackend.retrieve_matching_jwk')
+    @patch('mozilla_django_oidc.auth.requests')
+    def test_jwt_verify_sign_key(self, request_mock, jwk_mock, jws_mock):
+        """Test jwt verification signature."""
+        auth_request = RequestFactory().get('/foo', {'code': 'foo',
+                                                     'state': 'bar'})
+        auth_request.session = {}
+
+        jwk_mock_ret = {
+            "kty": "RSA",
+            "alg": "RS256",
+            "use": "sig",
+            "kid": "cc7d29c9cb3780741cc0876633c9107a0f33c289",
+            "n": "20LvblCBaPicNV3-NnJuahqbpi-b8hFD",
+            "e": "AQAB"
+        }
+        jwk_mock.return_value = jwk_mock_ret
+
+        jws_mock.return_value = json.dumps({
+            'aud': 'audience'
+        }).encode('utf-8')
+        get_json_mock = Mock()
+        get_json_mock.json.return_value = {
+            'nickname': 'username',
+            'email': 'email@example.com'
+        }
+        request_mock.get.return_value = get_json_mock
+        post_json_mock = Mock()
+        post_json_mock.json.return_value = {
+            'id_token': 'token',
+            'access_token': 'access_token'
+        }
+        request_mock.post.return_value = post_json_mock
+        self.backend.authenticate(request=auth_request)
+        calls = [
+            call(force_bytes('token'), jwk_mock_ret)
+        ]
+        jws_mock.assert_has_calls(calls)
+
+
 def dotted_username_algo_callback(email):
     return 'dotted_username_algo'