feat: 관리자 API — .env KIS 자격증명을 지정 사용자에게 (재)할당

POST /auth/admin/tenant/assign-env-credentials {username}: 동일 앱키 1사용자 제약을 지키며
현 보유자(부트스트랩 admin)에서 대상 사용자로 이동, 대상의 기존 자격증명 교체, 컨텍스트 캐시 무효화.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-21 10:50:32 +09:00
parent d2493473b8
commit 81c09ee0d9

View File

@@ -295,6 +295,92 @@ async def revoke_user(user_id: int, admin: str = Depends(require_admin), db: Asy
return UserOut.model_validate(user, from_attributes=True) 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) ───────────────────────── # ── 멀티테넌트 이관 관리자 API (설계문서 §4, §5.10) ─────────────────────────
# `ensure_bootstrap_admin()`(app/services/tenant/bootstrap.py, main.py lifespan 에서 # `ensure_bootstrap_admin()`(app/services/tenant/bootstrap.py, main.py lifespan 에서
# 앱 기동 시 1회 자동 실행)와 동일한 이관/백필 로직을 관리자가 수동으로 재실행할 수 # 앱 기동 시 1회 자동 실행)와 동일한 이관/백필 로직을 관리자가 수동으로 재실행할 수