Multi-Tenant Codebases: Make the Unsafe Query Impossible to Write

# Nothing stops a caller forgetting the filter.
def get_orders(session, status: str) -> list[Order]:
    return session.query(Order).filter_by(status=status).all()

# Tenant comes from the request context, not from an argument,
# and there is no code path that omits it.
def get_orders(status: str) -> list[Order]:
    return (
        tenant_session()                 # raises if no tenant in context
        .query(Order)
        .filter_by(status=status)
        .all()
    )

def tenant_session() -> Session:
    tid = current_tenant.get()           # ContextVar, set by middleware
    if tid is None:
        raise NoTenantContext("No tenant bound to this request.")
    return session_factory(tenant_id=tid)   # filter applied at the session

## Multi-tenancy — read before touching data access
Every row in `orders`, `invoices`, `documents`, `audit_log` and
`attachments` belongs to exactly one tenant.

- NEVER construct a query against these tables directly. Use the
  helpers in `db/tenant_scope.py`, which bind the tenant from the
  request context.
- NEVER pass tenant_id as a function argument. It comes from the
  authenticated principal, and an argument is something a caller
  can get wrong.
- NEVER use `session.execute(text(...))` on a tenant table. Raw SQL
  bypasses the scoping. If you genuinely need raw SQL, stop and ask.
- Cache keys MUST include the tenant. `f"orders:{status}"` is a
  cross-tenant cache poisoning bug. Use `cache_key()`.
- Background jobs: the tenant must be an explicit part of the job
  payload and re-bound at the start of the handler. Context does
  not survive the queue.

If a task seems to require reading across tenants, stop and say so.
That is an admin operation with a separate, audited code path.

## Cross-tenant code
Lives ONLY in `src/admin/`. Nothing outside that directory may
construct an unscoped session.

- Every function there is decorated `@requires_platform_role` and
  `@audited`. Both. No exceptions.
- The audit record includes: actor, tenants touched, justification
  string, timestamp. The justification is a required argument.
- Never import anything from `src/admin/` into request-path code.

Adding a function here needs a second reviewer. Say so in the PR.

@pytest.mark.parametrize("endpoint", ALL_TENANT_SCOPED_ENDPOINTS)
def test_cannot_read_other_tenants_resource(client, endpoint):
    a = make_tenant("a"); b = make_tenant("b")
    resource = create_resource(tenant=a)

    r = client.get(endpoint.format(id=resource.id), auth=token(b))

    # 404, not 403 — 403 confirms the resource exists.
    assert r.status_code == 404
    assert str(resource.id) not in r.text

#!/bin/bash
HITS=$(git diff origin/main...HEAD -U0 | grep -E '^\+' \
  | grep -E 'text\(|execute\(' \
  | grep -Ei 'orders|invoices|documents|attachments|audit_log' || true)

if [ -n "$HITS" ]; then
  echo "Raw SQL against a tenant-scoped table:" >&2
  echo "$HITS" >&2
  echo "Use db/tenant_scope.py, or add 'raw-sql-reviewed' with a reason." >&2
  exit 1
fi