[refresh-token] add refresh token endpoint

typo

[refresh-token] Refactor renew token stuff, add jwt_get_user_id_from_payload_handler for custom user_id format

[refresh-token] add tests for refresh token feature

some more api checks

Author: alvinchow86

rest_framework_jwt/authentication.py

@@ -13,6 +13,7 @@
 
 
 jwt_decode_handler = api_settings.JWT_DECODE_HANDLER
+jwt_get_user_id_from_payload = api_settings.JWT_PAYLOAD_GET_USER_ID_HANDLER
 
 
 class JSONWebTokenAuthentication(BaseAuthentication):
@@ -62,7 +63,7 @@ def authenticate_credentials(self, payload):
         Returns an active user that matches the payload's user id and email.
         """
         try:
-            user_id = payload.get('user_id')
+            user_id = jwt_get_user_id_from_payload(payload)
 
             if user_id:
                 user = User.objects.get(pk=user_id, is_active=True)

rest_framework_jwt/serializers.py

@@ -1,11 +1,17 @@
+from calendar import timegm
+from datetime import datetime
+import jwt
+
 from django.contrib.auth import authenticate, get_user_model
-from rest_framework import serializers
+from rest_framework import serializers, exceptions
 
 from rest_framework_jwt.settings import api_settings
 
 
 jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
 jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
+jwt_decode_handler = api_settings.JWT_DECODE_HANDLER
+jwt_get_user_id_from_payload = api_settings.JWT_PAYLOAD_GET_USER_ID_HANDLER
 
 
 class JSONWebTokenSerializer(serializers.Serializer):
@@ -41,6 +47,13 @@ def validate(self, attrs):
 
                 payload = jwt_payload_handler(user)
 
+                # Include original issued at time for a brand new token,
+                # to allow token refresh
+                if api_settings.JWT_ALLOW_TOKEN_RENEWAL:
+                    payload['orig_iat'] = timegm(
+                        datetime.utcnow().utctimetuple()
+                    )
+
                 return {
                     'token': jwt_encode_handler(payload)
                 }
@@ -50,3 +63,57 @@ def validate(self, attrs):
         else:
             msg = 'Must include "username" and "password"'
             raise serializers.ValidationError(msg)
+
+
+class RefreshJSONWebTokenSerializer(serializers.Serializer):
+    """
+    Check an access token
+    """
+    token = serializers.CharField()
+
+    def validate(self, attrs):
+        token = attrs['token']
+
+        # Check payload valid (based off of JSONWebTokenAuthentication,
+        # may want to refactor)
+        try:
+            payload = jwt_decode_handler(token)
+        except jwt.ExpiredSignature:
+            msg = 'Signature has expired.'
+            raise serializers.ValidationError(msg)
+        except jwt.DecodeError:
+            msg = 'Error decoding signature.'
+            raise serializers.ValidationError(msg)
+
+        # Make sure user exists (may want to refactor this)
+        User = get_user_model()
+        try:
+            user_id = jwt_get_user_id_from_payload(payload)
+            if user_id:
+                user = User.objects.get(pk=user_id, is_active=True)
+            else:
+                msg = 'Invalid payload'
+                raise serializers.ValidationError(msg)
+        except User.DoesNotExist:
+            raise serializers.ValidationError("User doesn't exist")
+
+        # Get and check 'orig_iat'
+        orig_iat = payload.get('orig_iat')
+        if orig_iat:
+            # Verify expiration
+            expiration_timestamp = (
+                orig_iat +
+                int(api_settings.JWT_TOKEN_RENEWAL_LIMIT.total_seconds())
+            )
+            now_timestamp = timegm(datetime.utcnow().utctimetuple())
+            if now_timestamp > expiration_timestamp:
+                raise serializers.ValidationError("Refresh has expired")
+        else:
+            raise serializers.ValidationError("orig_iat field is required")
+
+        new_payload = jwt_payload_handler(user)
+        new_payload['orig_iat'] = orig_iat
+
+        return {
+            'token': jwt_encode_handler(new_payload)
+        }

rest_framework_jwt/settings.py

@@ -16,19 +16,26 @@
     'JWT_PAYLOAD_HANDLER':
     'rest_framework_jwt.utils.jwt_payload_handler',
 
+    'JWT_PAYLOAD_GET_USER_ID_HANDLER':
+    'rest_framework_jwt.utils.jwt_get_user_id_from_payload_handler',
+
     'JWT_SECRET_KEY': settings.SECRET_KEY,
     'JWT_ALGORITHM': 'HS256',
     'JWT_VERIFY': True,
     'JWT_VERIFY_EXPIRATION': True,
     'JWT_LEEWAY': 0,
-    'JWT_EXPIRATION_DELTA': datetime.timedelta(seconds=300)
+    'JWT_EXPIRATION_DELTA': datetime.timedelta(seconds=300),
+
+    'JWT_ALLOW_TOKEN_RENEWAL': False,
+    'JWT_TOKEN_RENEWAL_LIMIT': datetime.timedelta(days=7),
 }
 
 # List of settings that may be in string import notation.
 IMPORT_STRINGS = (
     'JWT_ENCODE_HANDLER',
     'JWT_DECODE_HANDLER',
     'JWT_PAYLOAD_HANDLER',
+    'JWT_PAYLOAD_GET_USER_ID_HANDLER',
 )
 
 api_settings = APISettings(USER_SETTINGS, DEFAULTS, IMPORT_STRINGS)

rest_framework_jwt/tests/simtime.py

@@ -0,0 +1,48 @@
+"""
+Test utility to simulate changing the current time, namely overriding utcnow()
+"""
+
+import datetime
+from datetime import datetime as orig_datetime
+
+
+SIMTIME = None
+
+
+def set_simtime(simtime):
+    """
+    Argument should be a naive utc datetime
+    """
+    global SIMTIME
+    SIMTIME = simtime
+
+
+def clear_simtime():
+    global SIMTIME
+    SIMTIME = None
+
+
+class SimulationDatetimeMeta(type):
+    """
+    Need to override isinstance(<datetime.datetime obj>, SimulationDatetime) to
+    return True
+    """
+    def __instancecheck__(self, other):
+        if isinstance(other, datetime.datetime):
+            return True
+
+
+class SimulationDatetime(datetime.datetime):
+    """
+    Mock datetime object with patched utcnow() method
+    """
+
+    @classmethod
+    def utcnow(cls):
+        if SIMTIME:
+            # assert False, SIMTIME
+            return SIMTIME
+
+        return orig_datetime.utcnow()
+
+    __metaclass__ = SimulationDatetimeMeta

rest_framework_jwt/tests/test_views.py

@@ -1,20 +1,31 @@
+from calendar import timegm
+from datetime import datetime, timedelta
+
 from django.test import TestCase
 from django.conf import settings
 from django.contrib.auth.models import User
+import jwt
 from rest_framework import status
 from rest_framework.compat import patterns
 from rest_framework.test import APIClient
 
+import rest_framework_jwt.serializers
 from rest_framework_jwt import utils
 from rest_framework_jwt.runtests.models import CustomUser
+from rest_framework_jwt.settings import api_settings, DEFAULTS
+from rest_framework_jwt.tests import simtime
+from rest_framework_jwt.tests.simtime import SimulationDatetime
 
 urlpatterns = patterns(
     '',
     (r'^auth-token/$', 'rest_framework_jwt.views.obtain_jwt_token'),
+    (r'^auth-token-refresh/$', 'rest_framework_jwt.views.refresh_jwt_token'),
 )
 
+orig_datetime = datetime
+
 
-class ObtainJSONWebTokenTests(TestCase):
+class BaseTestCase(TestCase):
     urls = 'rest_framework_jwt.tests.test_views'
 
     def setUp(self):
@@ -29,6 +40,9 @@ def setUp(self):
             'password': self.password
         }
 
+
+class ObtainJSONWebTokenTests(BaseTestCase):
+
     def test_jwt_login_json(self):
         """
         Ensure JWT login view using JSON POST works.
@@ -145,3 +159,128 @@ def test_jwt_login_json_bad_creds(self):
 
     def tearDown(self):
         settings.AUTH_USER_MODEL = self.ORIG_AUTH_USER_MODEL
+
+
+class RefreshJSONWebTokenTests(BaseTestCase):
+    urls = 'rest_framework_jwt.tests.test_views'
+
+    def setUp(self):
+        super(RefreshJSONWebTokenTests, self).setUp()
+        api_settings.JWT_ALLOW_TOKEN_RENEWAL = True
+
+        # monkey patch datetime objects in places that use datetime.utcnow()
+        jwt.datetime = SimulationDatetime
+        rest_framework_jwt.serializers.datetime = SimulationDatetime
+        utils.datetime = SimulationDatetime
+
+    def get_token(self):
+        client = APIClient(enforce_csrf_checks=True)
+        response = client.post('/auth-token/', self.data, format='json')
+        return response.data['token']
+
+    def test_refresh_jwt(self):
+        """
+        Test getting a refreshed token from original token works
+        """
+        client = APIClient(enforce_csrf_checks=True)
+
+        # Set simulation time to now
+        currtime = datetime.utcnow()
+        simtime.set_simtime(currtime)
+
+        orig_token = self.get_token()
+        orig_token_decoded = utils.jwt_decode_handler(orig_token)
+
+        # Make sure 'orig_iat' exists and is the current time
+        orig_iat = orig_token_decoded['orig_iat']
+        self.assertEquals(orig_iat, timegm(currtime.utctimetuple()))
+
+        # Fast-forward to later time (but before first token expires)
+        currtime += api_settings.JWT_EXPIRATION_DELTA - timedelta(seconds=30)
+        simtime.set_simtime(currtime)
+
+        # Now try to get a refreshed token
+        response = client.post('/auth-token-refresh/', {'token': orig_token},
+                               format='json')
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+        new_token = response.data['token']
+        new_token_decoded = utils.jwt_decode_handler(new_token)
+
+        # Make sure 'orig_iat' on the new token is same as origina
+        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)
+        token = self.get_token()
+
+        # Fast-forward to after token expires
+        after_expire = (
+            datetime.utcnow() + api_settings.JWT_EXPIRATION_DELTA +
+            timedelta(seconds=10)
+        )
+        simtime.set_simtime(after_expire)
+
+        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_renewal_expiration(self):
+        """
+        Test that token can't be refreshed after token renewal limit
+        """
+        # For simpler test, make the RENEWAL_LIMIT just a bit larger than
+        # EXPIRATION_DELTA
+        api_settings.JWT_TOKEN_RENEWAL_LIMIT = (
+            api_settings.JWT_EXPIRATION_DELTA +
+            api_settings.JWT_EXPIRATION_DELTA / 2
+        )
+
+        client = APIClient(enforce_csrf_checks=True)
+
+        initial_time = datetime.utcnow()
+
+        token1 = self.get_token()
+
+        # Token1 refresh to Token2, just before it expires
+        currtime = (initial_time + api_settings.JWT_EXPIRATION_DELTA -
+                    timedelta(seconds=5))
+
+        simtime.set_simtime(currtime)
+
+        response = client.post('/auth-token-refresh/', {'token': token1},
+                               format='json')
+        token2 = response.data['token']
+
+        # Fast-forward to after token renewal expiration
+        # Token2 hasn't expired yet, but it can't be used to renew anymore!
+        currtime = (initial_time + api_settings.JWT_TOKEN_RENEWAL_LIMIT +
+                    timedelta(minutes=1))
+        simtime.set_simtime(currtime)
+
+        response = client.post('/auth-token-refresh/', {'token': token2},
+                               format='json')
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(response.data['non_field_errors'][0],
+                         'Refresh has expired')
+
+    def tearDown(self):
+        # Restore original settings
+        api_settings.JWT_ALLOW_TOKEN_RENEWAL = \
+            DEFAULTS['JWT_ALLOW_TOKEN_RENEWAL']
+
+        api_settings.JWT_TOKEN_RENEWAL_LIMIT = \
+            DEFAULTS['JWT_TOKEN_RENEWAL_LIMIT']
+
+        # Undo datetime monkeypatching
+        jwt.datetime = orig_datetime
+        rest_framework_jwt.serializers.datetime = orig_datetime
+        utils.datetime = orig_datetime
+
+        simtime.clear_simtime()

rest_framework_jwt/utils.py

@@ -1,4 +1,4 @@
-import datetime
+from datetime import datetime
 import jwt
 
 from rest_framework_jwt.settings import api_settings
@@ -9,10 +9,18 @@ def jwt_payload_handler(user):
         'user_id': user.pk,
         'email': user.email,
         'username': user.get_username(),
-        'exp': datetime.datetime.utcnow() + api_settings.JWT_EXPIRATION_DELTA
+        'exp': datetime.utcnow() + api_settings.JWT_EXPIRATION_DELTA
     }
 
 
+def jwt_get_user_id_from_payload_handler(payload):
+    """
+    Override this function if user_id is formatted differently in payload
+    """
+    user_id = payload.get('user_id')
+    return user_id
+
+
 def jwt_encode_handler(payload):
     return jwt.encode(
         payload,