133f565704
Why: - nothing over HTTP could create the first tenant: every /v1 route needs an API key, and a key cannot exist before its tenant. The service was unusable by a human without hand-written SQL. Changes: - add `provision_tenant`, owning tenant reuse-or-create, key generation and hashing, and domain registration in one transaction - expose it as `python -m src.cli.provision_tenant`, alongside `alembic upgrade head` and `qdrant_bootstrap` - add `tenants.get_by_slug`/`create` and `api_keys.create` - log `tenant.provisioned` / `api_key.provisioned` with the key prefix only Impact: - a third deployment step; the plaintext key is printed once and never logged or stored (ADR-0011, ADR-0009) Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
"""Tenant provisioning against real Postgres (ADR-0009).
|
|
|
|
The property worth a real database here is the one a fake cannot show: the
|
|
issued key authenticates through the *production* auth path, and the row it
|
|
authenticates against holds no plaintext.
|
|
"""
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from src.application.auth.service import resolve_auth_context
|
|
from src.application.domains import ensure_domain_allowed
|
|
from src.application.domains.errors import UnknownDomainError
|
|
from src.application.tenants import provision_tenant
|
|
from src.infrastructure.postgres.models.api_key import ApiKey
|
|
|
|
pytestmark = [
|
|
pytest.mark.integration,
|
|
pytest.mark.postgres,
|
|
pytest.mark.asyncio(loop_scope="session"),
|
|
]
|
|
|
|
|
|
async def test_provision_tenant_issues_a_key_that_authenticates(
|
|
db_sessionmaker: async_sessionmaker[AsyncSession],
|
|
) -> None:
|
|
result = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
|
|
|
|
auth = await resolve_auth_context(db_sessionmaker, result.api_key)
|
|
|
|
assert auth.tenant_id == result.tenant_id
|
|
assert auth.tenant_slug == "acme"
|
|
assert auth.api_key_id == result.api_key_id
|
|
assert "files:write" in auth.scopes
|
|
|
|
|
|
async def test_provision_tenant_persists_only_the_key_hash(
|
|
db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
|
|
) -> None:
|
|
"""A plaintext key in Postgres would make every later hashing decision moot."""
|
|
result = await provision_tenant(db_sessionmaker, slug="acme")
|
|
|
|
stored = (
|
|
await db_session.execute(select(ApiKey).where(ApiKey.id == result.api_key_id))
|
|
).scalar_one()
|
|
|
|
assert result.api_key not in stored.key_hash
|
|
assert stored.key_hash != result.api_key
|
|
assert stored.key_prefix == result.api_key_prefix
|
|
assert result.api_key.startswith(f"sk_{result.api_key_prefix}_")
|
|
|
|
|
|
async def test_provision_tenant_registers_domains_so_uploads_are_allowed(
|
|
db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
|
|
) -> None:
|
|
result = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
|
|
|
|
await ensure_domain_allowed(db_session, tenant_id=result.tenant_id, domain="fire")
|
|
with pytest.raises(UnknownDomainError):
|
|
await ensure_domain_allowed(db_session, tenant_id=result.tenant_id, domain="life")
|
|
|
|
assert result.domains_created == ("fire",)
|
|
|
|
|
|
async def test_provision_tenant_rerun_reuses_the_tenant_and_issues_a_new_key(
|
|
db_sessionmaker: async_sessionmaker[AsyncSession],
|
|
) -> None:
|
|
"""Adding a key to a live tenant must not need a different command."""
|
|
first = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
|
|
|
|
second = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire", "life"))
|
|
|
|
assert second.tenant_id == first.tenant_id
|
|
assert second.tenant_created is False
|
|
assert second.api_key_id != first.api_key_id
|
|
assert second.domains_created == ("life",)
|
|
assert second.domains_existing == ("fire",)
|
|
# Both keys stay valid -- reprovisioning adds a key, it does not rotate one.
|
|
assert (await resolve_auth_context(db_sessionmaker, first.api_key)).tenant_id == first.tenant_id
|
|
assert (
|
|
await resolve_auth_context(db_sessionmaker, second.api_key)
|
|
).tenant_id == first.tenant_id
|
|
|
|
|
|
async def test_provision_tenant_honours_requested_scopes(
|
|
db_sessionmaker: async_sessionmaker[AsyncSession],
|
|
) -> None:
|
|
"""A key scoped to uploads must not be able to manage the allowlist."""
|
|
result = await provision_tenant(db_sessionmaker, slug="acme", scopes=("files:write",))
|
|
|
|
auth = await resolve_auth_context(db_sessionmaker, result.api_key)
|
|
|
|
assert auth.scopes == frozenset({"files:write"})
|
|
assert not auth.has_scope("domains:write")
|