JWT Revoking / Blocklist¶

JWT revoking is a mechanism for preventing an otherwise valid JWT from accessing your routes while still letting other valid JWTs in. To utilize JWT revoking in this extension, you must define a callback function via the token_in_blocklist_loader() decorator. This function is called whenever a valid JWT is used to access a protected route. The callback will receive the JWT header and JWT payload as arguments, and must return True if the JWT has been revoked.

In production, you will want to use some form of persistent storage (database, redis, etc) to store your JWTs. It would be bad if your application forgot that a JWT was revoked if it was restarted. We can provide some general recommendations on what type of storage engine to use, but ultimately the choice will depend on your specific application and tech stack.

Redis¶

If your only requirements are to check if a JWT has been revoked, our recommendation is to use redis. It is blazing fast, can be configured to persist data to disc, and can automatically clear out JWTs after they expire by utilizing the Time To Live (TTL) functionality when storing a JWT. Here is an example using redis:

from datetime import timedelta

import redis
from flask import Flask
from flask import jsonify

from flask_jwt_extended import create_access_token
from flask_jwt_extended import get_jwt
from flask_jwt_extended import jwt_required
from flask_jwt_extended import JWTManager

ACCESS_EXPIRES = timedelta(hours=1)

app = Flask(__name__)
app.config["JWT_SECRET_KEY"] = "super-secret"  # Change this!
app.config["JWT_ACCESS_TOKEN_EXPIRES"] = ACCESS_EXPIRES
jwt = JWTManager(app)

# Setup our redis connection for storing the blocklisted tokens. You will probably
# want your redis instance configured to persist data to disk, so that a restart
# does not cause your application to forget that a JWT was revoked.
jwt_redis_blocklist = redis.StrictRedis(
    host="localhost", port=6379, db=0, decode_responses=True
)


# Callback function to check if a JWT exists in the redis blocklist
@jwt.token_in_blocklist_loader
def check_if_token_is_revoked(jwt_header, jwt_payload: dict):
    jti = jwt_payload["jti"]
    token_in_redis = jwt_redis_blocklist.get(jti)
    return token_in_redis is not None


@app.route("/login", methods=["POST"])
def login():
    access_token = create_access_token(identity="example_user")
    return jsonify(access_token=access_token)


# Endpoint for revoking the current users access token. Save the JWTs unique
# identifier (jti) in redis. Also set a Time to Live (TTL)  when storing the JWT
# so that it will automatically be cleared out of redis after the token expires.
@app.route("/logout", methods=["DELETE"])
@jwt_required()
def logout():
    jti = get_jwt()["jti"]
    jwt_redis_blocklist.set(jti, "", ex=ACCESS_EXPIRES)
    return jsonify(msg="Access token revoked")


# A blocklisted access token will not be able to access this any more
@app.route("/protected", methods=["GET"])
@jwt_required()
def protected():
    return jsonify(hello="world")


if __name__ == "__main__":
    app.run()

Warning

Note that configuring redis to be disk-persistent is an absolutely necessity for production use. Otherwise, events like power outages or server crashes/reboots would cause all invalidated tokens to become valid again (assuming the secret key does not change). This is especially concering for long-lived refresh tokens, discussed below.

Database¶

If you need to keep track of information about revoked JWTs our recommendation is to utilize a database. This allows you to easily store and utilize metadata for revoked tokens, such as when it was revoked, who revoked it, can it be un-revoked, etc. Here is an example using SQLAlchemy:

from datetime import datetime
from datetime import timedelta
from datetime import timezone

from flask import Flask
from flask import jsonify
from flask_sqlalchemy import SQLAlchemy

from flask_jwt_extended import create_access_token
from flask_jwt_extended import get_jwt
from flask_jwt_extended import jwt_required
from flask_jwt_extended import JWTManager

app = Flask(__name__)

ACCESS_EXPIRES = timedelta(hours=1)
app.config["JWT_SECRET_KEY"] = "super-secret"  # Change this!
app.config["JWT_ACCESS_TOKEN_EXPIRES"] = ACCESS_EXPIRES
jwt = JWTManager(app)

# We are using an in memory database here as an example. Make sure to use a
# database with persistent storage in production!
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite://"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db = SQLAlchemy(app)


# This could be expanded to fit the needs of your application. For example,
# it could track who revoked a JWT, when a token expires, notes for why a
# JWT was revoked, an endpoint to un-revoked a JWT, etc.
# Making jti an index can significantly speed up the search when there are
# tens of thousands of records. Remember this query will happen for every
# (protected) request,
# If your database supports a UUID type, this can be used for the jti column
# as well
class TokenBlocklist(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    jti = db.Column(db.String(36), nullable=False, index=True)
    created_at = db.Column(db.DateTime, nullable=False)


# Callback function to check if a JWT exists in the database blocklist
@jwt.token_in_blocklist_loader
def check_if_token_revoked(jwt_header, jwt_payload: dict) -> bool:
    jti = jwt_payload["jti"]
    token = db.session.query(TokenBlocklist.id).filter_by(jti=jti).scalar()

    return token is not None


@app.route("/login", methods=["POST"])
def login():
    access_token = create_access_token(identity="example_user")
    return jsonify(access_token=access_token)


# Endpoint for revoking the current users access token. Saved the unique
# identifier (jti) for the JWT into our database.
@app.route("/logout", methods=["DELETE"])
@jwt_required()
def modify_token():
    jti = get_jwt()["jti"]
    now = datetime.now(timezone.utc)
    db.session.add(TokenBlocklist(jti=jti, created_at=now))
    db.session.commit()
    return jsonify(msg="JWT revoked")


# A blocklisted access token will not be able to access this any more
@app.route("/protected", methods=["GET"])
@jwt_required()
def protected():
    return jsonify(hello="world")


if __name__ == "__main__":
    db.create_all()
    app.run()

Revoking Refresh Tokens¶

It is critical to note that a user’s refresh token must also be revoked when logging out; otherwise, this refresh token could just be used to generate a new access token. Usually this falls to the responsibility of the frontend application, which must send two separate requests to the backend in order to revoke these tokens.

This can be implemented via two separate routes marked with @jwt_required() and @jwt_required(refresh=True) to revoke access and refresh tokens, respectively. However, it is more convenient to provide a single endpoint where the frontend can send a DELETE for each token. The following is an example:

@app.route("/logout", methods=["DELETE"])
@jwt_required(verify_type=False)
def logout():
    token = get_jwt()
    jti = token["jti"]
    ttype = token["type"]
    jwt_redis_blocklist.set(jti, "", ex=ACCESS_EXPIRES)

    # Returns "Access token revoked" or "Refresh token revoked"
    return jsonify(msg=f"{ttype.capitalize()} token successfully revoked")

or, for the database format:

class TokenBlocklist(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    jti = db.Column(db.String(36), nullable=False, index=True)
    type = db.Column(db.String(16), nullable=False)
    user_id = db.Column(
        db.ForeignKey('person.id'),
        default=lambda: get_current_user().id,
        nullable=False,
    )
    created_at = db.Column(
        db.DateTime,
        server_default=func.now(),
        nullable=False,
    )

@app.route("/logout", methods=["DELETE"])
@jwt_required(verify_type=False)
def modify_token():
    token = get_jwt()
    jti = token["jti"]
    ttype = token["type"]
    now = datetime.now(timezone.utc)
    db.session.add(TokenBlocklist(jti=jti, type=ttype, created_at=now))
    db.session.commit()
    return jsonify(msg=f"{ttype.capitalize()} token successfully revoked")

Token type and user columns are not required and can be omitted. That being said, including these can help to audit that the frontend is performing its revoking job correctly and revoking both tokens.

Alternatively, there are a few ways to revoke both tokens at once:

Send the access token in the header (per usual), and send the refresh token in the DELETE request body. This saves a request but still needs frontend changes, so may not be worth implementing
Embed the refresh token’s jti in the access token. The revoke route should be authenticated with the access token. Upon revoking the access token, extract the refresh jti from it and invalidate both. This has the advantage of requiring no extra work from the frontend.
Store every generated tokens jti in a database upon creation. Have a boolean column to represent whether it is valid or not, which the token_in_blocklist_loader should respond based upon. Upon revoking a token, mark that token row as invalid, as well as all other tokens from the same user generated at the same time. This would also allow for a “log out everywhere” option where all tokens for a user are invalidated at once, which is otherwise not easily possibile

The best option of course depends and needs to be chosen based upon the circumstances. If there if ever a time where an unknown, untracked token needs to be immediately invalidated, this can be accomplished by changing the secret key.

JWT Revoking / Blocklist¶

Redis¶

Database¶

Revoking Refresh Tokens¶

Contents

Navigation