feat(REQ-012): implement service token rotation CLI

- Add service token rotation CLI command with force flag support
- Implement token rotation in auth service with expiration handling
- Add portfolio repository helper for transaction/batch operations
- Add cascade delete support in postgres repository
- Add comprehensive CLI tests for service token operations
- Add REQ-012 specification for service token rotation
- Create BAT-002 delivery batch for token rotation implementation
- Update SRS and batch indexes
This commit is contained in:
2026-05-02 12:43:43 +00:00
parent 8370d998d2
commit b4d964253b
12 changed files with 492 additions and 3 deletions
+33
View File
@@ -114,3 +114,36 @@ def revoke_service_token(
revoked_at=revoked_at,
)
render_service_token_result(ctx.obj.console, ctx.obj.output_format, response)
@token_app.command("rotate")
def rotate_service_token(
ctx: typer.Context,
name: str = typer.Argument(..., help="Service token name to rotate."),
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 _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,
token=issued.token,
)
if ctx.obj.output_format == "json":
from app.cli.core.utils import render_output
render_output(ctx.obj.console, "json", response.model_dump(mode="json"))
else:
ctx.obj.console.print("[bold yellow]⚠ 갱신된 토큰 원문은 지금 1회만 표시됩니다. 안전한 곳에 저장하세요.[/bold yellow]")
ctx.obj.console.print(f"[bold green]token:[/bold green] {issued.token}")
render_service_token_result(ctx.obj.console, ctx.obj.output_format, response)
+1 -1
View File
@@ -172,7 +172,7 @@ def print_service_token_list(console: Console, tokens: list[AuthTokenResponse])
created_at = token.created_at.isoformat(timespec="seconds") if token.created_at else "N/A"
lines = [
f"[bold cyan]{token.name}[/bold cyan] ({status})",
f" token : {token.token_prefix}",
f" token_prefix: {token.token_prefix} [dim](full token shown only at issue time)[/dim]",
f" id : {token.id}",
f" user_id : {token.user_id}",
f" perm : {token.permission}",
+8
View File
@@ -65,6 +65,14 @@ class PortfolioRepository(Protocol):
def revoke_auth_token(self, user_id: str, token_id: str) -> None: ...
def rotate_service_token(
self,
token_id: str,
token_hash: str,
token_prefix: str,
runtime_id: str,
) -> AuthToken: ...
def get_service_token_by_name(self, name: str) -> AuthToken | None: ...
def get_token_principal(self, token_hash: str) -> TokenPrincipal | None: ...
+25
View File
@@ -600,6 +600,31 @@ class PostgresPortfolioRepository:
if row is None:
raise ValueError("토큰을 찾을 수 없습니다")
def rotate_service_token(
self,
token_id: str,
token_hash: str,
token_prefix: str,
runtime_id: str,
) -> AuthToken:
self.ensure_schema()
with self._connect().cursor() as cur:
cur.execute(
"""
UPDATE auth_tokens
SET token_hash = %s, token_prefix = %s, runtime_id = %s
WHERE id = %s AND token_type = 'service' AND revoked_at IS NULL
RETURNING
id, user_id, name, token_prefix, permission, token_type,
expires_at, last_used_at, revoked_at, created_at, runtime_id
""",
(token_hash, token_prefix, runtime_id, token_id),
)
row = cur.fetchone()
if row is None:
raise ValueError("토큰을 찾을 수 없습니다")
return self._row_to_auth_token(row)
def get_service_token_by_name(self, name: str) -> AuthToken | None:
self.ensure_schema()
with self._connect().cursor() as cur:
+18
View File
@@ -155,6 +155,24 @@ class AuthService:
self._repository.revoke_auth_token(existing.user_id, existing.id)
return datetime.now(timezone.utc)
def rotate_service_token(self, name: str) -> IssuedServiceToken:
existing = self._repository.get_service_token_by_name(name)
if existing is None:
raise ValueError(f"서비스 토큰을 찾을 수 없습니다: {name}")
if existing.revoked_at is not None:
raise ValueError(f"이미 폐기된 서비스 토큰입니다: {name}")
if existing.expires_at is not None:
raise ValueError(f"만료일이 설정된 토큰은 rotate할 수 없습니다: {name}. revoke 후 재발급하세요.")
raw_token = generate_token()
new_runtime_id = str(uuid.uuid4())
auth_token = self._repository.rotate_service_token(
existing.id,
token_hash=hash_token(raw_token),
token_prefix=token_prefix(raw_token),
runtime_id=new_runtime_id,
)
return IssuedServiceToken(token=raw_token, auth_token=auth_token)
def list_service_tokens(
self,
username: str | None = None,