Abstracted out logic from the refresh serializer to make a verification serializer. Added a verification view. Added verification tests.

Author: Jwpe

rest_framework_jwt/serializers.py

@@ -81,16 +81,17 @@ def validate(self, attrs):
             raise serializers.ValidationError(msg)
 
 
-class RefreshJSONWebTokenSerializer(Serializer):
+class VerificationBaseSerializer(Serializer):
     """
-    Check an access token
+    Abstract serializer used for verifying and refreshing JWTs.
     """
     token = serializers.CharField()
 
     def validate(self, attrs):
-        User = utils.get_user_model()
-        token = attrs['token']
+        msg = _('Please define a validate method.')
+        raise NotImplementedError(msg)
 
+    def _check_payload(self, token):
         # Check payload valid (based off of JSONWebTokenAuthentication,
         # may want to refactor)
         try:
@@ -102,6 +103,10 @@ def validate(self, attrs):
             msg = _('Error decoding signature.')
             raise serializers.ValidationError(msg)
 
+        return payload
+
+    def _check_user(self, payload):
+        User = utils.get_user_model()
         # Make sure user exists (may want to refactor this)
         try:
             user_id = jwt_get_user_id_from_payload(payload)
@@ -115,6 +120,38 @@ def validate(self, attrs):
             msg = _("User doesn't exist.")
             raise serializers.ValidationError(msg)
 
+        return user
+
+
+class VerifyJSONWebTokenSerializer(VerificationBaseSerializer):
+    """
+    Check the veracity of an access token.
+    """
+
+    def validate(self, attrs):
+        token = attrs['token']
+
+        payload = self._check_payload(token=token)
+        user = self._check_user(payload=payload)
+
+        new_payload = jwt_payload_handler(user)
+
+        return {
+            'token': jwt_encode_handler(new_payload),
+            'user': user
+        }
+
+
+class RefreshJSONWebTokenSerializer(VerificationBaseSerializer):
+    """
+    Refresh an access token.
+    """
+
+    def validate(self, attrs):
+        token = attrs['token']
+
+        payload = self._check_payload(token=token)
+        user = self._check_user(payload=payload)
         # Get and check 'orig_iat'
         orig_iat = payload.get('orig_iat')
 

rest_framework_jwt/views.py

@@ -6,7 +6,10 @@
 
 from rest_framework_jwt.settings import api_settings
 
-from .serializers import JSONWebTokenSerializer, RefreshJSONWebTokenSerializer
+from .serializers import (
+    JSONWebTokenSerializer, RefreshJSONWebTokenSerializer,
+    VerifyJSONWebTokenSerializer
+)
 
 jwt_response_payload_handler = api_settings.JWT_RESPONSE_PAYLOAD_HANDLER
 
@@ -37,20 +40,17 @@ def post(self, request):
         return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
 
 
-class RefreshJSONWebToken(APIView):
+class VerifyJSONWebToken(APIView):
     """
-    API View that returns a refreshed token (with new expiration) based on
-    existing token
-
-    If 'orig_iat' field (original issued-at-time) is found, will first check
-    if it's within expiration window, then copy it to the new token
+    API View that checks the veracity of a token, returning the token if it
+    is valid.
     """
     throttle_classes = ()
     permission_classes = ()
     authentication_classes = ()
     parser_classes = (parsers.FormParser, parsers.JSONParser,)
     renderer_classes = (renderers.JSONRenderer,)
-    serializer_class = RefreshJSONWebTokenSerializer
+    serializer_class = VerifyJSONWebTokenSerializer
 
     def post(self, request):
         serializer = self.serializer_class(data=request.DATA)
@@ -65,5 +65,17 @@ def post(self, request):
         return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
 
 
+class RefreshJSONWebToken(VerifyJSONWebToken):
+    """
+    API View that returns a refreshed token (with new expiration) based on
+    existing token
+
+    If 'orig_iat' field (original issued-at-time) is found, will first check
+    if it's within expiration window, then copy it to the new token
+    """
+    serializer_class = RefreshJSONWebTokenSerializer
+
+
 obtain_jwt_token = ObtainJSONWebToken.as_view()
 refresh_jwt_token = RefreshJSONWebToken.as_view()
+verify_jwt_token = VerifyJSONWebToken.as_view()

tests/test_views.py

@@ -25,6 +25,8 @@
     '',
     (r'^auth-token/$', 'rest_framework_jwt.views.obtain_jwt_token'),
     (r'^auth-token-refresh/$', 'rest_framework_jwt.views.refresh_jwt_token'),
+    (r'^auth-token-verify/$', 'rest_framework_jwt.views.verify_jwt_token'),
+
 )
 
 orig_datetime = datetime
@@ -204,20 +206,18 @@ def test_jwt_login_json_bad_creds(self):
         self.assertEqual(response.status_code, 400)
 
 
-class RefreshJSONWebTokenTests(BaseTestCase):
-    urls = 'tests.test_views'
-
-    def setUp(self):
-        super(RefreshJSONWebTokenTests, self).setUp()
-        api_settings.JWT_ALLOW_REFRESH = True
+class TokenTestCase(BaseTestCase):
+    """
+    Handlers for getting tokens from the API, or creating arbitrary ones.
+    """
 
     def get_token(self):
         client = APIClient(enforce_csrf_checks=True)
         response = client.post('/auth-token/', self.data, format='json')
         return response.data['token']
 
     def create_token(self, user, exp=None, orig_iat=None):
-        payload = utils.jwt_payload_handler(self.user)
+        payload = utils.jwt_payload_handler(user)
         if exp:
             payload['exp'] = exp
 
@@ -227,6 +227,84 @@ def create_token(self, user, exp=None, orig_iat=None):
         token = utils.jwt_encode_handler(payload)
         return token
 
+
+class VerifyJSONWebTokenTests(TokenTestCase):
+
+    def test_verify_jwt(self):
+        """
+        Test that a valid, non-expired token will return a 200 response
+        and itself when passed to the validation endpoint.
+        """
+        client = APIClient(enforce_csrf_checks=True)
+
+        orig_token = self.get_token()
+
+        # Now try to get a refreshed token
+        response = client.post('/auth-token-verify/', {'token': orig_token},
+                               format='json')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        self.assertEqual(response.data['token'], orig_token)
+
+    def test_verify_jwt_fails_with_expired_token(self):
+        """
+        Test that an expired token will fail with the correct error.
+        """
+        client = APIClient(enforce_csrf_checks=True)
+
+        # Make an expired token..
+        token = self.create_token(
+            self.user,
+            exp=datetime.utcnow() - timedelta(seconds=5),
+            orig_iat=datetime.utcnow() - timedelta(hours=1)
+        )
+
+        response = client.post('/auth-token-verify/', {'token': token},
+                               format='json')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertRegexpMatches(response.data['non_field_errors'][0],
+                                 'Signature has expired')
+
+    def test_verify_jwt_fails_with_bad_token(self):
+        """
+        Test that an invalid token will fail with the correct error.
+        """
+        client = APIClient(enforce_csrf_checks=True)
+
+        token = "i am not a correctly formed token"
+
+        response = client.post('/auth-token-verify/', {'token': token},
+                               format='json')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertRegexpMatches(response.data['non_field_errors'][0],
+                                 'Error decoding signature')
+
+    def test_verify_jwt_fails_with_missing_user(self):
+        """
+        Test that an invalid token will fail with a user that does not exist.
+        """
+        client = APIClient(enforce_csrf_checks=True)
+
+        user = User.objects.create_user(
+            email='[email protected]', username='jsmith', password='password')
+
+        token = self.create_token(user)
+        # Delete the user used to make the token
+        user.delete()
+
+        response = client.post('/auth-token-verify/', {'token': token},
+                               format='json')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertRegexpMatches(response.data['non_field_errors'][0],
+                                 "User doesn't exist")
+
+
+class RefreshJSONWebTokenTests(TokenTestCase):
+
+    def setUp(self):
+        super(RefreshJSONWebTokenTests, self).setUp()
+        api_settings.JWT_ALLOW_REFRESH = True
+
     def test_refresh_jwt(self):
         """
         Test getting a refreshed token from original token works
@@ -257,25 +335,6 @@ def test_refresh_jwt(self):
         self.assertEquals(new_token_decoded['orig_iat'], orig_iat)
         self.assertGreater(new_token_decoded['exp'], orig_token_decoded['exp'])
 
-    def test_refresh_jwt_fails_with_expired_token(self):
-        """
-        Test that using an expired token to refresh won't work
-        """
-        client = APIClient(enforce_csrf_checks=True)
-
-        # Make an expired token..
-        token = self.create_token(
-            self.user,
-            exp=datetime.utcnow() - timedelta(seconds=5),
-            orig_iat=datetime.utcnow() - timedelta(hours=1)
-        )
-
-        response = client.post('/auth-token-refresh/', {'token': token},
-                               format='json')
-        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
-        self.assertRegexpMatches(response.data['non_field_errors'][0],
-                                 'Signature has expired')
-
     def test_refresh_jwt_after_refresh_expiration(self):
         """
         Test that token can't be refreshed after token refresh limit