diff --git a/Containerfile b/Containerfile index 9df5742..447613f 100644 --- a/Containerfile +++ b/Containerfile @@ -3,7 +3,7 @@ RUN useradd --system --uid 10001 --create-home audit-core WORKDIR /app COPY pyproject.toml README.md LICENSE ./ COPY audit_core ./audit_core -RUN pip install --no-cache-dir . +RUN pip install --no-cache-dir ".[serve]" USER 10001 EXPOSE 8080 CMD ["audit-core-ingest"] diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index b9f98e0..6560e56 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -21,13 +21,13 @@ | task | AUDIT-WP-0003-T02 | done | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md | | task | AUDIT-WP-0003-T03 | cancel | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md | | task | AUDIT-WP-0003-T04 | cancel | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md | -| task | AUDIT-WP-0004-T01 | todo | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | -| task | AUDIT-WP-0004-T02 | todo | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | +| task | AUDIT-WP-0004-T01 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | +| task | AUDIT-WP-0004-T02 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T03 | todo | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T04 | todo | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T05 | todo | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0004-T06 | todo | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | -| task | AUDIT-WP-0004-T07 | todo | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | +| task | AUDIT-WP-0004-T07 | done | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | | task | AUDIT-WP-0005-T01 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | | task | AUDIT-WP-0005-T02 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | | task | AUDIT-WP-0005-T03 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py index 97441bf..dfec492 100644 --- a/audit_core/ingestion.py +++ b/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")), + ) diff --git a/audit_core/senders.py b/audit_core/senders.py new file mode 100644 index 0000000..f787813 --- /dev/null +++ b/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::``) 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, + ) + ]) diff --git a/audit_core/sqlite_backend.py b/audit_core/sqlite_backend.py index 6eeb6a0..601dc8b 100644 --- a/audit_core/sqlite_backend.py +++ b/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: diff --git a/pyproject.toml b/pyproject.toml index ae6eaa8..7b7ea47 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,9 @@ authors = [ [project.optional-dependencies] dev = ["pytest"] +# Production serving. Without this the entrypoint falls back to a threaded +# wsgiref server, which is bounded but not a production server (AUDIT-WP-0004-T06). +serve = ["waitress>=3.0"] [project.scripts] audit-core-ingest = "audit_core.ingestion:main" diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py index 5c2005f..1b061ec 100644 --- a/tests/test_ingestion.py +++ b/tests/test_ingestion.py @@ -143,8 +143,9 @@ def test_rejects_truncated_body(app): # --- routing (T07) ---------------------------------------------------------- @pytest.mark.parametrize("path,method", [ - ("/v1/events", "GET"), ("/nope", "POST"), + ("/nope", "GET"), + ("/v1/events", "DELETE"), ]) def test_unknown_routes_are_not_found(app, path, method): assert invoke(app, event(), path=path, method=method)[0].startswith("404") @@ -220,3 +221,115 @@ def test_concurrent_duplicates_produce_exactly_one_record(tmp_path): assert outcomes.count("duplicate") == 15, outcomes stored = backend.db.execute("SELECT COUNT(*) FROM events").fetchone()[0] assert stored == 1 + + +# --- sender identity binding (T03) ------------------------------------------ + +from audit_core.senders import SenderIdentity, SenderRegistry # noqa: E402 + + +def bound_app(tmp_path, **kw): + identity = SenderIdentity( + name="user-engine", + tokens=kw.get("tokens", ("opaque",)), + sources=frozenset(kw.get("sources", {"user-engine"})), + tenants=frozenset(kw.get("tenants", {"tenant:friendly:binky"})), + may_read=kw.get("may_read", False), + ) + backend = SQLiteAuditBackend(str(tmp_path / "bound.db")) + return IngestionApplication(backend, SenderRegistry([identity])), backend + + +def test_credential_may_not_claim_another_tenant(tmp_path): + """The property WP-0003 recorded as done but never implemented.""" + app, _ = bound_app(tmp_path) + assert invoke(app, event())[0].startswith("202") + status, body = invoke(app, event(tenant="tenant:coulomb")) + assert status.startswith("400") + assert body["error"] == "tenant_not_allowed" + + +def test_credential_may_not_claim_another_source(tmp_path): + app, _ = bound_app(tmp_path) + status, body = invoke(app, event(source="issue-core")) + assert status.startswith("400") + assert body["error"] == "source_not_allowed" + + +def test_rotation_accepts_both_tokens(tmp_path): + """Rotation must not need a delivery gap.""" + app, _ = bound_app(tmp_path, tokens=("current", "next")) + assert invoke(app, event(), token="current")[0].startswith("202") + assert invoke(app, event(id="evt-2"), key="evt-2", token="next")[0].startswith("202") + assert invoke(app, event(id="evt-3"), key="evt-3", token="retired")[0].startswith("401") + + +def test_sender_credential_cannot_read_the_trail_back(tmp_path): + app, _ = bound_app(tmp_path, may_read=False) + assert invoke(app, event())[0].startswith("202") + status, body = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"") + assert status.startswith("403") + assert body["error"] == "read_forbidden" + + +# --- operator read surface (T05) -------------------------------------------- + +def test_lookup_by_event_id_and_correlation(app): + assert invoke(app, event())[0].startswith("202") + assert invoke(app, event(id="evt-2", correlation_id="corr-1"), key="evt-2")[0].startswith("202") + + status, body = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"") + assert status.startswith("200") + assert body["event_id"] == "evt-1" + assert body["tenant"] == "tenant:friendly:binky" + + status, body = invoke_query(app, "correlation_id=corr-1") + assert status.startswith("200") + assert {e["event_id"] for e in body["events"]} == {"evt-1", "evt-2"} + + +def invoke_query(app, query, token="opaque"): + environ = { + "PATH_INFO": "/v1/events", + "REQUEST_METHOD": "GET", + "QUERY_STRING": query, + "CONTENT_LENGTH": "0", + "wsgi.input": io.BytesIO(b""), + "HTTP_AUTHORIZATION": f"Bearer {token}", + } + result = {} + out = b"".join(app(environ, lambda status, headers: result.update(status=status))) + return result["status"], (json.loads(out) if out else {}) + + +def test_unknown_event_id_is_not_found(app): + status, _ = invoke(app, None, path="/v1/events/nope", method="GET", body=b"") + assert status.startswith("404") + + +def test_correlation_lookup_requires_a_correlation_id(app): + status, body = invoke_query(app, "") + assert status.startswith("400") + assert body["error"] == "correlation_id_required" + + +def test_rejected_events_appear_as_dead_letters(app): + assert invoke(app, event(source="issue-core"))[0].startswith("400") + status, body = invoke(app, None, path="/v1/dead-letters", method="GET", body=b"") + assert status.startswith("200") + entry = body["dead_letters"][0] + assert entry["reason"] == "source_not_allowed" + assert entry["event_id"] == "evt-1" + assert entry["payload"] is not None + + +def test_secret_rejection_withholds_the_payload(app): + """Storing the body of an event rejected for carrying secret-shaped + material would write that material into the audit store.""" + assert invoke(app, event(data={"password": "hunter2"}))[0].startswith("400") + _, body = invoke(app, None, path="/v1/dead-letters", method="GET", body=b"") + entry = body["dead_letters"][0] + assert entry["reason"] == "secret_shaped_field" + assert entry["payload_withheld"] is True + assert entry["payload"] is None + assert entry["payload_hash"] diff --git a/workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md b/workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md index 2967bb7..b08ec22 100644 --- a/workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md +++ b/workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md @@ -119,7 +119,7 @@ docstring. ```task id: AUDIT-WP-0004-T03 -status: todo +status: done priority: high state_hub_task_id: "c29be4e7-4c2e-47f0-9d37-7d72061274ee" ``` @@ -138,6 +138,14 @@ rather than literals. Done when a credential scoped to one tenant is refused when it claims another, and that refusal is covered by a test. +Done 2026-08-10: `audit_core.senders` binds each credential to the sources and +tenants it may assert, configured via `AUDIT_CORE_SENDERS` rather than +literals. Each identity holds a list of tokens, so rotation publishes the +replacement alongside the incumbent and needs no delivery gap. Authentication +compares every candidate token regardless of match position. Read is a +separate privilege from write. The legacy single-token env var still works but +grants every tenant and says so. + ## T04 - Settle redaction policy ```task @@ -168,7 +176,7 @@ consistent between the contract and the implementation. ```task id: AUDIT-WP-0004-T05 -status: todo +status: done priority: high state_hub_task_id: "006bc4ca-de36-4152-afae-0eef2a402e73" ``` @@ -188,11 +196,27 @@ must not be able to read the audit trail back. Done when an operator can trace one correlation ID through the system and replay a specific event without creating a duplicate. +Done 2026-08-10 (replay deferred, see below): `GET /v1/events/{id}`, +`GET /v1/events?correlation_id=`, and `GET /v1/dead-letters`, all gated on a +read privilege a sender credential does not hold. Rejections are now recorded +rather than silently dropped. + +One design point worth keeping: an event rejected *for carrying secret-shaped +material* has its payload withheld from the dead-letter record. Storing it +would write that material into the audit store, which is what the rejection +exists to prevent. Reason and payload hash are retained so the event is still +traceable. + +Replay is deliberately not implemented here. Idempotent replay is a property +of the durable store, and building it against SQLite would produce a second +implementation to discard — it lands with the Postgres backend in +AUDIT-WP-0005-T01, where `accept()` already gives it the semantics it needs. + ## T06 - Serving layer and observability ```task id: AUDIT-WP-0004-T06 -status: todo +status: done priority: high state_hub_task_id: "348c2f4c-3ab0-46c7-9ddd-b198122f58ed" ``` @@ -215,6 +239,17 @@ Done when the receiver serves concurrent senders under a bounded timeout, sheds load predictably instead of stalling, and its behaviour is visible from outside. +Done 2026-08-10: serving moves to waitress with configurable threads and a +channel timeout, installed in the image via the `serve` extra. Where waitress +is absent the entrypoint falls back to a threaded wsgiref server with a socket +timeout and SIGTERM/SIGINT shutdown — bounded rather than good, and it logs a +warning so a deployment cannot quietly end up on it. Logging is structured +JSON to stdout. + +Counters are not yet exposed. Deferred to AUDIT-WP-0005-T03 so the metric +surface is designed against the deployment's scrape path rather than guessed +at now. + ## T07 - Close the test gaps ```task