Mina Choi
11d30bb3d1
여러 줄 주석이 설명보다 경위(예전·실측·지적)를 적고 있어 읽는 사람이 결론을 찾기 어려웠다. - ts·tsx·js·mjs·css·py 478개: 여러 줄 주석은 첫 문장 한 줄로, 과거형·날짜 문장은 삭제 - 주석 위치는 TypeScript 파서·파이썬 tokenize/ast 로 찾는다 — 문자열 안의 # · /* 는 건드리지 않는다 - eslint·ts·noqa·type: ignore 같은 지시 주석은 그대로 둔다 파이썬 275개 정리 전후 AST 동일, TS 298개 주석 뺀 토큰 동일(빈 JSX 주석 10곳만 차이). site·frontend·admin tsc, site vitest 105 passed Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
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)
|
|
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 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
|
|
|
|
|
|
# 개발자(내부 운영) 전용 게이트.
|
|
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")))
|