Mina Choi
6784e59ca5
git 저장소가 없어 히스토리·협업 기반이 아예 없던 상태를 연다.
함께 문서를 재편했다. 그동안 문서가 있어도 "이 제품이 뭘 푸는가"와
"어떻게 도는가"를 담은 문서가 없어서, 목표 문장이 backend/frontend
README 두 곳에 복붙돼 있었다 — 상위 문서가 없어 아래로 샌 것이다.
신설
README.md 레포 진입점 + 문서 지도 + 문서 규칙 4가지
AGENTS.md 에이전트·신규 합류자용 함정 목록과 규약
(CLAUDE.md 는 여기로 걸린 심볼릭 링크)
docs/PRODUCT.md 제품 정의 — 문제·사용자·원칙·**non-goals**·성공 기준
docs/ARCHITECTURE.md payload 경계·발행 파이프라인·서빙 결정·앱 분리 설계
이동
backend/docs/DECISIONS.md → docs/DECISIONS.md
백엔드만의 결정이 아니다. 게다가 코드 주석 ~25곳이 이미
`docs/DECISIONS.md` 로 적고 있어 레포 루트 기준으로는 그게 맞다.
갱신
docs/DEPLOY.md 서빙 결정 반영 — nginx 정적 서빙이 지금 경로(3절),
Azure 는 나중에 켤 때(4절)로 분리
docs/ARCHITECTURE.md 사이트 = 한 장(2026-08-31) 구조 반영
docs/COLLECTION_SEO_AEO_FLOW.md
robots.txt·sitemap.xml 은 오리진 루트에만 굽는다는 점 명시
frontend/site/scripts/prerender.ts
헤더 주석의 렌더 보고서 경로가 실제(422줄)와 달라 수정
.gitignore
★ CLAUDE.md 를 더 이상 무시하지 않는다. 에이전트 지침은 팀과 모든
에이전트가 공유하는 규약이라 커밋해야 한다 — 무시하면 클론한 사람이
"배포 후 republish_all.py 필수" 같은 함정을 전달받지 못한다.
개인용 오버라이드는 ~/.claude/CLAUDE.md 에 둔다.
134 lines
5.3 KiB
Python
134 lines
5.3 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
|
|
|
|
|
|
# 개발자(내부 운영) 전용 게이트. 회사 스코프를 넘어 전 고객사 데이터를 보는 /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")))
|