82 lines
3.4 KiB
Python
82 lines
3.4 KiB
Python
"""
|
|
데코레이터 유틸리티 모듈
|
|
"""
|
|
|
|
from functools import wraps
|
|
from typing import Callable
|
|
from flask import request, g
|
|
from flask_jwt_extended import get_jwt_identity, get_jwt
|
|
from utils.response import *
|
|
import settings
|
|
from extensions import custom_logger
|
|
|
|
logger = custom_logger(f"{settings.LOG_PREFIX}_decorators")
|
|
|
|
# Lazy 로딩
|
|
def _get_user_service():
|
|
from services.user.user_service import UserService
|
|
return UserService()
|
|
|
|
def _get_account_lock_service():
|
|
from services.auth.cache_service import AccountLockService
|
|
return AccountLockService
|
|
|
|
def _get_jwt_blacklist_service():
|
|
from services.auth.cache_service import JWTBlacklistService
|
|
return JWTBlacklistService
|
|
|
|
def load_current_user(f: Callable) -> Callable:
|
|
"""
|
|
JWT identity(user_uuid)를 기반으로 User 객체를 로드함.
|
|
g.current_user에 저장
|
|
Redis 캐시를 활용
|
|
JWT 블랙리스트 체크 (로그아웃된 토큰 방지)
|
|
"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
try:
|
|
# JWT 페이로드 및 user_uuid 추출
|
|
jwt_payload = get_jwt()
|
|
user_uuid = get_jwt_identity()
|
|
jti = jwt_payload.get("jti")
|
|
|
|
if not user_uuid:
|
|
logger.warning("JWT에서 user_uuid를 찾을 수 없습니다.")
|
|
return unauthorized_response(message="토큰에서 사용자 정보를 찾을 수 없습니다.")
|
|
|
|
# JWT 블랙리스트 체크 (사용이 불가능한 토큰)
|
|
JWTBlacklistService = _get_jwt_blacklist_service()
|
|
if jti and JWTBlacklistService.is_blacklisted(jti):
|
|
logger.warning(f"블랙리스트된 토큰 접근 시도: jti={jti}, user_uuid={user_uuid}")
|
|
return unauthorized_response(message="사용이 불가능한 토큰입니다. 다시 로그인해주세요.")
|
|
|
|
# UserService를 통한 조회
|
|
UserService = _get_user_service()
|
|
user = UserService.get_user_by_uuid(user_uuid, use_cache=True)
|
|
|
|
if not user:
|
|
logger.warning(f"user_uuid에 해당하는 사용자를 찾을 수 없습니다: {user_uuid}")
|
|
return unauthorized_response(message="사용자를 찾을 수 없습니다.")
|
|
|
|
# 계정 잠금 체크 (Redis 기반)
|
|
AccountLockService = _get_account_lock_service()
|
|
|
|
if AccountLockService.is_locked(user_uuid):
|
|
logger.warning(f"잠긴 계정 접근 시도: {user_uuid}")
|
|
return unauthorized_response(message="계정이 잠겼습니다. 관리자에게 문의하세요.")
|
|
|
|
# DB의 로그인 실패 횟수가 최대 시도 횟수를 초과하면 Redis에 잠금 설정
|
|
if (user.count or 0) >= settings.MAX_LOGIN_ATTEMPTS:
|
|
AccountLockService.lock(user_uuid)
|
|
logger.warning(f"로그인 실패 횟수 초과로 계정 잠금: {user_uuid}")
|
|
return unauthorized_response(message="계정이 잠겼습니다. 관리자에게 문의하세요.")
|
|
|
|
# g에 현재 사용자 저장
|
|
g.current_user = user
|
|
|
|
return f(*args, **kwargs)
|
|
except Exception as e:
|
|
logger.exception(f"현재 사용자 로드에 실패하였습니다. error={str(e)}")
|
|
return error_response(message="현재 사용자 정보를 불러오는 데 실패하였습니다.")
|
|
|
|
return decorated_function |