Sem-Ver: bugfix Catch JtiUniquenessException and respond with a 401 inside _process_asap_token.

Signed-off-by: David Black <[email protected]>

Author: dbaxa

atlassian_jwt_auth/frameworks/common/asap.py

@@ -3,6 +3,7 @@
 from atlassian_jwt_auth.exceptions import (
     PublicKeyRetrieverException,
     NoTokenProvidedError,
+    JtiUniquenessException,
 )
 
 
@@ -52,6 +53,10 @@ def _process_asap_token(request, backend, settings, verifier=None):
         error_response = backend.get_401_response(
             'Unauthorized: Invalid token', request=request
         )
+    except JtiUniquenessException:
+        error_response = backend.get_401_response(
+            'Unauthorized: Invalid token - duplicate jti', request=request
+        )
 
     if error_response is not None and settings.ASAP_REQUIRED:
         return error_response

atlassian_jwt_auth/frameworks/django/tests/test_django.py

@@ -94,6 +94,16 @@ def check_response(self,
     def test_request_with_valid_token_is_allowed(self):
         self.check_response('needed', 'one', 200)
 
+    def test_request_with_duplicate_jti_is_rejected(self):
+        token = create_token(
+            issuer='client-app', audience='server-app',
+            key_id='client-app/key01', private_key=self._private_key_pem
+        )
+        str_auth = 'Bearer ' + token.decode(encoding='iso-8859-1')
+        self.check_response('needed', 'one', 200, authorization=str_auth)
+        self.check_response('needed', 'duplicate jti', 401,
+                            authorization=str_auth)
+
     def test_request_with_string_headers_is_allowed(self):
         token = create_token(
             issuer='client-app', audience='server-app',

atlassian_jwt_auth/frameworks/flask/tests/test_flask.py

@@ -64,6 +64,15 @@ def test_request_with_valid_token_is_allowed(self):
         )
         self.assertEqual(self.send_request(token).status_code, 200)
 
+    @unittest.expectedFailure
+    def test_request_with_duplicate_jti_is_rejected(self):
+        token = create_token(
+            'client-app', 'server-app',
+            'client-app/key01', self._private_key_pem
+        )
+        self.assertEqual(self.send_request(token).status_code, 200)
+        self.assertEqual(self.send_request(token).status_code, 401)
+
     def test_request_with_invalid_audience_is_rejected(self):
         token = create_token(
             'client-app', 'invalid-audience',

atlassian_jwt_auth/frameworks/wsgi/tests/test_wsgi.py

@@ -34,7 +34,7 @@ def setUp(self):
     def get_app_with_middleware(self, config):
         return ASAPMiddleware(app, config)
 
-    def send_request(self, url='/', config=None, token=None):
+    def send_request(self, url='/', config=None, token=None, application=None):
         """ returns the response of sending a request containing the given
             token sent in the Authorization header.
         """
@@ -48,9 +48,9 @@ def start_response(status, response_headers, exc_info=None):
         environ = {}
         if token:
             environ['HTTP_AUTHORIZATION'] = b'Bearer ' + token
-
-        app = self.get_app_with_middleware(config or self.config)
-        return app(environ, start_response), resp_info, environ
+        if application is None:
+            application = self.get_app_with_middleware(config or self.config)
+        return application(environ, start_response), resp_info, environ
 
     def test_request_with_valid_token_is_allowed(self):
         token = create_token(
@@ -61,6 +61,19 @@ def test_request_with_valid_token_is_allowed(self):
         self.assertEqual(resp_info['status'], '200 OK')
         self.assertIn('ATL_ASAP_CLAIMS', environ)
 
+    def test_request_with_duplicate_jti_is_rejected(self):
+        token = create_token(
+            'client-app', 'server-app',
+            'client-app/key01', self._private_key_pem
+        )
+        application = self.get_app_with_middleware(self.config)
+        body, resp_info, environ = self.send_request(
+            token=token, application=application)
+        self.assertEqual(resp_info['status'], '200 OK')
+        body, resp_info, environ = self.send_request(
+            token=token, application=application)
+        self.assertEqual(resp_info['status'], '401 Unauthorized')
+
     def test_request_with_invalid_audience_is_rejected(self):
         token = create_token(
             'client-app', 'invalid-audience',