kyu
6d3b0bacc0
- 19개 마이크로서비스 (news-collector, score-engine, ta-engine, dart-collector, aux-signal, us-market, graph-engine, telegram-bot, dashboard-api, kis-api 등) - 가치투자 스코어링 + 10공식 앙상블 보팅 (매직포뮬러/F-Score/Altman/PEG/ 모멘텀/Beneish/GP-A/G-Score/Amihud/BAB) - 뉴스 수집→형태소→임베딩→중복제거→AI분석 파이프라인 - 기술적분석 + GAT 그래프신경망 + 미증시 동조 시그널 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
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"))
|