Helper functions for verifying JWTs in a request (#131)

This is useful for using flask before_requests, or creating your own
decorators while utilizing flask_jwt_extended features.

Author: Landon Gilbert-Bland

docs/api.rst

@@ -34,6 +34,22 @@ Protected endpoint decorators
 .. autofunction:: jwt_optional
 
 
+.. _Verify Tokens in Request:
+
+Verify Tokens in Request
+~~~~~~~~~~~~~~~~~~~~~~~~
+These perform the same actions as the protected endpoint decorators, without
+actually decorating a function. These are very useful if you want to create
+your own decorators on top of flask jwt extended (such as role_required), or
+if you want to hook some of this extensions functionality into a flask
+before_request handler.
+
+.. autofunction:: verify_jwt_in_request
+.. autofunction:: verify_jwt_in_request_optional
+.. autofunction:: verify_fresh_jwt_in_request
+.. autofunction:: verify_jwt_refresh_token_in_request
+
+
 Utilities
 ~~~~~~~~~
 .. autofunction:: create_access_token

docs/custom_decorators.rst

@@ -0,0 +1,17 @@
+Custom Decorators
+=================
+
+You can create your own decorators that extend the functionality of the
+decorators provided by this extension. For example, you may want to create
+your own decorator that verifies a JWT is present as well as verifying that
+this token has sufficient permissions/roles to access an endpoint. 
+
+:ref:`Verify Tokens in Request` is a list of functions that can be
+used to build your own decorators (these are also what all the default
+decorators provided by this extension use internally).
+
+Here is an example of how this might look.
+
+.. literalinclude:: ../examples/custom_decorators.py
+
+

docs/index.rst

@@ -16,6 +16,7 @@ Flask-JWT-Extended's Documentation
    add_custom_data_claims
    tokens_from_complex_object
    complex_objects_from_token
+   custom_decorators
    refresh_tokens
    token_freshness
    changing_default_behavior

examples/custom_decorators.py

@@ -0,0 +1,52 @@
+from functools import wraps
+
+from flask import Flask, jsonify, request
+from flask_jwt_extended import (
+    JWTManager, verify_jwt_in_request, create_access_token,
+    get_jwt_claims
+)
+
+app = Flask(__name__)
+
+app.config['JWT_SECRET_KEY'] = 'super-secret'  # Change this!
+jwt = JWTManager(app)
+
+
+
+# Here is a custom decorator that verifies the JWT is present in
+# the request, as well as insuring that this user has a role of
+# `admin` in the access token
+def admin_required(fn):
+    @wraps(fn)
+    def wrapper(*args, **kwargs):
+        verify_jwt_in_request()
+        claims = get_jwt_claims()
+        if claims['roles'] != 'admin':
+            return jsonify(msg='Admins only!'), 403
+        else:
+            return fn(*args, **kwargs)
+    return wrapper
+
+
+@jwt.user_claims_loader
+def add_claims_to_access_token(identity):
+    if identity == 'admin':
+        return {'roles': 'admin'}
+    else:
+        return {'roles': 'pesant'}
+
+
+@app.route('/login', methods=['POST'])
+def login():
+    username = request.json.get('username', None)
+    access_token = create_access_token(username)
+    return jsonify(access_token=access_token)
+
+
+@app.route('/protected', methods=['GET'])
+@admin_required
+def protected():
+    return jsonify(secret_message="go banana!") 
+
+if __name__ == '__main__':
+    app.run()

flask_jwt_extended/__init__.py

@@ -1,13 +1,14 @@
 from .jwt_manager import JWTManager
 from .view_decorators import (
-    jwt_required, fresh_jwt_required, jwt_refresh_token_required, jwt_optional
+    fresh_jwt_required, jwt_optional, jwt_refresh_token_required, jwt_required,
+    verify_fresh_jwt_in_request, verify_jwt_in_request,
+    verify_jwt_in_request_optional, verify_jwt_refresh_token_in_request
 )
 from .utils import (
-    create_refresh_token, create_access_token, get_jwt_identity,
-    get_jwt_claims, set_access_cookies, set_refresh_cookies,
-    unset_jwt_cookies, unset_access_cookies, unset_refresh_cookies,
-    get_raw_jwt, get_current_user, current_user, get_jti, decode_token,
-    get_csrf_token
+    create_access_token, create_refresh_token, current_user, decode_token,
+    get_csrf_token, get_current_user, get_jti, get_jwt_claims, get_jwt_identity,
+    get_raw_jwt, set_access_cookies, set_refresh_cookies, unset_access_cookies,
+    unset_jwt_cookies, unset_refresh_cookies
 )
 
 __version__ = '3.8.2'

flask_jwt_extended/view_decorators.py

@@ -19,6 +19,72 @@
 )
 
 
+def verify_jwt_in_request():
+    """
+    Ensure that the requeste has a valid access token. This does not check the
+    freshness of the access token. Raises an appropiate exception there is
+    no token or if the token is invalid.
+    """
+    if request.method not in config.exempt_methods:
+        jwt_data = _decode_jwt_from_request(request_type='access')
+        ctx_stack.top.jwt = jwt_data
+        verify_token_claims(jwt_data)
+        _load_user(jwt_data[config.identity_claim_key])
+
+
+def verify_jwt_in_request_optional():
+    """
+    Optionally check if this request has a valid access token.  If an access
+    token in present in the request, :func:`~flask_jwt_extended.get_jwt_identity`
+    will return  the identity of the access token. If no access token is
+    present in the request, this simply returns, and
+    :func:`~flask_jwt_extended.get_jwt_identity` will return `None` instead.
+
+    If there is an invalid access token in the request (expired, tampered with,
+    etc), this will still raise the appropiate exception.
+    """
+    try:
+        if request.method not in config.exempt_methods:
+            jwt_data = _decode_jwt_from_request(request_type='access')
+            ctx_stack.top.jwt = jwt_data
+            verify_token_claims(jwt_data)
+            _load_user(jwt_data[config.identity_claim_key])
+    except (NoAuthorizationError, InvalidHeaderError):
+        pass
+
+
+def verify_fresh_jwt_in_request():
+    """
+    Ensure that the requeste has a valid and fresh access token. Raises an
+    appropiate exception if there is no token, the token is invalid, or the
+    token is not marked as fresh.
+    """
+    if request.method not in config.exempt_methods:
+        jwt_data = _decode_jwt_from_request(request_type='access')
+        ctx_stack.top.jwt = jwt_data
+        fresh = jwt_data['fresh']
+        if isinstance(fresh, bool):
+            if not fresh:
+                raise FreshTokenRequired('Fresh token required')
+        else:
+            now = timegm(datetime.utcnow().utctimetuple())
+            if fresh < now:
+                raise FreshTokenRequired('Fresh token required')
+        verify_token_claims(jwt_data)
+        _load_user(jwt_data[config.identity_claim_key])
+
+
+def verify_jwt_refresh_token_in_request():
+    """
+    Ensure that the requeste has a valid refresh token. Raises an appropiate
+    exception if there is no token or the token is invalid.
+    """
+    if request.method not in config.exempt_methods:
+        jwt_data = _decode_jwt_from_request(request_type='refresh')
+        ctx_stack.top.jwt = jwt_data
+        _load_user(jwt_data[config.identity_claim_key])
+
+
 def jwt_required(fn):
     """
     A decorator to protect a Flask endpoint.
@@ -31,11 +97,7 @@ def jwt_required(fn):
     """
     @wraps(fn)
     def wrapper(*args, **kwargs):
-        if request.method not in config.exempt_methods:
-            jwt_data = _decode_jwt_from_request(request_type='access')
-            ctx_stack.top.jwt = jwt_data
-            verify_token_claims(jwt_data)
-            _load_user(jwt_data[config.identity_claim_key])
+        verify_jwt_in_request()
         return fn(*args, **kwargs)
     return wrapper
 
@@ -56,13 +118,7 @@ def jwt_optional(fn):
     """
     @wraps(fn)
     def wrapper(*args, **kwargs):
-        try:
-            jwt_data = _decode_jwt_from_request(request_type='access')
-            ctx_stack.top.jwt = jwt_data
-            verify_token_claims(jwt_data)
-            _load_user(jwt_data[config.identity_claim_key])
-        except (NoAuthorizationError, InvalidHeaderError):
-            pass
+        verify_jwt_in_request_optional()
         return fn(*args, **kwargs)
     return wrapper
 
@@ -79,19 +135,7 @@ def fresh_jwt_required(fn):
     """
     @wraps(fn)
     def wrapper(*args, **kwargs):
-        if request.method not in config.exempt_methods:
-            jwt_data = _decode_jwt_from_request(request_type='access')
-            ctx_stack.top.jwt = jwt_data
-            fresh = jwt_data['fresh']
-            if isinstance(fresh, bool):
-                if not fresh:
-                    raise FreshTokenRequired('Fresh token required')
-            else:
-                now = timegm(datetime.utcnow().utctimetuple())
-                if fresh < now:
-                    raise FreshTokenRequired('Fresh token required')
-            verify_token_claims(jwt_data)
-            _load_user(jwt_data[config.identity_claim_key])
+        verify_fresh_jwt_in_request()
         return fn(*args, **kwargs)
     return wrapper
 
@@ -105,10 +149,7 @@ def jwt_refresh_token_required(fn):
     """
     @wraps(fn)
     def wrapper(*args, **kwargs):
-        if request.method not in config.exempt_methods:
-            jwt_data = _decode_jwt_from_request(request_type='refresh')
-            ctx_stack.top.jwt = jwt_data
-            _load_user(jwt_data[config.identity_claim_key])
+        verify_jwt_refresh_token_in_request()
         return fn(*args, **kwargs)
     return wrapper