Mina Choi
cfee6e89d0
- UserRole 1=일반 2=최고관리자 3=개발자(내부 운영). 개발자 계정은 회원 목록·총계에서 제외해 고객사에 노출하지 않음 - 계정 생성 시 권한 선택(일반/최고관리자) 추가, 개발자는 앱에서 부여 불가(DB 시드 전용) - 사이드바 '개발자' 그룹 신설 — 회사 설정·디자인 시스템을 개발자에게만 노출 - /dev/design 디자인 시스템 페이지: 색 토큰·타이포·버튼·배지·입력·반경, 목록 화면 구조, 반응형 기준, URL 상태 규칙, 회사 커스터마이징 훅 - 공급사 재협상 요청(#15) 플로우 문서 — sessions.custom 기반(DDL 0)
126 lines
4.8 KiB
Python
126 lines
4.8 KiB
Python
import asyncio
|
|
import json
|
|
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 __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
|
|
|
|
|
|
# ---- 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")))
|