Merge pull request #13 from s-hal/authz_code_owner

Fix code exchange ensure the authz code was issued to the requesting client

Author: Gijutsu

src/pyop/client_authentication.py

@@ -61,4 +61,4 @@ def verify_client_authentication(clients, parsed_request, authz_header=None):
         raise InvalidClientAuthentication(
             'Wrong authentication method used, MUST use \'{}\''.format(expected_authn_method))
 
-    return True
+    return client_id

src/pyop/provider.py

@@ -29,6 +29,7 @@
 from .exceptions import AuthorizationError
 from .exceptions import InvalidAccessToken
 from .exceptions import InvalidTokenRequest
+from .exceptions import InvalidAuthorizationCode
 from .request_validator import authorization_request_verify
 from .request_validator import client_id_is_known
 from .request_validator import client_preferences_match_provider_capabilities
@@ -339,7 +340,11 @@ def _do_code_exchange(self, request,  # type: Dict[str, str]
             raise InvalidTokenRequest(str(e), token_request) from e
 
         authentication_request = self.authz_state.get_authorization_request_for_code(token_request['code'])
-
+        
+        if token_request['client_id'] != authentication_request['client_id']:
+            logger.info('Authorization code \'%s\' belonging to \'%s\' was used by \'%s\'',
+                        token_request['code'], authentication_request['client_id'], token_request['client_id'])
+            raise InvalidAuthorizationCode('{} unknown'.format(token_request['code']))
         if token_request['redirect_uri'] != authentication_request['redirect_uri']:
             raise InvalidTokenRequest('Invalid redirect_uri: {} != {}'.format(token_request['redirect_uri'],
                                                                               authentication_request['redirect_uri']),
@@ -405,7 +410,7 @@ def _verify_client_authentication(self, request_body, http_headers=None):
         if http_headers is None:
             http_headers = {}
         token_request = dict(parse_qsl(request_body))
-        verify_client_authentication(self.clients, token_request, http_headers.get('Authorization'))
+        token_request['client_id']  = verify_client_authentication(self.clients, token_request, http_headers.get('Authorization'))
         return token_request
 
     def handle_userinfo_request(self, request=None, http_headers=None):

tests/pyop/test_client_authentication.py

@@ -41,21 +41,21 @@ def test_wrong_authentication_method(self):
     def test_authentication_method_defaults_to_client_secret_basic(self):
         del self.clients[TEST_CLIENT_ID]['token_endpoint_auth_method']
         authz_header = self.create_basic_auth()
-        assert verify_client_authentication(self.clients, self.token_request_args, authz_header)
+        assert verify_client_authentication(self.clients, self.token_request_args, authz_header) == TEST_CLIENT_ID
 
     def test_client_secret_post(self):
         self.clients[TEST_CLIENT_ID]['token_endpoint_auth_method'] = 'client_secret_post'
-        assert verify_client_authentication(self.clients, self.token_request_args)
+        assert verify_client_authentication(self.clients, self.token_request_args) == TEST_CLIENT_ID
 
     def test_client_secret_basic(self):
         self.clients[TEST_CLIENT_ID]['token_endpoint_auth_method'] = 'client_secret_basic'
         authz_header = self.create_basic_auth()
-        assert verify_client_authentication(self.clients, self.token_request_args, authz_header)
+        assert verify_client_authentication(self.clients, self.token_request_args, authz_header) == TEST_CLIENT_ID
 
     def test_unknown_client_id(self):
         self.token_request_args['client_id'] = 'unknown'
         with pytest.raises(InvalidClientAuthentication):
-            verify_client_authentication(self.clients, self.token_request_args)
+            verify_client_authentication(self.clients, self.token_request_args) == TEST_CLIENT_ID
 
     def test_wrong_client_secret(self):
         self.token_request_args['client_secret'] = 'foobar'
@@ -68,7 +68,7 @@ def test_public_client_no_auth(self):
         self.clients[TEST_CLIENT_ID]['token_endpoint_auth_method'] = 'none'
         del self.clients[TEST_CLIENT_ID]['client_secret']
 
-        assert verify_client_authentication(self.clients, self.token_request_args, None)
+        assert verify_client_authentication(self.clients, self.token_request_args, None) == TEST_CLIENT_ID
 
     def test_invalid_authorization_scheme(self):
         authz_header = self.create_basic_auth()
@@ -78,4 +78,4 @@ def test_invalid_authorization_scheme(self):
 
     def test_invalid_userid_password(self):
         with pytest.raises(InvalidClientAuthentication):
-            verify_client_authentication(self.clients, self.token_request_args, 'Basic invalid')
+            verify_client_authentication(self.clients, self.token_request_args, 'Basic invalid')

tests/pyop/test_provider.py

@@ -26,6 +26,8 @@
 
 TEST_CLIENT_ID = 'client1'
 TEST_CLIENT_SECRET = 'secret'
+INVALID_TEST_CLIENT_ID = 'invalid_client'
+INVALID_TEST_CLIENT_SECRET = 'invalid_secret'
 TEST_REDIRECT_URI = 'https://client.example.com'
 ISSUER = 'https://provider.example.com'
 TEST_USER_ID = 'user1'
@@ -71,7 +73,14 @@ def inject_provider(request):
             'client_secret': TEST_CLIENT_SECRET,
             'token_endpoint_auth_method': 'client_secret_post',
             'post_logout_redirect_uris': ['https://client.example.com/post_logout']
-
+        },
+        INVALID_TEST_CLIENT_ID: {
+            'subject_type': 'pairwise',
+            'redirect_uris': 'http://invalid.redirect.loc',
+            'response_types': ['code'],
+            'client_secret': INVALID_TEST_CLIENT_SECRET,
+            'token_endpoint_auth_method': 'client_secret_post',
+            'post_logout_redirect_uris': ['https://client.example.com/post_logout']
         }
     }
 
@@ -337,6 +346,13 @@ def test_handle_token_request_reject_invalid_client_authentication(self):
             self.provider.handle_token_request(urlencode(self.authorization_code_exchange_request_args),
                                                extra_id_token_claims={'foo': 'bar'})
 
+    def test_handle_token_request_reject_code_not_issued_to_client(self):
+        self.authorization_code_exchange_request_args['client_id'] = INVALID_TEST_CLIENT_ID
+        self.authorization_code_exchange_request_args['client_secret'] = INVALID_TEST_CLIENT_SECRET
+        self.authorization_code_exchange_request_args['code'] = self.create_authz_code()
+        with pytest.raises(InvalidAuthorizationCode):
+            self.provider.handle_token_request(urlencode(self.authorization_code_exchange_request_args))
+
     def test_handle_token_request_reject_invalid_redirect_uri_in_exchange_request(self):
         self.authorization_code_exchange_request_args['redirect_uri'] = 'https://invalid.com'
         self.authorization_code_exchange_request_args['code'] = self.create_authz_code()