"""대시보드 로그인 엔드포인트 (아이디/비밀번호 → 세션 토큰). 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"}