feat: tenancy — products + API keys

Product is the only tenant table this service owns (porte table:
organization_id -> product_id, everything below it -- tenant_ref/
branch_ref -- stays an opaque string owned by the consuming product, never
a row here). Bcrypt-hashed API keys (passlib, same CryptContext shape as
the auto's auth.service), generated once by scripts/create_product.py and
never persisted in the clear. require_product (X-Api-Key -> Product, 401
on missing/wrong/soft-deleted) is the porte adaptation of the auto's
require_permission — every /v1/* route will depend on it instead of a JWT
bearer token.

Also ports shared/base_model.py and shared/errors.py verbatim (Global
Constraints: soft delete mixins, structured 409 bodies).

products migration + real "alembic upgrade head" round-trip test
(tests/migrations/, new shared _helpers.py instead of the auto's ad hoc
cross-file _run_psql reuse). 13 tests green via `make k8s-test`.

alembic/env.py

@@ -12,7 +12,7 @@ from fiscal_svc.core.db import Base
 
 # Uncomment as modules gain SQLAlchemy models, so autogenerate can see them
 # (mirrors auto/backend/alembic/env.py's own convention):
-# from fiscal_svc.tenancy import models as tenancy_models  # noqa: F401
+from fiscal_svc.tenancy import models as tenancy_models  # noqa: F401
 # from fiscal_svc.documents import models as documents_models  # noqa: F401
 
 # this is the Alembic Config object, which provides

alembic/versions/fd0ffe65c993_products.py

@@ -0,0 +1,42 @@
+"""products (Task 2 -- tenancy): the only tenant table this service owns.
+`api_key_hash` is bcrypt ciphertext (never the plaintext key -- see
+`scripts/create_product.py`); `webhook_secret` is nullable, unused until F4.
+
+Revision ID: fd0ffe65c993
+Revises:
+Create Date: 2026-07-22 00:00:00.000000
+
+"""
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "fd0ffe65c993"
+down_revision: Union[str, Sequence[str], None] = None
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "products",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("name", sa.String(length=255), nullable=False),
+        sa.Column("api_key_hash", sa.String(length=255), nullable=False),
+        sa.Column("webhook_secret", sa.String(length=255), nullable=True),
+        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("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("products")

scripts/create_product.py

@@ -0,0 +1,50 @@
+"""CLI: provisions a new `Product` (a SowAI app consuming this service --
+auto, crm, ...) and prints its PLAINTEXT API key exactly ONCE. Only the
+bcrypt hash of the key is ever persisted (`tenancy.models.Product`'s
+docstring) -- if the printed key is lost, the only recovery is provisioning
+a new Product (or, once key rotation exists, rotating it; not built yet).
+
+There is no HTTP endpoint for this on purpose (design spec, decision #5:
+API keys are per-product and provisioning one is an operator action, not
+self-service) -- this script is the ONLY writer of `products`.
+
+Run against the target DB via the app's own session config (DATABASE_URL
+env, same convention as `auto/backend/scripts/seed_dev_admin.py`):
+
+    uv run python scripts/create_product.py --name auto
+"""
+import argparse
+import asyncio
+import secrets
+
+from fiscal_svc.core.db import async_session_maker
+from fiscal_svc.tenancy.service import create_product
+
+
+async def _main(name: str) -> None:
+    # `secrets.token_urlsafe` (CSPRNG, not `random`) -- this string IS the
+    # credential, same rigor as any password/token generated in this
+    # codebase. 32 bytes of entropy before urlsafe-base64 encoding.
+    api_key = secrets.token_urlsafe(32)
+    async with async_session_maker() as session:
+        product = await create_product(session, name=name, api_key=api_key)
+
+    print(f"Product criado: {product.name} (id={product.id})")
+    print()
+    print("API key -- copie agora. Só o hash bcrypt fica no banco; esta é a")
+    print("ÚNICA vez que o valor em claro é exibido:")
+    print()
+    print(api_key)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Provisiona um novo Product (consumidor da API do sowai-fiscal-svc)."
+    )
+    parser.add_argument("--name", required=True, help="Nome do produto SowAI (ex.: 'auto').")
+    args = parser.parse_args()
+    asyncio.run(_main(args.name))
+
+
+if __name__ == "__main__":
+    main()

src/fiscal_svc/shared/base_model.py

@@ -0,0 +1,30 @@
+"""Ported verbatim from `auto/backend/app/shared/base_model.py` (Task 2) --
+no adaptation needed, these three mixins carry no tenancy-shaped columns of
+their own."""
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, func
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+
+class UUIDPKMixin:
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+
+
+class TimestampMixin:
+    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()
+    )
+
+
+class SoftDeleteMixin:
+    deleted_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), default=None
+    )

src/fiscal_svc/shared/errors.py

@@ -0,0 +1,11 @@
+"""Ported verbatim from `auto/backend/app/shared/errors.py` (Task 2) --
+structured 409 error bodies for money/data-integrity conflicts. Every OTHER
+409 in this service (like the auto) is a plain `HTTPException(409,
+detail=str(exc))`; `conflict()` is for the ones a consumer needs to branch
+on programmatically (e.g. a produto's own retry logic reacting to
+`certificate_upload_conflict` vs `duplicate_fiscal_series` differently)."""
+from fastapi import HTTPException, status
+
+
+def conflict(code: str, message: str) -> HTTPException:
+    return HTTPException(status_code=status.HTTP_409_CONFLICT, detail={"code": code, "message": message})

src/fiscal_svc/tenancy/deps.py

@@ -0,0 +1,35 @@
+from fastapi import Depends, Header, HTTPException, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fiscal_svc.core.db import get_session
+from fiscal_svc.tenancy.models import Product
+from fiscal_svc.tenancy.service import authenticate_product
+
+
+async def require_product(
+    x_api_key: str | None = Header(default=None, alias="X-Api-Key"),
+    session: AsyncSession = Depends(get_session),
+) -> Product:
+    """Adaptation of the auto's `auth.deps.get_current_user`/
+    `require_permission` (porte table: `require_permission(...)` ->
+    `require_product(...)`, API key -> Product): every `/v1/*` route in this
+    service depends on this instead of a JWT bearer token -- there is no
+    per-request USER here, only a per-request PRODUCT (the SowAI app calling
+    on behalf of one of ITS OWN tenants, identified by the opaque
+    `tenant_ref`/`branch_ref` the request body/query carries, never resolved
+    by this service).
+
+    401 (not 403) for BOTH a missing header and a key that matches no live
+    product -- same "don't distinguish absent-credential from wrong-
+    credential" fail-closed shape as a bearer token that fails to decode in
+    `get_current_user`. A soft-deleted product's key also 401s here (
+    `authenticate_product` only looks at `deleted_at IS NULL` rows), same
+    "revoked = gone" semantics as a deactivated auto user."""
+    if x_api_key is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="API key ausente")
+
+    product = await authenticate_product(session, x_api_key)
+    if product is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="API key inválida")
+
+    return product

src/fiscal_svc/tenancy/models.py

@@ -0,0 +1,43 @@
+from sqlalchemy import String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from fiscal_svc.core.db import Base
+from fiscal_svc.shared.base_model import SoftDeleteMixin, TimestampMixin, UUIDPKMixin
+
+
+class Product(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin):
+    """A SowAI product consuming this service (auto, crm, ...) -- the ONLY
+    tenant concept this service itself owns (Task 2). Per the porte table
+    (2026-07-17-fiscal-svc-f2-servico.md): the auto's `Organization`/`Branch`
+    rows have no equivalent here -- `product_id` (this table's PK) is the
+    top of the tenancy hierarchy, and everything below it
+    (`tenant_ref`/`branch_ref`) is an OPAQUE string the product hands us,
+    never a row this service owns or validates the shape of (decision #4 of
+    the design spec: "o serviço não conhece o modelo de org de nenhum
+    produto").
+
+    `api_key_hash`: bcrypt (via passlib, same `CryptContext(schemes=
+    ["bcrypt"])` as the auto's `auth.service.hash_password`), generated by
+    `scripts/create_product.py` and never stored/logged in the clear -- see
+    that script's docstring for the "printed once" flow. Auth (Task 2's
+    `require_product`) has no per-product identifier to look the row up
+    BY (unlike the auto's `authenticate_user`, which first resolves the User
+    by `organization_slug` + `email` and THEN verifies the password against
+    THAT ONE row) -- an API key IS the only credential, so `require_product`
+    must verify the candidate key against every live product's hash instead
+    of a single one. That's fine at this service's cardinality: `products`
+    holds one row per SowAI PRODUCT (auto, crm, ...), not per end customer --
+    low tens at most, not the O(n) problem it would be for user accounts.
+
+    `webhook_secret`: nullable HMAC secret for the `documento.status_changed`
+    webhook (design spec's decision #5, F4) -- generated/rotated the same
+    way as `api_key_hash` once that feature lands; nullable now so this
+    migration doesn't need to change shape later, mirroring how the auto's
+    `Branch` fiscal columns were added nullable ahead of the feature that
+    fills them (Bloco A)."""
+
+    __tablename__ = "products"
+
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    api_key_hash: Mapped[str] = mapped_column(String(255), nullable=False)
+    webhook_secret: Mapped[str | None] = mapped_column(String(255), nullable=True)

src/fiscal_svc/tenancy/service.py

@@ -0,0 +1,50 @@
+from passlib.context import CryptContext
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fiscal_svc.tenancy.models import Product
+
+# Same scheme/config as the auto's `auth.service._pwd_context`
+# (`app/modules/auth/service.py`) -- bcrypt via passlib, `deprecated="auto"`
+# so passlib itself flags (and re-hashes on next write) any hash produced by
+# a scheme this CryptContext no longer lists as current.
+_pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def hash_api_key(plain: str) -> str:
+    return _pwd_context.hash(plain)
+
+
+def verify_api_key(plain: str, hashed: str) -> bool:
+    return _pwd_context.verify(plain, hashed)
+
+
+async def create_product(session: AsyncSession, name: str, api_key: str) -> Product:
+    """Used by `scripts/create_product.py` -- the ONLY writer of `products`
+    for now (no HTTP endpoint; provisioning a new SowAI product consumer is
+    an operator action, not self-service). `api_key` is the PLAINTEXT key
+    the caller generated; only its bcrypt hash is persisted."""
+    product = Product(name=name, api_key_hash=hash_api_key(api_key))
+    session.add(product)
+    await session.commit()
+    await session.refresh(product)
+    return product
+
+
+async def authenticate_product(session: AsyncSession, api_key: str) -> Product | None:
+    """Resolves a plaintext `X-Api-Key` to its `Product`, or None if it
+    matches no live product.
+
+    Unlike the auto's `authenticate_user` (which first narrows to ONE row
+    via `organization_slug` + `email`, THEN verifies a password against that
+    single hash), an API key carries no separate identifier to narrow the
+    lookup by -- bcrypt hashes are salted, so `WHERE api_key_hash = hash(
+    candidate)` can never match. This verifies the candidate against every
+    LIVE (`deleted_at IS NULL`) product's hash in turn, first match wins.
+    See `Product`'s docstring for why that's fine at this table's
+    cardinality (one row per SowAI PRODUCT, not per end customer)."""
+    result = await session.execute(select(Product).where(Product.deleted_at.is_(None)))
+    for product in result.scalars().all():
+        if verify_api_key(api_key, product.api_key_hash):
+            return product
+    return None

tests/migrations/_helpers.py

@@ -0,0 +1,49 @@
+"""Shared plumbing for the "real migration" tests (Global Constraints:
+every migration ships with a test that runs `alembic upgrade head` as a
+REAL subprocess against a disposable database -- never just
+`Base.metadata.create_all`, which builds the schema straight from the
+current model definitions and so can never catch a migration that drifted
+from them). Mirrors the technique `auto/backend/tests/migrations/` uses
+(e.g. `test_seed_cadastros_peca_pessoa_editar_permission.py`'s `_run_psql`),
+centralized here instead of re-imported test-file-to-test-file -- this repo
+starts that convention fresh rather than porting the auto's ad hoc
+cross-file reuse."""
+import os
+import subprocess
+from pathlib import Path
+
+REPO_DIR = Path(__file__).resolve().parents[2]
+
+_PSQL = "psql"
+
+
+def run_psql(*args: str) -> subprocess.CompletedProcess:
+    """Runs `psql -U postgres -h localhost <args>` against the pod's
+    Postgres sidecar (trust auth -- see `auto/k8s/test-runner.yaml`'s
+    header comment for why localhost/no-password is assumed) and asserts
+    it succeeded."""
+    result = subprocess.run(
+        [_PSQL, "-U", "postgres", "-h", "localhost", *args],
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+    assert result.returncode == 0, f"psql failed: {result.stderr}"
+    return result
+
+
+def run_alembic(database_url: str, *args: str) -> subprocess.CompletedProcess:
+    """Runs `uv run alembic <args>` as a subprocess with `DATABASE_URL`
+    pointed at `database_url` (never the service's own `settings.
+    database_url` default) -- `alembic/env.py` reads `DATABASE_URL` via
+    `fiscal_svc.core.config.settings`, which layers real env vars over
+    `.env`."""
+    env = {**os.environ, "DATABASE_URL": database_url}
+    return subprocess.run(
+        ["uv", "run", "alembic", *args],
+        cwd=str(REPO_DIR),
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=120,
+    )

tests/migrations/test_products_schema.py

@@ -0,0 +1,79 @@
+"""Migration test (Task 2): `products` -- confirms the table/columns exist
+via `information_schema` over the `create_all` schema (Teste A, same
+precedent as the auto's `test_fiscal_cadastro_schema.py`), and that a
+`Product` round-trips (INSERT cru via the real ORM/service) on a database
+built PURELY by `alembic upgrade head` (Teste B) -- never
+`Base.metadata.create_all`, which would never catch the migration itself
+drifting from the model."""
+import pytest
+import pytest_asyncio
+from sqlalchemy import select, text
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+# Import de registro: garante que Product está registrado em Base.metadata
+# antes do create_all da fixture test_engine, ao rodar este arquivo isolado.
+from fiscal_svc.tenancy import models as _tenancy_models  # noqa: F401
+from tests.migrations._helpers import run_alembic, run_psql
+
+_MIGRATION_DB_NAME = "fiscal_svc_test_products_schema"
+_MIGRATION_DB_URL = (
+    f"postgresql+asyncpg://postgres:postgres@localhost:5432/{_MIGRATION_DB_NAME}"
+)
+
+
+@pytest.mark.asyncio
+async def test_products_table_and_columns_exist(db_session):
+    rows = {
+        r[0]: r[1]
+        for r in (
+            await db_session.execute(
+                text(
+                    "select column_name, is_nullable from information_schema.columns "
+                    "where table_name='products'"
+                )
+            )
+        )
+    }
+    for required in ("id", "name", "api_key_hash", "webhook_secret", "created_at", "deleted_at"):
+        assert required in rows, f"coluna {required} ausente em products"
+    assert rows["name"] == "NO"
+    assert rows["api_key_hash"] == "NO"
+    assert rows["webhook_secret"] == "YES"
+    assert rows["deleted_at"] == "YES"
+
+
+@pytest_asyncio.fixture
+async def migration_database():
+    run_psql("-c", f"DROP DATABASE IF EXISTS {_MIGRATION_DB_NAME} WITH (FORCE);")
+    run_psql("-c", f"CREATE DATABASE {_MIGRATION_DB_NAME};")
+    yield
+    run_psql("-c", f"DROP DATABASE IF EXISTS {_MIGRATION_DB_NAME} WITH (FORCE);")
+
+
+@pytest.mark.asyncio
+async def test_product_round_trips_on_a_real_migrated_database(migration_database):
+    result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+    assert result.returncode == 0, result.stderr
+
+    from fiscal_svc.tenancy.service import create_product
+
+    engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        async with session_maker() as session:
+            product = await create_product(session, name="auto", api_key="plain-key-123")
+            product_id = product.id
+            # The plaintext key is NEVER what lands in the column.
+            assert product.api_key_hash != "plain-key-123"
+
+        async with session_maker() as session:
+            from fiscal_svc.tenancy.models import Product
+
+            reloaded = await session.get(Product, product_id)
+            assert reloaded.name == "auto"
+            assert reloaded.deleted_at is None
+
+            result_q = await session.execute(select(Product).where(Product.name == "auto"))
+            assert result_q.scalar_one().id == product_id
+    finally:
+        await engine.dispose()

tests/tenancy/test_tenancy.py

@@ -0,0 +1,132 @@
+"""Task 2: `Product` tenancy -- API key hashing/verification
+(`tenancy.service`) and the `require_product` FastAPI dependency
+(`tenancy.deps`), the adaptation of the auto's `require_permission(...)`
+(porte table: API key -> Product, no JWT/user here).
+
+Every key below is generated with a `uuid4` suffix (`_key()`), never a
+fixed literal: `products` has NO tenant scoping to isolate one test's rows
+from another's (unlike the auto's org-scoped tables, where a fixed literal
+is safe because every query also filters by that test's own
+`organization_id`) -- `authenticate_product` deliberately scans ALL live
+products (see its docstring), so a fixed key reused across two tests would
+have the OLDER, still-committed product (this suite's `db_session` fixture
+only rolls back; it does not truncate what earlier tests committed) answer
+for the newer test's assertions instead of a genuine 404/401."""
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+from fastapi import HTTPException
+
+from fiscal_svc.tenancy.deps import require_product
+from fiscal_svc.tenancy.service import authenticate_product, create_product, hash_api_key
+
+
+def _key() -> str:
+    return f"k-{uuid.uuid4().hex}"
+
+
+@pytest.mark.asyncio
+async def test_create_product_never_persists_the_plaintext_key(db_session):
+    key = _key()
+    product = await create_product(db_session, name="auto", api_key=key)
+    assert product.api_key_hash != key
+    # bcrypt's own marker -- proves passlib actually hashed it, not just
+    # stored some other opaque transform.
+    assert product.api_key_hash.startswith("$2b$")
+
+
+@pytest.mark.asyncio
+async def test_authenticate_product_returns_product_for_the_correct_key(db_session):
+    key_a, key_b = _key(), _key()
+    product = await create_product(db_session, name="auto", api_key=key_a)
+    await create_product(db_session, name="crm", api_key=key_b)
+
+    found = await authenticate_product(db_session, key_a)
+
+    assert found is not None
+    assert found.id == product.id
+
+
+@pytest.mark.asyncio
+async def test_authenticate_product_returns_none_for_a_wrong_key(db_session):
+    await create_product(db_session, name="auto", api_key=_key())
+
+    assert await authenticate_product(db_session, _key()) is None
+
+
+@pytest.mark.asyncio
+async def test_authenticate_product_ignores_soft_deleted_products(db_session):
+    """A revoked/deactivated Product's key must stop authenticating --
+    same 'revoked = gone' semantics as the auto's `get_active_user` ignoring
+    a soft-deleted User."""
+    key = _key()
+    product = await create_product(db_session, name="auto", api_key=key)
+    product.deleted_at = datetime.now(timezone.utc)
+    await db_session.commit()
+
+    assert await authenticate_product(db_session, key) is None
+
+
+@pytest.mark.asyncio
+async def test_authenticate_product_distinguishes_products_with_similar_keys(db_session):
+    """Not a narrowed lookup (no identifier besides the key itself) --
+    proves the "verify against every live product" loop in
+    `authenticate_product` returns the RIGHT product, not just A product."""
+    key_a, key_b = _key(), _key()
+    product_a = await create_product(db_session, name="auto", api_key=key_a)
+    product_b = await create_product(db_session, name="crm", api_key=key_b)
+
+    found_a = await authenticate_product(db_session, key_a)
+    found_b = await authenticate_product(db_session, key_b)
+
+    assert found_a.id == product_a.id
+    assert found_b.id == product_b.id
+
+
+@pytest.mark.asyncio
+async def test_require_product_missing_header_is_401(db_session):
+    with pytest.raises(HTTPException) as exc_info:
+        await require_product(x_api_key=None, session=db_session)
+    assert exc_info.value.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_require_product_wrong_key_is_401(db_session):
+    key = _key()
+    await create_product(db_session, name="auto", api_key=key)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await require_product(x_api_key=f"not-{key}", session=db_session)
+    assert exc_info.value.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_require_product_soft_deleted_product_is_401(db_session):
+    key = _key()
+    product = await create_product(db_session, name="auto", api_key=key)
+    product.deleted_at = datetime.now(timezone.utc)
+    await db_session.commit()
+
+    with pytest.raises(HTTPException) as exc_info:
+        await require_product(x_api_key=key, session=db_session)
+    assert exc_info.value.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_require_product_valid_key_returns_the_product(db_session):
+    key = _key()
+    product = await create_product(db_session, name="auto", api_key=key)
+
+    result = await require_product(x_api_key=key, session=db_session)
+
+    assert result.id == product.id
+
+
+def test_hash_api_key_is_salted_non_deterministic():
+    """Same input, two different hashes -- this is EXACTLY why
+    `authenticate_product` cannot do `WHERE api_key_hash = hash(candidate)`
+    and must verify against each live product's hash instead (see its
+    docstring)."""
+    same_input = _key()
+    assert hash_api_key(same_input) != hash_api_key(same_input)