[Flask] Flask-JWT-Extended JWT 토큰

Flask-JWT-Extended 기본 설정

• 설치 pip install flask-jwt-extended
• config.py에 jwt 토큰을 사용하기 위한 Secret Key와 토큰 만료시간을 등록한다.

JWT_SECRET_KEY = ''  # string 으로 직접 정의
JWT_ACCESS_TOKEN_EXPIRES = datetime.timedelta(minutes=int)  # access 토큰 만료 시간
JWT_REFRESH_TOKEN_EXPIRES = datetime.timedelta(minutes=int) # refresh 토큰 만료 시간

로그인

# 로그인 시도 횟수
login_attempts = {}
login_attempts_limit = 5 

@bp.route('/login', methods=['POST', 'GET'])
def login():
    if request.method == 'POST':
        user_id = request.get_json()['userId']
        user_pw = request.get_json()['pw']

        # 해당 유저 아이디 조회
        user = User.query.filter_by(user_id=user_id).first() 

        # 로그인 5회 실패 시 
        login_attempts[user_id] = login_attempts.get(user_id, 0)
        if login_attempts[user_id] >= login_attempts_limit:
            return 'Login attempts exceeded limit'

        if user:
            if check_password_hash(user.pwd.decode('utf-8'), user_pw):
                # 로그인 성공 시 해당 유저 아이디 로그인 시도 횟수 초기화
                login_attempts.pop(user_id, None)

                # 토큰 발급
                access_token = create_access_token(identity=user_id)
                refresh_token = create_refresh_token(identity=user_id)

                # 리프레시 토큰 정보
                refresh_token_data = decode_token(refresh_token)  # 토큰 디코딩
                created_at = datetime.fromtimestamp(refresh_token_data['iat'])  # 토큰 생성일
                expired_at = datetime.fromtimestamp(refresh_token_data['exp'])  # 토큰 만료일
                is_revoked = False  # 토큰 취소 여부

                # 리프레시 토큰 정보 DB저장  
                existing_token = Token.query.filter_by(user_id=user_id).first()  
                if existing_token:  # 해당 아이디로 저장된 토큰이 이미 있으면 업데이트
                    existing_token.token = refresh_token
                    existing_token.created_at = created_at
                    existing_token.expired_at = expired_at
                    existing_token.is_revoked = is_revoked
                    db.session.commit()
                else: 
                    new_token = Token(token=refresh_token, user_id=user_id, created_at=created_at, expired_at=expired_at, is_revoked=is_revoked)
                    db.session.add(new_token)
                    db.session.commit()
                    
                response={
                    'loginRes': "Y",
                    'access': access_token,
                    'refresh': refresh_token,
                    'resultCode': 200,
                    'resultDesc': "Success",
                    'resultMsg': f'{user.user_name}님 환영합니다.'}
                return  Response(response=json.dumps(response), status=200, mimetype="application/json")
            else:
                # 로그인 실패 시 로그인 시도 횟수 +1
                login_attempts[user_id] = login_attempts.get(user_id, 0) + 1
                response={
                'loginRes': "N",
                'access': "",
                'refresh': "",
                'resultCode': 401,
                'resultDesc': "Unauthorized",
                'resultMsg': "비밀번호를 다시 확인해주세요."
                }
                return Response(response=json.dumps(response), status=401, mimetype="application/json")

        else: 
            # 로그인 실패 시 로그인 시도 횟수 +1
            login_attempts[user_id] = login_attempts.get(user_id, 0) + 1
            
            response={
            'loginRes': "N",
            'access': "",
            'refresh': "",
            'resultCode': 404,
            'resultDesc': "Not Found",
            'resultMsg': "일치하는 회원 정보가 없습니다."
            }
            return Response(response=json.dumps(response), status=404, mimetype="application/json")
    else:
        return 'login.html 파일 요청'

• 로그인 성공

로그인이 필요한 API 요청한 경우

jwt_required()를 사용하여 api 함수를 실행하기 전 토큰이 유효성을 검증하고 유효한 경우에만 해당 함수를 실행한다.

@bp.route('/my_page', methods=['GET'])
@jwt_required()  # access 토큰의 유효성을 검증 후 토큰이 유효하면 코드 수행
def my_page():
    current_user = get_jwt_identity()
    return jsonify(msg=f'Welcome, {current_user}')

토큰이 만료된 경우

@bp.route('/refresh', methods=['POST'])
@jwt_required(refresh=True)
def refresh():
    refresh_token = request.headers['Authorization'].replace('Bearer ', '')

    # DB에 있는 리프레시 토큰 상태 가져오기
    is_revoked = Token.query.filter_by(token=refresh_token).first().is_revoked

    # 리프레시 토큰이 DB에서 취소된 경우(is_revoked==True)
    if is_revoked:
        response={
            'resultCode': 401,
            'resultDesc': "Unauthorized",
            'resultMsg': "로그인이 필요합니다."
            }
        return Response(response=json.dumps(response), status=401, mimetype="application/json")
        # return url_for()
    else: 
        current_user = get_jwt_identity()
        access = create_access_token(identity=current_user)
        response={
            'resultCode': 200,
            'resultDesc': "Success",
            'access': access
            }
        return Response(response=json.dumps(response), status=200, mimetype="application/json")

• Refresh Token 만료

• Access Token 재발급 성공

• Refresh Token 권한 취소한 경우

로그아웃

토큰을 저장하기 위한 jwt_blocklist를 생성하여 사용한다.

jwt_blocklist = set()

@bp.route('/logout', methods=['POST'])
@jwt_required()  
def logout():
    jti = get_jwt()['jti']  # jti : 토큰을 고유ID로 저장
    jwt_blocklist.add(jti)  # jwt_blocklist : 토큰의 고유ID, 토큰 유지 기간, 토큰 유지 기간 설정 여부
    return jsonify(msg='Successfully logged out.')  # jwt_bloacklist에 jti만 넣어주고 나머지 생략하면 토큰 즉시 파괴

• 로그아웃 성공