shuim
9063a9d01e
- auth.py 신규: POST /api/v1/auth/login → JWT 세션 토큰(HS256, SECRET_KEY 서명, 24h) GET /api/v1/auth/me 자격 확인. IP당 5회 실패 시 300초 잠금(인메모리) - config: ADMIN_USERNAME / ADMIN_PASSWORD / ADMIN_PASSWORD_HASH(bcrypt) / SESSION_TOKEN_EXPIRE_HOURS / LOGIN_MAX_FAILURES / LOGIN_LOCKOUT_SECONDS - security: create_session_token / verify_session_token / is_valid_credential 추가. require_api_key·optional_api_key 는 시그니처 유지, X-API-Key 값으로 API 키 또는 세션 토큰 모두 수용 → 프론트 api.ts·useWebSocket·MCP 서버 호출 경로 무변경 - main.py WebSocket 인증도 is_valid_credential 로 통일 - bcrypt 검증은 passlib 대신 bcrypt 직접 사용 (passlib 1.7.4 ↔ bcrypt 5.0 비호환) - Login.tsx: 아이디/비밀번호 폼, 401/429/503 별 안내 메시지 - .env.example 에 관리자 계정 항목 추가 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
"""대시보드 로그인 엔드포인트 (아이디/비밀번호 → 세션 토큰).
|
|
|
|
POST /api/v1/auth/login {username, password} → {token, expires_at, username}
|
|
GET /api/v1/auth/me (X-API-Key: 세션 토큰 또는 API 키) → 현재 인증 주체
|
|
|
|
- 자격 증명은 .env 의 ADMIN_USERNAME + ADMIN_PASSWORD_HASH(bcrypt) 또는 ADMIN_PASSWORD(평문)
|
|
- 발급 토큰은 기존 X-API-Key 헤더로 전송 → security.require_api_key 가 세션 토큰도 수용
|
|
- 무차별 대입 완화: IP당 연속 실패 LOGIN_MAX_FAILURES 회 초과 시 LOGIN_LOCKOUT_SECONDS 잠금(인메모리)
|
|
"""
|
|
import hmac
|
|
import logging
|
|
import time
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
import bcrypt
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.core.config import settings
|
|
from app.core.security import (
|
|
_get_api_key,
|
|
create_session_token,
|
|
require_api_key,
|
|
verify_session_token,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
# IP별 실패 기록: {ip: (연속 실패 횟수, 마지막 실패 시각)}
|
|
_failures: dict[str, tuple[int, float]] = {}
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str = Field(..., min_length=1, max_length=64)
|
|
password: str = Field(..., min_length=1, max_length=256)
|
|
|
|
|
|
class LoginResponse(BaseModel):
|
|
token: str
|
|
expires_at: datetime
|
|
username: str
|
|
|
|
|
|
def _login_enabled() -> bool:
|
|
return bool(settings.ADMIN_PASSWORD_HASH or settings.ADMIN_PASSWORD)
|
|
|
|
|
|
def _check_password(password: str) -> bool:
|
|
if settings.ADMIN_PASSWORD_HASH:
|
|
# passlib 1.7.4 는 bcrypt>=4.1 과 호환되지 않으므로 bcrypt 를 직접 사용
|
|
try:
|
|
return bcrypt.checkpw(password.encode()[:72], settings.ADMIN_PASSWORD_HASH.encode())
|
|
except ValueError:
|
|
logger.error("[Auth] ADMIN_PASSWORD_HASH 형식 오류 — bcrypt 해시가 아님")
|
|
return False
|
|
return hmac.compare_digest(password.encode(), settings.ADMIN_PASSWORD.encode())
|
|
|
|
|
|
def _client_ip(request: Request) -> str:
|
|
fwd = request.headers.get("x-forwarded-for")
|
|
if fwd:
|
|
return fwd.split(",")[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
def _lockout_remaining(ip: str) -> int:
|
|
rec = _failures.get(ip)
|
|
if not rec:
|
|
return 0
|
|
count, last = rec
|
|
if count < settings.LOGIN_MAX_FAILURES:
|
|
return 0
|
|
remaining = int(settings.LOGIN_LOCKOUT_SECONDS - (time.monotonic() - last))
|
|
if remaining <= 0:
|
|
_failures.pop(ip, None)
|
|
return 0
|
|
return remaining
|
|
|
|
|
|
def _record_failure(ip: str) -> None:
|
|
count, _ = _failures.get(ip, (0, 0.0))
|
|
_failures[ip] = (count + 1, time.monotonic())
|
|
# 오래된 항목 정리 (메모리 누수 방지)
|
|
if len(_failures) > 1000:
|
|
cutoff = time.monotonic() - settings.LOGIN_LOCKOUT_SECONDS
|
|
for k in [k for k, (_, t) in _failures.items() if t < cutoff]:
|
|
_failures.pop(k, None)
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
async def login(body: LoginRequest, request: Request):
|
|
if not _login_enabled():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="로그인이 설정되지 않았습니다. .env 에 ADMIN_PASSWORD 또는 ADMIN_PASSWORD_HASH 를 설정하세요.",
|
|
)
|
|
|
|
ip = _client_ip(request)
|
|
remaining = _lockout_remaining(ip)
|
|
if remaining > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail=f"로그인 시도가 너무 많습니다. {remaining}초 후 다시 시도하세요.",
|
|
)
|
|
|
|
user_ok = hmac.compare_digest(body.username.encode(), settings.ADMIN_USERNAME.encode())
|
|
pw_ok = _check_password(body.password)
|
|
if not (user_ok and pw_ok):
|
|
_record_failure(ip)
|
|
logger.warning(f"[Auth] 로그인 실패 ip={ip} user={body.username!r}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="아이디 또는 비밀번호가 올바르지 않습니다.",
|
|
)
|
|
|
|
_failures.pop(ip, None)
|
|
token, expires_at = create_session_token(settings.ADMIN_USERNAME)
|
|
logger.info(f"[Auth] 로그인 성공 user={settings.ADMIN_USERNAME} ip={ip}")
|
|
return LoginResponse(token=token, expires_at=expires_at, username=settings.ADMIN_USERNAME)
|
|
|
|
|
|
@router.get("/me")
|
|
async def me(credential: str = Depends(require_api_key)):
|
|
"""현재 자격 증명 확인 — 세션 토큰이면 username, API 키면 'api-key'."""
|
|
if hmac.compare_digest(credential.encode(), _get_api_key().encode()):
|
|
return {"username": "api-key", "auth_type": "api_key"}
|
|
return {"username": verify_session_token(credential), "auth_type": "session"}
|