Internal refactoring

Author: vimalloc

flask_jwt_extended/config.py

@@ -23,7 +23,8 @@ class _Config(object):
     should be done with flasks ```app.config```.
 
     Default values for the configuration options are set in the jwt_manager
-    object. All of these values are read only.
+    object. All of these values are read only. This is simply a loose wrapper
+    with some helper functionality for flasks `app.config`.
     """
 
     @property
@@ -224,11 +225,11 @@ def cookie_max_age(self):
         return None if self.session_cookie else 2147483647  # 2^31
 
     @property
-    def identity_claim(self):
+    def identity_claim_key(self):
         return current_app.config['JWT_IDENTITY_CLAIM']
 
     @property
-    def user_claims(self):
+    def user_claims_key(self):
         return current_app.config['JWT_USER_CLAIMS']
 
 config = _Config()

flask_jwt_extended/jwt_manager.py

@@ -381,7 +381,7 @@ def _create_refresh_token(self, identity, expires_delta=None):
             algorithm=config.algorithm,
             expires_delta=expires_delta,
             csrf=config.csrf_protect,
-            identity_claim=config.identity_claim
+            identity_claim_key=config.identity_claim_key,
         )
         return refresh_token
 
@@ -397,7 +397,8 @@ def _create_access_token(self, identity, fresh=False, expires_delta=None):
             fresh=fresh,
             user_claims=self._user_claims_callback(identity),
             csrf=config.csrf_protect,
-            identity_claim=config.identity_claim
+            identity_claim_key=config.identity_claim_key,
+            user_claims_key=config.user_claims_key
         )
         return access_token
 

flask_jwt_extended/tokens.py

@@ -2,9 +2,9 @@
 import uuid
 
 import jwt
+from werkzeug.security import safe_str_cmp
 
-from flask_jwt_extended.exceptions import JWTDecodeError
-from flask_jwt_extended.config import config
+from flask_jwt_extended.exceptions import JWTDecodeError, CSRFError
 
 
 def _create_csrf_token():
@@ -26,7 +26,7 @@ def _encode_jwt(additional_token_data, expires_delta, secret, algorithm):
 
 
 def encode_access_token(identity, secret, algorithm, expires_delta, fresh,
-                        user_claims, csrf, identity_claim):
+                        user_claims, csrf, identity_claim_key, user_claims_key):
     """
     Creates a new encoded (utf-8) access token.
 
@@ -41,26 +41,27 @@ def encode_access_token(identity, secret, algorithm, expires_delta, fresh,
                         be json serializable
     :param csrf: Whether to include a csrf double submit claim in this token
                  (boolean)
-    :param identity_claim: Which claim should be used to store the identity in
+    :param identity_claim_key: Which key should be used to store the identity
+    :param user_claims_key: Which key should be used to store the user claims
     :return: Encoded access token
     """
-    # Create the jwt
     token_data = {
-        identity_claim: identity,
+        identity_claim_key: identity,
         'fresh': fresh,
         'type': 'access',
     }
 
-    # Add `user_claims` only is not empty or None.
+    # Don't add extra data to the token if user_claims is empty.
     if user_claims:
-        token_data[config.user_claims] = user_claims
+        token_data[user_claims_key] = user_claims
 
     if csrf:
         token_data['csrf'] = _create_csrf_token()
     return _encode_jwt(token_data, expires_delta, secret, algorithm)
 
 
-def encode_refresh_token(identity, secret, algorithm, expires_delta, csrf, identity_claim):
+def encode_refresh_token(identity, secret, algorithm, expires_delta, csrf,
+                         identity_claim_key):
     """
     Creates a new encoded (utf-8) refresh token.
 
@@ -71,28 +72,29 @@ def encode_refresh_token(identity, secret, algorithm, expires_delta, csrf, ident
                                (datetime.timedelta)
     :param csrf: Whether to include a csrf double submit claim in this token
                  (boolean)
-    :param identity_claim: Which claim should be used to store the identity in
+    :param identity_claim_key: Which key should be used to store the identity
     :return: Encoded refresh token
     """
     token_data = {
-        identity_claim: identity,
+        identity_claim_key: identity,
         'type': 'refresh',
     }
     if csrf:
         token_data['csrf'] = _create_csrf_token()
     return _encode_jwt(token_data, expires_delta, secret, algorithm)
 
 
-def decode_jwt(encoded_token, secret, algorithm, csrf, identity_claim):
+def decode_jwt(encoded_token, secret, algorithm, identity_claim_key,
+               user_claims_key, csrf_value=None):
     """
     Decodes an encoded JWT
 
     :param encoded_token: The encoded JWT string to decode
     :param secret: Secret key used to encode the JWT
     :param algorithm: Algorithm used to encode the JWT
-    :param csrf: If this token is expected to have a CSRF double submit
-                 value present (boolean)
-    :param identity_claim: expected claim that is used to identify the subject
+    :param identity_claim_key: expected key that contains the identity
+    :param user_claims_key: expected key that contains the user claims
+    :param csrf_value: Expected double submit csrf value
     :return: Dictionary containing contents of the JWT
     """
     # This call verifies the ext, iat, and nbf claims
@@ -101,16 +103,18 @@ def decode_jwt(encoded_token, secret, algorithm, csrf, identity_claim):
     # Make sure that any custom claims we expect in the token are present
     if 'jti' not in data:
         raise JWTDecodeError("Missing claim: jti")
-    if identity_claim not in data:
-        raise JWTDecodeError("Missing claim: {}".format(identity_claim))
+    if identity_claim_key not in data:
+        raise JWTDecodeError("Missing claim: {}".format(identity_claim_key))
     if 'type' not in data or data['type'] not in ('refresh', 'access'):
         raise JWTDecodeError("Missing or invalid claim: type")
     if data['type'] == 'access':
         if 'fresh' not in data:
             raise JWTDecodeError("Missing claim: fresh")
-        if config.user_claims not in data:
-            data[config.user_claims] = {}
-    if csrf:
+        if user_claims_key not in data:
+            data[user_claims_key] = {}
+    if csrf_value:
         if 'csrf' not in data:
             raise JWTDecodeError("Missing claim: csrf")
+        if not safe_str_cmp(data['csrf'], csrf_value):
+            raise CSRFError("CSRF double submit tokens do not match")
     return data

flask_jwt_extended/utils.py

@@ -28,7 +28,7 @@ def get_jwt_identity():
     In a protected endpoint, this will return the identity of the JWT that is
     accessing this endpoint. If no JWT is present,`None` is returned instead.
     """
-    return get_raw_jwt().get(config.identity_claim, None)
+    return get_raw_jwt().get(config.identity_claim_key, None)
 
 
 def get_jwt_claims():
@@ -37,7 +37,7 @@ def get_jwt_claims():
     in the JWT that is accessing the endpoint. If no custom user claims are
     present, an empty dict is returned instead.
     """
-    return get_raw_jwt().get(config.user_claims, {})
+    return get_raw_jwt().get(config.user_claims_key, {})
 
 
 def get_current_user():
@@ -60,19 +60,21 @@ def get_jti(encoded_token):
     return decode_token(encoded_token).get('jti')
 
 
-def decode_token(encoded_token):
+def decode_token(encoded_token, csrf_value=None):
     """
     Returns the decoded token (python dict) from an encoded JWT. This does all
     the checks to insure that the decoded token is valid before returning it.
 
     :param encoded_token: The encoded JWT to decode into a python dict.
+    :param csrf_value: Expected CSRF double submit value (optional)
     """
     return decode_jwt(
         encoded_token=encoded_token,
         secret=config.decode_key,
         algorithm=config.algorithm,
-        csrf=config.csrf_protect,
-        identity_claim=config.identity_claim
+        identity_claim_key=config.identity_claim_key,
+        user_claims_key=config.user_claims_key,
+        csrf_value=csrf_value
     )
 
 
@@ -153,13 +155,7 @@ def verify_token_claims(*args, **kwargs):
 
 
 def get_csrf_token(encoded_token):
-    token = decode_jwt(
-        encoded_token,
-        config.decode_key,
-        config.algorithm,
-        csrf=True,
-        identity_claim=config.identity_claim
-    )
+    token = decode_token(encoded_token)
     return token['csrf']
 
 

flask_jwt_extended/view_decorators.py

@@ -1,7 +1,6 @@
 from functools import wraps
 
 from flask import request
-from werkzeug.security import safe_str_cmp
 try:
     from flask import _app_ctx_stack as ctx_stack
 except ImportError:  # pragma: no cover
@@ -13,9 +12,8 @@
     FreshTokenRequired, CSRFError, UserLoadError, RevokedTokenError,
     UserClaimsVerificationError
 )
-from flask_jwt_extended.tokens import decode_jwt
 from flask_jwt_extended.utils import (
-    has_user_loader, user_loader, token_in_blacklist,
+    has_user_loader, user_loader, token_in_blacklist, decode_token,
     has_token_in_blacklist_callback, verify_token_claims
 )
 
@@ -34,9 +32,9 @@ def jwt_required(fn):
     def wrapper(*args, **kwargs):
         jwt_data = _decode_jwt_from_request(request_type='access')
         ctx_stack.top.jwt = jwt_data
-        if not verify_token_claims(jwt_data[config.user_claims]):
+        if not verify_token_claims(jwt_data[config.user_claims_key]):
             raise UserClaimsVerificationError('User claims verification failed')
-        _load_user(jwt_data[config.identity_claim])
+        _load_user(jwt_data[config.identity_claim_key])
         return fn(*args, **kwargs)
     return wrapper
 
@@ -60,9 +58,9 @@ def wrapper(*args, **kwargs):
         try:
             jwt_data = _decode_jwt_from_request(request_type='access')
             ctx_stack.top.jwt = jwt_data
-            if not verify_token_claims(jwt_data[config.user_claims]):
+            if not verify_token_claims(jwt_data[config.user_claims_key]):
                 raise UserClaimsVerificationError('User claims verification failed')
-            _load_user(jwt_data[config.identity_claim])
+            _load_user(jwt_data[config.identity_claim_key])
         except (NoAuthorizationError, InvalidHeaderError):
             pass
         return fn(*args, **kwargs)
@@ -85,9 +83,9 @@ def wrapper(*args, **kwargs):
         ctx_stack.top.jwt = jwt_data
         if not jwt_data['fresh']:
             raise FreshTokenRequired('Fresh token required')
-        if not verify_token_claims(jwt_data[config.user_claims]):
+        if not verify_token_claims(jwt_data[config.user_claims_key]):
             raise UserClaimsVerificationError('User claims verification failed')
-        _load_user(jwt_data[config.identity_claim])
+        _load_user(jwt_data[config.identity_claim_key])
         return fn(*args, **kwargs)
     return wrapper
 
@@ -103,7 +101,7 @@ def jwt_refresh_token_required(fn):
     def wrapper(*args, **kwargs):
         jwt_data = _decode_jwt_from_request(request_type='refresh')
         ctx_stack.top.jwt = jwt_data
-        _load_user(jwt_data[config.identity_claim])
+        _load_user(jwt_data[config.identity_claim_key])
         return fn(*args, **kwargs)
     return wrapper
 
@@ -148,20 +146,14 @@ def _decode_jwt_from_headers():
         if len(parts) != 1:
             msg = "Bad {} header. Expected value '<JWT>'".format(header_name)
             raise InvalidHeaderError(msg)
-        token = parts[0]
+        encoded_token = parts[0]
     else:
         if parts[0] != header_type or len(parts) != 2:
             msg = "Bad {} header. Expected value '{} <JWT>'".format(header_name, header_type)
             raise InvalidHeaderError(msg)
-        token = parts[1]
+        encoded_token = parts[1]
 
-    return decode_jwt(
-        encoded_token=token,
-        secret=config.decode_key,
-        algorithm=config.algorithm,
-        csrf=False,
-        identity_claim=config.identity_claim
-    )
+    return decode_token(encoded_token)
 
 
 def _decode_jwt_from_cookies(request_type):
@@ -172,29 +164,18 @@ def _decode_jwt_from_cookies(request_type):
         cookie_key = config.refresh_cookie_name
         csrf_header_key = config.refresh_csrf_header_name
 
+    if config.csrf_protect and request.method in config.csrf_request_methods:
+        csrf_value = request.headers.get(csrf_header_key, None)
+        if not csrf_value:
+            raise CSRFError("Missing CSRF token in headers")
+    else:
+        csrf_value = None
+
     encoded_token = request.cookies.get(cookie_key)
     if not encoded_token:
         raise NoAuthorizationError('Missing cookie "{}"'.format(cookie_key))
 
-    decoded_token = decode_jwt(
-        encoded_token=encoded_token,
-        secret=config.decode_key,
-        algorithm=config.algorithm,
-        csrf=config.csrf_protect,
-        identity_claim=config.identity_claim
-    )
-
-    # Verify csrf double submit tokens match if required
-    if config.csrf_protect and request.method in config.csrf_request_methods:
-        csrf_token_in_token = decoded_token['csrf']
-        csrf_token_in_header = request.headers.get(csrf_header_key, None)
-
-        if not csrf_token_in_header:
-            raise CSRFError("Missing CSRF token in headers")
-        if not safe_str_cmp(csrf_token_in_header, csrf_token_in_token):
-            raise CSRFError("CSRF double submit tokens do not match")
-
-    return decoded_token
+    return decode_token(encoded_token, csrf_value=csrf_value)
 
 
 def _decode_jwt_from_request(request_type):

tests/test_config.py

@@ -52,8 +52,8 @@ def test_default_configs(app):
 
         assert config.cookie_max_age is None
 
-        assert config.identity_claim == 'identity'
-        assert config.user_claims == 'user_claims'
+        assert config.identity_claim_key == 'identity'
+        assert config.user_claims_key == 'user_claims'
 
 
 def test_override_configs(app):
@@ -125,8 +125,8 @@ def test_override_configs(app):
 
         assert config.cookie_max_age == 2147483647
 
-        assert config.identity_claim == 'foo'
-        assert config.user_claims == 'bar'
+        assert config.identity_claim_key == 'foo'
+        assert config.user_claims_key == 'bar'
 
 
 # noinspection PyStatementEffect

tests/test_decode_tokens.py

@@ -29,7 +29,7 @@ def default_access_token(app):
     with app.test_request_context():
         return {
             'jti': '1234',
-            config.identity_claim: 'username',
+            config.identity_claim_key: 'username',
             'type': 'access',
             'fresh': True,
             'csrf': 'abcd'
@@ -49,9 +49,9 @@ def empty_user_loader_return(identity):
     with app.test_request_context():
         token = create_access_token('username')
         pure_decoded = jwt.decode(token, config.decode_key, algorithms=[config.algorithm])
-        assert config.user_claims not in pure_decoded
+        assert config.user_claims_key not in pure_decoded
         extension_decoded = decode_token(token)
-        assert config.user_claims in extension_decoded
+        assert config.user_claims_key in extension_decoded
 
 
 @pytest.mark.parametrize("missing_claim", ['jti', 'type', 'identity', 'fresh', 'csrf'])
@@ -61,7 +61,7 @@ def test_missing_jti_claim(app, default_access_token, missing_claim):
 
     with pytest.raises(JWTDecodeError):
         with app.test_request_context():
-            decode_token(missing_jwt_token)
+            decode_token(missing_jwt_token, csrf_value='abcd')
 
 
 def test_bad_token_type(app, default_access_token):