Allow selective disabling of blocklist check (#501)

* Allow selective disabling of blocklist check

Fixes #499

* Update the documentation

---------

Co-authored-by: Indrajeet Khandekar <[email protected]>

Author: indrajeet307

flask_jwt_extended/view_decorators.py

@@ -48,6 +48,7 @@ def verify_jwt_in_request(
     refresh: bool = False,
     locations: Optional[LocationType] = None,
     verify_type: bool = True,
+    skip_revocation_check: bool = False,
 ) -> Optional[Tuple[dict, dict]]:
     """
     Verify that a valid JWT is present in the request, unless ``optional=True`` in
@@ -76,6 +77,14 @@ def verify_jwt_in_request(
         to the ``refresh`` argument. If ``False``, type will not be checked and both
         access and refresh tokens will be accepted.
 
+    :param skip_revocation_check:
+        If ``True``, revocation status of the token will be *not* checked. If ``False``,
+        revocation status of the token will be checked.
+
+    :param skip_revocation_check:
+        If ``True``, revocation status of the token will be *not* checked. If ``False``,
+        revocation status of the token will be checked.
+
     :return:
         A tuple containing the jwt_header and the jwt_data if a valid JWT is
         present in the request. If ``optional=True`` and no JWT is in the request,
@@ -87,7 +96,11 @@ def verify_jwt_in_request(
 
     try:
         jwt_data, jwt_header, jwt_location = _decode_jwt_from_request(
-            locations, fresh, refresh=refresh, verify_type=verify_type
+            locations,
+            fresh,
+            refresh=refresh,
+            verify_type=verify_type,
+            skip_revocation_check=skip_revocation_check,
         )
 
     except NoAuthorizationError:
@@ -115,6 +128,7 @@ def jwt_required(
     refresh: bool = False,
     locations: Optional[LocationType] = None,
     verify_type: bool = True,
+    skip_revocation_check: bool = False,
 ) -> Any:
     """
     A decorator to protect a Flask endpoint with JSON Web Tokens.
@@ -145,12 +159,18 @@ def jwt_required(
         If ``True``, the token type (access or refresh) will be checked according
         to the ``refresh`` argument. If ``False``, type will not be checked and both
         access and refresh tokens will be accepted.
+
+    :param skip_revocation_check:
+        If ``True``, revocation status of the token will be *not* checked. If ``False``,
+        revocation status of the token will be checked.
     """
 
     def wrapper(fn):
         @wraps(fn)
         def decorator(*args, **kwargs):
-            verify_jwt_in_request(optional, fresh, refresh, locations, verify_type)
+            verify_jwt_in_request(
+                optional, fresh, refresh, locations, verify_type, skip_revocation_check
+            )
             return current_app.ensure_sync(fn)(*args, **kwargs)
 
         return decorator
@@ -284,6 +304,7 @@ def _decode_jwt_from_request(
     fresh: bool,
     refresh: bool = False,
     verify_type: bool = True,
+    skip_revocation_check: bool = False,
 ) -> Tuple[dict, dict, str]:
     # Figure out what locations to look for the JWT in this request
     if isinstance(locations, str):
@@ -346,7 +367,10 @@ def _decode_jwt_from_request(
 
     if fresh:
         _verify_token_is_fresh(jwt_header, decoded_token)
-    verify_token_not_blocklisted(jwt_header, decoded_token)
+
+    if not skip_revocation_check:
+        verify_token_not_blocklisted(jwt_header, decoded_token)
+
     custom_verification_for_token(jwt_header, decoded_token)
 
     return decoded_token, jwt_header, jwt_location

tests/test_blocklist.py

@@ -21,6 +21,16 @@ def app():
     def access_protected():
         return jsonify(foo="bar")
 
+    @app.route("/protected_skip_blocklist", methods=["GET"])
+    @jwt_required(verify_type=False, skip_revocation_check=True)
+    def access_protected_skip_blocklist():
+        return jsonify(foo="bar")
+
+    @app.route("/protected_noskip_blocklist", methods=["GET"])
+    @jwt_required(verify_type=False)
+    def access_protected_no_skip_blocklist():
+        return jsonify(foo="bar")
+
     @app.route("/refresh_protected", methods=["GET"])
     @jwt_required(refresh=True)
     def refresh_protected():
@@ -29,6 +39,48 @@ def refresh_protected():
     return app
 
 
+@pytest.mark.parametrize("blocklist_type", [["access"], ["refresh", "access"]])
+def test_blocklisted_access_token_revocation_skip(app, blocklist_type):
+    jwt = get_jwt_manager(app)
+
+    @jwt.token_in_blocklist_loader
+    def check_blocklisted(jwt_header, jwt_data):
+        assert jwt_header["alg"] == "HS256"
+        assert jwt_data["sub"] == "username"
+        return True
+
+    with app.test_request_context():
+        access_token = create_access_token("username")
+
+    test_client = app.test_client()
+    response = test_client.get(
+        "/protected_skip_blocklist", headers=make_headers(access_token)
+    )
+    assert response.get_json() == {"foo": "bar"}
+    assert response.status_code == 200
+
+
+@pytest.mark.parametrize("blocklist_type", [["access"], ["refresh", "access"]])
+def test_blocklisted_access_token_revocation_no_skip(app, blocklist_type):
+    jwt = get_jwt_manager(app)
+
+    @jwt.token_in_blocklist_loader
+    def check_blocklisted(jwt_header, jwt_data):
+        assert jwt_header["alg"] == "HS256"
+        assert jwt_data["sub"] == "username"
+        return True
+
+    with app.test_request_context():
+        access_token = create_access_token("username")
+
+    test_client = app.test_client()
+    response = test_client.get(
+        "/protected_noskip_blocklist", headers=make_headers(access_token)
+    )
+    assert response.get_json() == {"msg": "Token has been revoked"}
+    assert response.status_code == 401
+
+
 @pytest.mark.parametrize("blocklist_type", [["access"], ["refresh", "access"]])
 def test_non_blocklisted_access_token(app, blocklist_type):
     jwt = get_jwt_manager(app)