genichin
ce63c75c9c
- Add V1 API endpoints: assets, ledger, user-info - Implement Fernet-based credential payload encryption (req-010) - Extend health response schema with runtime info (req-004) - Add service token runtime ID issuance (req-005) - Implement database schema migration for V1 API (req-009) - Add comprehensive test coverage for V1 endpoints - Update SRS with 10 new requirements (REQ-004 through REQ-011) - Create batch planning, design, and verification for BAT-001 - Integrate review-discovery StagePilot skill - Update documentation indices and baseline specs
112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
"""Service token CLI commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
import typer
|
|
|
|
from app.cli.core.output import mask_token_text, render_service_token_result
|
|
from app.cli.core.utils import render_output
|
|
from app.schemas.auth import AuthTokenResponse, 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("list")
|
|
def list_service_tokens(
|
|
ctx: typer.Context,
|
|
user: Optional[str] = typer.Option(None, "--user", help="Filter by username."),
|
|
show_revoked: bool = typer.Option(False, "--show-revoked", help="Include revoked tokens."),
|
|
) -> None:
|
|
def _list(service: AuthService, app_context):
|
|
tokens = service.list_service_tokens(user, include_revoked=show_revoked)
|
|
return [
|
|
AuthTokenResponse(
|
|
id=t.id,
|
|
user_id=t.user_id,
|
|
name=t.name,
|
|
token_prefix=t.token_prefix,
|
|
permission=t.permission,
|
|
token_type=t.token_type,
|
|
expires_at=t.expires_at,
|
|
last_used_at=t.last_used_at,
|
|
revoked_at=t.revoked_at,
|
|
created_at=t.created_at,
|
|
)
|
|
for t in tokens
|
|
]
|
|
|
|
result = _with_service(ctx, _list)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, result, empty_message="No service tokens found.")
|
|
|
|
|
|
@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,
|
|
runtime_id=issued.auth_token.runtime_id,
|
|
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)
|