feat: 주요 CLI, 인증, 서비스 로직 및 문서 개선

This commit is contained in:
2026-04-15 00:48:22 +09:00
parent 36fbb0879e
commit 0e9d2a16d8
17 changed files with 864 additions and 18 deletions
+54
View File
@@ -0,0 +1,54 @@
---
description: "Use when: /commit or /commit all to review changes, generate a commit message, commit, and optionally push with --push"
name: "commit"
argument-hint: "Optional: 'all' and/or '--push'. Examples: '/commit', '/commit all', '/commit --push', '/commit all --push'"
agent: "agent"
model: "GPT-4.1"
tools: [read, search, agent, todo, execute]
---
당신은 변경사항을 검토하고 커밋/푸시를 자동으로 수행하는 실행 도우미다.
반드시 GPT-4.1 모델 기준으로 동작한다.
반드시 아래 규칙을 지켜라.
1. 출력은 한국어로 작성한다.
2. 입력 해석 규칙:
- 빈 입력 또는 `all`: `/commit`과 동일하게 처리한다.
- `--push` 옵션이 있으면 커밋 후 원격 푸시까지 수행한다.
- 허용 예시: `/commit`, `/commit all`, `/commit --push`, `/commit all --push`
- 위 형식 외 인자는 오류로 보고하고 종료한다.
3. 실행 전 검증:
- git 저장소가 아니면 오류로 보고하고 종료한다.
- 변경 파일이 없으면 "커밋할 변경사항이 없다"고 보고하고 종료한다.
4. 변경 검토:
- `git status --short`와 `git diff`(필요 시 `git diff --staged`)로 변경을 파악한다.
- 변경 의도를 요약한 커밋 메시지를 1줄로 생성한다.
- 커밋 메시지 형식은 Conventional Commits를 우선 사용한다.
5. 커밋 수행:
- `/commit`, `/commit all` 모두 `git add -A` 후 `git commit -m "<생성 메시지>"`를 수행한다.
- 커밋 실패 시 오류 내용을 보고하고 종료한다.
6. 푸시 수행:
- `--push` 옵션이 있을 때만 `git push`를 수행한다.
- 기본 대상은 현재 체크아웃 브랜치의 upstream을 사용한다.
- upstream이 없으면 `origin <현재브랜치>`로 푸시를 시도하고 결과를 보고한다.
7. 안전 규칙:
- 파괴적 명령(`reset --hard`, `checkout --`, 강제 push)은 사용하지 않는다.
- 사용자가 명시하지 않은 변경을 되돌리지 않는다.
실행 절차:
1. 입력 옵션 파싱(`all`, `--push`)
2. 저장소/변경 여부 확인
3. 변경 내용 검토 및 커밋 메시지 생성
4. `git add -A` + `git commit`
5. 옵션이 있으면 `git push`
6. 결과 보고
최종 응답 형식:
1. 실행 옵션 요약
2. 생성된 커밋 메시지
3. 커밋 결과(해시, 변경 파일 수)
4. 푸시 결과(실행한 경우)
5. 다음 액션
사용자 입력:
{{input}}
+5 -1
View File
@@ -48,11 +48,15 @@ tools: [read, edit, search, todo, agent]
- {{DECIDE: 질문}}
- {{CONFIRM: 내용}}
- {{DATA: 필요한 정보}}
- {{DELEGATE_TO: agent}} (agent 자동 결정)
- 추가 의사결정이 필요한 경우 반드시 플레이스홀더를 삽입한다.
11. 플레이스홀더 집계 규칙:
- 본문에 미해결 플레이스홀더가 있으면 `# 9. 사용자 결정 필요 항목 요약`에 유형별로 집계한다.
- 집계 시 {{DECIDE}}, {{CONFIRM}}, {{DATA}}, {{DELEGATE_TO}}를 유형별로 분리하고 중복 문구는 1회만 남긴다.
- 본문에 미해결 플레이스홀더가 없으면 `현재 미해결 플레이스홀더 없음`으로 정규화한다.
12. 여러 문서를 처리할 때 문서별 결과를 분리해 보고한다.
13. 본문 플레이스홀더 정규화 규칙:
- {{DELEGATE_TO: agent}} 형식의 항목은 agent가 결정을 생성해 본문 문장으로 대체하고, 요약 섹션에서도 제거한다.
실행 절차:
1. 입력 ID 목록 정규화 및 형식 검증
@@ -72,7 +76,7 @@ tools: [read, edit, search, todo, agent]
5. 연결 Planning 미승인/미발견 목록(있으면)
6. 문서별 검토 결과
7. 문서별 수정 내역 요약
8. 문서별 추가된 사용자 결정 항목({{DECIDE}}/{{CONFIRM}}/{{DATA}})
8. 문서별 추가된 사용자 결정 항목({{DECIDE}}/{{CONFIRM}}/{{DATA}}/{{DELEGATE_TO}})
9. 업데이트된 파일 경로 목록
10. 다음 액션
+10 -6
View File
@@ -12,9 +12,10 @@ tools: [read, edit, search, create_file, agent, todo]
반드시 아래 규칙을 지켜라.
1. 출력은 한국어로 작성한다.
2. 입력 형식: `imp-<3자리숫자> <항목목록>`
2. 입력 형식: `imp-<3자리숫자>` 또는 `imp-<3자리숫자> <항목목록>`
- imp ID: `imp-<3자리숫자>` 형식 필수
- 항목 목록: `I-001, I-002` 또는 `P-001, P-002` 또는 혼합 가능
- 항목 목록(선택): `I-001, I-002` 또는 `P-001, P-002` 또는 혼합 가능
- 항목 목록이 생략되면 imp 문서 `# 3. 작업 단위 분해`의 미완료 항목 전체를 위에서 아래 순서대로 실행한다.
- 형식 오류 시 오류로 보고하고 종료한다.
3. imp 파일 검색 규칙:
- 경로: docs/sdlc/implementation/
@@ -23,6 +24,8 @@ tools: [read, edit, search, create_file, agent, todo]
4. imp 문서 상태 확인:
- 상태가 `draft`가 아니면 "이미 확정된 Implementation 문서다. 상태를 확인하라"로 보고하고 종료한다.
5. 항목 해석 규칙:
- 항목 목록이 없는 경우: `# 3. 작업 단위 분해` 표의 모든 I-ID 중 상태가 `완료`가 아닌 항목을 문서 순서대로 대상에 포함한다.
- 항목 목록이 없는데 미완료 I-ID가 없으면 "실행할 미완료 항목이 없다"고 보고하고 종료한다.
- I-ID 입력 시: imp 문서 `# 3. 작업 단위 분해` 표에서 해당 I-ID 행을 직접 찾는다.
- P-ID 입력 시: `# 3. 작업 단위 분해` 표에서 `연결 설계/AC` 열에 해당 P-ID가 포함된 모든 I-ID를 찾는다.
- 상태가 이미 `완료`인 항목은 건너뛰고 보고한다.
@@ -46,14 +49,15 @@ tools: [read, edit, search, create_file, agent, todo]
- 확인 필요: {{CONFIRM: 내용}}
- 추가 정보 필요: {{DATA: 필요한 정보}}
10. 범위 제한:
- 이번 명령에서 지정하지 않은 I-ID는 절대 수정하지 않는다.
- 항목 목록을 명시한 경우, 이번 명령에서 지정하지 않은 I-ID는 절대 수정하지 않는다.
- 항목 목록이 생략된 경우, 문서의 미완료 I-ID 전체가 이번 명령의 지정 항목으로 간주된다.
- imp 문서 외 SDLC 문서(Design, Planning 등)는 참조만 하고 수정하지 않는다.
- 지정한 파일 변경 계획 외의 추가 파일을 임의로 수정하지 않는다.
실행 절차:
1. 입력 파싱: imp ID + 항목 목록 형식 검증
1. 입력 파싱: imp ID 및 항목 목록(선택) 형식 검증
2. imp 파일 검색 및 상태 확인
3. 항목 해석: I-ID 직접 조회 또는 P-ID → I-ID 변환
3. 항목 해석: 항목 목록 없음 -> 미완료 I-ID 전체(문서 순서), 항목 목록 있음 -> I-ID 직접 조회 또는 P-ID → I-ID 변환
4. 상태가 이미 `완료`인 항목 필터링 및 보고
5. Design 문서 읽기 (실행 내용 근거 확보)
6. 항목별 순서대로 실행:
@@ -64,7 +68,7 @@ tools: [read, edit, search, create_file, agent, todo]
8. 최종 응답 제공
최종 응답 형식:
1. 실행 입력 요약 (imp ID, 지정 항목 목록)
1. 실행 입력 요약 (imp ID, 지정 항목 목록 또는 전체 미완료 자동 선택)
2. 항목별 실행 결과 (성공/건너뜀/실패)
3. 생성/수정된 실제 파일 경로 목록
4. imp 문서 갱신 내역 (상태 변경된 항목)
+12
View File
@@ -105,6 +105,9 @@ portfoliohub user create --username alice --password password123 --email alice@e
portfoliohub user get <user-id>
portfoliohub user update <user-id> --username alice-updated --force
portfoliohub user delete <user-id> --force
portfoliohub bot token issue <bot-id> --name nightly --ttl-days 30
portfoliohub bot token issue <bot-id> --name permanent --no-expiry
portfoliohub bot token revoke <bot-id> <token-id> --force
portfoliohub portfolio list
portfoliohub portfolio status 005930:tv --recent 10
portfoliohub portfolio create 005930:tv --symbol 005930 --cash-balance 10000000 --force
@@ -125,6 +128,8 @@ Supported commands:
- `user create`
- `user update`
- `user delete`
- `bot token issue`
- `bot token revoke`
- `portfolio list`
- `portfolio status`
- `portfolio create`
@@ -139,6 +144,13 @@ Supported commands:
Write commands ask for confirmation by default. Use `--force` to skip the prompt for scripts and automation.
Bot token issue policy:
- 기본값은 365일 TTL이다.
- `--ttl-days`와 `--no-expiry`는 동시에 사용할 수 없다.
- `--no-expiry`는 관리자 권한에서만 허용된다.
- 토큰 원문은 발급 응답 1회 이후 마스킹된 값으로만 출력/기록된다.
The CLI uses the same repository and settings layer as the HTTP API. To target the external TimescaleDB, set `PORTFOLIOHUB_DB_ENABLED=true` and provide `TIMESCALEDB_*` values in the environment or `.env`.
Initialize a new database explicitly before the first use:
+115
View File
@@ -0,0 +1,115 @@
"""Bot token CLI commands."""
from __future__ import annotations
from datetime import datetime
import typer
from app.cli.core.utils import execute_command
from app.cli.core.output import mask_token_text, render_bot_result
from app.schemas.auth import (
BotTokenIssueCliResponse,
BotTokenRevokeCliResponse,
CreateBotTokenRequest,
)
from app.services.auth_service import AuthService
bot_app = typer.Typer(help="Bot principal and token commands.")
token_app = typer.Typer(help="Bot token lifecycle commands.")
bot_app.add_typer(token_app, name="token")
_DEFAULT_BOT_TOKEN_TTL_DAYS = 365
def _confirm(force: bool, message: str) -> None:
if force:
return
confirmed = typer.confirm(message)
if not confirmed:
raise typer.Exit(code=0)
def _parse_issue_policy(ttl_days: int | None, no_expiry: bool) -> int | None:
if no_expiry and ttl_days is not None:
raise ValueError("--ttl-days와 --no-expiry는 동시에 사용할 수 없습니다")
if ttl_days is not None and ttl_days <= 0:
raise ValueError("--ttl-days는 1 이상의 정수여야 합니다")
if ttl_days is None and not no_expiry:
return _DEFAULT_BOT_TOKEN_TTL_DAYS
return ttl_days
def _to_issue_response(bot_id: str, issued, output_token: str | None) -> BotTokenIssueCliResponse:
return BotTokenIssueCliResponse(
bot_id=bot_id,
token_id=issued.auth_token.id,
expires_at=issued.auth_token.expires_at,
revoked=issued.auth_token.revoked_at is not None,
token=output_token,
)
def _to_revoke_response(bot_id: str, token_id: str, revoked_at: datetime) -> BotTokenRevokeCliResponse:
return BotTokenRevokeCliResponse(
bot_id=bot_id,
token_id=token_id,
revoked=True,
revoked_at=revoked_at,
)
def _with_service(ctx: typer.Context, callback):
app_context = ctx.obj
repository = app_context.repository_factory(app_context.settings)
service = AuthService(repository, app_context.settings)
try:
return execute_command(app_context.console, callback=lambda: callback(service, app_context))
finally:
if callable(getattr(type(repository), "close", None)):
repository.close()
@token_app.command("issue")
def issue_bot_token(
ctx: typer.Context,
bot_id: str = typer.Argument(..., help="Bot principal identifier."),
name: str = typer.Option("cli", help="Token display name."),
permission: str | None = typer.Option(None, help="Token permission: read or trade-write."),
ttl_days: int | None = typer.Option(None, "--ttl-days", help="Token TTL in days."),
no_expiry: bool = typer.Option(False, "--no-expiry", help="Issue a token with no expiry (admin only)."),
) -> None:
def _issue_with_policy(service: AuthService, app_context):
expires_in_days = _parse_issue_policy(ttl_days, no_expiry)
payload = CreateBotTokenRequest(
name=name,
permission=permission,
expires_in_days=expires_in_days,
no_expiry=no_expiry,
)
return service.issue_bot_token(bot_id, payload)
issued = _with_service(ctx, _issue_with_policy)
response = _to_issue_response(bot_id, issued, issued.token)
render_bot_result(ctx.obj.console, ctx.obj.output_format, response)
if ctx.obj.output_format != "json":
ctx.obj.console.print(f"masked_token: {mask_token_text(issued.token)}")
@token_app.command("revoke")
def revoke_bot_token(
ctx: typer.Context,
bot_id: str = typer.Argument(..., help="Bot principal identifier."),
token_id: str = typer.Argument(..., help="Token identifier to revoke."),
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
) -> None:
_confirm(force, f"Revoke bot token {token_id} for bot {bot_id}?")
revoked_at = _with_service(
ctx,
lambda service, app_context: service.revoke_bot_token(bot_id, token_id),
)
response = _to_revoke_response(bot_id, token_id, revoked_at)
render_bot_result(ctx.obj.console, ctx.obj.output_format, response)
+30
View File
@@ -0,0 +1,30 @@
"""Output helpers dedicated to auth/bot CLI flows."""
from __future__ import annotations
from typing import Any
from rich.console import Console
from app.cli.core.utils import render_output
from app.core.security import mask_secret
def mask_token_text(raw_token: str) -> str:
return mask_secret(raw_token, prefix=6, suffix=4)
def _mask_payload(payload: dict[str, Any]) -> dict[str, Any]:
masked = dict(payload)
token = masked.get("token")
if isinstance(token, str) and token:
masked["token"] = mask_token_text(token)
return masked
def render_bot_result(console: Console, output_format: str, payload: Any) -> None:
plain = payload.model_dump(mode="json") if hasattr(payload, "model_dump") else payload
if isinstance(plain, dict):
render_output(console, output_format, _mask_payload(plain))
return
render_output(console, output_format, plain)
+2
View File
@@ -9,6 +9,7 @@ import typer
from rich.console import Console
from app.cli.commands.health import register_health_commands
from app.cli.commands.bot_tokens import bot_app
from app.cli.commands.portfolio import portfolio_app
from app.cli.commands.strategy import strategy_app
from app.cli.commands.user import user_app
@@ -49,6 +50,7 @@ def create_app(
app.add_typer(portfolio_app, name="portfolio")
app.add_typer(strategy_app, name="strategy")
app.add_typer(user_app, name="user")
app.add_typer(bot_app, name="bot")
register_health_commands(app)
return app
+37
View File
@@ -6,6 +6,8 @@ import base64
import hashlib
import hmac
import secrets
from datetime import datetime, timezone
from typing import Any
def hash_password(password: str) -> str:
@@ -48,3 +50,38 @@ def hash_token(token: str) -> str:
def token_prefix(token: str) -> str:
return token[:8]
def mask_secret(value: str, *, prefix: int = 6, suffix: int = 4, mask_char: str = "*") -> str:
text = (value or "").strip()
if not text:
return ""
if len(text) <= prefix + suffix:
return mask_char * len(text)
hidden = len(text) - prefix - suffix
return f"{text[:prefix]}{mask_char * hidden}{text[-suffix:]}"
def build_audit_event(
*,
action: str,
actor_id: str,
bot_id: str,
token_id: str,
token_ref: str,
result: str,
detail: str | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"event_type": "auth.bot_token",
"action": action,
"actor_id": actor_id,
"bot_id": bot_id,
"token_id": mask_secret(token_id, prefix=6, suffix=4),
"token_ref": token_ref,
"result": result,
"occurred_at": datetime.now(timezone.utc).isoformat(),
}
if detail:
payload["detail"] = detail
return payload
+7
View File
@@ -15,6 +15,13 @@ _INVEST_ACTION_MAP = {
}
def normalize_expiry_days(expires_in_days: int | None, *, default_days: int) -> int:
days = default_days if expires_in_days is None else int(expires_in_days)
if days <= 0:
raise ValueError("만료일은 1일 이상이어야 합니다")
return days
class PortfolioDomainRepository(Protocol):
def get(self, portfolio_id: str) -> Portfolio: ...
def get_or_create(
+59
View File
@@ -345,6 +345,30 @@ class PostgresPortfolioRepository:
ON portfolio_tags (tag_value, portfolio_id)
"""
)
cur.execute(
"""
CREATE TABLE IF NOT EXISTS auth_audit_logs (
id TEXT PRIMARY KEY,
event_type TEXT NOT NULL,
action TEXT NOT NULL,
actor_id TEXT NOT NULL,
bot_id TEXT NOT NULL,
token_id_masked TEXT NOT NULL,
token_ref_masked TEXT NOT NULL,
result TEXT NOT NULL,
detail TEXT,
payload JSONB NOT NULL,
occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
)
"""
)
cur.execute(
"""
CREATE INDEX IF NOT EXISTS idx_auth_audit_logs_bot_occurred
ON auth_audit_logs (bot_id, occurred_at DESC)
"""
)
cur.execute("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'timescaledb')")
if cur.fetchone()[0]:
cur.execute(
@@ -821,6 +845,41 @@ class PostgresPortfolioRepository:
bot_principal_id=row[4],
)
def record_audit_event(self, payload: dict[str, Any]) -> None:
self.ensure_schema()
event_id = str(uuid.uuid4())
occurred_at_raw = payload.get("occurred_at")
occurred_at = datetime.now(timezone.utc)
if isinstance(occurred_at_raw, str):
try:
occurred_at = datetime.fromisoformat(occurred_at_raw)
except ValueError:
occurred_at = datetime.now(timezone.utc)
with self._connect().cursor() as cur:
cur.execute(
"""
INSERT INTO auth_audit_logs (
id, event_type, action, actor_id, bot_id,
token_id_masked, token_ref_masked, result, detail, payload, occurred_at
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s)
""",
(
event_id,
str(payload.get("event_type", "auth.bot_token")),
str(payload.get("action", "")),
str(payload.get("actor_id", "unknown")),
str(payload.get("bot_id", "unknown")),
str(payload.get("token_id", "")),
str(payload.get("token_ref", "")),
str(payload.get("result", "unknown")),
payload.get("detail"),
json.dumps(payload, ensure_ascii=True),
occurred_at,
),
)
def get(self, portfolio_id: str) -> Portfolio:
self.ensure_schema()
with self._connect().cursor() as cur:
+16
View File
@@ -118,6 +118,7 @@ class CreateBotTokenRequest(BaseModel):
name: str = Field(min_length=1)
permission: Literal["read", "trade-write"] | None = None
expires_in_days: int | None = Field(default=None, ge=1, le=3650)
no_expiry: bool = False
@field_validator("name")
@classmethod
@@ -132,3 +133,18 @@ class IssuedBotTokenResponse(BaseModel):
token: str
token_type: Literal["bearer"] = "bearer"
auth_token: AuthTokenResponse
class BotTokenIssueCliResponse(BaseModel):
bot_id: str
token_id: str
expires_at: datetime | None = None
revoked: bool
token: str | None = None
class BotTokenRevokeCliResponse(BaseModel):
bot_id: str
token_id: str
revoked: bool
revoked_at: datetime | None = None
+77 -8
View File
@@ -4,12 +4,14 @@ from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any
from fastapi import Depends
from app.core.auth import AuthContext, Permission, _has_permission
from app.core.config import Settings, get_settings
from app.core.security import generate_token, hash_token, token_prefix, verify_password
from app.core.security import build_audit_event, generate_token, hash_token, mask_secret, token_prefix, verify_password
from app.domain.services import normalize_expiry_days
from app.domain.models import AuthToken, BotPrincipal, User
from app.repositories.factory import get_portfolio_repository_from_settings
from app.repositories.portfolio_repository import PortfolioRepository
@@ -138,7 +140,14 @@ class AuthService:
def deactivate_bot_principal(self, bot_principal_id: str) -> BotPrincipal:
return self._repository.deactivate_bot_principal(bot_principal_id)
def issue_bot_token(self, bot_principal_id: str, payload: CreateBotTokenRequest) -> IssuedBotToken:
def issue_bot_token(
self,
bot_principal_id: str,
payload: CreateBotTokenRequest,
*,
actor_permission: Permission = Permission.ADMIN,
actor_id: str | None = None,
) -> IssuedBotToken:
bot = self._repository.get_bot_principal(bot_principal_id)
if not bot.is_active:
raise ValueError("비활성화된 봇에는 토큰을 발급할 수 없습니다")
@@ -149,7 +158,11 @@ class AuthService:
raise ValueError("요청한 토큰 권한이 봇 권한을 초과합니다")
raw_token = generate_token()
expires_at = self._build_expiration(payload.expires_in_days, self._settings.bot_token_default_ttl_days)
expires_at = self._build_bot_expiration(
payload.expires_in_days,
no_expiry=payload.no_expiry,
actor_permission=actor_permission,
)
auth_token = self._repository.create_auth_token(
bot.id,
name=payload.name,
@@ -160,6 +173,20 @@ class AuthService:
bot_principal_id=bot.id,
expires_at=expires_at,
)
try:
self._record_audit_event(
build_audit_event(
action="bot_token.issue",
actor_id=actor_id or bot.owner_user_id,
bot_id=bot.id,
token_id=auth_token.id,
token_ref=mask_secret(raw_token, prefix=6, suffix=4),
result="success",
)
)
except Exception as exc:
self._repository.revoke_bot_auth_token(bot.id, auth_token.id)
raise ValueError("감사 로그 기록에 실패하여 토큰 발급을 취소했습니다") from exc
return IssuedBotToken(token=raw_token, auth_token=auth_token)
def issue_owned_bot_token(
@@ -172,7 +199,12 @@ class AuthService:
bot = self._repository.get_bot_principal(bot_principal_id)
if bot.owner_user_id != user_id:
raise ValueError("봇 접근 권한이 없습니다")
return self.issue_bot_token(bot_principal_id, payload)
return self.issue_bot_token(
bot_principal_id,
payload,
actor_permission=auth_context.permission,
actor_id=user_id,
)
def list_bot_tokens(self, bot_principal_id: str) -> list[AuthToken]:
self._repository.get_bot_principal(bot_principal_id)
@@ -185,16 +217,31 @@ class AuthService:
raise ValueError("봇 접근 권한이 없습니다")
return self._repository.list_bot_auth_tokens(bot_principal_id)
def revoke_bot_token(self, bot_principal_id: str, token_id: str) -> None:
self._repository.get_bot_principal(bot_principal_id)
def revoke_bot_token(self, bot_principal_id: str, token_id: str, *, actor_id: str | None = None) -> datetime:
bot = self._repository.get_bot_principal(bot_principal_id)
self._repository.revoke_bot_auth_token(bot_principal_id, token_id)
revoked_at = datetime.now(timezone.utc)
try:
self._record_audit_event(
build_audit_event(
action="bot_token.revoke",
actor_id=actor_id or bot.owner_user_id,
bot_id=bot.id,
token_id=token_id,
token_ref=mask_secret(token_id, prefix=6, suffix=4),
result="success",
)
)
except Exception as exc:
raise ValueError("토큰은 폐기되었으나 감사 로그 기록에 실패했습니다") from exc
return revoked_at
def revoke_owned_bot_token(self, auth_context: AuthContext, bot_principal_id: str, token_id: str) -> None:
user_id = self._require_db_user_context(auth_context)
bot = self._repository.get_bot_principal(bot_principal_id)
if bot.owner_user_id != user_id:
raise ValueError("봇 접근 권한이 없습니다")
self._repository.revoke_bot_auth_token(bot_principal_id, token_id)
self.revoke_bot_token(bot_principal_id, token_id, actor_id=user_id)
@staticmethod
def _require_db_user_context(auth_context: AuthContext) -> str:
@@ -206,9 +253,31 @@ class AuthService:
@staticmethod
def _build_expiration(expires_in_days: int | None, default_days: int) -> datetime:
days = expires_in_days if expires_in_days is not None else default_days
days = normalize_expiry_days(expires_in_days, default_days=default_days)
return datetime.now(timezone.utc) + timedelta(days=days)
@staticmethod
def _build_bot_expiration(
expires_in_days: int | None,
*,
no_expiry: bool,
actor_permission: Permission,
) -> datetime | None:
if no_expiry:
if expires_in_days is not None:
raise ValueError("만료일 옵션과 무기한 옵션은 동시에 사용할 수 없습니다")
if actor_permission is not Permission.ADMIN:
raise ValueError("비관리자는 --no-expiry를 사용할 수 없습니다")
return None
days = normalize_expiry_days(expires_in_days, default_days=365)
return datetime.now(timezone.utc) + timedelta(days=days)
def _record_audit_event(self, payload: dict[str, Any]) -> None:
recorder = getattr(self._repository, "record_audit_event", None)
if callable(recorder):
recorder(payload)
def get_auth_service(
settings: Settings = Depends(get_settings),
@@ -0,0 +1,109 @@
# 0. 문서 상태
- 상태: draft
- 승인자: 고진환
- 승인 시각: {{CONFIRM: 승인 후 기재}}
# 1. Implementation 개요
- 입력 Design: dsn-001
- 목적: CLI bot 토큰 발급/삭제 기능을 설계 기준에 따라 작은 구현 단위로 분해하고, Verification 단계에 즉시 전달 가능한 검증 근거를 준비한다.
- 이번 구현 우선순위 작업:
- 1순위: P-002 만기 정책 및 권한 가드 구현
- 2순위: P-003 토큰 출력 마스킹 및 감사 로그 구현
- 3순위: P-001 CLI bot 토큰 발급/삭제 명령 추가
# 2. 이번 구현 범위
## In Scope
- `bot token issue`, `bot token revoke` CLI 명령 추가 및 help 노출
- `--ttl-days`/`--no-expiry` 입력 검증(상호배타, 값 범위)과 서비스 계층 정책 재검증
- 기본 TTL 365일 정책 반영 및 비관리자 무기한 토큰 금지
- issue/revoke 결과의 마스킹 출력 처리 및 감사 로그 기록 경로 구현
- revoke 확인 프롬프트/`--force` 분기와 표준 출력(table/json) 정합성 반영
- 정상/예외 테스트 및 문서 동기화 반영
## Out of Scope
- bot token list/rotate 등 신규 명령 확장
- 감사 로그 저장소 아키텍처 재설계(외부 로그 파이프라인 교체 포함)
- API 채널의 대규모 인터페이스 재정의
# 3. 작업 단위 분해
| Impl ID | 작업명 | 연결 설계/AC | 변경 유형 | 우선순위 | 상태 |
|---|---|---|---|---|---|
| I-001 | bot 토큰 CLI 그룹/서브커맨드 골격 추가 | P-001 AC-1 | 코드(명령 라우팅) | High | 완료 |
| I-002 | issue 입력 옵션 검증(상호배타/범위) | P-002 AC-3 | 코드(입력 검증) | High | 완료 |
| I-003 | 만기 정책 서비스 재검증 및 기본 TTL 365일 적용 | P-002 AC-1, AC-2 | 코드(도메인/서비스) | High | 완료 |
| I-004 | issue/revoke 서비스 호출 및 결과 DTO 정규화 | P-001 AC-2, AC-3 | 코드(서비스 통합) | High | 완료 |
| I-005 | 토큰 출력 마스킹 정책 적용(원문 1회 출력 후 차단) | P-003 AC-1, AC-3 | 코드(출력/보안) | High | 완료 |
| I-006 | 감사 로그 이벤트 기록 및 실패 분기 처리 | P-003 AC-2 | 코드(로깅/오류 처리) | Medium | 완료 |
| I-007 | revoke UX 가드(확인 프롬프트, --force, 오류 메시지) | P-004 AC-1, AC-2 | 코드(UX/CLI) | Medium | 완료 |
| I-008 | 테스트/문서 동기화 및 근거 번들 준비 | P-005 AC-1/AC-2 | 테스트+문서 | High | 완료 |
# 4. 파일 변경 계획
| Impl ID | 대상 파일 | 변경 목적 | 변경 종류(생성/수정) | 상태 |
|---|---|---|---|---|
| I-001 | app/cli/main.py | `bot` 그룹 등록 및 커맨드 엔트리 연결 | 수정 | 완료 |
| I-001 | app/cli/commands/bot_tokens.py | `token issue/revoke` 커맨드 골격 정의 | 생성 | 완료 |
| I-002 | app/cli/commands/bot_tokens.py | `--ttl-days`/`--no-expiry` 검증 로직 추가 | 수정 | 완료 |
| I-003 | app/services/auth_service.py | 만기 정책 재검증, 기본 TTL/권한 정책 강제 | 수정 | 완료 |
| I-003 | app/domain/services.py | 정책 객체 정규화 규칙 보완(필요 시) | 수정 | 완료 |
| I-004 | app/services/auth_service.py | issue/revoke 결과 DTO 필드 정규화 | 수정 | 완료 |
| I-004 | app/schemas/auth.py | 응답 스키마(`token_id/expires_at/revoked`) 정렬 | 수정 | 완료 |
| I-005 | app/cli/core/output.py | table/json 출력 시 마스킹 렌더링 추가 | 수정 | 완료 |
| I-006 | app/core/security.py | 감사 로그 payload 표준화 및 민감정보 필터링 | 수정 | 완료 |
| I-006 | app/repositories/postgres.py | 감사 로그 쓰기 경로(트랜잭션 연계) 보강 | 수정 | 완료 |
| I-007 | app/cli/commands/bot_tokens.py | revoke 확인 프롬프트, `--force` 분기, 사용자 지침 메시지 | 수정 | 완료 |
| I-008 | tests/test_cli.py | bot token help/성공/취소 흐름 테스트 추가 | 수정 | 완료 |
| I-008 | tests/test_auth.py | 정책 검증(관리자/비관리자/no-expiry) 테스트 추가 | 수정 | 완료 |
| I-008 | tests/test_app.py | 출력 마스킹 및 보안 회귀 시나리오 보강 | 수정 | 완료 |
| I-008 | README.md | CLI 사용 예시 및 정책 제약 문서화 | 수정 | 완료 |
| I-008 | docs/technical/auth-and-token-lifecycle.md | 토큰 수명주기/감사 로그 동작 반영 | 수정 | 완료 |
# 5. 테스트/검증 계획
| Impl ID | 검증 방법 | 성공 기준 | 근거 |
|---|---|---|---|
| I-001 | CLI help 테스트 | `bot token issue/revoke`가 help에 노출된다 | `tests/test_cli.py` |
| I-002 | 옵션 조합 예외 테스트 | `--ttl-days`와 `--no-expiry` 동시 입력 시 실패한다 | `tests/test_auth.py` |
| I-003 | 권한/기본값 테스트 | 기본 TTL 365일, 비관리자 `--no-expiry` 차단 | `tests/test_auth.py` |
| I-004 | 성공 경로 통합 테스트 | issue 응답에 `token_id/expires_at/revoked`가 존재한다 | `tests/test_cli.py` |
| I-005 | 출력 스냅샷/회귀 테스트 | 재출력/로그에 raw token이 노출되지 않는다 | `tests/test_app.py` |
| I-006 | 감사 로그 실패 분기 테스트 | issue는 롤백 실패 처리, revoke는 폐기 유지+경고를 만족한다 | `tests/test_auth.py` |
| I-007 | UX 동작 테스트 | revoke 시 확인 프롬프트, `--force` 시 생략, 취소는 code 0 | `tests/test_cli.py` |
| I-008 | 회귀+문서 동기화 확인 | 변경 경로 테스트 통과 및 문서 업데이트 반영 | 테스트 실행 로그 + 변경 문서 diff |
# 6. 영향도 및 의존성 점검
| 항목 | 영향 내용 | 대응 방안 |
|---|---|---|
| 코드 | CLI 진입점, 인증 서비스, 출력/로깅 경로가 동시에 변경되어 회귀 범위가 넓다 | 단위+통합 테스트를 분리 실행하고 민감정보 노출 회귀 테스트를 추가한다 |
| 문서 | CLI 사용법과 토큰 수명주기 운영 규칙이 불일치할 수 있다 | README/기술문서를 같은 변경 세트에서 동기 갱신한다 |
| 설정/의존성 | 로그 sink/DB 쓰기 실패 시 issue/revoke 실패 정책이 다르므로 운영 혼선 위험이 있다 | 실패 분기 정책을 테스트로 고정하고 운영 경고 기준을 문서에 명시한다 |
# 7. Verification 단계 전달 체크
- [x] 구현 범위가 설계 문서와 연결됨
- [x] 작업 단위별 검증 방법이 정의됨
- [x] 영향 파일과 의존성 점검 결과가 기록됨
- [x] Verification에 전달할 실행/검증 근거가 정의됨
- [ ] 최종 승인
- {{CONFIRM: Implementation 초안 승인 여부를 확인하라}}
# 8. 파일 처리 결과
- 생성 경로: app/cli/commands/bot_tokens.py
- 생성 경로: app/cli/core/output.py
- 수정 경로: app/cli/main.py
- 수정 경로: app/services/auth_service.py
- 수정 경로: app/domain/services.py
- 수정 경로: app/schemas/auth.py
- 수정 경로: app/core/security.py
- 수정 경로: app/repositories/postgres.py
- 수정 경로: tests/test_cli.py
- 수정 경로: tests/test_auth.py
- 수정 경로: tests/test_app.py
- 수정 경로: README.md
- 수정 경로: docs/technical/auth-and-token-lifecycle.md
- 수정 경로: docs/sdlc/implementation/imp-001_2026-04-15_cli-bot-token-lifecycle-cli.implementation.md
# 9. 사용자 결정 필요 항목 요약
## {{CONFIRM}} (2건)
- 승인 후 기재 (위치: #0)
- Implementation 초안 승인 여부를 확인하라. (위치: #7)
## {{DATA}}
- 현재 미해결 플레이스홀더 없음
@@ -64,6 +64,26 @@
3. 요청 처리 시 해시 비교와 만료/폐기 상태 검증
4. `DELETE /auth/tokens/{token_id}` 또는 봇 비활성화로 토큰 폐기
CLI 봇 토큰 라이프사이클:
1. `bot token issue <bot_id>` 실행 시 `--ttl-days`/`--no-expiry` 정책을 먼저 검증한다.
2. 정책 검증 후 서비스 계층에서 권한과 만기 정책을 재검증한다.
3. 발급 성공 시 응답은 마스킹된 토큰만 재출력 가능하며, 원문 토큰은 1회 출력 경로로 제한한다.
4. `bot token revoke <bot_id> <token_id>` 실행 시 확인 프롬프트(`--force`로 생략 가능)를 거쳐 폐기한다.
만기 정책:
- 기본 TTL은 365일이다.
- `--ttl-days`와 `--no-expiry` 동시 사용은 거부한다.
- 비관리자 컨텍스트에서는 `--no-expiry`를 거부한다.
감사 로그 정책:
- `issue/revoke`는 `auth.bot_token` 이벤트를 감사 로그로 남긴다.
- 감사 로그에는 마스킹된 `token_id/token_ref`만 기록한다.
- issue 경로에서 감사 로그 기록이 실패하면 발급 토큰을 즉시 폐기하고 실패 처리한다.
- revoke 경로에서 감사 로그 기록이 실패해도 폐기는 유지하고 명령은 실패로 반환한다.
## 실패 응답 기준
- bearer 헤더 누락 또는 형식 오류: `401`
+15
View File
@@ -2,6 +2,7 @@ from fastapi.testclient import TestClient
from app.core.config import Settings
from app.core.config import get_settings
from app.core.security import mask_secret
from app.domain.models import User
from app.main import app, create_app
@@ -104,3 +105,17 @@ def test_startup_does_not_bootstrap_when_users_exist() -> None:
assert repo.created == []
assert repo.closed is True
def test_mask_secret_keeps_prefix_and_suffix() -> None:
masked = mask_secret("abcdefghijklmnopqrstuvwxyz", prefix=6, suffix=4)
assert masked.startswith("abcdef")
assert masked.endswith("wxyz")
assert "ghijklmnopqrstuv" not in masked
def test_mask_secret_short_value_is_fully_masked() -> None:
masked = mask_secret("abcd", prefix=3, suffix=2)
assert masked == "****"
+142
View File
@@ -1,9 +1,17 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
from fastapi.testclient import TestClient
from app.core.auth import AuthContext, Permission
from app.core.config import Settings
from app.core.config import get_settings
from app.domain.models import AuthToken, BotPrincipal
from app.main import app
from app.schemas.auth import CreateBotTokenRequest
from app.services.auth_service import AuthService
def _client_with_tokens() -> TestClient:
@@ -63,3 +71,137 @@ def test_admin_route_accepts_admin_token(monkeypatch) -> None:
assert response.status_code == 501
assert response.json()["detail"] == "Portfolio persistence is not implemented yet"
@dataclass
class _AuthRepo:
fail_audit: bool = False
def __post_init__(self) -> None:
self.bot = BotPrincipal(
id="bot-1",
name="bot",
owner_user_id="owner-1",
permission="trade-write",
is_active=True,
)
self.tokens: dict[str, AuthToken] = {}
self._seq = 1
def get_bot_principal(self, bot_principal_id: str) -> BotPrincipal:
if bot_principal_id != self.bot.id:
raise ValueError("봇을 찾을 수 없습니다")
return self.bot
def create_auth_token(
self,
user_id: str,
*,
name: str,
token_hash: str,
token_prefix: str,
permission: str,
token_type: str,
bot_principal_id: str | None = None,
expires_at: datetime | None = None,
) -> AuthToken:
token = AuthToken(
id=f"token-{self._seq}",
user_id=user_id,
name=name,
token_prefix=token_prefix,
permission=permission,
token_type=token_type,
bot_principal_id=bot_principal_id,
expires_at=expires_at,
created_at=datetime.now(timezone.utc),
)
self._seq += 1
self.tokens[token.id] = token
return token
def revoke_bot_auth_token(self, bot_principal_id: str, token_id: str) -> None:
token = self.tokens.get(token_id)
if token is None or token.bot_principal_id != bot_principal_id:
raise ValueError("토큰을 찾을 수 없습니다")
token.revoked_at = datetime.now(timezone.utc)
def record_audit_event(self, payload: dict[str, object]) -> None:
if self.fail_audit:
raise RuntimeError("audit failed")
def _service(repo: _AuthRepo) -> AuthService:
return AuthService(repo, Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000"))
def test_issue_bot_token_defaults_to_365_days() -> None:
repo = _AuthRepo()
service = _service(repo)
result = service.issue_bot_token("bot-1", CreateBotTokenRequest(name="cli"))
assert result.auth_token.expires_at is not None
assert (result.auth_token.expires_at - datetime.now(timezone.utc)).days >= 364
def test_issue_owned_bot_token_rejects_no_expiry_for_non_admin() -> None:
repo = _AuthRepo()
service = _service(repo)
auth_context = AuthContext(
token="db-token",
permission=Permission.TRADE_WRITE,
user_id="owner-1",
token_source="db",
)
try:
service.issue_owned_bot_token(
auth_context,
"bot-1",
CreateBotTokenRequest(name="cli", no_expiry=True),
)
except ValueError as exc:
assert "비관리자" in str(exc)
else:
raise AssertionError("Expected ValueError for non-admin no-expiry")
def test_issue_bot_token_rolls_back_when_audit_fails() -> None:
repo = _AuthRepo(fail_audit=True)
service = _service(repo)
try:
service.issue_bot_token("bot-1", CreateBotTokenRequest(name="cli"))
except ValueError as exc:
assert "발급을 취소" in str(exc)
else:
raise AssertionError("Expected ValueError on audit failure")
assert len(repo.tokens) == 1
token = next(iter(repo.tokens.values()))
assert token.revoked_at is not None
def test_revoke_bot_token_keeps_revoke_when_audit_fails() -> None:
repo = _AuthRepo(fail_audit=True)
service = _service(repo)
created = repo.create_auth_token(
"bot-1",
name="cli",
token_hash="hash",
token_prefix="prefix",
permission="trade-write",
token_type="bot",
bot_principal_id="bot-1",
)
try:
service.revoke_bot_token("bot-1", created.id)
except ValueError as exc:
assert "토큰은 폐기" in str(exc)
else:
raise AssertionError("Expected ValueError on audit failure")
assert repo.tokens[created.id].revoked_at is not None
+152 -1
View File
@@ -9,7 +9,8 @@ from typer.testing import CliRunner
from app.cli.main import create_app
from app.core.config import Settings
from app.domain.models import LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, User
from app.core.security import hash_token, token_prefix
from app.domain.models import AuthToken, BotPrincipal, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, User
class FakeRepo:
@@ -22,6 +23,17 @@ class FakeRepo:
self._user_seq = 1
self._comments: dict[str, PortfolioComment] = {}
self._tags: dict[str, PortfolioTag] = {}
self._bot_principals: dict[str, BotPrincipal] = {
"bot-1": BotPrincipal(
id="bot-1",
name="cli-bot",
owner_user_id="owner-1",
permission="trade-write",
is_active=True,
)
}
self._auth_tokens: dict[str, AuthToken] = {}
self._token_seq = 1
self.closed = False
def close(self) -> None:
@@ -103,6 +115,49 @@ class FakeRepo:
raise ValueError("사용자를 찾을 수 없습니다")
del self._users[user_id]
def get_bot_principal(self, bot_principal_id: str) -> BotPrincipal:
bot = self._bot_principals.get(bot_principal_id)
if bot is None:
raise ValueError("봇을 찾을 수 없습니다")
return bot
def create_auth_token(
self,
user_id: str,
*,
name: str,
token_hash: str,
token_prefix: str,
permission: str,
token_type: str,
bot_principal_id: str | None = None,
expires_at: datetime | None = None,
) -> AuthToken:
token = AuthToken(
id=f"token-{self._token_seq}",
user_id=user_id,
name=name,
token_prefix=token_prefix,
permission=permission,
token_type=token_type,
bot_principal_id=bot_principal_id,
expires_at=expires_at,
created_at=datetime.now(timezone.utc),
)
self._token_seq += 1
self._auth_tokens[token.id] = token
return token
def revoke_bot_auth_token(self, bot_principal_id: str, token_id: str) -> None:
token = self._auth_tokens.get(token_id)
if token is None or token.bot_principal_id != bot_principal_id:
raise ValueError("토큰을 찾을 수 없습니다")
token.revoked_at = datetime.now(timezone.utc)
def record_audit_event(self, payload: dict[str, object]) -> None:
if payload.get("force_fail"):
raise RuntimeError("audit failure")
def get(self, portfolio_id: str) -> Portfolio:
if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id:
raise ValueError("포트폴리오를 찾을 수 없습니다")
@@ -769,3 +824,99 @@ def test_cli_tag_remove_flow() -> None:
assert result.exit_code == 0
assert "Removed tag: grade:A" in result.stdout
def test_cli_help_exposes_bot_token_commands() -> None:
repo = FakeRepo(_portfolio())
runner = build_runner(repo)
result = runner.invoke(runner.app, ["bot", "token", "--help"])
assert result.exit_code == 0
assert "issue" in result.stdout
assert "revoke" in result.stdout
def test_cli_bot_token_issue_supports_json() -> None:
repo = FakeRepo(_portfolio())
runner = build_runner(repo)
result = runner.invoke(
runner.app,
["--format", "json", "bot", "token", "issue", "bot-1", "--name", "nightly"],
)
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["bot_id"] == "bot-1"
assert payload["revoked"] is False
assert payload["expires_at"] is not None
assert payload["token"].endswith("...") is False
def test_cli_bot_token_issue_rejects_conflicting_expiry_options() -> None:
repo = FakeRepo(_portfolio())
runner = build_runner(repo)
result = runner.invoke(
runner.app,
[
"bot",
"token",
"issue",
"bot-1",
"--ttl-days",
"7",
"--no-expiry",
],
)
assert result.exit_code == 1
assert "동시에 사용할 수 없습니다" in result.stdout
def test_cli_bot_token_revoke_can_abort() -> None:
repo = FakeRepo(_portfolio())
token = repo.create_auth_token(
"bot-1",
name="to-revoke",
token_hash=hash_token("token-value"),
token_prefix=token_prefix("token-value"),
permission="trade-write",
token_type="bot",
bot_principal_id="bot-1",
)
runner = build_runner(repo)
result = runner.invoke(
runner.app,
["bot", "token", "revoke", "bot-1", token.id],
input="n\n",
)
assert result.exit_code == 0
assert repo._auth_tokens[token.id].revoked_at is None
def test_cli_bot_token_revoke_force_succeeds() -> None:
repo = FakeRepo(_portfolio())
token = repo.create_auth_token(
"bot-1",
name="to-revoke",
token_hash=hash_token("token-value"),
token_prefix=token_prefix("token-value"),
permission="trade-write",
token_type="bot",
bot_principal_id="bot-1",
)
runner = build_runner(repo)
result = runner.invoke(
runner.app,
["--format", "json", "bot", "token", "revoke", "bot-1", token.id, "--force"],
)
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["token_id"] == token.id
assert payload["revoked"] is True