1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60
|
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()
|