Bind sender identities, add operator read surface, real serving layer

AUDIT-WP-0004 T03, T05, T06.

T03 - the receiver accepted whatever tenant and source a caller sent as long
as it held the one shared token, despite WP-0003 recording tenant isolation as
delivered. audit_core.senders binds each credential to the sources and tenants
it may assert, driven by AUDIT_CORE_SENDERS rather than literals. Identities
hold a list of tokens so rotation publishes the replacement alongside the
incumbent and needs no delivery gap. Read is a separate privilege from write,
so a sender credential cannot read the audit trail back.

T05 - lookup by event id, lookup by correlation id, and a dead-letter view.
Rejections are recorded rather than silently dropped. An event rejected for
carrying secret-shaped material has its payload withheld: storing it would
write that material into the audit store, which is what the rejection exists
to prevent. Reason and payload hash are kept so it stays traceable.

Replay is deliberately not built here. Idempotent replay is a property of the
durable store and building it against SQLite would produce a second
implementation to throw away; it lands with the Postgres backend in
AUDIT-WP-0005-T01.

T06 - serving moves to waitress with configurable threads and channel timeout,
installed in the image via the serve extra. Without it the entrypoint falls
back to a threaded wsgiref server with a socket timeout and graceful shutdown
on SIGTERM, and logs a warning so a deployment cannot quietly land on the
fallback. Metric counters deferred to WP-0005-T03 to be designed against the
real scrape path.

Tests 36 -> 46, covering cross-tenant and cross-source refusal, token
rotation, read/write privilege separation, correlation lookup, and payload
withholding on secret rejection.

Remaining in WP-0004: T04 redaction policy, which needs a decision on whether
a secret-shaped field is a rejection or a redaction.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

audit_core/ingestion.py

@@ -21,15 +21,15 @@ these, so they are part of the interface, not an implementation detail:
 from __future__ import annotations
 
 import hashlib
-import hmac
 import json
 import logging
 import os
+import signal
 import sys
 from datetime import datetime, timezone
 from http import HTTPStatus
 from typing import Any
-from wsgiref.simple_server import make_server
+from urllib.parse import parse_qs
 
 from audit_core.interface import (
     AuditEvent,
@@ -38,6 +38,7 @@ from audit_core.interface import (
     EventValidationError,
     IdempotentAuditBackend,
 )
+from audit_core.senders import SenderIdentity, SenderRegistry, development_registry
 from audit_core.sqlite_backend import SQLiteAuditBackend
 
 MAX_BODY_BYTES = 256 * 1024
@@ -54,9 +55,9 @@ class IngestionApplication:
     event.
     """
 
-    def __init__(self, backend: IdempotentAuditBackend, bearer_token: str) -> None:
-        if not bearer_token:
-            raise ValueError("bearer token is required")
+    def __init__(
+        self, backend: IdempotentAuditBackend, senders: SenderRegistry | str
+    ) -> None:
         policy = backend.retention_policy
         if not policy.durable:
             # The mock file backend declares durable=False. Refusing it here is
@@ -66,8 +67,12 @@ class IngestionApplication:
                 f"backend custody_class={policy.custody_class!r} is not durable; "
                 "refusing to accept audit events against it"
             )
+        if isinstance(senders, str):
+            if not senders:
+                raise ValueError("bearer token is required")
+            senders = development_registry(senders)
         self.backend = backend
-        self.token = bearer_token
+        self.senders = senders
 
     def __call__(self, environ, start_response):
         try:
@@ -87,18 +92,30 @@ class IngestionApplication:
             return self._json(start_response, HTTPStatus.OK, {"status": "ok"})
         if path == "/readyz":
             return self._readiness(start_response)
-        if path != "/v1/events" or environ.get("REQUEST_METHOD") != "POST":
-            return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"})
-
-        if not self._authorized(environ):
+        method = environ.get("REQUEST_METHOD")
+        identity = self.senders.authenticate(environ.get("HTTP_AUTHORIZATION"))
+        if identity is None:
             return self._json(
                 start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"}
             )
 
+        if method == "GET" and (path.startswith("/v1/events") or path == "/v1/dead-letters"):
+            return self._read(start_response, environ, path, identity)
+
+        if path != "/v1/events" or method != "POST":
+            return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"})
+
+        if not identity.may_write:
+            return self._json(start_response, HTTPStatus.FORBIDDEN, {"error": "write_forbidden"})
+
+        raw = b""
         try:
             raw = self._read_body(environ)
-            event = normalize(json.loads(raw), environ.get("HTTP_IDEMPOTENCY_KEY"))
+            event = normalize(
+                json.loads(raw), environ.get("HTTP_IDEMPOTENCY_KEY"), identity
+            )
         except (ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
+            self._dead_letter(raw, str(exc), identity)
             return self._json(start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)})
 
         try:
@@ -123,15 +140,68 @@ class IngestionApplication:
             },
         )
 
-    def _authorized(self, environ) -> bool:
-        supplied = str(environ.get("HTTP_AUTHORIZATION", ""))
-        expected = f"Bearer {self.token}"
+    def _read(self, start_response, environ, path: str, identity):
+        """Operator read surface (AUDIT-WP-0004-T05).
+
+        Read is a distinct privilege from write: a sender credential must not
+        be able to read the audit trail back.
+        """
+        if not identity.may_read:
+            return self._json(start_response, HTTPStatus.FORBIDDEN, {"error": "read_forbidden"})
+
+        query = parse_qs(environ.get("QUERY_STRING", ""))
         try:
-            return hmac.compare_digest(supplied, expected)
-        except TypeError:
-            # compare_digest rejects non-ASCII str. A header containing one is
-            # simply not a valid credential.
-            return False
+            if path == "/v1/dead-letters":
+                return self._json(
+                    start_response, HTTPStatus.OK,
+                    {"dead_letters": self.backend.dead_letters(_limit(query))},
+                )
+            if path == "/v1/events":
+                correlation = (query.get("correlation_id") or [""])[0]
+                if not correlation:
+                    return self._json(
+                        start_response, HTTPStatus.BAD_REQUEST,
+                        {"error": "correlation_id_required"},
+                    )
+                return self._json(
+                    start_response, HTTPStatus.OK,
+                    {"events": self.backend.by_correlation(correlation, _limit(query))},
+                )
+            event_id = path[len("/v1/events/"):]
+            record = self.backend.get(event_id) if event_id else None
+            if record is None:
+                return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"})
+            return self._json(start_response, HTTPStatus.OK, record)
+        except BackendUnavailableError as exc:
+            log.error("read failed: %s", exc)
+            return self._json(
+                start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"error": "backend_unavailable"}
+            )
+
+    def _dead_letter(self, raw: bytes, reason: str, identity) -> None:
+        """Record a rejection so an operator can see what the sender dropped."""
+        recorder = getattr(self.backend, "record_rejection", None)
+        if not callable(recorder):
+            return
+        event_id = None
+        try:
+            parsed = json.loads(raw)
+            if isinstance(parsed, dict):
+                event_id = str(parsed.get("id") or "") or None
+        except (ValueError, TypeError):
+            pass
+        try:
+            recorder(
+                event_id=event_id,
+                reason=reason,
+                payload_hash=hashlib.sha256(raw).hexdigest(),
+                sender=identity.name,
+                payload=raw.decode("utf-8", "replace"),
+            )
+        except BackendUnavailableError as exc:
+            # A rejection we could not record is worth a log line, but it must
+            # not turn a 400 into a 503 — the event is still rejected.
+            log.error("could not record dead letter: %s", exc)
 
     def _read_body(self, environ) -> bytes:
         try:
@@ -174,7 +244,11 @@ class IngestionApplication:
         return [body]
 
 
-def normalize(payload: dict[str, Any], idempotency_key: str | None) -> AuditEvent:
+def normalize(
+    payload: dict[str, Any],
+    idempotency_key: str | None,
+    identity: SenderIdentity | None = None,
+) -> AuditEvent:
     required = (
         "id", "type", "source", "subject", "tenant", "correlation_id", "occurred_at", "data",
     )
@@ -182,7 +256,16 @@ def normalize(payload: dict[str, Any], idempotency_key: str | None) -> AuditEven
         raise ValueError("invalid_event")
     if idempotency_key != payload["id"]:
         raise ValueError("idempotency_key_mismatch")
-    if payload["source"] != "user-engine":
+    source = str(payload["source"])
+    tenant = str(payload["tenant"])
+    # The claimed source and tenant are checked against what this credential is
+    # permitted to assert, not against a literal (AUDIT-WP-0004-T03).
+    if identity is not None:
+        if not identity.permits_source(source):
+            raise ValueError("source_not_allowed")
+        if not identity.permits_tenant(tenant):
+            raise ValueError("tenant_not_allowed")
+    elif source != "user-engine":
         raise ValueError("source_not_allowed")
     observed_at = _normalize_timestamp(payload["occurred_at"])
     if _contains_secret(payload["data"]):
@@ -190,9 +273,9 @@ def normalize(payload: dict[str, Any], idempotency_key: str | None) -> AuditEven
     return AuditEvent(
         event_id=str(payload["id"]),
         observed_at=observed_at,
-        tenant=str(payload["tenant"]),
+        tenant=tenant,
         scope="tenant",
-        source="user-engine",
+        source=source,
         action=str(payload["type"]),
         resource=str(payload["subject"]),
         outcome="recorded",
@@ -228,6 +311,62 @@ def _contains_secret(value: Any) -> bool:
     return False
 
 
+def _limit(query: dict[str, list[str]], default: int = 100, ceiling: int = 1000) -> int:
+    try:
+        return max(1, min(int((query.get("limit") or [default])[0]), ceiling))
+    except (TypeError, ValueError):
+        return default
+
+
+def serve(app, host: str, port: int, threads: int, timeout: int) -> None:
+    """Serve ``app``, preferring a production WSGI server (T06).
+
+    waitress is the intended production server and is installed in the image
+    via the ``serve`` extra. The fallback is a threaded wsgiref server with a
+    socket timeout — bounded rather than good, and loud about which one is in
+    use so a deployment cannot quietly end up on the fallback.
+    """
+    try:
+        from waitress import serve as waitress_serve
+    except ImportError:
+        log.warning(
+            "waitress not installed — falling back to a threaded wsgiref server. "
+            "Install the 'serve' extra for production (AUDIT-WP-0004-T06)."
+        )
+        _serve_fallback(app, host, port, timeout)
+        return
+
+    log.info("serving on waitress host=%s port=%s threads=%s", host, port, threads)
+    waitress_serve(
+        app, host=host, port=port, threads=threads,
+        channel_timeout=timeout, ident="audit-core",
+    )
+
+
+def _serve_fallback(app, host: str, port: int, timeout: int) -> None:
+    from socketserver import ThreadingMixIn
+    from wsgiref.simple_server import WSGIServer, make_server
+
+    class ThreadedWSGIServer(ThreadingMixIn, WSGIServer):
+        daemon_threads = True
+        # Without this a slow or idle client holds a worker indefinitely; the
+        # original single-threaded server let one such client block every
+        # sender.
+        timeout = timeout
+
+    with make_server(host, port, app, server_class=ThreadedWSGIServer) as server:
+        server.socket.settimeout(timeout)
+
+        def shutdown(signum, _frame):
+            log.info("received signal %s, shutting down", signum)
+            server.shutdown()
+
+        for sig in (signal.SIGTERM, signal.SIGINT):
+            signal.signal(sig, shutdown)
+        log.info("serving on threaded wsgiref host=%s port=%s", host, port)
+        server.serve_forever()
+
+
 def main() -> None:
     logging.basicConfig(
         level=os.environ.get("AUDIT_CORE_LOG_LEVEL", "INFO"),
@@ -237,11 +376,11 @@ def main() -> None:
     backend = SQLiteAuditBackend(
         os.environ.get("AUDIT_CORE_DATABASE_PATH", "/data/audit-core.db")
     )
-    app = IngestionApplication(backend, os.environ["AUDIT_CORE_INGEST_TOKEN"].strip())
-    host = os.environ.get("AUDIT_CORE_HOST", "0.0.0.0")
-    port = int(os.environ.get("AUDIT_CORE_HTTP_PORT", "8080"))
-    # NOTE: wsgiref is single-threaded and has no request timeout. Replacing it
-    # with a production WSGI server is AUDIT-WP-0004-T06; do not deploy on this.
-    log.warning("serving on wsgiref development server — not for production (AUDIT-WP-0004-T06)")
-    with make_server(host, port, app) as server:
-        server.serve_forever()
+    app = IngestionApplication(backend, SenderRegistry.from_env())
+    serve(
+        app,
+        host=os.environ.get("AUDIT_CORE_HOST", "0.0.0.0"),
+        port=int(os.environ.get("AUDIT_CORE_HTTP_PORT", "8080")),
+        threads=int(os.environ.get("AUDIT_CORE_THREADS", "8")),
+        timeout=int(os.environ.get("AUDIT_CORE_REQUEST_TIMEOUT", "30")),
+    )

audit_core/senders.py

@@ -0,0 +1,164 @@
+"""Sender identities and what they are permitted to claim.
+
+AUDIT-WP-0004-T03. WP-0003 recorded tenant isolation as delivered, but the
+receiver accepted whatever ``tenant`` and ``source`` a caller sent as long as
+it held the one shared token. This module binds a credential to the sources and
+tenants it may write for, and drives that binding from configuration rather
+than literals.
+
+Each identity carries a *list* of tokens so a credential can be rotated without
+a delivery gap: publish the replacement alongside the incumbent, move the
+sender, then drop the old one.
+"""
+
+from __future__ import annotations
+
+import hmac
+import json
+import os
+from dataclasses import dataclass, field
+from typing import Any, Iterable
+
+WILDCARD = "*"
+
+
+@dataclass(frozen=True)
+class SenderIdentity:
+    """A credential holder and the claims it is allowed to make."""
+
+    name: str
+    tokens: tuple[str, ...]
+    sources: frozenset[str]
+    tenants: frozenset[str] = field(default_factory=lambda: frozenset({WILDCARD}))
+    may_write: bool = True
+    may_read: bool = False
+
+    def __post_init__(self) -> None:
+        if not self.name:
+            raise ValueError("sender identity needs a name")
+        if not self.tokens or any(not t for t in self.tokens):
+            raise ValueError(f"sender {self.name!r} needs at least one non-empty token")
+        if not self.sources:
+            raise ValueError(f"sender {self.name!r} needs at least one permitted source")
+
+    def permits_source(self, source: str) -> bool:
+        return WILDCARD in self.sources or source in self.sources
+
+    def permits_tenant(self, tenant: str) -> bool:
+        """Whether this identity may write for ``tenant``.
+
+        Tenant identifiers are opaque here. Their shape
+        (``tenant:<grouping>:<name>``) is owned by the IAM Profile and
+        tenant-engine; matching is exact, never parsed or pattern-derived.
+        """
+        return WILDCARD in self.tenants or tenant in self.tenants
+
+
+class SenderRegistry:
+    """Resolves a credential to the identity that holds it."""
+
+    def __init__(self, identities: Iterable[SenderIdentity]) -> None:
+        self.identities = tuple(identities)
+        if not self.identities:
+            raise ValueError("at least one sender identity is required")
+
+    def authenticate(self, authorization_header: str | None) -> SenderIdentity | None:
+        """Return the matching identity, or None.
+
+        Every candidate token is compared even after a match is found, so the
+        work done does not depend on which credential was supplied or how many
+        identities precede it.
+        """
+        supplied = str(authorization_header or "")
+        matched: SenderIdentity | None = None
+        for identity in self.identities:
+            for token in identity.tokens:
+                try:
+                    hit = hmac.compare_digest(supplied, f"Bearer {token}")
+                except TypeError:
+                    # compare_digest rejects non-ASCII str; such a header is
+                    # simply not a valid credential.
+                    hit = False
+                if hit and matched is None:
+                    matched = identity
+        return matched
+
+    @classmethod
+    def from_env(cls, env: dict[str, str] | None = None) -> "SenderRegistry":
+        """Build a registry from ``AUDIT_CORE_SENDERS`` (JSON).
+
+        Shape::
+
+            [{"name": "user-engine",
+              "tokens": ["current", "next"],
+              "sources": ["user-engine"],
+              "tenants": ["tenant:friendly:binky"],
+              "may_read": false}]
+
+        ``tenants`` defaults to ``["*"]``. Omitting it is a deliberate choice to
+        accept any tenant from that sender and should be justified per sender,
+        not adopted by default.
+        """
+        env = env if env is not None else dict(os.environ)
+        raw = env.get("AUDIT_CORE_SENDERS")
+        if raw:
+            return cls(_parse_identities(raw))
+
+        legacy = (env.get("AUDIT_CORE_INGEST_TOKEN") or "").strip()
+        if not legacy:
+            raise ValueError(
+                "no sender configuration: set AUDIT_CORE_SENDERS (preferred) "
+                "or AUDIT_CORE_INGEST_TOKEN"
+            )
+        # Legacy single-token form. Kept so an existing deployment keeps
+        # working, but it grants every tenant — which is what T03 exists to
+        # stop, so it is deliberately noisy about what it is.
+        return cls([
+            SenderIdentity(
+                name="legacy-ingest-token",
+                tokens=(legacy,),
+                sources=frozenset({"user-engine"}),
+                tenants=frozenset({WILDCARD}),
+            )
+        ])
+
+
+def _parse_identities(raw: str) -> list[SenderIdentity]:
+    try:
+        entries = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"AUDIT_CORE_SENDERS is not valid JSON: {exc}") from exc
+    if not isinstance(entries, list) or not entries:
+        raise ValueError("AUDIT_CORE_SENDERS must be a non-empty JSON list")
+    return [_identity_from(entry) for entry in entries]
+
+
+def _identity_from(entry: Any) -> SenderIdentity:
+    if not isinstance(entry, dict):
+        raise ValueError("each sender entry must be an object")
+    tokens = entry.get("tokens") or ([entry["token"]] if entry.get("token") else [])
+    return SenderIdentity(
+        name=str(entry.get("name") or ""),
+        tokens=tuple(str(t) for t in tokens),
+        sources=frozenset(str(s) for s in (entry.get("sources") or [])),
+        tenants=frozenset(str(t) for t in (entry.get("tenants") or [WILDCARD])),
+        may_write=bool(entry.get("may_write", True)),
+        may_read=bool(entry.get("may_read", False)),
+    )
+
+
+def development_registry(token: str) -> SenderRegistry:
+    """A single permissive sender, for tests and local development.
+
+    Accepts any tenant. Not a production configuration — production binds each
+    sender to the tenants it may write for via ``AUDIT_CORE_SENDERS``.
+    """
+    return SenderRegistry([
+        SenderIdentity(
+            name="development",
+            tokens=(token,),
+            sources=frozenset({"user-engine"}),
+            tenants=frozenset({WILDCARD}),
+            may_read=True,
+        )
+    ])

audit_core/sqlite_backend.py

@@ -34,8 +34,25 @@ CREATE TABLE IF NOT EXISTS events (
 );
 CREATE INDEX IF NOT EXISTS events_correlation_idx ON events (correlation_id);
 CREATE INDEX IF NOT EXISTS events_tenant_idx ON events (tenant);
+
+CREATE TABLE IF NOT EXISTS dead_letters (
+  id            INTEGER PRIMARY KEY AUTOINCREMENT,
+  event_id      TEXT,
+  received_at   TEXT NOT NULL,
+  sender        TEXT,
+  reason        TEXT NOT NULL,
+  payload_hash  TEXT NOT NULL,
+  payload       TEXT,
+  payload_withheld INTEGER NOT NULL DEFAULT 0
+);
+CREATE INDEX IF NOT EXISTS dead_letters_event_idx ON dead_letters (event_id);
 """
 
+# Rejection reasons whose payload must never be persisted. Storing the body of
+# an event rejected *for containing secret-shaped material* would write that
+# material into the audit store — the precise outcome the rejection prevents.
+WITHHOLD_PAYLOAD_REASONS = frozenset({"secret_shaped_field"})
+
 
 class SQLiteAuditBackend:
     """Store audit events in SQLite with idempotent accept semantics.
@@ -147,6 +164,86 @@ class SQLiteAuditBackend:
             )
         return AcceptResult(duplicate=True, reference=reference)
 
+    # --- operator read surface (AUDIT-WP-0004-T05) --------------------------
+
+    def get(self, event_id: str) -> dict | None:
+        """Return one stored event record, or None."""
+        row = self._query(
+            "SELECT record, accepted_at FROM events WHERE event_id = ?", (event_id,)
+        )
+        if not row:
+            return None
+        return {"accepted_at": row[0][1], **json.loads(row[0][0])}
+
+    def by_correlation(self, correlation_id: str, limit: int = 100) -> list[dict]:
+        """Return every stored event carrying ``correlation_id``, oldest first."""
+        rows = self._query(
+            "SELECT record, accepted_at FROM events WHERE correlation_id = ? "
+            "ORDER BY accepted_at, event_id LIMIT ?",
+            (correlation_id, int(limit)),
+        )
+        return [{"accepted_at": at, **json.loads(rec)} for rec, at in rows]
+
+    def record_rejection(
+        self,
+        *,
+        event_id: str | None,
+        reason: str,
+        payload_hash: str,
+        sender: str | None = None,
+        payload: str | None = None,
+    ) -> None:
+        """Record a rejected event so it is visible to an operator.
+
+        A rejection is not a silent drop: the sender dead-letters the event and
+        somebody has to be able to see why. The payload is withheld when the
+        rejection reason implies it carries secret-shaped material.
+        """
+        withheld = reason in WITHHOLD_PAYLOAD_REASONS
+        try:
+            self.db.execute(
+                "INSERT INTO dead_letters "
+                "(event_id, received_at, sender, reason, payload_hash, payload, payload_withheld) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (
+                    event_id,
+                    datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
+                    sender,
+                    reason,
+                    payload_hash,
+                    None if withheld else payload,
+                    1 if withheld else 0,
+                ),
+            )
+        except sqlite3.Error as exc:
+            raise BackendUnavailableError(str(exc)) from exc
+
+    def dead_letters(self, limit: int = 100) -> list[dict]:
+        """Return recent rejections, newest first."""
+        rows = self._query(
+            "SELECT event_id, received_at, sender, reason, payload_hash, payload, "
+            "payload_withheld FROM dead_letters ORDER BY id DESC LIMIT ?",
+            (int(limit),),
+        )
+        return [
+            {
+                "event_id": r[0],
+                "received_at": r[1],
+                "sender": r[2],
+                "reason": r[3],
+                "payload_hash": r[4],
+                "payload": r[5],
+                "payload_withheld": bool(r[6]),
+            }
+            for r in rows
+        ]
+
+    def _query(self, sql: str, params: tuple) -> list:
+        try:
+            return self.db.execute(sql, params).fetchall()
+        except sqlite3.Error as exc:
+            raise BackendUnavailableError(str(exc)) from exc
+
     def health(self) -> None:
         """Raise :class:`BackendUnavailableError` if the store is unusable."""
         try: