Allow change of header name and type, cookie header initial commit

Author: vimalloc

flask_jwt_extended/config.py

@@ -1,46 +1,67 @@
 import datetime
 from flask import current_app
 
-# Defaults
+# TODO move this to the docs
+# blacklist storage options (simplekv). If using a storage option that supports
+# the simplekv.TimeToLiveMixin (example: redis, memcached), the TTL will be
+# automatically set to 15 minutes after the token expires (to account for
+# clock drift between different jwt providers/consumers).
+#
+# See: http://pythonhosted.org/simplekv/index.html#simplekv.TimeToLiveMixin
 
-# Authorize header type, what we are expecting to see in the auth header
-AUTH_HEADER = 'Bearer'
 
-# How long an access token will live before it expires.
-ACCESS_TOKEN_EXPIRES = datetime.timedelta(minutes=15)
+# Where to look for the JWT. Available options are cookie and header
+REQUEST_JWT_LOCATION = 'header'
+
+# Options for where to get the JWT if using a header approach
+HEADER_NAME = 'Authorization'
+HEADER_TYPE = 'Bearer'
+
+# Options for where to get and handling JWTs if using a cookie approach
+COOKIE_ACCESS_TOKEN_NAME = 'access_token'
+COOKIE_REFRESH_TOKEN_NAME = 'refresh_token'
+COOKIE_CSRF_DOUBLE_SUBMIT = False
+COOKIE_XSRF_ACCESS_NAME = 'xsrf_access_token'
+COOKIE_XSRF_REFRESH_NAME = 'xsrf_refresh_token'
 
-# How long the refresh token will live before it expires
+# How long an a token will live before they expire.
+ACCESS_TOKEN_EXPIRES = datetime.timedelta(minutes=15)
 REFRESH_TOKEN_EXPIRES = datetime.timedelta(days=30)
 
 # What algorithm to use to sign the token. See here for a list of options:
-# https://github.com/jpadilla/pyjwt/blob/master/jwt/api_jwt.py
+# https://github.com/jpadilla/pyjwt/blob/master/jwt/api_jwt.py (note that
+# public private key is not yet supported)
 ALGORITHM = 'HS256'
 
-# Blacklist enabled
+# Options for blacklisting/revoking tokens
 BLACKLIST_ENABLED = False
-
-# blacklist storage options (simplekv). If using a storage option that supports
-# the simplekv.TimeToLiveMixin (example: redis, memcached), the TTL will be
-# automatically set to 15 minutes after the token expires (to account for
-# clock drift between different jwt providers/consumers).
-#
-# See: http://pythonhosted.org/simplekv/index.html#simplekv.TimeToLiveMixin
 BLACKLIST_STORE = None
+BLACKLIST_TOKEN_CHECKS = 'refresh'  # valid options are 'all', and 'refresh'
+
 
-# blacklist check requests. Possible values are all and refresh
-BLACKLIST_TOKEN_CHECKS = 'refresh'
+def get_jwt_header_name():
+    name = current_app.config.get('JWT_HEADER_NAME', HEADER_NAME)
+    if not name:
+        raise RuntimeError("JWT_HEADER_NAME must be set")
+    return name
 
 
-def get_auth_header():
-    return current_app.config.get('JWT_AUTH_HEADER', AUTH_HEADER)
+def get_jwt_header_type():
+    return current_app.config.get('JWT_HEADER_TYPE', HEADER_TYPE)
 
 
 def get_access_expires():
-    return current_app.config.get('JWT_ACCESS_TOKEN_EXPIRES', ACCESS_TOKEN_EXPIRES)
+    delta = current_app.config.get('JWT_ACCESS_TOKEN_EXPIRES', ACCESS_TOKEN_EXPIRES)
+    if not isinstance(delta, datetime.timedelta):
+        raise RuntimeError('JWT_ACCESS_TOKEN_EXPIRES must be a datetime.timedelta')
+    return delta
 
 
 def get_refresh_expires():
-    return current_app.config.get('JWT_REFRESH_TOKEN_EXPIRES', REFRESH_TOKEN_EXPIRES)
+    delta = current_app.config.get('JWT_REFRESH_TOKEN_EXPIRES', REFRESH_TOKEN_EXPIRES)
+    if not isinstance(delta, datetime.timedelta):
+        raise RuntimeError('JWT_REFRESH_TOKEN_EXPIRES must be a datetime.timedelta')
+    return delta
 
 
 def get_algorithm():

flask_jwt_extended/utils.py

@@ -14,7 +14,7 @@
     from flask import _request_ctx_stack as ctx_stack
 
 from flask_jwt_extended.config import get_access_expires, get_refresh_expires, \
-    get_algorithm, get_blacklist_enabled, get_blacklist_checks, get_auth_header
+    get_algorithm, get_blacklist_enabled, get_blacklist_checks, get_jwt_header_type
 from flask_jwt_extended.exceptions import JWTEncodeError, JWTDecodeError, \
     InvalidHeaderError, NoAuthHeaderError, WrongTokenError, RevokedTokenError, \
     FreshTokenRequired
@@ -143,7 +143,7 @@ def _decode_jwt_from_request():
         raise NoAuthHeaderError("Missing Authorization Header")
 
     # Make sure the header is valid
-    expected_header = get_auth_header()
+    expected_header = get_jwt_header_type()
     parts = auth_header.split()
     if not expected_header:
         if len(parts) != 1:

tests/test_config.py

@@ -6,7 +6,7 @@
 
 from flask_jwt_extended.config import get_access_expires, get_refresh_expires, \
     get_algorithm, get_blacklist_enabled, get_blacklist_store, \
-    get_blacklist_checks, get_auth_header
+    get_blacklist_checks, get_jwt_header_type, get_jwt_header_name
 from flask_jwt_extended import JWTManager
 
 
@@ -26,7 +26,8 @@ def test_default_configs(self):
             self.assertEqual(get_blacklist_enabled(), False)
             self.assertEqual(get_blacklist_store(), None)
             self.assertEqual(get_blacklist_checks(), 'refresh')
-            self.assertEqual(get_auth_header(), 'Bearer')
+            self.assertEqual(get_jwt_header_name(), 'Authorization')
+            self.assertEqual(get_jwt_header_type(), 'Bearer')
 
     def test_override_configs(self):
         self.app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(minutes=5)
@@ -35,7 +36,8 @@ def test_override_configs(self):
         self.app.config['JWT_BLACKLIST_ENABLED'] = True
         self.app.config['JWT_BLACKLIST_STORE'] = simplekv.memory.DictStore()
         self.app.config['JWT_BLACKLIST_TOKEN_CHECKS'] = 'all'
-        self.app.config['JWT_AUTH_HEADER'] = 'JWT'
+        self.app.config['JWT_HEADER_NAME'] = 'Auth'
+        self.app.config['JWT_HEADER_TYPE'] = 'JWT'
 
         with self.app.test_request_context():
             self.assertEqual(get_access_expires(), timedelta(minutes=5))
@@ -44,4 +46,18 @@ def test_override_configs(self):
             self.assertEqual(get_blacklist_enabled(), True)
             self.assertIsInstance(get_blacklist_store(), simplekv.memory.DictStore)
             self.assertEqual(get_blacklist_checks(), 'all')
-            self.assertEqual(get_auth_header(), 'JWT')
+            self.assertEqual(get_jwt_header_name(), 'Auth')
+            self.assertEqual(get_jwt_header_type(), 'JWT')
+
+        self.app.config['JWT_HEADER_NAME'] = ''
+        self.app.config['JWT_ACCESS_TOKEN_EXPIRES'] = 'banana'
+        self.app.config['JWT_REFRESH_TOKEN_EXPIRES'] = 'banana'
+
+        self.app.testing = True  # Propagate exceptions
+        with self.app.test_request_context():
+            with self.assertRaises(RuntimeError):
+                get_jwt_header_name()
+            with self.assertRaises(RuntimeError):
+                get_access_expires()
+            with self.assertRaises(RuntimeError):
+                get_refresh_expires()

tests/test_protected_endpoints.py

@@ -62,9 +62,9 @@ def _jwt_post(self, url, jwt):
         data = json.loads(response.get_data(as_text=True))
         return status_code, data
 
-    def _jwt_get(self, url, jwt, auth_header='Bearer'):
-        auth_header = '{} {}'.format(auth_header, jwt).strip()
-        response = self.client.get(url, headers={'Authorization': auth_header})
+    def _jwt_get(self, url, jwt, header_name='Authorization', header_type='Bearer'):
+        header_type = '{} {}'.format(header_type, jwt).strip()
+        response = self.client.get(url, headers={header_name: header_type})
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         return status_code, data
@@ -279,22 +279,32 @@ def claims():
         self.assertEqual(status, 200)
         self.assertEqual(data, {'username': 'test', 'claims': {'foo': 'bar'}})
 
-    def test_different_auth_header(self):
+    def test_different_headers(self):
         response = self.client.post('/auth/login')
         data = json.loads(response.get_data(as_text=True))
         access_token = data['access_token']
 
-        self.app.config['JWT_AUTH_HEADER'] = 'JWT'
-        status, data = self._jwt_get('/protected', access_token, auth_header='JWT')
+        self.app.config['JWT_HEADER_TYPE'] = 'JWT'
+        status, data = self._jwt_get('/protected', access_token, header_type='JWT')
         self.assertEqual(data, {'msg': 'hello world'})
         self.assertEqual(status, 200)
 
-        self.app.config['JWT_AUTH_HEADER'] = ''
-        status, data = self._jwt_get('/protected', access_token, auth_header='')
+        self.app.config['JWT_HEADER_TYPE'] = ''
+        status, data = self._jwt_get('/protected', access_token, header_type='')
         self.assertEqual(data, {'msg': 'hello world'})
         self.assertEqual(status, 200)
 
-        self.app.config['JWT_AUTH_HEADER'] = ''
-        status, data = self._jwt_get('/protected', access_token, auth_header='Bearer')
+        self.app.config['JWT_HEADER_TYPE'] = ''
+        status, data = self._jwt_get('/protected', access_token, header_type='Bearer')
         self.assertIn('msg', data)
         self.assertEqual(status, 422)
+
+        self.app.config['JWT_HEADER_TYPE'] = 'Bearer'
+        self.app.config['JWT_HEADER_NAME'] = 'Auth'
+        status, data = self._jwt_get('/protected', access_token, header_name='Auth')
+        self.assertIn('msg', data)
+        self.assertEqual(status, 401)
+
+        status, data = self._jwt_get('/protected', access_token, header_name='Authorization')
+        self.assertEqual(data, {'msg': 'hello world'})
+        self.assertEqual(status, 200)