genichinandClaude Sonnet 4.6
c815ce39b6
BAT-006 (REQ-016/017/018): - client token 발급 경로 제거 (issue_client_token, POST /auth/tokens, 관련 스키마/테스트) - service-token rotate unique index revoked_at IS NULL 조건 추가 (Alembic CONCURRENTLY 마이그레이션) - POST /auth/service-tokens body daemon_id forward 누락 보정 (1-line fix) BAT-007 (REQ-022/023): - daemon deactivate 시 바인딩 service token 자동 revoke (CLI --force 옵션 + revoke_tokens_by_daemon_id) - NULL daemon_id legacy token 진단 CLI (migrate-null-daemon --dry-run) 거버넌스 문서: dcy-003/004 Discovery, REQ-016~023/025~032 신규, BAT-006/007 전체 산출물 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
255 lines
9.1 KiB
Python
255 lines
9.1 KiB
Python
"""REQ-023: NULL daemon_id legacy service token 마이그레이션 CLI 검증 (AC-1~5)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from app.cli.main import create_app
|
|
from app.core.config import Settings
|
|
from app.domain.models import AuthToken, Daemon, User
|
|
from app.services.auth_service import AuthService
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fake repository
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class _FakeRepo:
|
|
def __post_init__(self) -> None:
|
|
self._users: dict[str, User] = {}
|
|
self._tokens: dict[str, AuthToken] = {}
|
|
self._daemons: dict[str, Daemon] = {}
|
|
self._seq = 1
|
|
|
|
def ping(self) -> None:
|
|
pass
|
|
|
|
def ensure_schema(self) -> None:
|
|
pass
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def get_user_by_username(self, username: str) -> User | None:
|
|
return next((u for u in self._users.values() if u.username == username), None)
|
|
|
|
def get_user(self, user_id: str) -> User:
|
|
u = self._users.get(user_id)
|
|
if u is None:
|
|
raise ValueError("사용자를 찾을 수 없습니다")
|
|
return u
|
|
|
|
def create_auth_token(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
name: str,
|
|
token_hash: str,
|
|
token_prefix: str,
|
|
permission: str,
|
|
token_type: str,
|
|
expires_at: datetime | None = None,
|
|
runtime_id: str | None = None,
|
|
daemon_id: str | None = None,
|
|
) -> AuthToken:
|
|
token = AuthToken(
|
|
id=f"token-{self._seq}",
|
|
user_id=user_id,
|
|
name=name,
|
|
token_prefix=token_prefix,
|
|
permission=permission,
|
|
token_type=token_type,
|
|
expires_at=expires_at,
|
|
created_at=datetime.now(timezone.utc),
|
|
runtime_id=runtime_id,
|
|
daemon_id=daemon_id,
|
|
)
|
|
self._seq += 1
|
|
self._tokens[token.id] = token
|
|
return token
|
|
|
|
def get_service_token_by_name(self, name: str) -> AuthToken | None:
|
|
return next(
|
|
(t for t in self._tokens.values()
|
|
if t.name == name and t.token_type == "service" and t.revoked_at is None),
|
|
None,
|
|
)
|
|
|
|
def revoke_auth_token(self, user_id: str, token_id: str) -> None:
|
|
token = self._tokens.get(token_id)
|
|
if token is None or token.user_id != user_id:
|
|
raise ValueError("토큰을 찾을 수 없습니다")
|
|
token.revoked_at = datetime.now(timezone.utc)
|
|
|
|
def list_service_tokens_with_null_daemon(self) -> list[AuthToken]:
|
|
return [
|
|
t for t in self._tokens.values()
|
|
if t.token_type == "service" and t.daemon_id is None and t.revoked_at is None
|
|
]
|
|
|
|
def list_daemons_for_user(self, owner_user_id: str) -> list[Daemon]:
|
|
return [d for d in self._daemons.values() if d.owner_user_id == owner_user_id]
|
|
|
|
def rebind_service_token(self, name: str, daemon_id: str) -> AuthToken:
|
|
token = self.get_service_token_by_name(name)
|
|
if token is None:
|
|
raise ValueError(f"서비스 토큰을 찾을 수 없습니다: {name}")
|
|
daemon = self._daemons.get(daemon_id)
|
|
if daemon is None or not daemon.is_active:
|
|
raise ValueError(f"존재하지 않거나 비활성 데몬입니다: {daemon_id}")
|
|
if daemon.owner_user_id != token.user_id:
|
|
raise ValueError("데몬이 해당 토큰 사용자에 속하지 않습니다")
|
|
token.daemon_id = daemon_id
|
|
return token
|
|
|
|
def revoke_tokens_by_daemon_id(self, daemon_id: str) -> int:
|
|
count = 0
|
|
for t in self._tokens.values():
|
|
if t.daemon_id == daemon_id and t.revoked_at is None:
|
|
t.revoked_at = datetime.now(timezone.utc)
|
|
count += 1
|
|
return count
|
|
|
|
def get_daemon_by_id(self, daemon_id: str) -> Daemon | None:
|
|
return self._daemons.get(daemon_id)
|
|
|
|
|
|
def _make_runner(repo: _FakeRepo) -> CliRunner:
|
|
app = create_app(
|
|
settings_factory=lambda: Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000"),
|
|
repository_factory=lambda _: repo,
|
|
)
|
|
runner = CliRunner()
|
|
runner.app = app
|
|
return runner
|
|
|
|
|
|
def _add_null_daemon_token(repo: _FakeRepo, name: str = "orphan-tok", user_id: str = "user-1") -> AuthToken:
|
|
from app.core.security import generate_token, hash_token, token_prefix
|
|
raw = generate_token()
|
|
return repo.create_auth_token(
|
|
user_id,
|
|
name=name,
|
|
token_hash=hash_token(raw),
|
|
token_prefix=token_prefix(raw),
|
|
permission="read",
|
|
token_type="service",
|
|
daemon_id=None,
|
|
runtime_id="rt-1",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-023 AC-1: dry-run with NULL daemon tokens → inventory 출력
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migrate_null_daemon_dry_run_shows_inventory() -> None:
|
|
repo = _FakeRepo()
|
|
repo._users["user-1"] = User(id="user-1", username="alice", permission="read")
|
|
token = _add_null_daemon_token(repo, name="orphan-tok", user_id="user-1")
|
|
runner = _make_runner(repo)
|
|
|
|
result = runner.invoke(runner.app, ["service-token", "migrate-null-daemon", "--dry-run"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "orphan-tok" in result.output
|
|
assert token.id in result.output or "user-1" in result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-023 AC-2: dry-run with 0 NULL daemon tokens → "no NULL daemon tokens"
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migrate_null_daemon_dry_run_no_tokens_prints_empty_message() -> None:
|
|
repo = _FakeRepo()
|
|
runner = _make_runner(repo)
|
|
|
|
result = runner.invoke(runner.app, ["service-token", "migrate-null-daemon", "--dry-run"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "no NULL daemon tokens" in result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-023 AC-3: --dry-run 없이 실행 → non-zero exit + 수동 안내
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migrate_null_daemon_without_dry_run_exits_nonzero() -> None:
|
|
repo = _FakeRepo()
|
|
runner = _make_runner(repo)
|
|
|
|
result = runner.invoke(runner.app, ["service-token", "migrate-null-daemon"])
|
|
|
|
assert result.exit_code != 0
|
|
assert "rebind" in result.output or "revoke" in result.output or "dry-run" in result.output
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-023 AC-4: rebind 후 dry-run에서 해당 token 제외
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migrate_null_daemon_after_rebind_token_excluded_from_inventory() -> None:
|
|
repo = _FakeRepo()
|
|
repo._users["user-1"] = User(id="user-1", username="alice", permission="read")
|
|
daemon = Daemon(
|
|
id="daemon-1",
|
|
owner_user_id="user-1",
|
|
daemon_slug="alpha",
|
|
is_active=True,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
repo._daemons["daemon-1"] = daemon
|
|
_add_null_daemon_token(repo, name="orphan-tok", user_id="user-1")
|
|
service = AuthService(repo, Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000"))
|
|
service.rebind_service_token("orphan-tok", "daemon-1")
|
|
|
|
null_tokens = repo.list_service_tokens_with_null_daemon()
|
|
|
|
assert null_tokens == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-023 AC-5: revoke 후 dry-run에서 해당 token 제외
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migrate_null_daemon_after_revoke_token_excluded_from_inventory() -> None:
|
|
repo = _FakeRepo()
|
|
repo._users["user-1"] = User(id="user-1", username="alice", permission="read")
|
|
token = _add_null_daemon_token(repo, name="admin-fix", user_id="user-1")
|
|
service = AuthService(repo, Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000"))
|
|
service.revoke_service_token("admin-fix")
|
|
|
|
null_tokens = repo.list_service_tokens_with_null_daemon()
|
|
|
|
assert null_tokens == []
|
|
assert repo._tokens[token.id].revoked_at is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-023: dry-run 출력에 매핑 후보 daemon 포함 여부
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migrate_null_daemon_dry_run_shows_candidate_daemon() -> None:
|
|
repo = _FakeRepo()
|
|
repo._users["user-1"] = User(id="user-1", username="alice", permission="read")
|
|
daemon = Daemon(
|
|
id="daemon-1",
|
|
owner_user_id="user-1",
|
|
daemon_slug="alpha",
|
|
is_active=True,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
repo._daemons["daemon-1"] = daemon
|
|
_add_null_daemon_token(repo, name="orphan-tok", user_id="user-1")
|
|
runner = _make_runner(repo)
|
|
|
|
result = runner.invoke(runner.app, ["service-token", "migrate-null-daemon", "--dry-run"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "daemon-1" in result.output
|