설계: docs/multi_tenant_design.md / 검수: docs/multi_tenant_phase1_review.md (High 0, MT=False 배포 가능)
데이터·인프라
- user_kis_credentials(Fernet 암호화, PBKDF2 키 파생, key_version), stock_bans_v2 신설,
portfolio/trade_history/active_trades/trade_journals/trading_sessions 등 user_id nullable 추가
- alembic 0009~0011(idempotent, legacy owner 백필). main.py 기동 시 portfolio.user_id 스키마 검사(H1)
- Jenkinsfile: DB Migration 을 Deploy 앞으로, alembic 이력 없으면 stamp 0008 후 upgrade(N1)
- bootstrap: users 비어 있으면 BOOTSTRAP_ADMIN_* 로 관리자 시드 + .env KIS 자격증명 이관 + 백필(SAVEPOINT)
백엔드
- core/crypto.py, core/tenant_auth.py(get_current_user — 비활성 사용자 401, API 키→legacy owner,
deny_non_owner_session_when_mt), core/dt_utils.py
- services/tenant: context, kis_pool, registry(UserContextRegistry, 락 최소화·복호화 to_thread), risk_factory
- KoreaInvestmentAPI.for_user(사용자별 인스턴스, 토큰캐시 users/{id}/, 시세는 공용 위임)
- trade_executor.safe_place_order_v2(사용자별 락/세마포어, legacy ctx 는 기존 키)
- mdd_manager/loss_tracker _v2 + ban_until tz-aware/naive 경계 정규화(레거시 경로 버그 포함 수정)
- db_worker: payload user_id 옵션 필터, upsert_stock_ban_v2, legacy ActiveTrade 조회 LIMIT 1
- API: /account/credentials(등록·마스킹조회·삭제·연결테스트·자동매매), /auth/admin/tenant/*,
/portfolio/v2(조회·요약·sync·buy·sell), /trades/v2(+pending). PortfolioResponse Optional 수정
프론트
- /account 내 계좌 설정, useTenantMode(세션=v2 / API키=legacy, 미확정 시 쿼리·버튼 비활성),
계좌 미등록 인라인 안내, v2 매수/매도/거래이력 분기
테스트: tests/integration(격리 Postgres, KIS 전면 mock) 20 passed / 1 xfailed(0012 이전 제약)
2단계(스케줄러 사용자화·legacy 라우터 격리) 전까지 MULTI_TENANT_ENABLED=True 전환 금지
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""사용자별 KIS(한국투자증권) 자격증명 관리 API (멀티테넌트 1단계, 설계문서 §5.3).
|
|
|
|
- 저장/삭제는 "조회 → upsert/삭제 → 응답"이 한 요청 안에서 원자적으로 끝나야 하므로
|
|
auth.py 의 register/login 과 같은 성격으로 엔드포인트에서 직접 세션을 커밋한다
|
|
(CLAUDE.md §3.2 예외 — 실시간 계층이 아님).
|
|
- 평문 app_key/app_secret/account_no 는 어떤 응답·로그에도 노출하지 않는다
|
|
(app_key_hint·마스킹된 계좌번호만 노출).
|
|
- API 키(X-API-Key=SECRET_KEY) 호출자는 `get_current_user` 가 legacy owner 로
|
|
매핑하므로(§5.4), legacy owner 도 없는 최초 상태(`user.id is None`)에서만 400을
|
|
반환하고 그 외에는 정상적으로 legacy owner 본인의 자격증명을 다룬다.
|
|
"""
|
|
import asyncio
|
|
import hashlib
|
|
import logging
|
|
import time
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.crypto import CredentialDecryptError, current_key_version, decrypt, encrypt, is_encryption_configured
|
|
from app.core.database import get_db
|
|
from app.core.tenant_auth import CurrentUser, get_current_user
|
|
from app.models.models import UserKisCredential
|
|
from app.services.korea_investment_api import KoreaInvestmentAPI
|
|
from app.services.tenant.registry import user_context_registry
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
_VERIFY_COOLDOWN_SEC = 60
|
|
# user_id -> 마지막 verify 시도 monotonic 시각 (인메모리, 프로세스 재시작 시 리셋되어도 무방)
|
|
_last_verify_attempt: dict[int, float] = {}
|
|
|
|
|
|
# ── 스키마 ────────────────────────────────────────────────────────────────
|
|
class CredentialsIn(BaseModel):
|
|
mode: str
|
|
app_key: str = Field(..., min_length=1)
|
|
app_secret: str = Field(..., min_length=1)
|
|
account_no: str = Field(..., min_length=1)
|
|
|
|
@field_validator("mode")
|
|
@classmethod
|
|
def _mode_valid(cls, v: str) -> str:
|
|
if v not in ("virtual", "real"):
|
|
raise ValueError("mode는 virtual 또는 real 이어야 합니다.")
|
|
return v
|
|
|
|
@field_validator("account_no")
|
|
@classmethod
|
|
def _account_no_valid(cls, v: str) -> str:
|
|
digits = v.replace("-", "")
|
|
if not digits.isdigit() or len(digits) != 10:
|
|
raise ValueError("계좌번호는 하이픈 제거 후 숫자 10자리(계좌 8자리+상품코드 2자리)여야 합니다.")
|
|
return digits
|
|
|
|
|
|
class CredentialsOut(BaseModel):
|
|
mode: str
|
|
app_key_hint: str
|
|
account_masked: str
|
|
is_enabled: bool
|
|
auto_trade_enabled: bool
|
|
last_verified_at: datetime | None
|
|
updated_at: datetime | None
|
|
decrypt_error: bool = False
|
|
|
|
|
|
class AutoTradeIn(BaseModel):
|
|
enabled: bool
|
|
|
|
|
|
class VerifyOut(BaseModel):
|
|
ok: bool
|
|
balance_summary: dict
|
|
verified_at: datetime
|
|
|
|
|
|
# ── 헬퍼 ─────────────────────────────────────────────────────────────────
|
|
def _require_user_id(user: CurrentUser) -> int:
|
|
if user.id is None:
|
|
raise HTTPException(status_code=400, detail="관리자 계정을 먼저 생성하세요.")
|
|
return user.id
|
|
|
|
|
|
async def _get_own(db: AsyncSession, user_id: int) -> UserKisCredential | None:
|
|
result = await db.execute(select(UserKisCredential).where(UserKisCredential.user_id == user_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
def _mask_account(account_no_plain: str) -> str:
|
|
"""계좌 8자리 기준 마스킹: '******' + 뒤 2자리 (상품코드 제외)."""
|
|
acct = account_no_plain[:8]
|
|
return "******" + acct[-2:]
|
|
|
|
|
|
async def _to_out(cred: UserKisCredential) -> CredentialsOut:
|
|
"""계좌번호 마스킹 표시를 위해 복호화(M4: 스레드로 위임)한다.
|
|
|
|
복호화 실패(L2, 키 로테이션 실패 등)는 500 대신 `decrypt_error=true`로 응답해
|
|
마스킹 계좌 대신 원인 진단이 가능함을 알린다.
|
|
"""
|
|
try:
|
|
account_masked = _mask_account(
|
|
await asyncio.to_thread(decrypt, cred.account_no_enc, cred.key_version)
|
|
)
|
|
decrypt_error = False
|
|
except CredentialDecryptError as e:
|
|
logger.error("[KISCredentials] 계좌번호 복호화 실패 user_id=%s: %s", cred.user_id, e)
|
|
account_masked = "******"
|
|
decrypt_error = True
|
|
|
|
return CredentialsOut(
|
|
mode=cred.mode,
|
|
app_key_hint=(cred.app_key_hint or "") + "…",
|
|
account_masked=account_masked,
|
|
is_enabled=cred.is_enabled,
|
|
auto_trade_enabled=cred.auto_trade_enabled,
|
|
last_verified_at=cred.last_verified_at,
|
|
updated_at=cred.updated_at,
|
|
decrypt_error=decrypt_error,
|
|
)
|
|
|
|
|
|
def _pick_float(d: dict, keys: list[str]) -> float:
|
|
for k in keys:
|
|
v = d.get(k)
|
|
if v not in (None, ""):
|
|
try:
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return 0.0
|
|
|
|
|
|
# ── 엔드포인트 ────────────────────────────────────────────────────────────
|
|
@router.get("/credentials", response_model=CredentialsOut)
|
|
async def get_credentials(
|
|
user: CurrentUser = Depends(get_current_user), db: AsyncSession = Depends(get_db)
|
|
):
|
|
user_id = _require_user_id(user)
|
|
cred = await _get_own(db, user_id)
|
|
if cred is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="등록된 KIS 자격증명이 없습니다.")
|
|
return await _to_out(cred)
|
|
|
|
|
|
@router.put("/credentials", response_model=CredentialsOut)
|
|
async def put_credentials(
|
|
body: CredentialsIn,
|
|
user: CurrentUser = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
user_id = _require_user_id(user)
|
|
if not is_encryption_configured():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="CREDENTIALS_ENCRYPTION_KEY/SECRET_KEY 미설정 또는 기본값 — 자격증명 암호화 저장 불가",
|
|
)
|
|
|
|
app_key_hash = hashlib.sha256(body.app_key.encode()).hexdigest()
|
|
dup = await db.execute(
|
|
select(UserKisCredential).where(
|
|
UserKisCredential.app_key_hash == app_key_hash,
|
|
UserKisCredential.user_id != user_id,
|
|
)
|
|
)
|
|
if dup.scalar_one_or_none() is not None:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="이미 다른 사용자가 등록한 앱키입니다.")
|
|
|
|
now = datetime.utcnow()
|
|
cred = await _get_own(db, user_id)
|
|
is_new = cred is None
|
|
if is_new:
|
|
cred = UserKisCredential(user_id=user_id, auto_trade_enabled=False)
|
|
db.add(cred)
|
|
|
|
cred.mode = body.mode
|
|
cred.app_key_enc = encrypt(body.app_key)
|
|
cred.app_secret_enc = encrypt(body.app_secret)
|
|
cred.account_no_enc = encrypt(body.account_no)
|
|
cred.account_product_code = body.account_no[8:10]
|
|
cred.app_key_hash = app_key_hash
|
|
cred.app_key_hint = body.app_key[:6]
|
|
cred.is_enabled = False
|
|
cred.last_verified_at = None
|
|
cred.key_version = current_key_version()
|
|
cred.updated_at = now
|
|
|
|
try:
|
|
await db.commit()
|
|
except IntegrityError:
|
|
await db.rollback()
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="이미 다른 사용자가 등록한 앱키입니다.")
|
|
await db.refresh(cred)
|
|
|
|
await user_context_registry.invalidate(user_id)
|
|
logger.info(
|
|
"[KISCredentials] 자격증명 %s user_id=%s mode=%s app_key=%s…",
|
|
"생성" if is_new else "교체", user_id, body.mode, body.app_key[:6],
|
|
)
|
|
return await _to_out(cred)
|
|
|
|
|
|
@router.delete("/credentials", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_credentials(
|
|
user: CurrentUser = Depends(get_current_user), db: AsyncSession = Depends(get_db)
|
|
):
|
|
user_id = _require_user_id(user)
|
|
cred = await _get_own(db, user_id)
|
|
if cred is not None:
|
|
await db.delete(cred)
|
|
await db.commit()
|
|
await user_context_registry.invalidate(user_id)
|
|
logger.info("[KISCredentials] 자격증명 삭제 user_id=%s", user_id)
|
|
|
|
|
|
@router.post("/credentials/verify", response_model=VerifyOut)
|
|
async def verify_credentials(
|
|
user: CurrentUser = Depends(get_current_user), db: AsyncSession = Depends(get_db)
|
|
):
|
|
user_id = _require_user_id(user)
|
|
cred = await _get_own(db, user_id)
|
|
if cred is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="등록된 KIS 자격증명이 없습니다.")
|
|
|
|
now_mono = time.monotonic()
|
|
last = _last_verify_attempt.get(user_id)
|
|
if last is not None:
|
|
remaining = _VERIFY_COOLDOWN_SEC - (now_mono - last)
|
|
if remaining > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail=f"{int(remaining) + 1}초 후 다시 시도하세요.",
|
|
)
|
|
_last_verify_attempt[user_id] = now_mono
|
|
|
|
try:
|
|
app_key = await asyncio.to_thread(decrypt, cred.app_key_enc, cred.key_version)
|
|
app_secret = await asyncio.to_thread(decrypt, cred.app_secret_enc, cred.key_version)
|
|
account_no = await asyncio.to_thread(decrypt, cred.account_no_enc, cred.key_version)
|
|
except CredentialDecryptError as e:
|
|
logger.error("[KISCredentials] verify 복호화 실패 user_id=%s: %s", user_id, e)
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="자격증명 복호화 실패 — 서버 키 변경 여부 확인")
|
|
|
|
temp_kis = KoreaInvestmentAPI.for_user(
|
|
user_id=user_id, app_key=app_key, app_secret=app_secret, account_no=account_no, mode=cred.mode,
|
|
)
|
|
try:
|
|
balance = await temp_kis.get_balance()
|
|
except Exception as e:
|
|
# L5: KIS 예외 원문 대신 예외 타입명만 노출(방어적 — 현재 KIS 오류 본문에 키는 없음).
|
|
logger.warning("[KISCredentials] verify 실패 user_id=%s: %s", user_id, e)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"KIS 연결 확인 실패: {type(e).__name__} — 연결 실패",
|
|
)
|
|
finally:
|
|
await temp_kis.close()
|
|
|
|
if balance.get("rt_cd") != "0":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=balance.get("msg1") or "KIS 잔고 조회 실패",
|
|
)
|
|
|
|
output2 = balance.get("output2") or {}
|
|
if isinstance(output2, list):
|
|
output2 = output2[0] if output2 else {}
|
|
|
|
balance_summary = {
|
|
"available_cash": _pick_float(output2, ["dnca_tot_amt", "d2_deposit", "nass_amt"]),
|
|
"total_assets": _pick_float(output2, ["tot_evlu_amt", "tot_evlu_amt_smtl"]),
|
|
"stock_value": _pick_float(output2, ["scts_evlu_amt", "evlu_amt_smtl_amt"]),
|
|
}
|
|
|
|
now = datetime.utcnow()
|
|
cred.last_verified_at = now
|
|
cred.is_enabled = True
|
|
cred.updated_at = now
|
|
await db.commit()
|
|
|
|
logger.info("[KISCredentials] verify 성공 user_id=%s mode=%s", user_id, cred.mode)
|
|
return VerifyOut(ok=True, balance_summary=balance_summary, verified_at=now)
|
|
|
|
|
|
@router.patch("/auto-trade", response_model=CredentialsOut)
|
|
async def patch_auto_trade(
|
|
body: AutoTradeIn,
|
|
user: CurrentUser = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
user_id = _require_user_id(user)
|
|
cred = await _get_own(db, user_id)
|
|
if cred is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="등록된 KIS 자격증명이 없습니다.")
|
|
|
|
if body.enabled and not cred.is_enabled:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="먼저 연결 테스트(verify)를 통과해야 자동매매를 활성화할 수 있습니다.",
|
|
)
|
|
|
|
cred.auto_trade_enabled = body.enabled
|
|
cred.updated_at = datetime.utcnow()
|
|
await db.commit()
|
|
# H3: 캐시된 UserContext.auto_trade_enabled 스냅샷이 갱신되도록 무효화(다음 호출에서 재조립).
|
|
await user_context_registry.invalidate(user_id)
|
|
logger.info("[KISCredentials] auto_trade_enabled=%s user_id=%s", body.enabled, user_id)
|
|
return await _to_out(cred)
|