feat: add jwt_auth_failed custom attribute

Adds a jwt_auth_failed custom attribute that can be used for debugging,
but also will help with some planned cleanup and refactoring in
ARCHBOM-1218.

ARCHBOM-1218

CHANGELOG.rst

@@ -12,14 +12,31 @@ Change Log
 Unreleased
 ----------
 
+[6.5.0] - 2021-02-12
+--------------------
+
+Added
+~~~~~
+
+* Added a new custom attribute `jwt_auth_failed` to both monitor failures, and to help prepare for future refactors.
+
+
 [6.4.0] - 2021-01-19
 --------------------
 
-Updated
-~~~~~~~
+Added
+~~~~~
 
 * Added a new custom attribute `request_is_staff_or_superuser`
 
+[6.3.0] - 2021-01-12
+--------------------
+
+Removed
+~~~~~~~~
+
+* Drop support for Python 3.5
+
 [6.2.0] - 2020-08-24
 --------------------
 

edx_rest_framework_extensions/__init__.py

@@ -1,3 +1,3 @@
 """ edx Django REST Framework extensions. """
 
-__version__ = '6.4.0'  # pragma: no cover
+__version__ = '6.5.0'  # pragma: no cover

edx_rest_framework_extensions/auth/jwt/authentication.py

@@ -5,6 +5,7 @@
 import jwt
 from django.contrib.auth import get_user_model
 from django.middleware.csrf import CsrfViewMiddleware
+from edx_django_utils.monitoring import set_custom_attribute
 from rest_framework import exceptions
 from rest_framework_jwt.authentication import JSONWebTokenAuthentication
 
@@ -77,11 +78,16 @@ def authenticate(self, request):
             return user_and_auth
 
         except jwt.InvalidTokenError as token_error:
+            # Note: I think this case is not used, but will monitor the custom attribute to verify.
+            set_custom_attribute('jwt_auth_failed', 'InvalidTokenError:{}'.format(repr(token_error)))
             raise exceptions.AuthenticationFailed() from token_error
-        except Exception as ex:
+        except Exception as exception:
             # Errors in production do not need to be logged (as they may be noisy),
             # but debug logging can help quickly resolve issues during development.
-            logger.debug(ex)
+            logger.debug('Failed JWT Authentication,', exc_info=exception)
+            # Note: I think this case should only include AuthenticationFailed and PermissionDenied,
+            #   but will monitor the custom attribute to verify.
+            set_custom_attribute('jwt_auth_failed', 'Exception:{}'.format(repr(exception)))
             raise
 
     def authenticate_credentials(self, payload):

edx_rest_framework_extensions/auth/jwt/tests/test_authentication.py

@@ -174,19 +174,22 @@ def test_authenticate_credentials_no_usernames(self):
         with self.assertRaises(AuthenticationFailed):
             JwtAuthentication().authenticate_credentials({'email': 'test@example.com'})
 
-    def test_authenticate_csrf_protected(self):
+    @mock.patch('edx_rest_framework_extensions.auth.jwt.authentication.set_custom_attribute')
+    def test_authenticate_csrf_protected(self, mock_set_custom_attribute):
         """ Verify authenticate exception for CSRF protected cases. """
         request = RequestFactory().post('/')
 
         request.META[USE_JWT_COOKIE_HEADER] = 'true'
 
-        with mock.patch.object(JSONWebTokenAuthentication, 'authenticate', return_value=('mock-user', "mock-auth")):  # noqa E501 line too long
-            with mock.patch.object(Logger, 'debug') as debug_logger:
-                with self.assertRaises(PermissionDenied) as context_manager:
-                    JwtAuthentication().authenticate(request)
+        with mock.patch.object(JSONWebTokenAuthentication, 'authenticate', return_value=('mock-user', "mock-auth")):
+            with self.assertRaises(PermissionDenied) as context_manager:
+                JwtAuthentication().authenticate(request)
 
-        self.assertEqual(context_manager.exception.detail, 'CSRF Failed: CSRF cookie not set.')
-        self.assertTrue(debug_logger.called)
+        assert context_manager.exception.detail.startswith('CSRF Failed')
+        mock_set_custom_attribute.assert_called_once_with(
+            'jwt_auth_failed',
+            "Exception:PermissionDenied('CSRF Failed: CSRF cookie not set.')",
+        )
 
     @ddt.data(True, False)
     def test_get_decoded_jwt_from_auth(self, is_jwt_authentication):
@@ -195,25 +198,40 @@ def test_get_decoded_jwt_from_auth(self, is_jwt_authentication):
         # Mock out the `is_jwt_authenticated` method
         authentication.is_jwt_authenticated = lambda request: is_jwt_authentication
 
-        user = factories.UserFactory()
-        payload = generate_latest_version_payload(user)
-        jwt = generate_jwt_token(payload)
-        mock_request_with_cookie = mock.Mock(COOKIES={}, auth=jwt)
+        jwt_token = self._get_test_jwt_token()
+        mock_request_with_cookie = mock.Mock(COOKIES={}, auth=jwt_token)
 
-        expected_decoded_jwt = jwt_decode_handler(jwt) if is_jwt_authentication else None
+        expected_decoded_jwt = jwt_decode_handler(jwt_token) if is_jwt_authentication else None
 
         decoded_jwt = authentication.get_decoded_jwt_from_auth(mock_request_with_cookie)
         self.assertEqual(expected_decoded_jwt, decoded_jwt)
 
-    def test_with_explicitly_jwt_authorization(self):
+    def test_authenticate_with_correct_jwt_authorization(self):
+        """
+        With JWT header it continues and validates the credentials and throws error.
+
+        Note: CSRF protection should be skipped for this case, with no PermissionDenied.
+        """
+        jwt_token = self._get_test_jwt_token()
+        request = RequestFactory().get('/', HTTP_AUTHORIZATION=jwt_token)
+        JwtAuthentication().authenticate(request)
+
+    def test_authenticate_with_incorrect_jwt_authorization(self):
         """ With JWT header it continues and validates the credentials and throws error. """
         auth_header = '{token_name} {token}'.format(token_name='JWT', token='wrongvalue')
         request = RequestFactory().get('/', HTTP_AUTHORIZATION=auth_header)
         with self.assertRaises(AuthenticationFailed):
             JwtAuthentication().authenticate(request)
 
-    def test_jwt_returns_none_for_bearer_header(self):
+    def test_authenticate_with_bearer_token(self):
         """ Returns a None for bearer header request. """
         auth_header = '{token_name} {token}'.format(token_name='Bearer', token='abc123')
         request = RequestFactory().get('/', HTTP_AUTHORIZATION=auth_header)
         self.assertIsNone(JwtAuthentication().authenticate(request))
+
+    def _get_test_jwt_token(self):
+        """ Returns a user and jwt token """
+        user = factories.UserFactory()
+        payload = generate_latest_version_payload(user)
+        jwt_token = generate_jwt_token(payload)
+        return jwt_token