feat(python): add a customer directory example

AGENTS.md

@@ -96,6 +96,7 @@ Every npm package except EQL lives under `languages/typescript/`: packages in `l
 - `languages/python`: The Python SDK, `stackencrypt`, a pyo3 extension over `stack-encrypt`, built by maturin. **The fourth Cargo workspace**, excluded from the root one. It has two crates. `crates/core` has no pyo3: the engine-side logic, linked and tested without a Python interpreter. `crates/engine` is the `_engine` cdylib, `abi3-py39`, one wheel per platform for Python 3.9 to 3.14. `crates/core/fuzz` is a detached cargo-fuzz crate over the value conversion. The root `mise.toml`'s `fuzz:python-values` task runs it, from `fuzz.yml`, so it is the one Python task outside `languages/python/mise.toml`. `crates/core/testdata/value-conformance.json` is a copy of vitaminc's shared value corpus; its `README.md` records the commit it came from. `crates/core` compiles the Go guest's EQL resolver, `languages/golang/encrypt/guest/src/targets.rs`, in place through `#[path]`. So both SDKs run one resolver, and a change to that file is a change to both. The `extension-module` feature is set only by maturin in `pyproject.toml`, never in a crate manifest, so `cargo test` links libpython. Its tasks live in its own `mise.toml` and run from that folder. `mise run build` builds the wheel into `.venv`. `mise run test` runs pytest. `mise run test:cargo` runs plain `cargo test --locked` for both crates, with libpython on the loader path. Its extension tests include `crates/engine/src/heap_hygiene.rs`: a spy allocator fails the test when a block the extension frees during `encrypt`, `query` or `decrypt` still contains text or decimal plaintext. Python's own objects are out of its scope. It runs on Unix, and CI runs it on Linux. `mise run test:deterministic` rebuilds the extension with the `deterministic-kms` test feature, whose keys come from a seed through stack-kms's `DeterministicSource`. It then runs `tests_deterministic/` in its own pytest process: Python opens the EQL rows the Go SDK stored under that seed in `packages/eql/tests/encryption/fixtures/`, the frozen `stored_v4_rows_v1.json` among them, and seals the same values to the same terms. It does not use nextest, and its CI job sets no `MISE_ENV`. The nextest rule under "Working on the Rust crates" is for the root workspace, whose `stack-kms` tests set environment variables. No test in this workspace does, and nextest comes from `mise.test.toml`, which this workspace's CI does not load. The wheel's version is `crates/engine/Cargo.toml`'s. `stackencrypt.__engine__` is the commit of this checkout, recorded by `crates/core/build.rs`. Its `Cargo.lock` is seeded from the root one, so every crate it shares with the root workspace resolves to the version the root tests. After a root dependency moves, re-seed it the same way: copy the root `Cargo.lock` over it and run `cargo metadata --offline` there. That command prunes the copy to what the workspace uses. Until then, `tests-python.yml` fails on `--locked`. Nothing is published. CI is `tests-python.yml`. It runs the Rust checks and pytest on Python 3.9 and 3.14, where `MISE_PYTHON_VERSION` moves a leg off the local version. A `live` job runs `mise run test:live`: `encrypt`, `decrypt` and `query` against ZeroKMS with the repository's `CS_*` secrets, and the rows inserted into, searched in and read back from a `postgres:17` service with the EQL v4 bundle built from the tree. It skips on a fork's pull request, and the task fails on any skipped test. It then builds one abi3 wheel per FFI platform, uploads them as artifacts, and writes each build time to the job summary. `scripts/__tests__/python-ci.test.mjs` compares its path filter with the workspace's `Cargo.lock` and its wheel matrix with the protect-ffi platform packages. `languages/python/uv.lock` locks the `test` extra: pytest, attrs, Pydantic, SQLAlchemy, and psycopg. The package has no runtime dependency, so nothing else is locked. The middle three are for the declaration tests. psycopg is for the live tests. `mise run build` installs them with `uv sync --locked`, so a lock that does not match `pyproject.toml` fails the build. After a change to the `test` extra, run `uv lock` in `languages/python`. None has an upper bound. Dependabot's `uv` entry proposes each new minor or patch release as a lock update. maturin, the build backend, is not locked. `python/stackencrypt/_eql.py`, the EQL type markers, is generated from the EQL catalog by `eql-codegen py-eql`. To regenerate it, run `mise run types:generate` in `packages/eql`. Never edit it by hand: `tests-python.yml` runs its parity tests and fails on a hand edit. Design: the Python ADRs under `languages/python/docs/adr/` (ADR-0001, the declaration; ADR-0002, the call shape of `query`), the spec in #1341. `languages/python/docs/scorecard.md` marks the eight shared principles of `docs/sdk-design-principles.md` against the code. The README's SQL and Python blocks run in order in the `live` job, through `tests/test_readme_live.py`, so a change to a README step needs that test to pass.
 - `e2e/*`: Cross-package end-to-end tests (package managers, supply chain, Prisma example README)
 - `languages/typescript/examples/*`: Working apps (basic, prisma, supabase-worker)
+- `languages/python/examples/customer-directory`: An internal FastAPI example of the Python SDK, written twice: with psycopg (`app.py`) and with SQLAlchemy (`app_sqlalchemy.py`). It has its own `pyproject.toml` and `uv.lock`, and installs `stackencrypt` from this checkout. Its tests need real credentials, and CI does not run them.
 - `docs/plans/*`: Internal design plans. User-facing documentation lives at https://cipherstash.com/docs (not in this repo).
 - `skills/*`: Agent skills (`stash-cli`, `stash-encryption`, `stash-indexing`, `stash-deployment`, `stash-zerokms`, `stash-auth`, `stash-postgres`, `stash-edge`, `stash-drizzle`, `stash-dynamodb`, `stash-supabase`, `stash-prisma`, `stash-managed-platforms`, `stash-supply-chain-security`)
 

SECURITY.md

@@ -151,7 +151,7 @@ The following are **in scope**:
 
 The following are **out of scope**:
 
-- Example applications in the `languages/typescript/examples` dir (though we are still grateful for any relevant disclosures there)
+- Example applications in the `languages/typescript/examples` and `languages/python/examples` dirs (though we are still grateful for any relevant disclosures there)
 - Social engineering, physical attacks, or denial-of-service
 - Attacks requiring privileged access to developer machines or CI/CD infrastructure
 

languages/python/examples/customer-directory/README.md

@@ -0,0 +1,63 @@
+# Customer directory
+
+This is an internal example of the `stackencrypt` Python SDK. It is a FastAPI service that stores customers in PostgreSQL.
+
+| Field | Stored as |
+|---|---|
+| `email` | Encrypted, with exact-match search |
+| `name` | Encrypted, with no search |
+| `plan` | Plaintext |
+| `created_at` | Plaintext |
+
+The SDK is not released, so its API can change. This example installs the SDK from this checkout.
+
+## Before you start
+
+You need Docker, mise, and a Rust toolchain. mise installs the toolchain.
+
+You also need CipherStash credentials. Set `CS_CLIENT_ACCESS_KEY`, `CS_WORKSPACE_CRN`, `CS_CLIENT_ID`, and `CS_CLIENT_KEY`. Or log in with `stash auth login`.
+
+## Run it
+
+Run these commands from this folder.
+
+`mise run setup` deletes this example's database and creates it again. Nothing else is deleted.
+
+```sh
+mise run setup
+mise run dev
+```
+
+Add a customer, then find the customer by exact email:
+
+```sh
+curl -X POST http://127.0.0.1:8000/customers \
+  -H 'content-type: application/json' \
+  -d '{"email":"ada@example.com","name":"Ada Lovelace","plan":"free"}'
+
+curl -X POST http://127.0.0.1:8000/customers/search \
+  -H 'content-type: application/json' \
+  -d '{"email":"ada@example.com"}'
+
+curl 'http://127.0.0.1:8000/customers?limit=10'
+```
+
+The search is a `POST` so that the email stays out of URLs and access logs.
+
+## The SQLAlchemy version
+
+`app_sqlalchemy.py` is the same service with the model declared in SQLAlchemy. It uses the same table, so the two apps share rows. You can run both at the same time.
+
+```sh
+mise run dev:sqlalchemy   # http://127.0.0.1:8001
+```
+
+A SQLAlchemy object contains plaintext, and the columns must receive ciphertext. So the app calls `encrypt` before an `INSERT` and `decrypt` after a `SELECT`. It does not use `Session`.
+
+## Run the tests
+
+The tests use the real ZeroKMS service and the example's PostgreSQL. They run against both apps. They empty the `customers` table. CI does not run them.
+
+```sh
+mise run test
+```

languages/python/examples/customer-directory/app.py

@@ -0,0 +1,94 @@
+"""A customer directory: a FastAPI service that stores encrypted customers.
+
+`email` is encrypted and searchable by exact match. `name` is encrypted with
+no search. `plan` and `created_at` are stored as plaintext.
+
+The handlers are plain `def` functions. FastAPI runs them on a thread pool, so
+requests share the one `Cipher` below at the same time.
+"""
+
+import os
+from datetime import datetime, timezone
+from typing import Annotated, List, Optional
+
+import psycopg
+import stackencrypt
+from fastapi import FastAPI, HTTPException, Query, Request
+from fastapi.responses import JSONResponse
+from psycopg.rows import dict_row
+from pydantic import BaseModel, ConfigDict, Field
+from stackencrypt import Passthrough, Text, TextEq
+
+from errors import install_error_handlers, log_error, problem
+
+DATABASE_URL = os.environ["DATABASE_URL"]
+SELECT_CUSTOMER = "SELECT email, name, plan, created_at FROM customers"
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+@stackencrypt.model("customers")
+class Customer(BaseModel):
+    # A Pydantic error prints the input by default, and the input is plaintext.
+    model_config = ConfigDict(hide_input_in_errors=True)
+
+    email: Annotated[str, TextEq]
+    name: Annotated[str, Text]
+    plan: Annotated[str, Passthrough]
+    created_at: Annotated[datetime, Passthrough] = Field(default_factory=utc_now)
+
+
+class Search(BaseModel):
+    model_config = ConfigDict(hide_input_in_errors=True)
+
+    email: str
+
+
+# One Cipher for the process: making one checks the model and the credentials.
+cipher = stackencrypt.Cipher(Customer)
+
+app = FastAPI(title="Customer directory")
+
+
+install_error_handlers(app)
+
+
+@app.exception_handler(psycopg.errors.UniqueViolation)
+def duplicate(request: Request, error: psycopg.errors.UniqueViolation) -> JSONResponse:
+    log_error(error)
+    return problem(409, "A customer with this email already exists.")
+
+
+@app.post("/customers", status_code=201, response_model=Customer)
+def add_customer(customer: Customer) -> Customer:
+    row = cipher.encrypt([customer])[0]
+    with psycopg.connect(DATABASE_URL) as conn:
+        conn.execute(
+            "INSERT INTO customers (email, name, plan, created_at)"
+            " VALUES (%(email)s, %(name)s, %(plan)s, %(created_at)s)",
+            row,
+        )
+    return customer
+
+
+@app.post("/customers/search", response_model=Customer)
+def find_customer(search: Search) -> Customer:
+    # The email is in the body so that it stays out of URLs and access logs.
+    term = cipher.query("email", search.email)
+    with psycopg.connect(DATABASE_URL) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
+            cur.execute(f"{SELECT_CUSTOMER} WHERE email = %s::eql_v4.query_text_eq", (term,))
+            found: Optional[dict] = cur.fetchone()
+    if found is None:
+        raise HTTPException(status_code=404, detail="No customer has this email.")
+    return cipher.decrypt([found])[0]
+
+
+@app.get("/customers", response_model=List[Customer])
+def list_customers(limit: int = Query(default=100, ge=1, le=500)) -> List[Customer]:
+    with psycopg.connect(DATABASE_URL) as conn:
+        with conn.cursor(row_factory=dict_row) as cur:
+            cur.execute(f"{SELECT_CUSTOMER} ORDER BY id LIMIT %s", (limit,))
+            return cipher.decrypt(cur.fetchall())

languages/python/examples/customer-directory/app_sqlalchemy.py

@@ -0,0 +1,132 @@
+"""The customer directory again, with the model declared in SQLAlchemy.
+
+It uses the table of `app.py`, so the two apps share rows.
+
+A SQLAlchemy object contains plaintext, and the columns must receive
+ciphertext. So the app calls `encrypt` before an INSERT and `decrypt` after a
+SELECT. It does not use `Session`.
+"""
+
+import os
+from datetime import datetime, timezone
+from typing import Annotated, List, Optional
+
+import sqlalchemy as sa
+import stackencrypt
+from fastapi import FastAPI, HTTPException, Query, Request
+from fastapi.responses import JSONResponse
+from psycopg.errors import UniqueViolation
+from pydantic import BaseModel, ConfigDict, Field
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+from sqlalchemy.types import UserDefinedType
+from stackencrypt import Ignore, Passthrough, Text, TextEq
+
+from errors import install_error_handlers, log_error, problem
+
+# hide_parameters leaves the bound values out of SQLAlchemy's error text.
+# PostgreSQL's own detail can still quote the start of a rejected value.
+engine = sa.create_engine(
+    os.environ["DATABASE_URL"].replace("postgresql://", "postgresql+psycopg://", 1),
+    hide_parameters=True,
+)
+
+
+class EqlType(UserDefinedType):
+    """An EQL column type. The tests fail with `sa.Text`, `sa.String`, or `sa.JSON`."""
+
+    cache_ok = True
+
+    def __init__(self, sql_name: str) -> None:
+        self.sql_name = sql_name
+
+    def get_col_spec(self, **kwargs) -> str:
+        return self.sql_name
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+@stackencrypt.model("customers")
+class Customer(Base):
+    __tablename__ = "customers"
+
+    id: Mapped[Annotated[int, Ignore]] = mapped_column(
+        sa.BigInteger, sa.Identity(always=True), primary_key=True
+    )
+    email: Mapped[Annotated[str, TextEq]] = mapped_column(EqlType("public.eql_v4_text_eq"))
+    name: Mapped[Annotated[str, Text]] = mapped_column(EqlType("public.eql_v4_text"))
+    plan: Mapped[Annotated[str, Passthrough]]
+    created_at: Mapped[Annotated[datetime, Passthrough]] = mapped_column(
+        sa.DateTime(timezone=True)
+    )
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class CustomerBody(BaseModel):
+    """The JSON shape of a customer, for requests and responses."""
+
+    # A Pydantic error prints the input by default, and the input is plaintext.
+    model_config = ConfigDict(hide_input_in_errors=True, from_attributes=True)
+
+    email: str
+    name: str
+    plan: str
+    created_at: datetime = Field(default_factory=utc_now)
+
+
+class Search(BaseModel):
+    model_config = ConfigDict(hide_input_in_errors=True)
+
+    email: str
+
+
+# `id` is left out: the model marks it Ignore, and `decrypt` does not want it.
+COLUMNS = (Customer.email, Customer.name, Customer.plan, Customer.created_at)
+
+# One Cipher for the process: making one checks the model and the credentials.
+cipher = stackencrypt.Cipher(Customer)
+
+app = FastAPI(title="Customer directory (SQLAlchemy)")
+install_error_handlers(app)
+
+
+@app.exception_handler(IntegrityError)
+def integrity_error(request: Request, error: IntegrityError) -> JSONResponse:
+    log_error(error)
+    if isinstance(error.orig, UniqueViolation):
+        return problem(409, "A customer with this email already exists.")
+    return problem(500, "The request failed.")
+
+
+@app.post("/customers", status_code=201, response_model=CustomerBody)
+def add_customer(body: CustomerBody) -> CustomerBody:
+    rows = cipher.encrypt([Customer(**body.model_dump())])
+    with engine.begin() as conn:
+        conn.execute(sa.insert(Customer.__table__), rows)
+    return body
+
+
+@app.post("/customers/search", response_model=CustomerBody)
+def find_customer(search: Search) -> Customer:
+    # The email is in the body so that it stays out of URLs and access logs.
+    term = sa.literal(cipher.query("email", search.email), sa.Text)
+    query = sa.select(*COLUMNS).where(
+        Customer.email == sa.cast(term, EqlType("eql_v4.query_text_eq"))
+    )
+    with engine.connect() as conn:
+        found: Optional[sa.RowMapping] = conn.execute(query).mappings().first()
+    if found is None:
+        raise HTTPException(status_code=404, detail="No customer has this email.")
+    return cipher.decrypt([found])[0]
+
+
+@app.get("/customers", response_model=List[CustomerBody])
+def list_customers(limit: int = Query(default=100, ge=1, le=500)) -> List[Customer]:
+    query = sa.select(*COLUMNS).order_by(Customer.id).limit(limit)
+    with engine.connect() as conn:
+        return cipher.decrypt(conn.execute(query).mappings().all())

languages/python/examples/customer-directory/compose.yaml

@@ -0,0 +1,13 @@
+services:
+  db:
+    image: postgres:17
+    environment:
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: customer_directory
+    ports:
+      - 127.0.0.1:5438:5432
+    healthcheck:
+      test: [CMD-SHELL, pg_isready -U postgres -d customer_directory]
+      interval: 1s
+      timeout: 3s
+      retries: 30

languages/python/examples/customer-directory/errors.py

@@ -0,0 +1,60 @@
+"""The error handling that both apps share."""
+
+import logging
+
+import stackencrypt
+from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+
+log = logging.getLogger("customer_directory")
+
+
+def log_error(error: Exception) -> None:
+    """Log what identifies an error, and never the error's own text.
+
+    An SDK error's message never contains a value, so `code` and `field` are
+    safe. PostgreSQL's primary message has no value in it. Its detail can
+    quote the value, so the detail is not logged. Any other exception's text
+    could contain plaintext, so only its type is logged.
+    """
+    # SQLAlchemy wraps the driver's error and keeps it in `orig`.
+    cause = getattr(error, "orig", error)
+    diag = getattr(cause, "diag", None)
+    log.error(
+        "customer_error type=%s code=%s field=%s sqlstate=%s message=%s",
+        type(error).__name__,
+        getattr(error, "code", None) if cause is error else None,
+        getattr(error, "field", None),
+        getattr(cause, "sqlstate", None),
+        getattr(diag, "message_primary", None),
+    )
+
+
+def problem(status: int, detail: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"detail": detail})
+
+
+def install_error_handlers(app: FastAPI) -> None:
+    """Give an app fixed error responses, with the details in the log."""
+
+    @app.middleware("http")
+    async def fixed_500(request: Request, call_next):
+        # This catches the error here because an `Exception` handler would
+        # still re-raise it, and Uvicorn would then log its text.
+        try:
+            return await call_next(request)
+        except Exception as error:
+            log_error(error)
+            return problem(500, "The request failed.")
+
+    @app.exception_handler(RequestValidationError)
+    def invalid_request(request: Request, error: RequestValidationError) -> JSONResponse:
+        # FastAPI's default response repeats the rejected input.
+        log_error(error)
+        return problem(422, "A field has an invalid value.")
+
+    @app.exception_handler(stackencrypt.InvalidValueError)
+    def invalid_value(request: Request, error: stackencrypt.InvalidValueError) -> JSONResponse:
+        log_error(error)
+        return problem(422, "A field has an invalid value.")