Merge pull request #106 from James1345/develop

Release 3.2.0

CHANGELOG.rst

@@ -1,3 +1,8 @@
+######
+3.2.0
+######
+- Introduce new setting AUTO_REFRESH for controlling if token expiry time should be extended automatically
+
 ######
 3.1.5
 ######

docs/changes.md

@@ -1,5 +1,8 @@
 #Changelog
 
+## 3.2.0
+- Introduce new setting AUTO_REFRESH for controlling if token expiry time should be extended automatically
+
 ## 3.1.5
 - Make AuthTokenAdmin more compatible with big user tables
 - Extend docs regarding usage of Token Authentication as single authentication method.

docs/settings.md

@@ -14,6 +14,7 @@ REST_KNOX = {
   'AUTH_TOKEN_CHARACTER_LENGTH': 64,
   'TOKEN_TTL': timedelta(hours=10),
   'USER_SERIALIZER': 'knox.serializers.UserSerializer',
+  'AUTO_REFRESH': FALSE,
 }
 #...snip...
 ```
@@ -57,6 +58,10 @@ the system will not prevent you setting this.
 This is the reference to the class used to serialize the `User` objects when
 succesfully returning from `LoginView`. The default is `knox.serializers.UserSerializer`
 
+## AUTO_REFRESH
+This defines if the token expiry time is extended by TOKEN_TTL each time the token
+is used.
+
 # Constants `knox.settings`
 Knox also provides some constants for information. These must not be changed in
 external code; they are used in the model definitions in knox and an error will
@@ -75,4 +80,8 @@ This is the length of the digest that will be stored in the database for each to
 ## SALT_LENGTH
 This is the length of the [salt][salt] that will be stored in the database for each token.
 
+## MIN_REFRESH_INTERVAL
+This is the minimum time in seconds that needs to pass for the token expiry to be updated 
+in the database.
+
 [salt]: https://en.wikipedia.org/wiki/Salt_(cryptography)

knox/auth.py

@@ -18,7 +18,7 @@ def compare_digest(a, b):
 
 from knox.crypto import hash_token
 from knox.models import AuthToken
-from knox.settings import CONSTANTS
+from knox.settings import CONSTANTS, knox_settings
 
 User = settings.AUTH_USER_MODEL
 
@@ -64,23 +64,27 @@ def authenticate_credentials(self, token):
         token = token.decode("utf-8")
         for auth_token in AuthToken.objects.filter(
                 token_key=token[:CONSTANTS.TOKEN_KEY_LENGTH]):
-            for other_token in auth_token.user.auth_token_set.all():
-                if other_token.digest != auth_token.digest and other_token.expires is not None:
-                    if other_token.expires < timezone.now():
-                        other_token.delete()
-            if auth_token.expires is not None:
-                if auth_token.expires < timezone.now():
-                    auth_token.delete()
-                    continue
+            if self._cleanup_token(auth_token):
+                continue
+
             try:
                 digest = hash_token(token, auth_token.salt)
             except (TypeError, binascii.Error):
                 raise exceptions.AuthenticationFailed(msg)
             if compare_digest(digest, auth_token.digest):
+                if settings.REST_KNOX["AUTO_REFRESH"]:
+                    self.renew_token(auth_token)
                 return self.validate_user(auth_token)
-        # Authentication with this token has failed
         raise exceptions.AuthenticationFailed(msg)
 
+    def renew_token(self, auth_token):
+        current_expiry = auth_token.expires
+        new_expiry = timezone.now() + knox_settings.TOKEN_TTL
+        auth_token.expires = new_expiry
+        # Throttle refreshing of token to avoid db writes
+        if (new_expiry - current_expiry).total_seconds() > CONSTANTS.MIN_REFRESH_INTERVAL:
+            auth_token.save(update_fields=('expires',))
+
     def validate_user(self, auth_token):
         if not auth_token.user.is_active:
             raise exceptions.AuthenticationFailed(
@@ -89,3 +93,15 @@ def validate_user(self, auth_token):
 
     def authenticate_header(self, request):
         return 'Token'
+
+    def _cleanup_token(self, auth_token):
+        for other_token in auth_token.user.auth_token_set.all():
+            if other_token.digest != auth_token.digest and other_token.expires is not None:
+                if other_token.expires < timezone.now():
+                    other_token.delete()
+        if auth_token.expires is not None:
+            if auth_token.expires < timezone.now():
+                auth_token.delete()
+                return True
+        return False
+

knox/settings.py

@@ -10,6 +10,7 @@
     'AUTH_TOKEN_CHARACTER_LENGTH': 64,
     'TOKEN_TTL': timedelta(hours=10),
     'USER_SERIALIZER': 'knox.serializers.UserSerializer',
+    'AUTO_REFRESH': False,
 }
 
 IMPORT_STRINGS = {
@@ -36,6 +37,7 @@ class CONSTANTS:
     TOKEN_KEY_LENGTH = 8
     DIGEST_LENGTH = 128
     SALT_LENGTH = 16
+    MIN_REFRESH_INTERVAL = 60
 
     def __setattr__(self, *args, **kwargs):
         raise RuntimeException('''

knox_project/settings.py

@@ -57,3 +57,7 @@
 STATIC_URL = '/static/'
 
 TEST_RUNNER = 'django_nose.NoseTestSuiteRunner'
+
+REST_KNOX = {
+    'AUTO_REFRESH': True
+}

setup.py

@@ -16,7 +16,7 @@
     # Versions should comply with PEP440.  For a discussion on single-sourcing
     # the version across setup.py and the project code, see
     # https://packaging.python.org/en/latest/single_source_version.html
-    version='3.1.5',
+    version='3.2.0',
     description='Authentication for django rest framework',
     long_description=long_description,
 

tests/tests.py

@@ -1,7 +1,9 @@
 import base64
-import datetime
+from datetime import datetime, timedelta
 
+from django.conf import settings
 from django.contrib.auth import get_user_model
+from django.test import override_settings
 
 try:
     # For django >= 2.0
@@ -10,19 +12,28 @@
     # For django < 2.0
     from django.conf.urls import reverse
 
-from rest_framework.test import APIRequestFactory, APITestCase as TestCase
+from freezegun import freeze_time
+
+from rest_framework.test import (
+    APIRequestFactory,
+    APITestCase as TestCase
+)
 
 from knox.auth import TokenAuthentication
 from knox.models import AuthToken
-from knox.settings import CONSTANTS
+from knox.settings import CONSTANTS, knox_settings
 
 User = get_user_model()
+root_url = reverse('api-root')
 
 
 def get_basic_auth_header(username, password):
     return 'Basic %s' % base64.b64encode(
         ('%s:%s' % (username, password)).encode('ascii')).decode()
 
+no_auto_refresh_knox = settings.REST_KNOX.copy()
+no_auto_refresh_knox["AUTO_REFRESH"] = False
+
 
 class AuthTestCase(TestCase):
 
@@ -70,7 +81,7 @@ def test_logout_all_deletes_only_targets_keys(self):
         self.assertEqual(AuthToken.objects.count(), 0)
         for _ in range(10):
             token = AuthToken.objects.create(user=self.user)
-            token2 = AuthToken.objects.create(user=self.user2)
+            AuthToken.objects.create(user=self.user2)
         self.assertEqual(AuthToken.objects.count(), 20)
 
         url = reverse('knox_logoutall')
@@ -81,10 +92,9 @@ def test_logout_all_deletes_only_targets_keys(self):
     def test_expired_tokens_login_fails(self):
         self.assertEqual(AuthToken.objects.count(), 0)
         token = AuthToken.objects.create(
-            user=self.user, expires=datetime.timedelta(seconds=0))
-        url = reverse('api-root')
+            user=self.user, expires=timedelta(seconds=0))
         self.client.credentials(HTTP_AUTHORIZATION=('Token %s' % token))
-        response = self.client.post(url, {}, format='json')
+        response = self.client.post(root_url, {}, format='json')
         self.assertEqual(response.status_code, 401)
         self.assertEqual(response.data, {"detail": "Invalid token."})
 
@@ -93,7 +103,7 @@ def test_expired_tokens_deleted(self):
         for _ in range(10):
             # 0 TTL gives an expired token
             token = AuthToken.objects.create(
-                user=self.user, expires=datetime.timedelta(seconds=0))
+                user=self.user, expires=timedelta(seconds=0))
         self.assertEqual(AuthToken.objects.count(), 10)
 
         # Attempting a single logout should delete all tokens
@@ -116,17 +126,81 @@ def test_update_token_key(self):
 
     def test_invalid_token_length_returns_401_code(self):
         invalid_token = "1" * (CONSTANTS.TOKEN_KEY_LENGTH - 1)
-        url = reverse('api-root')
         self.client.credentials(HTTP_AUTHORIZATION=('Token %s' % invalid_token))
-        response = self.client.post(url, {}, format='json')
+        response = self.client.post(root_url, {}, format='json')
         self.assertEqual(response.status_code, 401)
         self.assertEqual(response.data, {"detail": "Invalid token."})
 
     def test_invalid_odd_length_token_returns_401_code(self):
         token = AuthToken.objects.create(self.user)
         odd_length_token = token + '1'
-        url = reverse('api-root')
         self.client.credentials(HTTP_AUTHORIZATION=('Token %s' % odd_length_token))
-        response = self.client.post(url, {}, format='json')
+        response = self.client.post(root_url, {}, format='json')
         self.assertEqual(response.status_code, 401)
         self.assertEqual(response.data, {"detail": "Invalid token."})
+
+    def test_token_expiry_is_extended_with_auto_refresh_activated(self):
+        self.assertEqual(settings.REST_KNOX["AUTO_REFRESH"], True)
+        self.assertEqual(knox_settings.TOKEN_TTL, timedelta(hours=10))
+        ttl = knox_settings.TOKEN_TTL
+        original_time = datetime(2018, 7, 25, 0, 0, 0, 0)
+
+        with freeze_time(original_time):
+            token_key = AuthToken.objects.create(user=self.user)
+
+        self.client.credentials(HTTP_AUTHORIZATION=('Token %s' % token_key))
+        five_hours_later = original_time + timedelta(hours=5)
+        with freeze_time(five_hours_later):
+            response = self.client.get(root_url, {}, format='json')
+        self.assertEqual(response.status_code, 200)
+
+        # original expiry date was extended:
+        new_expiry = AuthToken.objects.get().expires
+        self.assertEqual(new_expiry.replace(tzinfo=None),
+                         original_time + ttl + timedelta(hours=5))
+
+        # token works after orignal expiry:
+        after_original_expiry = original_time + ttl + timedelta(hours=1)
+        with freeze_time(after_original_expiry):
+            response = self.client.get(root_url, {}, format='json')
+            self.assertEqual(response.status_code, 200)
+
+        # token does not work after new expiry:
+        new_expiry = AuthToken.objects.get().expires
+        with freeze_time(new_expiry + timedelta(seconds=1)):
+            response = self.client.get(root_url, {}, format='json')
+            self.assertEqual(response.status_code, 401)
+
+    @override_settings(REST_KNOX=no_auto_refresh_knox)
+    def test_token_expiry_is_not_extended_with_auto_refresh_deativated(self):
+        self.assertEqual(knox_settings.TOKEN_TTL, timedelta(hours=10))
+
+        now = datetime.now()
+        with freeze_time(now):
+            token_key = AuthToken.objects.create(user=self.user)
+
+        original_expiry = AuthToken.objects.get().expires
+
+        self.client.credentials(HTTP_AUTHORIZATION=('Token %s' % token_key))
+        with freeze_time(now + timedelta(hours=1)):
+            response = self.client.get(root_url, {}, format='json')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(original_expiry, AuthToken.objects.get().expires)
+
+    def test_token_expiry_is_not_extended_within_MIN_REFRESH_INTERVAL(self):
+        self.assertEqual(settings.REST_KNOX["AUTO_REFRESH"], True)
+
+        now = datetime.now()
+        with freeze_time(now):
+            token_key = AuthToken.objects.create(user=self.user)
+
+        original_expiry = AuthToken.objects.get().expires
+
+        self.client.credentials(HTTP_AUTHORIZATION=('Token %s' % token_key))
+        in_min_interval = now + timedelta(seconds=CONSTANTS.MIN_REFRESH_INTERVAL - 10)
+        with freeze_time(in_min_interval):
+            response = self.client.get(root_url, {}, format='json')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(original_expiry, AuthToken.objects.get().expires)

tox.ini

@@ -16,6 +16,7 @@ deps =
     django-nose
     djangorestframework
     flake8
+    freezegun
     mkdocs
     pyOpenSSL
     pytest-django