shuim
81c09ee0d9
POST /auth/admin/tenant/assign-env-credentials {username}: 동일 앱키 1사용자 제약을 지키며
현 보유자(부트스트랩 admin)에서 대상 사용자로 이동, 대상의 기존 자격증명 교체, 컨텍스트 캐시 무효화.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
477 lines
20 KiB
Python
477 lines
20 KiB
Python
"""대시보드 계정 인증 엔드포인트 (DB users 테이블 기반 회원가입/로그인).
|
|
|
|
POST /api/v1/auth/register {username, password} → 계정 생성 (201)
|
|
POST /api/v1/auth/login {username, password} → {token, expires_at, username, is_admin}
|
|
GET /api/v1/auth/me (X-API-Key: 세션 토큰 또는 API 키) → 현재 인증 주체
|
|
GET /api/v1/auth/users (관리자) 계정 목록
|
|
POST /api/v1/auth/users/{id}/approve (관리자) 가입 승인
|
|
POST /api/v1/auth/users/{id}/revoke (관리자) 비활성화
|
|
|
|
정책
|
|
- 첫 가입자는 자동으로 관리자 + 활성. 이후 가입자는 SIGNUP_AUTO_APPROVE=False 면
|
|
비활성 상태로 생성되어 관리자 승인 전까지 로그인 불가(403).
|
|
- 비밀번호는 bcrypt 해시로만 저장. 발급 토큰은 기존 X-API-Key 헤더로 전송되며
|
|
security.require_api_key 가 세션 토큰도 수용한다. API 키(SECRET_KEY)는 관리자로 간주.
|
|
- 무차별 대입 완화: IP당 연속 실패 LOGIN_MAX_FAILURES 회 초과 시 LOGIN_LOCKOUT_SECONDS 잠금.
|
|
|
|
DB 접근
|
|
- 회원가입/로그인은 "중복 확인 → INSERT/UPDATE → 응답" 이 한 요청 안에서 원자적으로
|
|
끝나야 하므로 sync_portfolio 와 같은 예외로 엔드포인트에서 직접 세션을 사용한다
|
|
(db_worker 큐는 결과를 응답에 되돌릴 수 없음). 실시간 계층과는 무관한 경로.
|
|
"""
|
|
import asyncio
|
|
import hmac
|
|
import logging
|
|
import re
|
|
import time
|
|
from datetime import datetime
|
|
|
|
import bcrypt
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from sqlalchemy import func, select, text
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import get_db
|
|
from app.core.security import (
|
|
_get_api_key,
|
|
create_session_token,
|
|
require_api_key,
|
|
verify_session_token,
|
|
)
|
|
from app.models.models import User
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
_USERNAME_RE = re.compile(r"^[a-zA-Z0-9_]{3,32}$")
|
|
|
|
# IP별 로그인 실패 기록: {ip: (연속 실패 횟수, 마지막 실패 시각 monotonic)}
|
|
_failures: dict[str, tuple[int, float]] = {}
|
|
|
|
|
|
# ── 스키마 ────────────────────────────────────────────────────────────────
|
|
class Credentials(BaseModel):
|
|
username: str = Field(..., min_length=3, max_length=32)
|
|
password: str = Field(..., min_length=8, max_length=128)
|
|
|
|
@field_validator("username")
|
|
@classmethod
|
|
def _username_format(cls, v: str) -> str:
|
|
v = v.strip()
|
|
if not _USERNAME_RE.match(v):
|
|
raise ValueError("아이디는 영문/숫자/밑줄 3~32자여야 합니다.")
|
|
return v
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
"""로그인 요청 — 길이 정책은 가입 시에만 적용(부트스트랩 시드 계정 등 짧은 비밀번호 허용)."""
|
|
username: str = Field(..., min_length=1, max_length=32)
|
|
password: str = Field(..., min_length=1, max_length=128)
|
|
|
|
|
|
class LoginResponse(BaseModel):
|
|
token: str
|
|
expires_at: datetime
|
|
username: str
|
|
is_admin: bool
|
|
|
|
|
|
class RegisterResponse(BaseModel):
|
|
username: str
|
|
is_active: bool
|
|
is_admin: bool
|
|
message: str
|
|
|
|
|
|
class UserOut(BaseModel):
|
|
id: int
|
|
username: str
|
|
is_active: bool
|
|
is_admin: bool
|
|
created_at: datetime
|
|
last_login_at: datetime | None
|
|
|
|
|
|
class TenantMigrateOut(BaseModel):
|
|
legacy_owner: str | None
|
|
credentials: str # created | exists | skipped_no_env | skipped_no_encryption
|
|
backfilled: dict[str, int]
|
|
|
|
|
|
class TenantStatusOut(BaseModel):
|
|
legacy_owner: str | None
|
|
multi_tenant_enabled: bool
|
|
credentials_registered: int
|
|
null_user_id_counts: dict[str, int]
|
|
|
|
|
|
# ── 헬퍼 ─────────────────────────────────────────────────────────────────
|
|
def _hash_password(password: str) -> str:
|
|
# passlib 1.7.4 는 bcrypt>=4.1 과 호환되지 않으므로 bcrypt 를 직접 사용 (72바이트 제한)
|
|
return bcrypt.hashpw(password.encode()[:72], bcrypt.gensalt()).decode()
|
|
|
|
|
|
_DUMMY_HASH = _hash_password("not-a-real-password") # 미존재 계정 타이밍 균등화용
|
|
|
|
|
|
def _check_password(password: str, password_hash: str) -> bool:
|
|
try:
|
|
return bcrypt.checkpw(password.encode()[:72], password_hash.encode())
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
# bcrypt 는 의도적으로 느린(≈100~300ms) CPU 연산 → 이벤트 루프를 막지 않도록 스레드로 위임
|
|
async def _hash_password_async(password: str) -> str:
|
|
return await asyncio.to_thread(_hash_password, password)
|
|
|
|
|
|
async def _check_password_async(password: str, password_hash: str) -> bool:
|
|
return await asyncio.to_thread(_check_password, password, password_hash)
|
|
|
|
|
|
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)
|
|
|
|
|
|
async def _resolve_user(credential: str, db: AsyncSession) -> User | None:
|
|
"""세션 토큰 → User. API 키면 None (호출 측에서 관리자 취급)."""
|
|
username = verify_session_token(credential)
|
|
if not username:
|
|
return None
|
|
return (await db.execute(select(User).where(User.username == username))).scalar_one_or_none()
|
|
|
|
|
|
async def require_admin(
|
|
credential: str = Depends(require_api_key), db: AsyncSession = Depends(get_db)
|
|
) -> str:
|
|
"""관리자 전용 의존성. API 키(SECRET_KEY) 또는 is_admin 세션 토큰만 통과."""
|
|
if hmac.compare_digest(credential.encode(), _get_api_key().encode()):
|
|
return "api-key"
|
|
user = await _resolve_user(credential, db)
|
|
if user is None or not user.is_active or not user.is_admin:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="관리자 권한이 필요합니다.")
|
|
return user.username
|
|
|
|
|
|
# ── 엔드포인트 ────────────────────────────────────────────────────────────
|
|
@router.post("/register", response_model=RegisterResponse, status_code=status.HTTP_201_CREATED)
|
|
async def register(body: Credentials, request: Request, db: AsyncSession = Depends(get_db)):
|
|
if not settings.SIGNUP_ENABLED:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="회원가입이 비활성화되어 있습니다.")
|
|
|
|
total = (await db.execute(select(func.count()).select_from(User))).scalar() or 0
|
|
is_first = total == 0
|
|
is_active = is_first or settings.SIGNUP_AUTO_APPROVE
|
|
|
|
user = User(
|
|
username=body.username,
|
|
password_hash=await _hash_password_async(body.password),
|
|
is_active=is_active,
|
|
is_admin=is_first,
|
|
)
|
|
db.add(user)
|
|
try:
|
|
await db.commit()
|
|
except IntegrityError:
|
|
await db.rollback()
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="이미 사용 중인 아이디입니다.")
|
|
|
|
logger.info(
|
|
f"[Auth] 회원가입 user={body.username} admin={is_first} active={is_active} ip={_client_ip(request)}"
|
|
)
|
|
if is_first:
|
|
message = "첫 계정이 관리자로 생성되었습니다. 바로 로그인할 수 있습니다."
|
|
elif is_active:
|
|
message = "가입이 완료되었습니다. 바로 로그인할 수 있습니다."
|
|
else:
|
|
message = "가입 신청이 접수되었습니다. 관리자 승인 후 로그인할 수 있습니다."
|
|
return RegisterResponse(username=user.username, is_active=is_active, is_admin=is_first, message=message)
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
async def login(body: LoginRequest, request: Request, db: AsyncSession = Depends(get_db)):
|
|
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 = (await db.execute(select(User).where(User.username == body.username.strip()))).scalar_one_or_none()
|
|
# 존재하지 않는 계정도 해시 검증을 수행해 응답 시간 차이로 계정 유무가 드러나지 않게 함
|
|
pw_ok = await _check_password_async(body.password, user.password_hash if user else _DUMMY_HASH)
|
|
|
|
if user is None or not pw_ok:
|
|
_record_failure(ip)
|
|
logger.warning(f"[Auth] 로그인 실패 ip={ip} user={body.username!r}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="아이디 또는 비밀번호가 올바르지 않습니다.",
|
|
)
|
|
if not user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="관리자 승인 대기 중인 계정입니다.",
|
|
)
|
|
|
|
_failures.pop(ip, None)
|
|
user.last_login_at = datetime.utcnow()
|
|
await db.commit()
|
|
|
|
token, expires_at = create_session_token(user.username)
|
|
logger.info(f"[Auth] 로그인 성공 user={user.username} ip={ip}")
|
|
return LoginResponse(token=token, expires_at=expires_at, username=user.username, is_admin=user.is_admin)
|
|
|
|
|
|
@router.get("/me")
|
|
async def me(credential: str = Depends(require_api_key), db: AsyncSession = Depends(get_db)):
|
|
"""현재 자격 증명 확인 — 세션 토큰이면 계정 정보, API 키면 'api-key'(관리자)."""
|
|
if hmac.compare_digest(credential.encode(), _get_api_key().encode()):
|
|
return {"username": "api-key", "auth_type": "api_key", "is_admin": True}
|
|
user = await _resolve_user(credential, db)
|
|
if user is None or not user.is_active:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="세션이 유효하지 않습니다.")
|
|
return {"username": user.username, "auth_type": "session", "is_admin": user.is_admin}
|
|
|
|
|
|
@router.get("/users", response_model=list[UserOut], dependencies=[Depends(require_admin)])
|
|
async def list_users(db: AsyncSession = Depends(get_db)):
|
|
rows = (await db.execute(select(User).order_by(User.id))).scalars().all()
|
|
return [UserOut.model_validate(u, from_attributes=True) for u in rows]
|
|
|
|
|
|
@router.post("/users/{user_id}/approve", response_model=UserOut)
|
|
async def approve_user(user_id: int, admin: str = Depends(require_admin), db: AsyncSession = Depends(get_db)):
|
|
user = await db.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="계정을 찾을 수 없습니다.")
|
|
user.is_active = True
|
|
await db.commit()
|
|
logger.info(f"[Auth] 계정 승인 user={user.username} by={admin}")
|
|
return UserOut.model_validate(user, from_attributes=True)
|
|
|
|
|
|
@router.post("/users/{user_id}/revoke", response_model=UserOut)
|
|
async def revoke_user(user_id: int, admin: str = Depends(require_admin), db: AsyncSession = Depends(get_db)):
|
|
user = await db.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="계정을 찾을 수 없습니다.")
|
|
if user.is_admin and user.username == admin:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="자기 자신은 비활성화할 수 없습니다.")
|
|
user.is_active = False
|
|
await db.commit()
|
|
logger.info(f"[Auth] 계정 비활성화 user={user.username} by={admin}")
|
|
return UserOut.model_validate(user, from_attributes=True)
|
|
|
|
|
|
class AssignEnvCredentialsIn(BaseModel):
|
|
username: str = Field(..., min_length=1, max_length=32)
|
|
|
|
|
|
@router.post(
|
|
"/admin/tenant/assign-env-credentials",
|
|
dependencies=[Depends(require_admin)],
|
|
)
|
|
async def assign_env_credentials(body: AssignEnvCredentialsIn, db: AsyncSession = Depends(get_db)):
|
|
""".env 의 KIS 자격증명(운영자가 테스트하던 계좌)을 지정 사용자에게 (재)할당.
|
|
|
|
- 동일 앱키는 한 사용자에게만 등록 가능하므로, 현재 보유자(보통 부트스트랩 admin)가
|
|
있으면 그 행을 대상 사용자로 **이동**한다. 대상 사용자가 다른 자격증명을 갖고 있으면 교체.
|
|
- 이동/교체된 사용자들의 UserContext 캐시는 무효화한다.
|
|
- 단일 트랜잭션 직접 접근 허용 경로(CLAUDE.md §3 예외: credentials).
|
|
"""
|
|
import hashlib
|
|
|
|
from app.core.crypto import current_key_version, encrypt, is_encryption_configured
|
|
from app.models.models import UserKisCredential
|
|
from app.services.tenant.registry import user_context_registry
|
|
|
|
target = (await db.execute(select(User).where(User.username == body.username.strip()))).scalar_one_or_none()
|
|
if target is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="사용자를 찾을 수 없습니다.")
|
|
|
|
app_key = settings.KOREA_INVESTMENT_APP_KEY
|
|
app_secret = settings.KOREA_INVESTMENT_APP_SECRET
|
|
account_no = (settings.KOREA_INVESTMENT_ACCOUNT_NUMBER or "").replace("-", "")
|
|
if not (app_key and app_secret and account_no):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=".env 에 KIS 자격증명이 설정되어 있지 않습니다.")
|
|
if not is_encryption_configured():
|
|
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="자격증명 암호화 키가 설정되지 않았습니다.")
|
|
|
|
app_key_hash = hashlib.sha256(app_key.encode()).hexdigest()
|
|
holder = (await db.execute(
|
|
select(UserKisCredential).where(UserKisCredential.app_key_hash == app_key_hash)
|
|
)).scalar_one_or_none()
|
|
previous_owner_id = holder.user_id if holder else None
|
|
|
|
if holder is not None and holder.user_id == target.id:
|
|
return {"username": target.username, "action": "already_assigned", "app_key_hint": app_key[:6]}
|
|
|
|
# 대상 사용자의 기존(다른 키) 자격증명은 제거 — 사용자당 1세트
|
|
existing_target = (await db.execute(
|
|
select(UserKisCredential).where(UserKisCredential.user_id == target.id)
|
|
)).scalar_one_or_none()
|
|
if existing_target is not None and existing_target is not holder:
|
|
await db.delete(existing_target)
|
|
await db.flush()
|
|
|
|
if holder is not None:
|
|
holder.user_id = target.id
|
|
action = "moved"
|
|
else:
|
|
db.add(UserKisCredential(
|
|
user_id=target.id,
|
|
mode=settings.KIS_MODE,
|
|
app_key_enc=encrypt(app_key),
|
|
app_secret_enc=encrypt(app_secret),
|
|
account_no_enc=encrypt(account_no),
|
|
account_product_code=account_no[8:10] if len(account_no) >= 10 else "01",
|
|
app_key_hash=app_key_hash,
|
|
app_key_hint=app_key[:6],
|
|
is_enabled=True,
|
|
auto_trade_enabled=False,
|
|
key_version=current_key_version(),
|
|
))
|
|
action = "created"
|
|
await db.commit()
|
|
|
|
for uid in {previous_owner_id, target.id} - {None}:
|
|
await user_context_registry.invalidate(uid)
|
|
|
|
logger.info(
|
|
f"[Auth] .env KIS 자격증명 {action}: → {target.username} (from user_id={previous_owner_id}, app_key={app_key[:6]}…)"
|
|
)
|
|
return {
|
|
"username": target.username,
|
|
"action": action,
|
|
"previous_owner_id": previous_owner_id,
|
|
"mode": settings.KIS_MODE,
|
|
"app_key_hint": app_key[:6],
|
|
}
|
|
|
|
|
|
# ── 멀티테넌트 이관 관리자 API (설계문서 §4, §5.10) ─────────────────────────
|
|
# `ensure_bootstrap_admin()`(app/services/tenant/bootstrap.py, main.py lifespan 에서
|
|
# 앱 기동 시 1회 자동 실행)와 동일한 이관/백필 로직을 관리자가 수동으로 재실행할 수
|
|
# 있게 노출한다. 예: 관리자 계정이 앱 기동 이후 처음 생성된 경우(부트스트랩 시점에는
|
|
# legacy owner 가 없어 자동 이관이 스킵됨) 이 API로 지연 이관을 수행한다.
|
|
@router.post(
|
|
"/admin/tenant/migrate-legacy",
|
|
response_model=TenantMigrateOut,
|
|
dependencies=[Depends(require_admin)],
|
|
)
|
|
async def migrate_legacy_tenant(db: AsyncSession = Depends(get_db)):
|
|
""".env KIS 자격증명을 legacy owner 로 이관 + user_id NULL 행 백필 (idempotent)."""
|
|
from app.core.crypto import is_encryption_configured
|
|
from app.models.models import UserKisCredential
|
|
from app.services.tenant.bootstrap import (
|
|
_BACKFILL_TABLES,
|
|
_backfill_legacy_owner_ids,
|
|
_get_legacy_owner,
|
|
_seed_legacy_kis_credential,
|
|
)
|
|
|
|
owner = await _get_legacy_owner(db)
|
|
if owner is None:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="관리자 계정이 없습니다. 먼저 회원가입하세요.")
|
|
|
|
existing = (
|
|
await db.execute(select(UserKisCredential).where(UserKisCredential.user_id == owner.id))
|
|
).scalar_one_or_none()
|
|
|
|
if existing is not None:
|
|
cred_result = "exists"
|
|
elif not (
|
|
settings.KOREA_INVESTMENT_APP_KEY
|
|
and settings.KOREA_INVESTMENT_APP_SECRET
|
|
and settings.KOREA_INVESTMENT_ACCOUNT_NUMBER
|
|
):
|
|
cred_result = "skipped_no_env"
|
|
elif not is_encryption_configured():
|
|
cred_result = "skipped_no_encryption"
|
|
else:
|
|
await _seed_legacy_kis_credential(db, owner)
|
|
cred_result = "created"
|
|
|
|
# 백필 대상 행 수는 백필 "직전" 상태를 반환(백필 후에는 0이 되므로 결과 확인용으로 직전 값 사용)
|
|
# M8: 문장별 SAVEPOINT — 한 테이블(컬럼 미존재 등)의 SELECT 실패가 PostgreSQL
|
|
# 트랜잭션을 InFailedSqlTransaction으로 만들어 이후 테이블 카운트까지 연쇄로
|
|
# 0 오보되는 것을 방지한다.
|
|
backfilled: dict[str, int] = {}
|
|
for table in _BACKFILL_TABLES:
|
|
try:
|
|
async with db.begin_nested():
|
|
backfilled[table] = (
|
|
await db.execute(text(f"SELECT COUNT(*) FROM {table} WHERE user_id IS NULL"))
|
|
).scalar() or 0
|
|
except Exception:
|
|
backfilled[table] = 0
|
|
|
|
await _backfill_legacy_owner_ids(db, owner.id)
|
|
|
|
logger.info(
|
|
f"[Auth] 레거시 이관 실행 owner={owner.username} credentials={cred_result} backfilled={backfilled}"
|
|
)
|
|
return TenantMigrateOut(legacy_owner=owner.username, credentials=cred_result, backfilled=backfilled)
|
|
|
|
|
|
@router.get(
|
|
"/admin/tenant/status",
|
|
response_model=TenantStatusOut,
|
|
dependencies=[Depends(require_admin)],
|
|
)
|
|
async def tenant_status(db: AsyncSession = Depends(get_db)):
|
|
"""legacy owner, MULTI_TENANT_ENABLED, 자격증명 등록 수, 테이블별 user_id NULL 행 수."""
|
|
from app.models.models import UserKisCredential
|
|
from app.services.tenant.bootstrap import _BACKFILL_TABLES, _get_legacy_owner
|
|
|
|
owner = await _get_legacy_owner(db)
|
|
cred_count = (await db.execute(select(func.count()).select_from(UserKisCredential))).scalar() or 0
|
|
|
|
null_counts: dict[str, int] = {}
|
|
for table in _BACKFILL_TABLES:
|
|
try:
|
|
null_counts[table] = (
|
|
await db.execute(text(f"SELECT COUNT(*) FROM {table} WHERE user_id IS NULL"))
|
|
).scalar() or 0
|
|
except Exception:
|
|
null_counts[table] = 0
|
|
|
|
return TenantStatusOut(
|
|
legacy_owner=owner.username if owner else None,
|
|
multi_tenant_enabled=settings.MULTI_TENANT_ENABLED,
|
|
credentials_registered=cred_count,
|
|
null_user_id_counts=null_counts,
|
|
)
|