chpark
d16456cb92
# 사용자별 격리 - JWT 토큰에 uid 추가 (auth.get_uid 헬퍼) - PostgreSQL — exchange_credentials/automation_config/trades/signal_events 에 user_id BIGINT - SQLite user_settings 테이블 신설 (글로벌 settings 는 옛 호환) - 모든 DB 함수 시그니처에 user_id 인자 추가 — 다른 사용자 데이터 절대 접근 불가 - alert_state — 모든 dict key 가 (user_id, ...) tuple 로 계층화 - core_logic alert_loop — 활성 사용자 순회 + 각자 settings/symbol/텔레그램 적용 - ensure_user_defaults() / ensure_user_automation() — 첫 사용 시 자동 시드 # 사용자 관리 (admin only) - users_db: delete_user / admin_reset_password / set_role - /api/users POST DELETE PUT password PUT role (본인 강등 / 마지막 admin 보호) - /admin/users 페이지 — 등록/삭제/role 토글/비번 reset 모달 - 사이드바 adminOnly 필터 — admin role 만 메뉴 노출 # 대시보드 개선 - 모바일 / 범례 토글 (모바일 60 캔들, 데스크톱 200) - 트레이드 이력: open 트레이드 실시간 PnL% (Binance ticker 호출 + 방향별 계산) - 메트릭 카드 분리 (실거래 vs 실시간 open) # 안정성 - api.ts: error.detail array/object 안전 처리 ([object Object] 방지) - Chart.tsx: Plotly yaxis title 객체 형태 + 모바일 height 동적 조정 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
"""JWT 인증."""
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
from fastapi import HTTPException, status, Depends
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from jose import jwt, JWTError
|
|
|
|
import users_db
|
|
|
|
JWT_SECRET = os.environ.get("JWT_SECRET", "change-me-in-production-please")
|
|
JWT_ALG = "HS256"
|
|
JWT_EXP_HOURS = 24 * 7
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False)
|
|
|
|
|
|
def create_token(user_id: int, username: str, role: str) -> str:
|
|
payload = {
|
|
"sub": username,
|
|
"uid": int(user_id),
|
|
"role": role,
|
|
"exp": datetime.utcnow() + timedelta(hours=JWT_EXP_HOURS),
|
|
"iat": datetime.utcnow(),
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALG)
|
|
|
|
|
|
def get_uid(payload: dict) -> int:
|
|
"""JWT payload 에서 user_id 추출. 옛 토큰 호환 — uid 없으면 username 으로 lookup."""
|
|
if "uid" in payload:
|
|
return int(payload["uid"])
|
|
# 옛 토큰 fallback
|
|
try:
|
|
import users_db
|
|
username = payload.get("sub")
|
|
for u in users_db.list_users():
|
|
if u.get("username") == username:
|
|
return int(u["id"])
|
|
except Exception:
|
|
pass
|
|
return 0
|
|
|
|
|
|
def decode_token(token: str) -> Optional[dict]:
|
|
try:
|
|
return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG])
|
|
except JWTError:
|
|
return None
|
|
|
|
|
|
def require_user(token: Optional[str] = Depends(oauth2_scheme)) -> dict:
|
|
if not token:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="not authenticated")
|
|
payload = decode_token(token)
|
|
if not payload:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token")
|
|
return payload
|
|
|
|
|
|
def require_admin(payload: dict = Depends(require_user)) -> dict:
|
|
if payload.get("role") != "admin":
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="admin only")
|
|
return payload
|