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

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

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

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import places
from common.enums import (
    DBWRType,
    ErrorType,
    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)
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)까지만 사는 접근 토큰 — 미니 블로그 메일의 "수정하기" 링크 전용."""
    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 용 토큰 검증기
async def _place_owner(place_id: str) -> str | None:
    try:
        pid = uuid.UUID(place_id)
    except ValueError:
        return None
    err, rows = await DB_SESSION_MNG.execute_lambda(
        places.DBType(),
        DBWRType.DB_READ.value,
        lambda s: DB_SESSION_MNG.execute(
            s, select(places.owner_user_id).where(places.place_id == pid, places.deleted == False)  # noqa: E712
        ),
    )
    if err != ErrorType.SUCCESS or not rows:
        return None
    return str(rows[0])


async def _act_as_place_owner(request: Request, user_info: UserInfo) -> UserInfo:
    if (user_info.role or 0) < UserRole.DEVELOPER.value:
        return user_info
    place_id = request.path_params.get("place_id") or request.query_params.get("place_id")
    if not place_id:
        return user_info
    owner = await _place_owner(str(place_id))
    if owner is None or owner == str(user_info.user_id):
        return user_info
    LOG.i(f"[admin-access] {user_info.id} → place={place_id} owner={owner} {request.method} {request.url.path}")
    return UserInfo(user_id=owner, id=user_info.id, role=user_info.role, token_version=user_info.token_version)


async def IsValidAccessToken(
    request: Request, credentials: HTTPAuthorizationCredentials = Depends(security)
) -> UserInfo:
    return await _act_as_place_owner(request, 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


# 개발자(내부 운영) 전용 게이트.
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 처리
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 안전한 문자열로 변환한다.
    return JSONResponse(content=RemoveNoneValues(obj.model_dump(mode="json")))