o2o-site-AEO/solution/backend/router/v1/validator/dependencies.py

import asyncio
import json
from datetime import datetime, timedelta, timezone
from typing import Any, Union

from fastapi import Depends
from fastapi.responses import JSONResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
import bcrypt
from jose import jwt, JWTError, ExpiredSignatureError

from common.enums import (
    EXCEPTION_ACCESS_TOKEN_EXPIRED,
    EXCEPTION_FORBIDDEN,
    EXCEPTION_INVALID_CLIENT_ACCESS,
    EXCEPTION_REFRESH_TOKEN_EXPIRED,
    UserRole,
)
from common.logger import LOG
from common.models.gmodel import UserInfo
from common.utils.gtime import GTime
from config.server_configs import jwt_token_config

security = HTTPBearer()


# ---- 비밀번호 해시 (bcrypt) ------------------------------------------------
# bcrypt 는 CPU 바운드 동기 작업이라 그대로 호출하면 asyncio 이벤트 루프를 막아
# 같은 워커의 다른 요청(healthz 등)까지 멈춘다. 스레드풀(asyncio.to_thread)로 보낸다.
# bcrypt 는 해싱 중 GIL 을 해제하므로 스레드들이 여러 코어에서 실제 병렬로 돈다.
# 입력은 최대 72 bytes 까지만 사용하므로 사전에 잘라준다.
def _hash_pw(pw: str) -> str:
    return bcrypt.hashpw(pw.encode("utf-8")[:72], bcrypt.gensalt()).decode("utf-8")


def _verify_pw(pw: str, hashed_pw: str) -> bool:
    try:
        return bcrypt.checkpw(pw.encode("utf-8")[:72], hashed_pw.encode("utf-8"))
    except (ValueError, TypeError):
        return False


async def GetHashedPW(pw: str) -> str:
    return await asyncio.to_thread(_hash_pw, pw)


async def VerifyPW(pw: str, hashed_pw: str) -> bool:
    return await asyncio.to_thread(_verify_pw, pw, hashed_pw)


# ---- JWT 토큰 발급/검증 ----------------------------------------------------
JWT_ALGORITHM = "HS256"
JWT_ACCESS_SECRET = jwt_token_config.access_key
JWT_REFRESH_SECRET = jwt_token_config.refresh_key
ACCESS_TOKEN_EXPIRE_MIN = jwt_token_config.access_expire_min
REFRESH_TOKEN_EXPIRE_MIN = 60 * 24 * jwt_token_config.refresh_expire_day


def __create_token(subject: Union[str, Any], secret_key: str, expire_min: int) -> str:
    to_encode = {
        "sub": str(subject),
        "exp": GTime.AddMinutes(expire_min),  # jose 가 exp 클레임을 자동 검증
    }
    return jwt.encode(to_encode, secret_key, JWT_ALGORITHM)


def CreateAccessToken(subject: UserInfo) -> str:
    return __create_token(subject.to_json(), JWT_ACCESS_SECRET, ACCESS_TOKEN_EXPIRE_MIN)


def CreateRefreshToken(subject: UserInfo) -> str:
    return __create_token(subject.to_json(), JWT_REFRESH_SECRET, REFRESH_TOKEN_EXPIRE_MIN)


def CreateDayPassToken(subject: UserInfo) -> str:
    """그날 자정(KST)까지만 사는 접근 토큰 — 미니 블로그 메일의 "수정하기" 링크 전용.

    ★ 일반 로그인 세션과 다르다 — 사장님이 메일에서 그 글 하나를 고치러 들어오는 맥락에서만
      쓰이고, 유효기간도 그만큼 짧다(2026-09-17, 사장님 지시: "로그인도 크레덴셜로 자동으로
      되게 (그날까지만)"). refresh 토큰은 안 준다 — 그날이 지나면 다시 메일을 받아야 한다."""
    now_kst = datetime.now(timezone(timedelta(hours=9)))
    midnight_kst = (now_kst + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
    expire_min = max(1, int((midnight_kst - now_kst).total_seconds() // 60))
    return __create_token(subject.to_json(), JWT_ACCESS_SECRET, expire_min)


def __decode_token(jwt_token: str, secret_key: str, expired_exception) -> UserInfo:
    try:
        decoded = jwt.decode(jwt_token, secret_key, algorithms=[JWT_ALGORITHM])
        return UserInfo(**json.loads(decoded.get("sub")))
    except ExpiredSignatureError:
        raise expired_exception
    except JWTError as ex:
        LOG.e_no_callstack(ex)
        raise EXCEPTION_INVALID_CLIENT_ACCESS


def DecodeAccessToken(jwt_token: str) -> UserInfo:
    return __decode_token(jwt_token, JWT_ACCESS_SECRET, EXCEPTION_ACCESS_TOKEN_EXPIRED)


def DecodeRefreshToken(jwt_token: str) -> UserInfo:
    return __decode_token(jwt_token, JWT_REFRESH_SECRET, EXCEPTION_REFRESH_TOKEN_EXPIRED)


# ---- Depends 용 토큰 검증기 ------------------------------------------------
# 보호된 엔드포인트에서 dependencies=[Depends(IsValidAccessToken)] 로 사용.
async def IsValidAccessToken(credentials: HTTPAuthorizationCredentials = Depends(security)) -> UserInfo:
    return DecodeAccessToken(credentials.credentials)


async def IsValidRefreshToken(credentials: HTTPAuthorizationCredentials = Depends(security)) -> UserInfo:
    return DecodeRefreshToken(credentials.credentials)


# 최고관리자 이상(OWNER/DEVELOPER) 게이트. 회원관리·회사설정에 건다.
async def RequireOwner(user_info: UserInfo = Depends(IsValidAccessToken)) -> UserInfo:
    if (user_info.role or 0) < UserRole.OWNER.value:
        raise EXCEPTION_FORBIDDEN
    return user_info


# 개발자(내부 운영) 전용 게이트. 회사 스코프를 넘어 전 고객사 데이터를 보는 /v1/admin 에만 건다.
# OWNER 는 자기 회사 최상위일 뿐 남의 회사를 볼 권한이 아니라서 여기선 막힌다.
async def RequireDeveloper(user_info: UserInfo = Depends(IsValidAccessToken)) -> UserInfo:
    if (user_info.role or 0) < UserRole.DEVELOPER.value:
        raise EXCEPTION_FORBIDDEN
    return user_info


# ---- ResponseNone 처리 -----------------------------------------------------
# 응답 객체에서 값이 None 인 필드를 재귀적으로 제거하여 페이로드를 줄인다.
# 모든 라우터는 return RemoveNoneResponse(await service....) 형태로 반환한다.
def RemoveNoneValues(obj: Any) -> Any:
    if isinstance(obj, dict):
        return {k: RemoveNoneValues(v) for k, v in obj.items() if v is not None}
    if isinstance(obj, list):
        return [RemoveNoneValues(v) for v in obj]
    return obj


def RemoveNoneResponse(obj) -> JSONResponse:
    # mode="json": uuid/datetime 등 DB 네이티브 타입(asyncpg.UUID 포함)을 pydantic 단에서
    # JSON 안전한 문자열로 변환한다. content 가 이미 JSON-safe dict 이므로 표준 JSONResponse 사용
    # (ORJSONResponse 는 최신 FastAPI 에서 deprecated).
    return JSONResponse(content=RemoveNoneValues(obj.model_dump(mode="json")))