Clean up database blocklist example

Author: Landon Gilbert-Bland

docs/blocklist_and_token_revoking.rst

@@ -10,27 +10,26 @@ 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.
 
-Here is a basic example of this in action.
-
-.. literalinclude:: ../examples/blocklist.py
-
 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.
+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.
 
-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.
+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:
 
+.. literalinclude:: ../examples/blocklist_redis.py
+
+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.
-
-Ultimately the choice of what persistent storage engine to use will depend on
-your specific application and tech stack. For some examples of JWT revoking using
-redist or a database, check out:
+etc. Here is an example using SQLAlchemy:
 
-- `Redis Example <https://github.com/vimalloc/flask-jwt-extended/blob/master/examples/redis_blocklist.py>`_
-- `Database Example <https://github.com/vimalloc/flask-jwt-extended/tree/master/examples/database_blocklist>`_
+.. literalinclude:: ../examples/blocklist_database.py

examples/blocklist.py

@@ -0,0 +1,72 @@
+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.
+class TokenBlocklist(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    jti = db.Column(db.String(36), nullable=False)
+    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):
+    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()

examples/blocklist_database.py

@@ -16,7 +16,9 @@
 app.config["JWT_ACCESS_TOKEN_EXPIRES"] = ACCESS_EXPIRES
 jwt = JWTManager(app)
 
-# Setup our redis connection for storing the blocklisted tokens
+# 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
 )
@@ -36,14 +38,14 @@ def login():
     return jsonify(access_token=access_token)
 
 
-# Endpoint for revoking the current users access token. Set a Time to Live (TTL)
-# when storing the token so that it will automatically be cleared out of redis
-# after the token expires.
+# 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, "", ACCESS_EXPIRES)
+    jwt_redis_blocklist.set(jti, "", ex=ACCESS_EXPIRES)
     return jsonify(msg="Access token revoked")