From e9e83b3a2624e8529c2c8e30530bcf06285349d7 Mon Sep 17 00:00:00 2001 From: Ali Zarinkolah Date: Thu, 20 Aug 2026 18:20:24 +0330 Subject: [PATCH] feat(tenant): add tenant_domains allowlist and /v1/domains management API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: - Domain values are denormalized into every Qdrant point payload. Without validation, an unregistered or typo'd domain (e.g. "fier" for "fire") silently creates a new partition that retrieval never queries — the file ends up invisible rather than rejected. Tenants also need independently sized domain sets (one may run 14 insurance lines, another 6), which rules out an enum. Changes: - tenant_domains table (migration 41335d162de8) + repository, unique on (tenant_id, domain). - src/application/domains/: ensure_domain_allowed() is the strict-allowlist check now run inside upload_source_file()'s first transaction, before any MinIO object, job row, or Qdrant point is written. - /v1/domains (list/create/patch/disable/enable) gated on its own domains:read/domains:write scopes, deliberately separate from files:write so an upload key cannot create partitions. domain itself is immutable (denormalized into every point payload); only display_name is editable. Disable blocks new uploads without touching already-indexed points. Impact: - BREAKING: POST /v1/files now rejects any domain without an active tenant_domains row (400, unknown_domain). A domain must be created via POST /v1/domains before the first upload to it. --- CLAUDE.md | 17 +- README.md | 13 +- .../41335d162de8_create_tenant_domains.py | 48 +++++ ...0009-postgres-sqlalchemy-alembic-schema.md | 41 +++- src/api/errors.py | 3 + src/api/router.py | 2 + src/api/routers/domains.py | 110 ++++++++++ src/api/schemas/domains.py | 63 ++++++ src/application/domains/__init__.py | 27 +++ src/application/domains/errors.py | 21 ++ src/application/domains/models.py | 16 ++ src/application/domains/service.py | 142 +++++++++++++ src/application/files/upload.py | 7 + .../postgres/models/__init__.py | 2 + .../postgres/models/tenant_domain.py | 52 +++++ .../postgres/repositories/tenant_domains.py | 73 +++++++ .../integration/postgres/test_domains_api.py | 193 ++++++++++++++++++ .../postgres/test_domains_service.py | 184 +++++++++++++++++ tests/integration/postgres/test_migrations.py | 15 ++ .../postgres/test_upload_service.py | 60 +++++- tests/support/factories.py | 25 +++ 21 files changed, 1102 insertions(+), 12 deletions(-) create mode 100644 alembic/versions/41335d162de8_create_tenant_domains.py create mode 100644 src/api/routers/domains.py create mode 100644 src/api/schemas/domains.py create mode 100644 src/application/domains/__init__.py create mode 100644 src/application/domains/errors.py create mode 100644 src/application/domains/models.py create mode 100644 src/application/domains/service.py create mode 100644 src/infrastructure/postgres/models/tenant_domain.py create mode 100644 src/infrastructure/postgres/repositories/tenant_domains.py create mode 100644 tests/integration/postgres/test_domains_api.py create mode 100644 tests/integration/postgres/test_domains_service.py diff --git a/CLAUDE.md b/CLAUDE.md index 0049058..04cbbbc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -22,10 +22,14 @@ and re-benchmark rather than tune them in place (ADR-0005). Also working: the `chunks` collection bootstrap (`src/infrastructure/qdrant/collection.py`, run as a deployment step via `uv run python -m src.cli.qdrant_bootstrap` — never at startup) and tenant-scoped point upserts (`src/application/points/` behind the -`PointStorage` port), so an upload is searchable by the time `201` returns. Not -built yet: `/v1/points` CRUD and keyword search (plan 002), `tenant_domains` -validation of the `domain` field, and `src/agent/`. That maps to plan 001 -Phases 1-5 done. +`PointStorage` port), so an upload is searchable by the time `201` returns. +Also working: `tenant_domains` plus `/v1/domains` (`src/application/domains/`), +a strict per-tenant allowlist — `POST /v1/files` rejects an unregistered or +disabled `domain` with `400` before anything is written, and domain management +sits behind its own `domains:read`/`domains:write` scopes, never `files:write`. +Not built yet: `/v1/points` CRUD and keyword search (plan 002), and +`src/agent/`. That maps to plan 001 Phases 1-5 done, Phase 6 (runbook, e2e +tests, Compose smoke test) not started. Architecture decisions live in `docs/adr/` (18 ADRs plus the 0000 template; 0001–0004 are `Accepted` — 0004 amended by 0018; 0014 is `Superseded by 0017`; @@ -229,6 +233,11 @@ content_sha256)` idempotency, no terminal job returning to `running`. ### Postgres conventions (ADR-0009) +`domain` is never free-form: it must match an `active` `tenant_domains` row for +the authenticated tenant (ADR-0009). Domain sets are per-tenant and vary in +size. The key itself is immutable — it is denormalized into every Qdrant point +payload and into `source_files`, so renaming it is a migration, not an edit. + UUID primary keys (app-generated), `timestamptz` for all timestamps, `Numeric(18, 8)` for money (never floats), `JSONB` for flexible metadata but typed/indexed columns for query-critical fields, string status columns with diff --git a/README.md b/README.md index 9a8938b..3d851c6 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,18 @@ uv run python -m src.cli.qdrant_bootstrap # the `chunks` collection uv run fastapi dev src/main.py ``` -Both commands are idempotent and safe to re-run. `qdrant_bootstrap` verifies an +Before a tenant can upload, its domains must be registered — `POST /v1/files` +rejects an unregistered or disabled `domain` with `400`. The calling backend +manages them over `/v1/domains` using a key with the `domains:write` scope: + +```bash +curl -X POST http://localhost:8000/v1/domains \ + -H "Authorization: Bearer $API_KEY" \ + -H 'Content-Type: application/json' \ + -d '{"domain": "fire", "display_name": "Fire insurance"}' +``` + +Both bootstrap commands are idempotent and safe to re-run. `qdrant_bootstrap` verifies an existing collection against the pinned schema and exits non-zero on a mismatch, rather than leaving a silently degraded sparse index in place. diff --git a/alembic/versions/41335d162de8_create_tenant_domains.py b/alembic/versions/41335d162de8_create_tenant_domains.py new file mode 100644 index 0000000..84fc0b3 --- /dev/null +++ b/alembic/versions/41335d162de8_create_tenant_domains.py @@ -0,0 +1,48 @@ +"""create tenant_domains + +Revision ID: 41335d162de8 +Revises: bfc6c81c2542 +Create Date: 2026-08-20 17:48:29.443293 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = '41335d162de8' +down_revision: Union[str, Sequence[str], None] = 'bfc6c81c2542' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('tenant_domains', + sa.Column('id', sa.Uuid(), nullable=False), + sa.Column('tenant_id', sa.Uuid(), nullable=False), + sa.Column('domain', sa.String(length=80), nullable=False), + sa.Column('display_name', sa.String(length=200), nullable=False), + sa.Column('status', sa.String(length=20), server_default='active', nullable=False), + sa.Column('metadata', postgresql.JSONB(astext_type=sa.Text()), server_default='{}', nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('disabled_at', sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("status IN ('active', 'disabled')", name='ck_tenant_domains_status'), + sa.ForeignKeyConstraint(['tenant_id'], ['tenants.id'], ondelete='CASCADE'), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('tenant_id', 'domain', name='uq_tenant_domains_tenant_id_domain') + ) + op.create_index(op.f('ix_tenant_domains_tenant_id'), 'tenant_domains', ['tenant_id'], unique=False) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f('ix_tenant_domains_tenant_id'), table_name='tenant_domains') + op.drop_table('tenant_domains') + # ### end Alembic commands ### diff --git a/docs/adr/0009-postgres-sqlalchemy-alembic-schema.md b/docs/adr/0009-postgres-sqlalchemy-alembic-schema.md index d0196a1..87ad528 100644 --- a/docs/adr/0009-postgres-sqlalchemy-alembic-schema.md +++ b/docs/adr/0009-postgres-sqlalchemy-alembic-schema.md @@ -98,13 +98,16 @@ One row per customer/tenant. | `slug` | Stable short name, unique, human-readable. | | `name` | Display name. | | `status` | `active` \| `suspended` \| `deleted`. Suspended tenants authenticate to a clear error but cannot run work. | -| `settings` | JSONB for tenant-level feature flags/limits (max upload size, enabled file types, allowed domains, etc.). | +| `settings` | JSONB for tenant-level feature flags/limits (max upload size, enabled file types, etc.). Allowed domains were previously listed here as well; they live in `tenant_domains` instead, per this ADR's own rule that query-critical fields get typed columns — `domain` is validated on every upload and filtered on every query. | | `created_at`, `updated_at`, `deleted_at` | Audit/soft-delete timestamps. | #### `tenant_domains` -Optional but recommended. Validates the `domain` values used throughout Qdrant -payloads (`car`, `fire`, etc.) per tenant. +**Required.** (Previously "optional but recommended"; implemented and made +mandatory alongside `/v1/domains`.) Validates the `domain` values used +throughout Qdrant payloads (`car`, `fire`, etc.) per tenant. Domain sets are +per-tenant and differ in size — one tenant may run 14 insurance lines and +another 6 — so this is data, not an enum. | Column | Notes | |---|---| @@ -116,7 +119,37 @@ payloads (`car`, `fire`, etc.) per tenant. | `metadata` | JSONB for domain-specific ingestion/retrieval settings. | This prevents arbitrary caller-supplied domains from silently creating new -partitions in Qdrant. +partitions in Qdrant. The failure it guards against is quiet: a typo such as +`fier` for `fire` produces no error anywhere — the file is stored, parsed, +embedded, and indexed into a partition retrieval never queries, so it is +invisible rather than failed. + +##### Enforcement and management + +- **Strict allowlist.** `POST /v1/files` rejects a domain with no `active` row + for the tenant (`400`, error code `unknown_domain`). There is no auto-create + on first use: that would record the typo rather than prevent it. The check + runs inside the upload's first transaction, before any MinIO object, job row, + or Qdrant point is written. +- **Managed over the API, not by an operator.** `/v1/domains` (list, create, + update, disable, enable) is the surface the calling backend uses. Domains are + created by an explicit, scoped call rather than as a side effect of an upload + — that distinction, not who makes the call, is what "strict" means here. +- **Its own scope.** `domains:read`/`domains:write`, deliberately separate from + `files:write`. Folding domain creation into the upload scope would let an + upload key create partitions again, which is the exact hole this closes. + `api_keys.scopes` is already a free JSONB list, so this needs no schema change. +- **`tenant_id` stays derived from the API key.** One key per tenant; nothing + request-suppliable. A platform key acting across tenants would need a real + actor model and is not adopted. +- **`domain` is immutable; `display_name` is not.** The key is denormalized into + every Qdrant point payload and into `source_files`, so renaming it means + rewriting all of them — a migration, not a `PATCH`. The update schema + therefore has no `domain` field. +- **Disable is not delete.** `status='disabled'` blocks new uploads and hides + the domain from listings, leaving already-indexed points intact and + retrievable. Actual removal needs the retention/erasure workflow this ADR and + plan 001 defer. #### `api_keys` diff --git a/src/api/errors.py b/src/api/errors.py index 9832671..5b4a99a 100644 --- a/src/api/errors.py +++ b/src/api/errors.py @@ -16,6 +16,7 @@ from src.application.auth.errors import ( MissingScopeError, TenantInactiveError, ) +from src.application.domains.errors import DomainAlreadyExistsError, UnknownDomainError from src.application.files.errors import FileTooLargeError, InvalidUploadError from src.application.ingestion.errors import ( ChunkLimitExceededError, @@ -40,6 +41,8 @@ _MAPPING: tuple[tuple[type[Exception], int, str], ...] = ( (TenantInactiveError, status.HTTP_401_UNAUTHORIZED, "tenant_not_found"), (MissingScopeError, status.HTTP_403_FORBIDDEN, "missing_scope"), (InvalidUploadError, status.HTTP_400_BAD_REQUEST, "validation_error"), + (UnknownDomainError, status.HTTP_400_BAD_REQUEST, "unknown_domain"), + (DomainAlreadyExistsError, status.HTTP_409_CONFLICT, "conflict"), (DocumentParseError, status.HTTP_400_BAD_REQUEST, "validation_error"), (UnsupportedSourceTypeError, status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, "unsupported_media_type"), (FileTooLargeError, status.HTTP_413_CONTENT_TOO_LARGE, "payload_too_large"), diff --git a/src/api/router.py b/src/api/router.py index ea50999..f806747 100644 --- a/src/api/router.py +++ b/src/api/router.py @@ -1,6 +1,8 @@ from fastapi import APIRouter +from src.api.routers.domains import router as domains_router from src.api.routers.files import router as files_router router = APIRouter() +router.include_router(domains_router) router.include_router(files_router) diff --git a/src/api/routers/domains.py b/src/api/routers/domains.py new file mode 100644 index 0000000..695c750 --- /dev/null +++ b/src/api/routers/domains.py @@ -0,0 +1,110 @@ +"""`/v1/domains` (ADR-0008, ADR-0009). + +The management surface for a tenant's domain allowlist, used by the calling +backend rather than by an operator with a psql prompt. + +Gated on `domains:read`/`domains:write`, deliberately **not** on `files:write`: +if an upload key could create domains, the allowlist would no longer prevent a +typo'd `domain` from creating a Qdrant partition, which is its only purpose. +""" + +from typing import Annotated + +from fastapi import APIRouter, Depends, status +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.api.dependencies.auth import require_scope +from src.api.schemas.domains import ( + CreateDomainRequest, + DomainListResponse, + DomainResponse, + UpdateDomainRequest, +) +from src.application.auth.context import AuthContext +from src.application.domains import ( + create_domain, + list_domains, + set_domain_status, + update_domain, +) +from src.bootstrap.dependencies import get_sessionmaker + +router = APIRouter(prefix="/domains", tags=["domains"]) + +_RequireDomainsRead = Annotated[AuthContext, Depends(require_scope("domains:read"))] +_RequireDomainsWrite = Annotated[AuthContext, Depends(require_scope("domains:write"))] +_SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)] + + +@router.get("") +async def list_tenant_domains( + auth: _RequireDomainsRead, + sessionmaker: _SessionmakerDep, + include_disabled: bool = False, +) -> DomainListResponse: + results = await list_domains( + sessionmaker, tenant_id=auth.tenant_id, include_disabled=include_disabled + ) + return DomainListResponse(domains=[DomainResponse.from_result(item) for item in results]) + + +@router.post("", status_code=status.HTTP_201_CREATED) +async def create_tenant_domain( + request: CreateDomainRequest, + auth: _RequireDomainsWrite, + sessionmaker: _SessionmakerDep, +) -> DomainResponse: + result = await create_domain( + sessionmaker, + tenant_id=auth.tenant_id, + domain=request.domain, + display_name=request.display_name, + metadata=request.metadata, + ) + return DomainResponse.from_result(result) + + +@router.patch("/{domain}") +async def update_tenant_domain( + domain: str, + request: UpdateDomainRequest, + auth: _RequireDomainsWrite, + sessionmaker: _SessionmakerDep, +) -> DomainResponse: + result = await update_domain( + sessionmaker, + tenant_id=auth.tenant_id, + domain=domain, + display_name=request.display_name, + ) + return DomainResponse.from_result(result) + + +@router.delete("/{domain}") +async def disable_tenant_domain( + domain: str, + auth: _RequireDomainsWrite, + sessionmaker: _SessionmakerDep, +) -> DomainResponse: + """Disable, not delete. + + Blocks new uploads and drops the domain from pickers while leaving the + points already indexed under it intact and retrievable. Actually removing + them needs the tenant-erasure workflow plan 001 defers. + """ + result = await set_domain_status( + sessionmaker, tenant_id=auth.tenant_id, domain=domain, status="disabled" + ) + return DomainResponse.from_result(result) + + +@router.post("/{domain}/enable") +async def enable_tenant_domain( + domain: str, + auth: _RequireDomainsWrite, + sessionmaker: _SessionmakerDep, +) -> DomainResponse: + result = await set_domain_status( + sessionmaker, tenant_id=auth.tenant_id, domain=domain, status="active" + ) + return DomainResponse.from_result(result) diff --git a/src/api/schemas/domains.py b/src/api/schemas/domains.py new file mode 100644 index 0000000..2462cfc --- /dev/null +++ b/src/api/schemas/domains.py @@ -0,0 +1,63 @@ +"""Public request/response models for `/v1/domains` (ADR-0008, ADR-0009). + +`tenant_id` appears in none of these: it comes from the authenticated key, and +accepting it from a body would break the isolation boundary (ADR-0002). +""" + +import uuid +from datetime import datetime + +from pydantic import BaseModel, Field, field_validator + +from src.application.domains.models import DomainResult + +# Lowercase alphanumerics plus - and _; the key is embedded in every Qdrant +# payload and filtered on as a keyword, so it stays boring on purpose. +_DOMAIN_PATTERN = r"^[a-z0-9][a-z0-9_-]*$" + + +class DomainResponse(BaseModel): + id: uuid.UUID + domain: str + display_name: str + status: str + metadata: dict[str, object] + created_at: datetime + updated_at: datetime + + @classmethod + def from_result(cls, result: DomainResult) -> "DomainResponse": + return cls( + id=result.id, + domain=result.domain, + display_name=result.display_name, + status=result.status, + metadata=result.metadata, + created_at=result.created_at, + updated_at=result.updated_at, + ) + + +class DomainListResponse(BaseModel): + domains: list[DomainResponse] + + +class CreateDomainRequest(BaseModel): + domain: str = Field(min_length=1, max_length=80, pattern=_DOMAIN_PATTERN) + display_name: str = Field(min_length=1, max_length=200) + metadata: dict[str, object] = Field(default_factory=dict) + + @field_validator("domain") + @classmethod + def _normalize(cls, value: str) -> str: + return value.strip() + + +class UpdateDomainRequest(BaseModel): + """`domain` is absent by design — the key is immutable. + + It is denormalized into every point payload and into `source_files`, so + renaming it is a migration rather than an edit (ADR-0009). + """ + + display_name: str = Field(min_length=1, max_length=200) diff --git a/src/application/domains/__init__.py b/src/application/domains/__init__.py new file mode 100644 index 0000000..32cefa7 --- /dev/null +++ b/src/application/domains/__init__.py @@ -0,0 +1,27 @@ +"""Tenant-domain management and the upload-time allowlist check (ADR-0009).""" + +from src.application.domains.errors import ( + DomainAlreadyExistsError, + DomainsError, + UnknownDomainError, +) +from src.application.domains.models import DomainResult +from src.application.domains.service import ( + create_domain, + ensure_domain_allowed, + list_domains, + set_domain_status, + update_domain, +) + +__all__ = [ + "DomainAlreadyExistsError", + "DomainResult", + "DomainsError", + "UnknownDomainError", + "create_domain", + "ensure_domain_allowed", + "list_domains", + "set_domain_status", + "update_domain", +] diff --git a/src/application/domains/errors.py b/src/application/domains/errors.py new file mode 100644 index 0000000..073cae2 --- /dev/null +++ b/src/application/domains/errors.py @@ -0,0 +1,21 @@ +"""Domain-management failures (ADR-0009). No HTTP knowledge here — +`src/api/errors.py` maps these to status codes. +""" + + +class DomainsError(Exception): + """Base class for tenant-domain failures.""" + + +class UnknownDomainError(DomainsError): + """The upload named a domain the tenant has not registered, or one that is + disabled. Maps to `400`. + + Rejecting is the whole point: an unrecognized `domain` would otherwise + create a new Qdrant partition silently, and a file in a partition nothing + queries is invisible rather than failed (ADR-0009). + """ + + +class DomainAlreadyExistsError(DomainsError): + """The tenant already has a domain with this key. Maps to `409`.""" diff --git a/src/application/domains/models.py b/src/application/domains/models.py new file mode 100644 index 0000000..d1cc26a --- /dev/null +++ b/src/application/domains/models.py @@ -0,0 +1,16 @@ +"""Transport-agnostic results for the domain-management service.""" + +import uuid +from dataclasses import dataclass +from datetime import datetime + + +@dataclass(frozen=True) +class DomainResult: + id: uuid.UUID + domain: str + display_name: str + status: str + metadata: dict[str, object] + created_at: datetime + updated_at: datetime diff --git a/src/application/domains/service.py b/src/application/domains/service.py new file mode 100644 index 0000000..90eba71 --- /dev/null +++ b/src/application/domains/service.py @@ -0,0 +1,142 @@ +"""Tenant-domain management (ADR-0009). + +A tenant's domain set is per-tenant and varies in size — one may run 14 +insurance lines, another 6 — so it is data, not an enum. + +`ensure_domain_allowed` is the reason this package exists: it is the strict +allowlist check the upload path runs before anything is written. Everything +else here is the management surface the calling backend uses to populate that +allowlist, under its own `domains:write` scope so an upload key cannot create +partitions. + +`tenant_id` is always a required parameter taken from `AuthContext`, never from +a request body (ADR-0002). +""" + +import uuid + +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.domains.errors import DomainAlreadyExistsError, UnknownDomainError +from src.application.domains.models import DomainResult +from src.infrastructure.postgres.models.tenant_domain import TenantDomain +from src.infrastructure.postgres.repositories import tenant_domains as repo + + +async def _flush_and_refresh(session: AsyncSession, tenant_domain: TenantDomain) -> None: + """Materialize server-generated columns before the row leaves the session. + + `updated_at` is `onupdate=func.now()`, so after an UPDATE its value lives in + the database, not in the instance. Reading it later would trigger a lazy + load outside any greenlet context (`MissingGreenlet`), so it is fetched here + while the session is still open. + """ + await session.flush() + await session.refresh(tenant_domain) + + +def _to_result(tenant_domain: TenantDomain) -> DomainResult: + return DomainResult( + id=tenant_domain.id, + domain=tenant_domain.domain, + display_name=tenant_domain.display_name, + status=tenant_domain.status, + metadata=tenant_domain.metadata_, + created_at=tenant_domain.created_at, + updated_at=tenant_domain.updated_at, + ) + + +async def ensure_domain_allowed( + session: AsyncSession, *, tenant_id: uuid.UUID, domain: str +) -> None: + """Raise `UnknownDomainError` unless the tenant has this domain active. + + Takes a session rather than a sessionmaker: the upload path calls this + inside its existing txn A, so the check costs no extra connection and + cannot pass and then go stale before the row is written. + """ + tenant_domain = await repo.get(session, tenant_id=tenant_id, domain=domain) + if tenant_domain is None: + raise UnknownDomainError( + f"domain '{domain}' is not registered for this tenant; " + f"create it via POST /v1/domains before uploading to it" + ) + if tenant_domain.status != "active": + raise UnknownDomainError(f"domain '{domain}' is disabled for this tenant") + + +async def list_domains( + sessionmaker: async_sessionmaker[AsyncSession], + *, + tenant_id: uuid.UUID, + include_disabled: bool = False, +) -> list[DomainResult]: + async with sessionmaker() as session: + found = await repo.list_for_tenant( + session, tenant_id=tenant_id, include_disabled=include_disabled + ) + return [_to_result(item) for item in found] + + +async def create_domain( + sessionmaker: async_sessionmaker[AsyncSession], + *, + tenant_id: uuid.UUID, + domain: str, + display_name: str, + metadata: dict[str, object] | None = None, +) -> DomainResult: + async with sessionmaker() as session: + if await repo.get(session, tenant_id=tenant_id, domain=domain) is not None: + raise DomainAlreadyExistsError(f"domain '{domain}' already exists for this tenant") + created = repo.create( + session, + tenant_id=tenant_id, + domain=domain, + display_name=display_name, + metadata=metadata, + ) + await session.commit() + return _to_result(created) + + +async def update_domain( + sessionmaker: async_sessionmaker[AsyncSession], + *, + tenant_id: uuid.UUID, + domain: str, + display_name: str, +) -> DomainResult: + """Only the label is mutable — see `repo.update_display_name`.""" + async with sessionmaker() as session: + found = await repo.get(session, tenant_id=tenant_id, domain=domain) + if found is None: + raise UnknownDomainError(f"domain '{domain}' is not registered for this tenant") + repo.update_display_name(found, display_name=display_name) + await _flush_and_refresh(session, found) + await session.commit() + return _to_result(found) + + +async def set_domain_status( + sessionmaker: async_sessionmaker[AsyncSession], + *, + tenant_id: uuid.UUID, + domain: str, + status: str, +) -> DomainResult: + """Disable or re-enable a domain. + + Disabling blocks new uploads and hides the domain from pickers. It does not + touch the points already indexed under it — removing those needs the + tenant-erasure workflow plan 001 defers. + """ + async with sessionmaker() as session: + found = await repo.get(session, tenant_id=tenant_id, domain=domain) + if found is None: + raise UnknownDomainError(f"domain '{domain}' is not registered for this tenant") + repo.set_status(found, status=status) + await _flush_and_refresh(session, found) + await session.commit() + return _to_result(found) diff --git a/src/application/files/upload.py b/src/application/files/upload.py index 36f59cd..fb63214 100644 --- a/src/application/files/upload.py +++ b/src/application/files/upload.py @@ -30,6 +30,7 @@ from anyio import CapacityLimiter, Semaphore, fail_after, to_thread from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.application.auth.context import AuthContext +from src.application.domains import ensure_domain_allowed from src.application.files.errors import InvalidUploadError from src.application.files.models import UploadResult from src.application.files.storage_keys import source_file_object_key @@ -118,6 +119,12 @@ async def upload_source_file( async with acquire_ingestion_slot(concurrency_limiter): async with sessionmaker() as session: + # Strict allowlist, checked inside txn A before anything is written + # (ADR-0009). An unregistered domain would otherwise create a new + # Qdrant partition silently, leaving the file invisible to + # retrieval rather than failing. + await ensure_domain_allowed(session, tenant_id=auth.tenant_id, domain=domain) + existing = await source_files_repo.find_active_by_content_hash( session, tenant_id=auth.tenant_id, diff --git a/src/infrastructure/postgres/models/__init__.py b/src/infrastructure/postgres/models/__init__.py index ea82abc..91a50e2 100644 --- a/src/infrastructure/postgres/models/__init__.py +++ b/src/infrastructure/postgres/models/__init__.py @@ -4,6 +4,7 @@ from src.infrastructure.postgres.models.ingestion_job import IngestionJob from src.infrastructure.postgres.models.ingestion_job_event import IngestionJobEvent from src.infrastructure.postgres.models.source_file import SourceFile from src.infrastructure.postgres.models.tenant import Tenant +from src.infrastructure.postgres.models.tenant_domain import TenantDomain __all__ = [ "ApiKey", @@ -12,4 +13,5 @@ __all__ = [ "IngestionJobEvent", "SourceFile", "Tenant", + "TenantDomain", ] diff --git a/src/infrastructure/postgres/models/tenant_domain.py b/src/infrastructure/postgres/models/tenant_domain.py new file mode 100644 index 0000000..cc4fa4a --- /dev/null +++ b/src/infrastructure/postgres/models/tenant_domain.py @@ -0,0 +1,52 @@ +import uuid +from datetime import datetime + +from sqlalchemy import CheckConstraint, DateTime, ForeignKey, String, UniqueConstraint, func +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from src.infrastructure.postgres.models.base import Base + +TENANT_DOMAIN_STATUSES = ("active", "disabled") + + +class TenantDomain(Base): + """A domain a tenant is allowed to ingest into (ADR-0009). + + Tenants do not share a domain list — one may run 14 insurance lines and + another 6 — so this is a per-tenant table rather than an enum or a global + lookup. + + Its purpose is to stop an arbitrary caller-supplied `domain` from silently + creating a new Qdrant partition. `domain` is denormalized into every point's + payload and into `source_files`, and a typo like `fier` for `fire` produces + no error anywhere: the file indexes into a partition retrieval never queries, + so it is invisible rather than failed. + + `domain` is the immutable key. Renaming it would mean rewriting every point + payload that carries it, which is a migration, not an edit — `display_name` + is the mutable human-facing label instead. + """ + + __tablename__ = "tenant_domains" + __table_args__ = ( + UniqueConstraint("tenant_id", "domain", name="uq_tenant_domains_tenant_id_domain"), + CheckConstraint(f"status IN {TENANT_DOMAIN_STATUSES}", name="ck_tenant_domains_status"), + ) + + id: Mapped[uuid.UUID] = mapped_column(primary_key=True) + tenant_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("tenants.id", ondelete="CASCADE"), index=True + ) + domain: Mapped[str] = mapped_column(String(80)) + display_name: Mapped[str] = mapped_column(String(200)) + status: Mapped[str] = mapped_column(String(20), default="active", server_default="active") + metadata_: Mapped[dict[str, object]] = mapped_column( + "metadata", JSONB, default=dict, server_default="{}" + ) + + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + disabled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None) diff --git a/src/infrastructure/postgres/repositories/tenant_domains.py b/src/infrastructure/postgres/repositories/tenant_domains.py new file mode 100644 index 0000000..955d607 --- /dev/null +++ b/src/infrastructure/postgres/repositories/tenant_domains.py @@ -0,0 +1,73 @@ +"""`tenant_domains` persistence (ADR-0009). + +Plain functions over an `AsyncSession` the caller owns. No function here +commits, rolls back, or closes the session (ADR-0012). Every read and write is +tenant-scoped by a required `tenant_id` argument, so a missing filter is a +signature error rather than a cross-tenant leak. +""" + +import uuid +from datetime import UTC, datetime + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.infrastructure.postgres.models.tenant_domain import TenantDomain + + +async def get(session: AsyncSession, *, tenant_id: uuid.UUID, domain: str) -> TenantDomain | None: + result = await session.execute( + select(TenantDomain).where( + TenantDomain.tenant_id == tenant_id, TenantDomain.domain == domain + ) + ) + return result.scalar_one_or_none() + + +async def list_for_tenant( + session: AsyncSession, *, tenant_id: uuid.UUID, include_disabled: bool = False +) -> list[TenantDomain]: + statement = select(TenantDomain).where(TenantDomain.tenant_id == tenant_id) + if not include_disabled: + statement = statement.where(TenantDomain.status == "active") + result = await session.execute(statement.order_by(TenantDomain.domain)) + return list(result.scalars().all()) + + +def create( + session: AsyncSession, + *, + tenant_id: uuid.UUID, + domain: str, + display_name: str, + metadata: dict[str, object] | None = None, +) -> TenantDomain: + tenant_domain = TenantDomain( + id=uuid.uuid4(), + tenant_id=tenant_id, + domain=domain, + display_name=display_name, + metadata_=metadata or {}, + ) + session.add(tenant_domain) + return tenant_domain + + +def update_display_name(tenant_domain: TenantDomain, *, display_name: str) -> TenantDomain: + """`domain` itself is deliberately not updatable. + + It is denormalized into every Qdrant point payload and into `source_files`, + so changing the key would mean rewriting all of them — a migration, not an + edit. The label is what callers actually want to change. + """ + tenant_domain.display_name = display_name + return tenant_domain + + +def set_status(tenant_domain: TenantDomain, *, status: str) -> TenantDomain: + """Disable/re-enable a domain. Existing points are untouched either way — + disabling blocks new uploads, it is not a delete (ADR-0002). + """ + tenant_domain.status = status + tenant_domain.disabled_at = datetime.now(UTC) if status == "disabled" else None + return tenant_domain diff --git a/tests/integration/postgres/test_domains_api.py b/tests/integration/postgres/test_domains_api.py new file mode 100644 index 0000000..93ee049 --- /dev/null +++ b/tests/integration/postgres/test_domains_api.py @@ -0,0 +1,193 @@ +"""`/v1/domains` over HTTP, against real Postgres (ADR-0008, ADR-0009). + +The property worth testing at this layer is the scope boundary: an upload key +must not be able to create domains, or the allowlist stops preventing anything. +""" + +from collections.abc import AsyncIterator + +import pytest +import pytest_asyncio +from asgi_lifespan import LifespanManager +from httpx import ASGITransport, AsyncClient +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.bootstrap.dependencies import get_sessionmaker +from src.config import Settings +from src.main import create_app +from tests.support.factories import create_api_key, create_tenant, create_tenant_domain + +pytestmark = [ + pytest.mark.integration, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + + +@pytest_asyncio.fixture(loop_scope="session") +async def api_client( + settings: Settings, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> AsyncIterator[AsyncClient]: + """The real app, with only the database swapped for the test's session + factory -- so routing, auth, scopes, and the error envelope are exercised. + """ + app = create_app(settings) + app.dependency_overrides[get_sessionmaker] = lambda: db_sessionmaker + async with ( + LifespanManager(app) as manager, + AsyncClient(transport=ASGITransport(app=manager.app), base_url="http://test") as client, + ): + yield client + + +def _auth(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +async def test_create_domain_returns_201_and_lists_it( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + tenant = await create_tenant(db_session) + _, token = await create_api_key( + db_session, tenant=tenant, scopes=["domains:read", "domains:write"] + ) + await db_session.commit() + + created = await api_client.post( + "/v1/domains", + json={"domain": "fire", "display_name": "Fire insurance"}, + headers=_auth(token), + ) + listed = await api_client.get("/v1/domains", headers=_auth(token)) + + assert created.status_code == 201 + assert created.json()["domain"] == "fire" + assert [item["domain"] for item in listed.json()["domains"]] == ["fire"] + + +async def test_create_domain_requires_the_domains_write_scope( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + """An upload key creating domains would defeat the allowlist entirely.""" + tenant = await create_tenant(db_session) + _, token = await create_api_key(db_session, tenant=tenant, scopes=["files:write"]) + await db_session.commit() + + response = await api_client.post( + "/v1/domains", + json={"domain": "fire", "display_name": "Fire"}, + headers=_auth(token), + ) + + assert response.status_code == 403 + assert response.json()["error"]["code"] == "missing_scope" + + +async def test_list_domains_never_shows_another_tenants_domains( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + owner = await create_tenant(db_session) + other = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=owner, domain="fire") + _, other_token = await create_api_key(db_session, tenant=other, scopes=["domains:read"]) + await db_session.commit() + + response = await api_client.get("/v1/domains", headers=_auth(other_token)) + + assert response.status_code == 200 + assert response.json()["domains"] == [] + + +async def test_create_domain_rejects_a_duplicate_with_409( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + _, token = await create_api_key(db_session, tenant=tenant, scopes=["domains:write"]) + await db_session.commit() + + response = await api_client.post( + "/v1/domains", + json={"domain": "fire", "display_name": "Fire"}, + headers=_auth(token), + ) + + assert response.status_code == 409 + assert response.json()["error"]["code"] == "conflict" + + +async def test_create_domain_rejects_a_malformed_key( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + tenant = await create_tenant(db_session) + _, token = await create_api_key(db_session, tenant=tenant, scopes=["domains:write"]) + await db_session.commit() + + response = await api_client.post( + "/v1/domains", + json={"domain": "Fire Insurance!", "display_name": "Fire"}, + headers=_auth(token), + ) + + assert response.status_code == 422 + + +async def test_patch_domain_cannot_rename_the_key( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + """`domain` is not part of the update schema -- it is denormalized into + every point payload, so renaming it is a migration, not an edit. + """ + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + _, token = await create_api_key( + db_session, tenant=tenant, scopes=["domains:read", "domains:write"] + ) + await db_session.commit() + + response = await api_client.patch( + "/v1/domains/fire", + json={"display_name": "Fire & perils", "domain": "renamed"}, + headers=_auth(token), + ) + + assert response.status_code == 200 + assert response.json()["domain"] == "fire" + assert response.json()["display_name"] == "Fire & perils" + + +async def test_delete_domain_disables_it_without_removing_it( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + _, token = await create_api_key( + db_session, tenant=tenant, scopes=["domains:read", "domains:write"] + ) + await db_session.commit() + + deleted = await api_client.delete("/v1/domains/fire", headers=_auth(token)) + default_list = await api_client.get("/v1/domains", headers=_auth(token)) + full_list = await api_client.get( + "/v1/domains", params={"include_disabled": True}, headers=_auth(token) + ) + + assert deleted.status_code == 200 + assert deleted.json()["status"] == "disabled" + assert default_list.json()["domains"] == [] + assert [item["domain"] for item in full_list.json()["domains"]] == ["fire"] + + +async def test_patch_unknown_domain_returns_400( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + tenant = await create_tenant(db_session) + _, token = await create_api_key(db_session, tenant=tenant, scopes=["domains:write"]) + await db_session.commit() + + response = await api_client.patch( + "/v1/domains/absent", json={"display_name": "x"}, headers=_auth(token) + ) + + assert response.status_code == 400 + assert response.json()["error"]["code"] == "unknown_domain" diff --git a/tests/integration/postgres/test_domains_service.py b/tests/integration/postgres/test_domains_service.py new file mode 100644 index 0000000..8309c3a --- /dev/null +++ b/tests/integration/postgres/test_domains_service.py @@ -0,0 +1,184 @@ +"""Tenant-domain management and the upload-time allowlist (ADR-0009).""" + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.domains import ( + DomainAlreadyExistsError, + UnknownDomainError, + create_domain, + ensure_domain_allowed, + list_domains, + set_domain_status, + update_domain, +) +from tests.support.factories import create_tenant, create_tenant_domain + +pytestmark = [ + pytest.mark.integration, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + + +async def test_ensure_domain_allowed_passes_for_a_registered_active_domain( + db_session: AsyncSession, +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + + await ensure_domain_allowed(db_session, tenant_id=tenant.id, domain="fire") + + +async def test_ensure_domain_allowed_rejects_an_unregistered_domain( + db_session: AsyncSession, +) -> None: + """The typo case: `fier` must not silently become a new Qdrant partition.""" + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + + with pytest.raises(UnknownDomainError, match="fier"): + await ensure_domain_allowed(db_session, tenant_id=tenant.id, domain="fier") + + +async def test_ensure_domain_allowed_rejects_a_disabled_domain( + db_session: AsyncSession, +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire", status="disabled") + + with pytest.raises(UnknownDomainError, match="disabled"): + await ensure_domain_allowed(db_session, tenant_id=tenant.id, domain="fire") + + +async def test_ensure_domain_allowed_rejects_another_tenants_domain( + db_session: AsyncSession, +) -> None: + """Domain lists are per-tenant; one tenant's `fire` is not another's.""" + owner = await create_tenant(db_session) + other = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=owner, domain="fire") + + with pytest.raises(UnknownDomainError): + await ensure_domain_allowed(db_session, tenant_id=other.id, domain="fire") + + +async def test_tenants_hold_independent_domain_sets_of_different_sizes( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + big = await create_tenant(db_session) + small = await create_tenant(db_session) + for index in range(14): + await create_tenant_domain(db_session, tenant=big, domain=f"line-{index:02d}") + for index in range(6): + await create_tenant_domain(db_session, tenant=small, domain=f"line-{index:02d}") + await db_session.commit() + + assert len(await list_domains(db_sessionmaker, tenant_id=big.id)) == 14 + assert len(await list_domains(db_sessionmaker, tenant_id=small.id)) == 6 + + +async def test_create_domain_then_upload_is_allowed( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await db_session.commit() + + created = await create_domain( + db_sessionmaker, tenant_id=tenant.id, domain="car", display_name="Car insurance" + ) + + assert created.domain == "car" + assert created.status == "active" + async with db_sessionmaker() as session: + await ensure_domain_allowed(session, tenant_id=tenant.id, domain="car") + + +async def test_create_domain_rejects_a_duplicate_key_for_the_same_tenant( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + await db_session.commit() + + with pytest.raises(DomainAlreadyExistsError): + await create_domain( + db_sessionmaker, tenant_id=tenant.id, domain="fire", display_name="Fire again" + ) + + +async def test_create_domain_allows_the_same_key_for_different_tenants( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + first = await create_tenant(db_session) + second = await create_tenant(db_session) + await db_session.commit() + + await create_domain(db_sessionmaker, tenant_id=first.id, domain="fire", display_name="Fire") + await create_domain(db_sessionmaker, tenant_id=second.id, domain="fire", display_name="Fire") + + assert len(await list_domains(db_sessionmaker, tenant_id=first.id)) == 1 + assert len(await list_domains(db_sessionmaker, tenant_id=second.id)) == 1 + + +async def test_update_domain_changes_only_the_display_name( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + await db_session.commit() + + updated = await update_domain( + db_sessionmaker, tenant_id=tenant.id, domain="fire", display_name="Fire & perils" + ) + + assert updated.display_name == "Fire & perils" + # The key is immutable: it is denormalized into every point payload. + assert updated.domain == "fire" + + +async def test_disabling_a_domain_blocks_new_uploads_without_deleting_it( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire") + await db_session.commit() + + disabled = await set_domain_status( + db_sessionmaker, tenant_id=tenant.id, domain="fire", status="disabled" + ) + + assert disabled.status == "disabled" + async with db_sessionmaker() as session: + with pytest.raises(UnknownDomainError): + await ensure_domain_allowed(session, tenant_id=tenant.id, domain="fire") + # Still there, just hidden from the default listing. + assert await list_domains(db_sessionmaker, tenant_id=tenant.id) == [] + assert len(await list_domains(db_sessionmaker, tenant_id=tenant.id, include_disabled=True)) == 1 + + +async def test_re_enabling_a_domain_restores_uploads( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=tenant, domain="fire", status="disabled") + await db_session.commit() + + await set_domain_status(db_sessionmaker, tenant_id=tenant.id, domain="fire", status="active") + + async with db_sessionmaker() as session: + await ensure_domain_allowed(session, tenant_id=tenant.id, domain="fire") + + +async def test_update_domain_rejects_another_tenants_domain( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + owner = await create_tenant(db_session) + other = await create_tenant(db_session) + await create_tenant_domain(db_session, tenant=owner, domain="fire") + await db_session.commit() + + with pytest.raises(UnknownDomainError): + await update_domain( + db_sessionmaker, tenant_id=other.id, domain="fire", display_name="hijacked" + ) diff --git a/tests/integration/postgres/test_migrations.py b/tests/integration/postgres/test_migrations.py index 5eea6ab..436804d 100644 --- a/tests/integration/postgres/test_migrations.py +++ b/tests/integration/postgres/test_migrations.py @@ -10,6 +10,7 @@ pytestmark = [ EXPECTED_TABLES = { "tenants", + "tenant_domains", "api_keys", "source_files", "ingestion_jobs", @@ -25,3 +26,17 @@ async def test_migrations_create_schema_from_empty_database(postgres_engine: Asy ) assert EXPECTED_TABLES.issubset(set(table_names)) + + +async def test_tenant_domains_enforces_one_row_per_tenant_and_key( + postgres_engine: AsyncEngine, +) -> None: + """The unique constraint is what stops the same domain being registered + twice for a tenant while still letting two tenants share a key. + """ + async with postgres_engine.connect() as connection: + constraints = await connection.run_sync( + lambda sync_conn: inspect(sync_conn).get_unique_constraints("tenant_domains") + ) + + assert any(constraint["column_names"] == ["tenant_id", "domain"] for constraint in constraints) diff --git a/tests/integration/postgres/test_upload_service.py b/tests/integration/postgres/test_upload_service.py index 64b982c..c042eec 100644 --- a/tests/integration/postgres/test_upload_service.py +++ b/tests/integration/postgres/test_upload_service.py @@ -11,6 +11,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.application.auth.context import AuthContext +from src.application.domains import UnknownDomainError from src.application.files.models import UploadResult from src.application.files.upload import upload_source_file from src.application.ingestion.errors import ( @@ -28,7 +29,7 @@ from tests.fakes import ( FakePointStorage, FakeSparseEmbedder, ) -from tests.support.factories import create_api_key, create_tenant +from tests.support.factories import create_api_key, create_tenant, create_tenant_domain pytestmark = [ pytest.mark.integration, @@ -70,9 +71,12 @@ async def _upload( ) -async def _auth_for(db_session: AsyncSession) -> AuthContext: +async def _auth_for(db_session: AsyncSession, *, domain: str = "general") -> AuthContext: tenant = await create_tenant(db_session) api_key, _ = await create_api_key(db_session, tenant=tenant) + # Uploads reject an unregistered domain (ADR-0009), so register the one the + # helper below uploads to. + await create_tenant_domain(db_session, tenant=tenant, domain=domain) await db_session.commit() return AuthContext( tenant_id=tenant.id, @@ -289,7 +293,7 @@ async def test_upload_source_file_indexes_points_and_records_real_counters( async def test_upload_source_file_indexes_points_under_the_authenticated_tenant( db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] ) -> None: - auth = await _auth_for(db_session) + auth = await _auth_for(db_session, domain="fire") point_storage = FakePointStorage() result = await _upload( @@ -395,3 +399,53 @@ async def test_upload_source_file_retry_after_index_failure_produces_no_duplicat assert len(tenant_jobs) == 2 assert {job.status for job in tenant_jobs} == {"failed", "succeeded"} + +async def test_upload_source_file_rejects_an_unregistered_domain_before_any_write( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """A typo'd domain must fail loudly, not create a new Qdrant partition + whose contents retrieval never queries (ADR-0009). + """ + auth = await _auth_for(db_session, domain="fire") + storage = FakeObjectStorage() + point_storage = FakePointStorage() + + with pytest.raises(UnknownDomainError, match="fier"): + await _upload( + sessionmaker=db_sessionmaker, + storage=storage, + auth=auth, + point_storage=point_storage, + domain="fier", + ) + + # Nothing was written anywhere: no object, no points, and no job row. + assert storage.objects == {} + assert point_storage.points == {} + async with db_sessionmaker() as verify_session: + jobs = (await verify_session.execute(select(IngestionJob))).scalars().all() + assert [job for job in jobs if job.tenant_id == auth.tenant_id] == [] + + +async def test_upload_source_file_rejects_a_disabled_domain( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + api_key, _ = await create_api_key(db_session, tenant=tenant) + await create_tenant_domain(db_session, tenant=tenant, domain="fire", status="disabled") + await db_session.commit() + auth = AuthContext( + tenant_id=tenant.id, + tenant_slug=tenant.slug, + api_key_id=api_key.id, + scopes=frozenset({"files:write"}), + actor_type="backend", + ) + + with pytest.raises(UnknownDomainError, match="disabled"): + await _upload( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + domain="fire", + ) diff --git a/tests/support/factories.py b/tests/support/factories.py index d4cad2c..638781e 100644 --- a/tests/support/factories.py +++ b/tests/support/factories.py @@ -12,6 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from src.application.auth.keys import generate_api_key, hash_secret from src.infrastructure.postgres.models.api_key import ApiKey from src.infrastructure.postgres.models.tenant import Tenant +from src.infrastructure.postgres.models.tenant_domain import TenantDomain async def create_tenant( @@ -47,3 +48,27 @@ async def create_api_key( session.add(api_key) await session.flush() return api_key, full_key + + +async def create_tenant_domain( + session: AsyncSession, + *, + tenant: Tenant, + domain: str = "general", + status: str = "active", +) -> TenantDomain: + """Register a domain so an upload to it passes the allowlist check. + + Uploads reject an unregistered domain (ADR-0009), so any test that uploads + needs one of these. + """ + tenant_domain = TenantDomain( + id=uuid.uuid4(), + tenant_id=tenant.id, + domain=domain, + display_name=domain, + status=status, + ) + session.add(tenant_domain) + await session.flush() + return tenant_domain