Merge pull request #200 from willseward/master

Add support for private/public keys pairs

Author: Carlton Gibson

docs/index.md

@@ -163,6 +163,8 @@ JWT_AUTH = {
     'rest_framework_jwt.utils.jwt_response_payload_handler',
 
     'JWT_SECRET_KEY': settings.SECRET_KEY,
+    'JWT_PUBLIC_KEY': None,
+    'JWT_PRIVATE_KEY': None,
     'JWT_ALGORITHM': 'HS256',
     'JWT_VERIFY': True,
     'JWT_VERIFY_EXPIRATION': True,
@@ -184,6 +186,16 @@ This is the secret key used to sign the JWT. Make sure this is safe and not shar
 
 Default is your project's `settings.SECRET_KEY`.
 
+### JWT_PUBLIC_KEY
+This is an object of type `cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey`. It will be used to verify the signature of the incoming JWT. Will override `JWT_SECRET_KEY` when set. Read the [documentation](https://cryptography.io/en/latest/hazmat/primitives/asymmetric/rsa/#cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey) for more details. Please note that `JWT_ALGORITHM` must be set to one of `RS256`, `RS384`, or `RS512`.
+
+Default is `None`.
+
+### JWT_PRIVATE_KEY
+This is an object of type `cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey`. It will be used to sign the signature component of the JWT. Will override `JWT_SECRET_KEY` when set. Read the [documentation](https://cryptography.io/en/latest/hazmat/primitives/asymmetric/rsa/#cryptography.hazmat.primitives.asymmetric.rsa.RSAPrivateKey) for more details. Please note that `JWT_ALGORITHM` must be set to one of `RS256`, `RS384`, or `RS512`.
+
+Default is `None`.
+
 ### JWT_ALGORITHM
 
 Possible values are any of the [supported algorithms](https://github.com/jpadilla/pyjwt#algorithms) for cryptographic signing in PyJWT.

requirements/testing.txt

@@ -4,4 +4,4 @@ pytest-django==2.8.0
 pytest-cov==1.6
 
 # Mocking the datetime module.
-freezegun==0.3.2
+cryptography==1.2.2

rest_framework_jwt/settings.py

@@ -19,6 +19,12 @@
     'JWT_PAYLOAD_GET_USER_ID_HANDLER':
     'rest_framework_jwt.utils.jwt_get_user_id_from_payload_handler',
 
+    'JWT_PRIVATE_KEY':
+    None,
+
+    'JWT_PUBLIC_KEY':
+    None,
+
     'JWT_PAYLOAD_GET_USERNAME_HANDLER':
     'rest_framework_jwt.utils.jwt_get_username_from_payload_handler',
 

rest_framework_jwt/utils.py

@@ -68,7 +68,7 @@ def jwt_get_username_from_payload_handler(payload):
 def jwt_encode_handler(payload):
     return jwt.encode(
         payload,
-        api_settings.JWT_SECRET_KEY,
+        api_settings.JWT_PRIVATE_KEY or api_settings.JWT_SECRET_KEY,
         api_settings.JWT_ALGORITHM
     ).decode('utf-8')
 
@@ -80,7 +80,7 @@ def jwt_decode_handler(token):
 
     return jwt.decode(
         token,
-        api_settings.JWT_SECRET_KEY,
+        api_settings.JWT_PUBLIC_KEY or api_settings.JWT_SECRET_KEY,
         api_settings.JWT_VERIFY,
         options=options,
         leeway=api_settings.JWT_LEEWAY,

tests/test_views.py

@@ -1,19 +1,22 @@
 import unittest
 from calendar import timegm
 from datetime import datetime, timedelta
+import time
 
 from django import get_version
 from django.test import TestCase
 from django.test.utils import override_settings
 from django.conf.urls import patterns
-from freezegun import freeze_time
 from rest_framework import status
 from rest_framework.test import APIClient
 
 from rest_framework_jwt import utils
 from rest_framework_jwt.compat import get_user_model
 from rest_framework_jwt.settings import api_settings, DEFAULTS
 
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.asymmetric import rsa
+
 from . import utils as test_utils
 
 User = get_user_model()
@@ -255,6 +258,9 @@ class TokenTestCase(BaseTestCase):
     Handlers for getting tokens from the API, or creating arbitrary ones.
     """
 
+    def setUp(self):
+        super(TokenTestCase, self).setUp()
+
     def get_token(self):
         client = APIClient(enforce_csrf_checks=True)
         response = client.post('/auth-token/', self.data, format='json')
@@ -272,22 +278,20 @@ def create_token(self, user, exp=None, orig_iat=None):
         return token
 
 
-class VerifyJSONWebTokenTests(TokenTestCase):
+class VerifyJSONWebTokenTestsSymmetric(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()
 
-        with freeze_time('2015-01-01 00:00:01'):
-            orig_token = self.get_token()
+        # Now try to get a refreshed token
+        response = client.post('/auth-token-verify/', {'token': orig_token},
+                               format='json')
 
-            with freeze_time('2015-01-01 00:00:10'):
-                # 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)
@@ -345,6 +349,102 @@ def test_verify_jwt_fails_with_missing_user(self):
                                  "User doesn't exist")
 
 
+class VerifyJSONWebTokenTestsAsymmetric(TokenTestCase):
+
+    def setUp(self):
+
+        super(VerifyJSONWebTokenTestsAsymmetric, self).setUp()
+
+        private_key = rsa.generate_private_key(public_exponent=65537,
+                                               key_size=2048,
+                                               backend=default_backend())
+        public_key = private_key.public_key()
+
+        api_settings.JWT_PRIVATE_KEY = private_key
+        api_settings.JWT_PUBLIC_KEY = public_key
+        api_settings.JWT_ALGORITHM = 'RS512'
+
+    def test_verify_jwt_with_pub_pvt_key(self):
+        """
+        Test that a token can be signed with asymmetrics keys
+        """
+        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_bad_pvt_key(self):
+        """
+        Test that an mismatched private key token will fail with
+        the correct error.
+        """
+
+        # Generate a new private key
+        private_key = rsa.generate_private_key(public_exponent=65537,
+                                               key_size=2048,
+                                               backend=default_backend())
+
+        # Don't set the private key
+        api_settings.JWT_PRIVATE_KEY = private_key
+
+        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_400_BAD_REQUEST)
+        self.assertRegexpMatches(response.data['non_field_errors'][0],
+                                 'Error decoding signature')
+
+    def tearDown(self):
+        # Restore original settings
+        api_settings.JWT_ALGORITHM = DEFAULTS['JWT_ALGORITHM']
+        api_settings.JWT_PRIVATE_KEY = DEFAULTS['JWT_PRIVATE_KEY']
+        api_settings.JWT_PUBLIC_KEY = DEFAULTS['JWT_PUBLIC_KEY']
+
+
 class RefreshJSONWebTokenTests(TokenTestCase):
 
     def setUp(self):
@@ -354,28 +454,29 @@ def setUp(self):
     def test_refresh_jwt(self):
         """
         Test getting a refreshed token from original token works
+
+        No date/time modifications are neccessary because it is assumed
+        that this operation will take less than 300 seconds.
         """
         client = APIClient(enforce_csrf_checks=True)
+        orig_token = self.get_token()
+        orig_token_decoded = utils.jwt_decode_handler(orig_token)
 
-        with freeze_time('2015-01-01 00:00:01'):
-            orig_token = self.get_token()
-            orig_token_decoded = utils.jwt_decode_handler(orig_token)
-
-            expected_orig_iat = timegm(datetime.utcnow().utctimetuple())
+        expected_orig_iat = timegm(datetime.utcnow().utctimetuple())
 
-            # Make sure 'orig_iat' exists and is the current time (give some slack)
-            orig_iat = orig_token_decoded['orig_iat']
-            self.assertLessEqual(orig_iat - expected_orig_iat, 1)
+        # Make sure 'orig_iat' exists and is the current time (give some slack)
+        orig_iat = orig_token_decoded['orig_iat']
+        self.assertLessEqual(orig_iat - expected_orig_iat, 1)
 
-            with freeze_time('2015-01-01 00:00:03'):
+        time.sleep(1)
 
-                # 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)
+        # 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)
+        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 original
         self.assertEquals(new_token_decoded['orig_iat'], orig_iat)