portfolio-hub/app/cli/commands/service_tokens.py

"""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, ServiceTokenRebindCliResponse, 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,
                daemon_id=t.daemon_id,
                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."),
    daemon_id: Optional[str] = typer.Option(None, "--daemon-id", help="Daemon ID to bind to this token."),
) -> 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,
            daemon_id=daemon_id,
        )

    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,
        daemon_id=issued.auth_token.daemon_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("rotate")
def rotate_service_token(
    ctx: typer.Context,
    name: str = typer.Argument(..., help="Service token name to rotate."),
) -> None:
    """Rotate a service token: revoke old and issue new with same daemon_id binding."""
    def _rotate(service: AuthService, app_context):
        return service.rotate_service_token(name)

    issued = _with_service(ctx, _rotate)
    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,
        daemon_id=issued.auth_token.daemon_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("rebind")
def rebind_service_token(
    ctx: typer.Context,
    name: str = typer.Argument(..., help="Service token name to rebind."),
    daemon_id: str = typer.Option(..., "--daemon-id", help="Existing daemon ID to bind to this token."),
) -> None:
    """Rebind an active service token to an existing daemon."""

    def _rebind(service: AuthService, app_context):
        return service.rebind_service_token(name, daemon_id)

    previous_daemon_id, rebound = _with_service(ctx, _rebind)
    response = ServiceTokenRebindCliResponse(
        token_id=rebound.id,
        name=name,
        expires_at=rebound.expires_at,
        revoked=rebound.revoked_at is not None,
        runtime_id=rebound.runtime_id,
        daemon_id=rebound.daemon_id,
        previous_daemon_id=previous_daemon_id,
    )
    render_service_token_result(ctx.obj.console, ctx.obj.output_format, response)


@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)


@token_app.command("migrate-null-daemon")
def migrate_null_daemon(
    ctx: typer.Context,
    dry_run: bool = typer.Option(False, "--dry-run", help="Inspect NULL daemon tokens without making changes."),
) -> None:
    """Diagnose and report active service tokens with no daemon binding (NULL daemon_id)."""
    if not dry_run:
        ctx.obj.console.print(
            "[red]자동 변경은 지원하지 않습니다.[/red] "
            "먼저 --dry-run으로 inventory를 확인한 후 "
            "token별로 `pfhub service-token rebind NAME --daemon-id ID` 또는 "
            "`pfhub service-token revoke NAME`을 수동 실행하세요."
        )
        raise typer.Exit(code=1)

    def _inspect(service: AuthService, app_context):
        tokens = service.list_null_daemon_service_tokens()
        if not tokens:
            app_context.console.print("no NULL daemon tokens found.")
            return

        for token in tokens:
            candidates = service.find_candidate_daemons_for_user(token.user_id)
            candidate_ids = [d.id for d in candidates]
            match_note = f"candidates={candidate_ids}" if candidate_ids else "[yellow]no candidate daemons for this user[/yellow]"
            app_context.console.print(
                f"id={token.id} user_id={token.user_id} name={token.name} "
                f"runtime_id={token.runtime_id} created_at={token.created_at} {match_note}"
            )

    _with_service(ctx, _inspect)