Flask JWT 인증 API 로그인 갱신 블랙리스트 가드 데코레이터 예제로 배우는 완벽 가이드

from flask import Flask, request, jsonify
from werkzeug.security import check_password_hash
import jwt, datetime

app = Flask(__name__)
app.config['SECRET_KEY'] = "your_secret_key"

@app.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    user = find_user_by_email(data['email'])  # DB 조회
    
    if not user or not check_password_hash(user.password, data['password']):
        return jsonify({"msg": "Invalid credentials"}), 401
    
    access_token = jwt.encode(
        {"user_id": user.id, "exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=15)},
        app.config['SECRET_KEY'],
        algorithm="HS256"
    )
    
    refresh_token = jwt.encode(
        {"user_id": user.id, "exp": datetime.datetime.utcnow() + datetime.timedelta(days=7)},
        app.config['SECRET_KEY'],
        algorithm="HS256"
    )
    
    return jsonify({"access_token": access_token, "refresh_token": refresh_token}), 200

@app.route('/refresh', methods=['POST'])
def refresh():
    data = request.get_json()
    refresh_token = data.get("refresh_token")
    
    try:
        payload = jwt.decode(refresh_token, app.config['SECRET_KEY'], algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        return jsonify({"msg": "Refresh token expired"}), 401
    except jwt.InvalidTokenError:
        return jsonify({"msg": "Invalid token"}), 401
    
    new_access_token = jwt.encode(
        {"user_id": payload["user_id"], "exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=15)},
        app.config['SECRET_KEY'],
        algorithm="HS256"
    )
    
    return jsonify({"access_token": new_access_token}), 200

blacklist = set()

@app.route('/logout', methods=['POST'])
def logout():
    token = request.headers.get("Authorization").split()[1]
    blacklist.add(token)
    return jsonify({"msg": "Successfully logged out"}), 200

@app.before_request
def check_blacklist():
    auth_header = request.headers.get("Authorization")
    if auth_header:
        token = auth_header.split()[1]
        if token in blacklist:
            return jsonify({"msg": "Token has been revoked"}), 401

from functools import wraps

def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        auth_header = request.headers.get("Authorization")
        if not auth_header:
            return jsonify({"msg": "Token is missing"}), 401
        
        token = auth_header.split()[1]
        if token in blacklist:
            return jsonify({"msg": "Token revoked"}), 401
        
        try:
            payload = jwt.decode(token, app.config['SECRET_KEY'], algorithms=["HS256"])
        except jwt.ExpiredSignatureError:
            return jsonify({"msg": "Token expired"}), 401
        except jwt.InvalidTokenError:
            return jsonify({"msg": "Invalid token"}), 401
        
        return f(payload["user_id"], *args, **kwargs)
    return decorated

@app.route('/protected', methods=['GET'])
@token_required
def protected_route(user_id):
    return jsonify({"msg": f"Hello user {user_id}, this is a protected route"}), 200