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>
416 lines
15 KiB
Python
416 lines
15 KiB
Python
"""Tests for daemon registry CRUD and service-token daemon binding."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
import pytest
|
|
|
|
from app.domain.models import AuthToken, Daemon, User
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
from app.services.auth_service import AuthService
|
|
from app.core.config import Settings
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Minimal in-memory repo for daemon registry tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class _DaemonRepo:
|
|
def __post_init__(self) -> None:
|
|
self._users: dict[str, User] = {}
|
|
self._tokens: dict[str, AuthToken] = {}
|
|
self._daemons: dict[str, Daemon] = {}
|
|
self._daemons_by_slug: dict[tuple[str, str], Daemon] = {}
|
|
self._seq = 1
|
|
self._daemon_seq = 1
|
|
|
|
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 sorted(
|
|
self._tokens.values(),
|
|
key=lambda token: token.created_at or datetime.min.replace(tzinfo=timezone.utc),
|
|
reverse=True,
|
|
)
|
|
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 get_daemon_by_id(self, daemon_id: str) -> Daemon | None:
|
|
return self._daemons.get(daemon_id)
|
|
|
|
def get_daemon_by_slug(self, owner_user_id: str, daemon_slug: str) -> Daemon | None:
|
|
return self._daemons_by_slug.get((owner_user_id, daemon_slug))
|
|
|
|
def create_daemon(self, owner_user_id: str, *, daemon_slug: str, display_name: str | None = None) -> Daemon:
|
|
key = (owner_user_id, daemon_slug)
|
|
if key in self._daemons_by_slug:
|
|
raise ValueError(f"이미 존재하는 daemon slug입니다: {daemon_slug}")
|
|
daemon = Daemon(
|
|
id=f"daemon-{self._daemon_seq}",
|
|
owner_user_id=owner_user_id,
|
|
daemon_slug=daemon_slug,
|
|
display_name=display_name,
|
|
is_active=True,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
self._daemon_seq += 1
|
|
self._daemons[daemon.id] = daemon
|
|
self._daemons_by_slug[key] = daemon
|
|
return daemon
|
|
|
|
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 update_daemon(self, daemon_id: str, *, display_name: str | None = None, is_active: bool | None = None) -> Daemon:
|
|
daemon = self._daemons.get(daemon_id)
|
|
if daemon is None:
|
|
raise ValueError("데몬을 찾을 수 없습니다")
|
|
updated = Daemon(
|
|
id=daemon.id,
|
|
owner_user_id=daemon.owner_user_id,
|
|
daemon_slug=daemon.daemon_slug,
|
|
display_name=display_name if display_name is not None else daemon.display_name,
|
|
is_active=is_active if is_active is not None else daemon.is_active,
|
|
created_at=daemon.created_at,
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
self._daemons[daemon_id] = updated
|
|
self._daemons_by_slug[(updated.owner_user_id, updated.daemon_slug)] = updated
|
|
return updated
|
|
|
|
def rotate_service_token(self, name: str) -> tuple[str, AuthToken]:
|
|
existing = self.get_service_token_by_name(name)
|
|
if existing is None:
|
|
raise ValueError(f"서비스 토큰을 찾을 수 없습니다: {name}")
|
|
existing.revoked_at = datetime.now(timezone.utc)
|
|
import uuid as _uuid
|
|
from app.core.security import generate_token, hash_token, token_prefix as _tp
|
|
raw = generate_token()
|
|
new_token = AuthToken(
|
|
id=f"token-{self._seq}",
|
|
user_id=existing.user_id,
|
|
name=name,
|
|
token_prefix=_tp(raw),
|
|
permission=existing.permission,
|
|
token_type="service",
|
|
expires_at=existing.expires_at,
|
|
created_at=datetime.now(timezone.utc),
|
|
runtime_id=str(_uuid.uuid4()),
|
|
daemon_id=existing.daemon_id,
|
|
)
|
|
self._seq += 1
|
|
self._tokens[new_token.id] = new_token
|
|
return raw, new_token
|
|
|
|
def rebind_service_token(self, name: str, daemon_id: str) -> AuthToken:
|
|
existing = self.get_service_token_by_name(name)
|
|
if existing is None:
|
|
raise ValueError(f"서비스 토큰을 찾을 수 없습니다: {name}")
|
|
existing.daemon_id = daemon_id
|
|
return existing
|
|
|
|
def revoke_tokens_by_daemon_id(self, daemon_id: str) -> int:
|
|
count = 0
|
|
for token in self._tokens.values():
|
|
if token.daemon_id == daemon_id and token.revoked_at is None:
|
|
token.revoked_at = datetime.now(timezone.utc)
|
|
count += 1
|
|
return count
|
|
|
|
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 _repo_with_user(username: str = "alice") -> _DaemonRepo:
|
|
repo = _DaemonRepo()
|
|
repo._users["user-1"] = User(id="user-1", username=username, permission="read")
|
|
return repo
|
|
|
|
|
|
def _service(repo: _DaemonRepo) -> AuthService:
|
|
return AuthService(repo, Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-023: Daemon registry CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_create_daemon_returns_daemon_id() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
|
|
assert daemon.id is not None
|
|
assert daemon.daemon_slug == "alpha-bot"
|
|
assert daemon.owner_user_id == "user-1"
|
|
assert daemon.is_active is True
|
|
|
|
|
|
def test_create_daemon_with_display_name() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="beta-bot", display_name="Beta Trading Bot")
|
|
|
|
assert daemon.display_name == "Beta Trading Bot"
|
|
|
|
|
|
def test_update_daemon_display_name() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot", display_name="Old Name")
|
|
|
|
updated = repo.update_daemon(daemon.id, display_name="New Name")
|
|
|
|
assert updated.display_name == "New Name"
|
|
assert updated.daemon_slug == "alpha-bot"
|
|
|
|
|
|
def test_create_daemon_duplicate_slug_raises() -> None:
|
|
repo = _repo_with_user()
|
|
repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
|
|
with pytest.raises(ValueError, match="alpha-bot"):
|
|
repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
|
|
|
|
def test_list_daemons_scoped_to_user() -> None:
|
|
repo = _repo_with_user()
|
|
repo._users["user-2"] = User(id="user-2", username="bob", permission="read")
|
|
repo.create_daemon("user-1", daemon_slug="alice-bot")
|
|
repo.create_daemon("user-2", daemon_slug="bob-bot")
|
|
|
|
alice_daemons = repo.list_daemons_for_user("user-1")
|
|
bob_daemons = repo.list_daemons_for_user("user-2")
|
|
|
|
assert len(alice_daemons) == 1
|
|
assert alice_daemons[0].daemon_slug == "alice-bot"
|
|
assert len(bob_daemons) == 1
|
|
assert bob_daemons[0].daemon_slug == "bob-bot"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-005: service-token issue daemon binding
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_issue_service_token_with_daemon_id() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
service = _service(repo)
|
|
|
|
result = service.issue_service_token("my-token", "alice", daemon_id=daemon.id)
|
|
|
|
assert result.auth_token.daemon_id == daemon.id
|
|
|
|
|
|
def test_issue_service_token_unknown_daemon_raises() -> None:
|
|
repo = _repo_with_user()
|
|
service = _service(repo)
|
|
|
|
with pytest.raises(ValueError, match="존재하지 않거나 비활성"):
|
|
service.issue_service_token("my-token", "alice", daemon_id="nonexistent-daemon-id")
|
|
|
|
|
|
def test_issue_service_token_inactive_daemon_raises() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
repo.update_daemon(daemon.id, is_active=False)
|
|
service = _service(repo)
|
|
|
|
with pytest.raises(ValueError, match="존재하지 않거나 비활성"):
|
|
service.issue_service_token("my-token", "alice", daemon_id=daemon.id)
|
|
|
|
|
|
def test_issue_service_token_without_daemon_id_succeeds() -> None:
|
|
repo = _repo_with_user()
|
|
service = _service(repo)
|
|
|
|
result = service.issue_service_token("my-token", "alice")
|
|
|
|
assert result.auth_token.daemon_id is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-012: service-token rotate daemon_id invariant
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_rotate_service_token_preserves_daemon_id() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
service = _service(repo)
|
|
|
|
issued = service.issue_service_token("my-token", "alice", daemon_id=daemon.id)
|
|
original_daemon_id = issued.auth_token.daemon_id
|
|
original_runtime_id = issued.auth_token.runtime_id
|
|
|
|
rotated = service.rotate_service_token("my-token")
|
|
|
|
assert rotated.auth_token.daemon_id == original_daemon_id
|
|
assert rotated.auth_token.runtime_id != original_runtime_id
|
|
assert rotated.auth_token.name == "my-token"
|
|
|
|
|
|
def test_rotate_service_token_generates_new_raw_token() -> None:
|
|
repo = _repo_with_user()
|
|
service = _service(repo)
|
|
|
|
issued = service.issue_service_token("my-token", "alice")
|
|
rotated = service.rotate_service_token("my-token")
|
|
|
|
assert rotated.token != issued.token
|
|
|
|
|
|
def test_rotate_nonexistent_token_raises() -> None:
|
|
repo = _repo_with_user()
|
|
service = _service(repo)
|
|
|
|
with pytest.raises(ValueError, match="nonexistent"):
|
|
service.rotate_service_token("nonexistent")
|
|
|
|
|
|
def test_rebind_service_token_updates_daemon_id_without_changing_runtime_id() -> None:
|
|
repo = _repo_with_user()
|
|
source_daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
target_daemon = repo.create_daemon("user-1", daemon_slug="beta-bot")
|
|
service = _service(repo)
|
|
|
|
issued = service.issue_service_token("my-token", "alice", daemon_id=source_daemon.id)
|
|
|
|
previous_daemon_id, rebound = service.rebind_service_token("my-token", target_daemon.id)
|
|
|
|
assert previous_daemon_id == source_daemon.id
|
|
assert rebound.daemon_id == target_daemon.id
|
|
assert rebound.runtime_id == issued.auth_token.runtime_id
|
|
|
|
|
|
def test_rebind_service_token_rejects_daemon_from_other_user() -> None:
|
|
repo = _repo_with_user()
|
|
repo._users["user-2"] = User(id="user-2", username="bob", permission="read")
|
|
source_daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
foreign_daemon = repo.create_daemon("user-2", daemon_slug="bob-bot")
|
|
service = _service(repo)
|
|
service.issue_service_token("my-token", "alice", daemon_id=source_daemon.id)
|
|
|
|
with pytest.raises(ValueError, match="해당 토큰 사용자에 속하지 않습니다"):
|
|
service.rebind_service_token("my-token", foreign_daemon.id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-022: daemon deactivation 시 바인딩 service token 자동 revoke
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_deactivate_daemon_revokes_bound_tokens() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
service = _service(repo)
|
|
issued = service.issue_service_token("tok1", "alice", daemon_id=daemon.id, no_expiry=True)
|
|
|
|
repo.update_daemon(daemon.id, is_active=False)
|
|
revoked_count = repo.revoke_tokens_by_daemon_id(daemon.id)
|
|
|
|
assert revoked_count == 1
|
|
assert repo._tokens[issued.auth_token.id].revoked_at is not None
|
|
assert not repo._daemons[daemon.id].is_active
|
|
|
|
|
|
def test_deactivate_daemon_revokes_multiple_bound_tokens() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
service = _service(repo)
|
|
issued1 = service.issue_service_token("tok1", "alice", daemon_id=daemon.id, no_expiry=True)
|
|
issued2 = service.issue_service_token("tok2", "alice", daemon_id=daemon.id, no_expiry=True)
|
|
|
|
repo.update_daemon(daemon.id, is_active=False)
|
|
revoked_count = repo.revoke_tokens_by_daemon_id(daemon.id)
|
|
|
|
assert revoked_count == 2
|
|
assert repo._tokens[issued1.auth_token.id].revoked_at is not None
|
|
assert repo._tokens[issued2.auth_token.id].revoked_at is not None
|
|
|
|
|
|
def test_deactivate_daemon_with_no_tokens_is_noop() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
|
|
repo.update_daemon(daemon.id, is_active=False)
|
|
revoked_count = repo.revoke_tokens_by_daemon_id(daemon.id)
|
|
|
|
assert revoked_count == 0
|
|
assert not repo._daemons[daemon.id].is_active
|
|
|
|
|
|
def test_reactivate_daemon_does_not_unrevoke_tokens() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
service = _service(repo)
|
|
issued = service.issue_service_token("tok1", "alice", daemon_id=daemon.id, no_expiry=True)
|
|
|
|
repo.update_daemon(daemon.id, is_active=False)
|
|
repo.revoke_tokens_by_daemon_id(daemon.id)
|
|
repo.update_daemon(daemon.id, is_active=True)
|
|
|
|
assert repo._daemons[daemon.id].is_active
|
|
assert repo._tokens[issued.auth_token.id].revoked_at is not None
|
|
|
|
|
|
def test_revoke_tokens_for_daemon_via_service() -> None:
|
|
repo = _repo_with_user()
|
|
daemon = repo.create_daemon("user-1", daemon_slug="alpha-bot")
|
|
service = _service(repo)
|
|
issued = service.issue_service_token("tok1", "alice", daemon_id=daemon.id, no_expiry=True)
|
|
|
|
count = service.revoke_tokens_for_daemon(daemon.id)
|
|
|
|
assert count == 1
|
|
assert repo._tokens[issued.auth_token.id].revoked_at is not None
|