60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
|
"""인증 유틸 - bcrypt 해시 + JWT 토큰 (이메일/비밀번호)"""
|
||
|
|
import os
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
import bcrypt
|
||
|
|
from jose import jwt, JWTError
|
||
|
|
from fastapi import Header, HTTPException, status
|
||
|
|
|
||
|
|
JWT_SECRET = os.getenv("JWT_SECRET", "")
|
||
|
|
JWT_ALGORITHM = "HS256"
|
||
|
|
JWT_EXPIRE_DAYS = int(os.getenv("JWT_EXPIRE_DAYS", "7"))
|
||
|
|
|
||
|
|
if not JWT_SECRET:
|
||
|
|
raise RuntimeError("JWT_SECRET 환경변수가 설정되지 않았습니다")
|
||
|
|
|
||
|
|
def hash_password(plain: str) -> str:
|
||
|
|
# bcrypt는 72바이트 제한 — 안전을 위해 잘라서 해시
|
||
|
|
pw_bytes = plain.encode("utf-8")[:72]
|
||
|
|
return bcrypt.hashpw(pw_bytes, bcrypt.gensalt()).decode("utf-8")
|
||
|
|
|
||
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
||
|
|
try:
|
||
|
|
pw_bytes = plain.encode("utf-8")[:72]
|
||
|
|
return bcrypt.checkpw(pw_bytes, hashed.encode("utf-8"))
|
||
|
|
except Exception:
|
||
|
|
return False
|
||
|
|
|
||
|
|
def create_token(user_id: int, email: str) -> str:
|
||
|
|
now = datetime.now(timezone.utc)
|
||
|
|
payload = {
|
||
|
|
"sub": str(user_id),
|
||
|
|
"email": email,
|
||
|
|
"iat": int(now.timestamp()),
|
||
|
|
"exp": int((now + timedelta(days=JWT_EXPIRE_DAYS)).timestamp()),
|
||
|
|
}
|
||
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
||
|
|
|
||
|
|
def decode_token(token: str) -> dict:
|
||
|
|
try:
|
||
|
|
return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
||
|
|
except JWTError as e:
|
||
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=f"invalid token: {e}")
|
||
|
|
|
||
|
|
async def current_user(authorization: str = Header(default="")) -> dict:
|
||
|
|
"""Authorization: Bearer <token> → {"id": int, "email": str}"""
|
||
|
|
if not authorization.lower().startswith("bearer "):
|
||
|
|
raise HTTPException(status_code=401, detail="missing bearer token")
|
||
|
|
token = authorization.split(" ", 1)[1].strip()
|
||
|
|
payload = decode_token(token)
|
||
|
|
try:
|
||
|
|
return {"id": int(payload["sub"]), "email": payload.get("email", "")}
|
||
|
|
except (KeyError, ValueError):
|
||
|
|
raise HTTPException(status_code=401, detail="malformed token payload")
|
||
|
|
|
||
|
|
# 타이밍 공격 방지용 더미 해시 (이메일이 없을 때도 verify를 호출해 응답시간 균일화)
|
||
|
|
_DUMMY_HASH = bcrypt.hashpw(b"dummy_password_for_timing", bcrypt.gensalt()).decode("utf-8")
|
||
|
|
|
||
|
|
def dummy_verify():
|
||
|
|
"""존재하지 않는 사용자에 대해서도 bcrypt 비용을 동일하게 지불"""
|
||
|
|
bcrypt.checkpw(b"dummy_password_for_timing", _DUMMY_HASH.encode("utf-8"))
|