From 81c09ee0d9b9648ad5c73e497eac982205d3fa07 Mon Sep 17 00:00:00 2001 From: shuim Date: Mon, 21 Sep 2026 10:50:32 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EA=B4=80=EB=A6=AC=EC=9E=90=20API=20?= =?UTF-8?q?=E2=80=94=20.env=20KIS=20=EC=9E=90=EA=B2=A9=EC=A6=9D=EB=AA=85?= =?UTF-8?q?=EC=9D=84=20=EC=A7=80=EC=A0=95=20=EC=82=AC=EC=9A=A9=EC=9E=90?= =?UTF-8?q?=EC=97=90=EA=B2=8C=20(=EC=9E=AC)=ED=95=A0=EB=8B=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /auth/admin/tenant/assign-env-credentials {username}: 동일 앱키 1사용자 제약을 지키며 현 보유자(부트스트랩 admin)에서 대상 사용자로 이동, 대상의 기존 자격증명 교체, 컨텍스트 캐시 무효화. Co-Authored-By: Claude Opus 5 (1M context) --- backend/app/api/endpoints/auth.py | 86 +++++++++++++++++++++++++++++++ 1 file changed, 86 insertions(+) diff --git a/backend/app/api/endpoints/auth.py b/backend/app/api/endpoints/auth.py index 8948855..a32db4e 100644 --- a/backend/app/api/endpoints/auth.py +++ b/backend/app/api/endpoints/auth.py @@ -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) +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회 자동 실행)와 동일한 이관/백필 로직을 관리자가 수동으로 재실행할 수