diff --git a/app/cli/commands/service_tokens.py b/app/cli/commands/service_tokens.py index 8307121..83206fb 100644 --- a/app/cli/commands/service_tokens.py +++ b/app/cli/commands/service_tokens.py @@ -2,10 +2,13 @@ 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.schemas.auth import ServiceTokenIssueCliResponse, ServiceTokenRevokeCliResponse +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.") @@ -23,6 +26,34 @@ def _with_service(ctx: typer.Context, callback): 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, diff --git a/app/cli/core/utils.py b/app/cli/core/utils.py index 7817c93..a6b436c 100644 --- a/app/cli/core/utils.py +++ b/app/cli/core/utils.py @@ -13,6 +13,7 @@ from rich.panel import Panel from rich.table import Table from app.core.errors import FeatureNotReadyError +from app.schemas.auth import AuthTokenResponse from app.schemas.portfolio import CommentResponse, PortfolioDetailResponse, PortfolioListItem from app.schemas.user import UserResponse @@ -162,6 +163,28 @@ def print_user_detail(console: Console, user: UserResponse) -> None: ) +def print_service_token_table(console: Console, tokens: list[AuthTokenResponse]) -> None: + table = Table(title="Service Tokens", expand=True) + table.add_column("Name", style="cyan", no_wrap=True) + table.add_column("Permission", no_wrap=True) + table.add_column("User ID", no_wrap=True) + table.add_column("Prefix", no_wrap=True) + table.add_column("Expires At", no_wrap=True) + table.add_column("Revoked At", no_wrap=True) + table.add_column("Created At", no_wrap=True) + for token in tokens: + table.add_row( + token.name, + token.permission, + token.user_id, + token.token_prefix, + token.expires_at.isoformat(timespec="seconds") if token.expires_at else "never", + token.revoked_at.isoformat(timespec="seconds") if token.revoked_at else "-", + token.created_at.isoformat(timespec="seconds") if token.created_at else "N/A", + ) + console.print(table) + + def print_comment_table(console: Console, comments: list[CommentResponse]) -> None: table = Table(title="Portfolio Comments", expand=True) table.add_column("ID", style="cyan", no_wrap=True) @@ -219,6 +242,9 @@ def render_output(console: Console, output_format: str, payload: Any, *, empty_m if all(isinstance(item, CommentResponse) for item in payload): print_comment_table(console, payload) return + if all(isinstance(item, AuthTokenResponse) for item in payload): + print_service_token_table(console, payload) + return return if isinstance(payload, PortfolioDetailResponse): diff --git a/app/repositories/portfolio_repository.py b/app/repositories/portfolio_repository.py index cf21876..441bd43 100644 --- a/app/repositories/portfolio_repository.py +++ b/app/repositories/portfolio_repository.py @@ -56,6 +56,13 @@ class PortfolioRepository(Protocol): def list_auth_tokens(self, user_id: str) -> list[AuthToken]: ... + def list_service_tokens( + self, + user_id: str | None = None, + *, + include_revoked: bool = False, + ) -> list[AuthToken]: ... + def revoke_auth_token(self, user_id: str, token_id: str) -> None: ... def get_service_token_by_name(self, name: str) -> AuthToken | None: ... diff --git a/app/repositories/postgres.py b/app/repositories/postgres.py index da6aa4d..e9fa057 100644 --- a/app/repositories/postgres.py +++ b/app/repositories/postgres.py @@ -564,6 +564,26 @@ class PostgresPortfolioRepository: rows = cur.fetchall() return [self._row_to_auth_token(row) for row in rows] + def list_service_tokens( + self, + user_id: str | None = None, + *, + include_revoked: bool = False, + ) -> list[AuthToken]: + self.ensure_schema() + conditions = ["token_type = 'service'"] + params: list = [] + if user_id is not None: + conditions.append("user_id = %s") + params.append(user_id) + if not include_revoked: + conditions.append("revoked_at IS NULL") + where = "WHERE " + " AND ".join(conditions) + " " + with self._connect().cursor() as cur: + cur.execute(_SELECT_AUTH_TOKEN + where + "ORDER BY created_at DESC, id ASC", params) + rows = cur.fetchall() + return [self._row_to_auth_token(row) for row in rows] + def revoke_auth_token(self, user_id: str, token_id: str) -> None: self.ensure_schema() with self._connect().cursor() as cur: diff --git a/app/services/auth_service.py b/app/services/auth_service.py index 56492a4..be201b5 100644 --- a/app/services/auth_service.py +++ b/app/services/auth_service.py @@ -152,6 +152,20 @@ class AuthService: self._repository.revoke_auth_token(existing.user_id, existing.id) return datetime.now(timezone.utc) + def list_service_tokens( + self, + username: str | None = None, + *, + include_revoked: bool = False, + ) -> list[AuthToken]: + user_id: str | None = None + if username is not None: + user = self._repository.get_user_by_username(username) + if user is None: + raise ValueError(f"사용자를 찾을 수 없습니다: {username}") + user_id = user.id + return self._repository.list_service_tokens(user_id, include_revoked=include_revoked) + @staticmethod def _build_expiration(expires_in_days: int | None, default_days: int) -> datetime: days = normalize_expiry_days(expires_in_days, default_days=default_days) diff --git a/docs/qsdlc/qsdlc-001_20260423_add-pfhub-service-token-list.md b/docs/qsdlc/qsdlc-001_20260423_add-pfhub-service-token-list.md new file mode 100644 index 0000000..a80740a --- /dev/null +++ b/docs/qsdlc/qsdlc-001_20260423_add-pfhub-service-token-list.md @@ -0,0 +1,72 @@ + +## Plan: `pfhub service-token list` 명령 추가 + +`pfhub service-token list` 명령을 추가한다. 레포지토리 → 서비스 → 출력 → CLI 순서로 구현하며, 기존 `user list` 패턴을 템플릿으로 사용한다. + +--- + +### Phase 1 — Repository 레이어 + +**1.** portfolio_repository.py — `PortfolioRepository` 프로토콜에 추상 메서드 추가: +``` +list_service_tokens(user_id: str | None, *, include_revoked: bool) -> list[AuthToken] +``` + +**2.** postgres.py — `PostgresRepository`에 구현체 추가: +- 기본 쿼리: `SELECT … FROM auth_tokens WHERE token_type='service'` +- `user_id` 지정 시 `AND user_id = %s` 조건 추가 +- `include_revoked=False`(기본)이면 `AND revoked_at IS NULL` 조건 추가 +- `ORDER BY created_at DESC` +- 기존 `list_auth_tokens()` 의 `_row_to_auth_token()` 재사용 + +--- + +### Phase 2 — Service 레이어 + +**3.** auth_service.py — `AuthService`에 메서드 추가: +``` +list_service_tokens(username: str | None, *, include_revoked: bool = False) -> list[AuthToken] +``` +- `username` 지정 시 `get_user_by_username()` → `user_id` 변환 후 리포지토리 위임 +- `username` 없으면 `user_id=None` 으로 전체 조회 + +--- + +### Phase 3 — Schema / 출력 레이어 + +**4.** auth.py — 기존 `AuthTokenResponse` 재사용 (이미 token_hash 미포함, 안전) + +**5.** utils.py — `print_service_token_table(console, tokens)` 함수 추가: +- Rich Table 컬럼: `name`, `permission`, `user_id`, `expires_at`, `revoked_at`, `created_at`, `token_prefix` +- `render_output()` dispatch에 `list[AuthTokenResponse]` 분기 추가 + +--- + +### Phase 4 — CLI 레이어 + +**6.** service_tokens.py — `@token_app.command("list")` 추가: +- `--user ` (Optional[str], default=None) — 특정 사용자 토큰만 조회 +- `--show-revoked` (bool flag, default=False) — revoked 토큰 포함 +- `_with_service()` + lambda 패턴 (user.py 참조) +- `render_output(..., empty_message="No service tokens found.")` + +--- + +### Verification + +1. `pfhub service-token list` → 활성 서비스 토큰 테이블 출력 +2. `pfhub service-token list --show-revoked` → revoked 토큰 포함 출력 +3. `pfhub service-token list --user alice` → alice 소유 토큰만 필터링 +4. `pfhub service-token list --output json` → JSON 포맷 출력 +5. 토큰 없을 때 → `"No service tokens found."` 메시지 출력 +6. test_cli.py 또는 test_auth.py에 신규 테스트 추가 후 전체 테스트 통과 확인 + +--- + +### Decisions + +- revoked 토큰: 기본 제외, `--show-revoked` 옵션으로 포함 +- `--user` 없으면 전체 서비스 토큰 조회 +- 스키마: `AuthTokenResponse` 재사용 (신규 모델 불필요) +- `token_hash`는 절대 노출하지 않음 + diff --git a/tests/test_cli.py b/tests/test_cli.py index 2fcaa20..88573cd 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -143,6 +143,20 @@ class FakeRepo: raise ValueError("토큰을 찾을 수 없습니다") token.revoked_at = datetime.now(timezone.utc) + def list_service_tokens( + self, + user_id: str | None = None, + *, + include_revoked: bool = False, + ) -> list[AuthToken]: + results = [ + t for t in self._auth_tokens.values() + if t.token_type == "service" + and (user_id is None or t.user_id == user_id) + and (include_revoked or t.revoked_at is None) + ] + return sorted(results, key=lambda t: t.created_at or datetime.min, reverse=True) + def get(self, portfolio_id: str) -> Portfolio: if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id: raise ValueError("포트폴리오를 찾을 수 없습니다") @@ -906,3 +920,125 @@ def test_cli_service_token_revoke_force_succeeds() -> None: payload = json.loads(result.stdout) assert payload["name"] == "to-revoke" assert payload["revoked"] is True + + +def test_cli_service_token_list_returns_active_tokens() -> None: + repo = FakeRepo(_portfolio()) + user = repo.create_user("alice", permission="read") + repo.create_auth_token( + user.id, + name="active-token", + token_hash=hash_token("tok-1"), + token_prefix=token_prefix("tok-1"), + permission="read", + token_type="service", + ) + runner = build_runner(repo) + + result = runner.invoke(runner.app, ["--format", "json", "service-token", "list"]) + + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert isinstance(payload, list) + assert len(payload) == 1 + assert payload[0]["name"] == "active-token" + + +def test_cli_service_token_list_excludes_revoked_by_default() -> None: + repo = FakeRepo(_portfolio()) + user = repo.create_user("alice", permission="read") + token = repo.create_auth_token( + user.id, + name="old-token", + token_hash=hash_token("tok-2"), + token_prefix=token_prefix("tok-2"), + permission="read", + token_type="service", + ) + repo.revoke_auth_token(user.id, token.id) + runner = build_runner(repo) + + result = runner.invoke(runner.app, ["--format", "json", "service-token", "list"]) + + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert payload == [] + + +def test_cli_service_token_list_show_revoked_includes_revoked() -> None: + repo = FakeRepo(_portfolio()) + user = repo.create_user("alice", permission="read") + token = repo.create_auth_token( + user.id, + name="old-token", + token_hash=hash_token("tok-3"), + token_prefix=token_prefix("tok-3"), + permission="read", + token_type="service", + ) + repo.revoke_auth_token(user.id, token.id) + runner = build_runner(repo) + + result = runner.invoke( + runner.app, ["--format", "json", "service-token", "list", "--show-revoked"] + ) + + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert len(payload) == 1 + assert payload[0]["name"] == "old-token" + assert payload[0]["revoked_at"] is not None + + +def test_cli_service_token_list_filter_by_user() -> None: + repo = FakeRepo(_portfolio()) + alice = repo.create_user("alice", permission="read") + bob = repo.create_user("bob", permission="read") + repo.create_auth_token( + alice.id, + name="alice-token", + token_hash=hash_token("tok-a"), + token_prefix=token_prefix("tok-a"), + permission="read", + token_type="service", + ) + repo.create_auth_token( + bob.id, + name="bob-token", + token_hash=hash_token("tok-b"), + token_prefix=token_prefix("tok-b"), + permission="read", + token_type="service", + ) + runner = build_runner(repo) + + result = runner.invoke( + runner.app, ["--format", "json", "service-token", "list", "--user", "alice"] + ) + + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert len(payload) == 1 + assert payload[0]["name"] == "alice-token" + + +def test_cli_service_token_list_unknown_user_exits_error() -> None: + repo = FakeRepo(_portfolio()) + runner = build_runner(repo) + + result = runner.invoke( + runner.app, ["service-token", "list", "--user", "nobody"] + ) + + assert result.exit_code == 1 + assert "nobody" in result.stdout + + +def test_cli_service_token_list_empty_message() -> None: + repo = FakeRepo(_portfolio()) + runner = build_runner(repo) + + result = runner.invoke(runner.app, ["service-token", "list"]) + + assert result.exit_code == 0 + assert "No service tokens found." in result.stdout