작업 내용 저장
This commit is contained in:
@@ -1,115 +0,0 @@
|
||||
"""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)
|
||||
@@ -0,0 +1,79 @@
|
||||
"""Service token CLI commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import typer
|
||||
|
||||
from app.cli.core.output import mask_token_text, render_service_token_result
|
||||
from app.schemas.auth import ServiceTokenIssueCliResponse, ServiceTokenRevokeCliResponse
|
||||
from app.services.auth_service import AuthService
|
||||
|
||||
token_app = typer.Typer(help="Service token lifecycle commands.")
|
||||
|
||||
|
||||
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:
|
||||
from app.cli.core.utils import execute_command
|
||||
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_service_token(
|
||||
ctx: typer.Context,
|
||||
name: str = typer.Argument(..., help="Unique service token name."),
|
||||
user: str = typer.Option(..., "--user", help="Username to associate with this service token."),
|
||||
permission: str = typer.Option("read", "--permission", help="Token permission: read or trade-write."),
|
||||
ttl_days: int | None = typer.Option(None, "--ttl-days", help="Token TTL in days. Default: 365."),
|
||||
no_expiry: bool = typer.Option(False, "--no-expiry", help="Issue a token with no expiry."),
|
||||
) -> None:
|
||||
def _issue(service: AuthService, app_context):
|
||||
return service.issue_service_token(
|
||||
name,
|
||||
user,
|
||||
permission=permission,
|
||||
expires_in_days=ttl_days,
|
||||
no_expiry=no_expiry,
|
||||
)
|
||||
|
||||
issued = _with_service(ctx, _issue)
|
||||
response = ServiceTokenIssueCliResponse(
|
||||
token_id=issued.auth_token.id,
|
||||
name=name,
|
||||
expires_at=issued.auth_token.expires_at,
|
||||
revoked=issued.auth_token.revoked_at is not None,
|
||||
token=issued.token,
|
||||
)
|
||||
render_service_token_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_service_token(
|
||||
ctx: typer.Context,
|
||||
name: str = typer.Argument(..., help="Service token name to revoke."),
|
||||
force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
if not force:
|
||||
confirmed = typer.confirm(f"서비스 토큰 '{name}'을 폐기하시겠습니까?")
|
||||
if not confirmed:
|
||||
raise typer.Exit(code=0)
|
||||
|
||||
def _revoke(service: AuthService, app_context):
|
||||
return service.revoke_service_token(name)
|
||||
|
||||
revoked_at = _with_service(ctx, _revoke)
|
||||
response = ServiceTokenRevokeCliResponse(
|
||||
token_id="",
|
||||
name=name,
|
||||
revoked=True,
|
||||
revoked_at=revoked_at,
|
||||
)
|
||||
render_service_token_result(ctx.obj.console, ctx.obj.output_format, response)
|
||||
Reference in New Issue
Block a user