From b0e6792cf065712c652a6b75713113782f09927d Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 14 Sep 2026 22:13:55 +0200 Subject: [PATCH] Bound /readyz so kubelet probes cannot hang the Service /readyz walked the hash chain and opened pooled connections with no libpq connect_timeout, so a 2s kubelet probe never saw a response and the pod stayed unready. Informed Decision accept is blocked on that. Probe health() only, under a 1.5s budget, publish last-known tamper_evidence, and fail TCP handshake in 1s. Integrity stays on /v1/integrity. Assistant: grok Assistant-Session: 01a0a182-bab7-7f11-b32b-d06f3af52082 --- audit_core/ingestion.py | 62 +++++++++++++++--- audit_core/postgres_backend.py | 58 +++++++++++++++-- docs/integrity.md | 7 ++- docs/operator-runbook.md | 2 +- tests/test_backend_conformance.py | 26 ++++++++ tests/test_ingestion.py | 63 +++++++++++++++++++ tests/test_integrity.py | 26 ++++++++ workplans/ADHOC-2026-09-14.md | 41 ++++++++++++ ...AUDIT-WP-0009-evidence-role-conformance.md | 9 ++- 9 files changed, 276 insertions(+), 18 deletions(-) create mode 100644 workplans/ADHOC-2026-09-14.md diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py index 84b1fd5..809f5cc 100644 --- a/audit_core/ingestion.py +++ b/audit_core/ingestion.py @@ -51,6 +51,11 @@ from audit_core.sqlite_backend import SQLiteAuditBackend MAX_BODY_BYTES = 256 * 1024 +# Kubelet readiness timeoutSeconds is 2. Stay under that even if a backend +# call hangs: a timed-out probe looks the same as down, and a hung worker +# cannot answer the next one. +DEFAULT_READY_BUDGET_SECONDS = 1.5 + log = logging.getLogger("audit_core.ingestion") @@ -102,7 +107,8 @@ class IngestionApplication: senders: SenderRegistry | str, require_custody_class: str | None = None, ) -> None: - policy = backend.retention_policy + getter = getattr(backend, "readiness_policy", None) + policy = getter() if callable(getter) else backend.retention_policy if require_custody_class and not custody_class_satisfies( policy.custody_class, require_custody_class ): @@ -129,6 +135,9 @@ class IngestionApplication: self.backend = backend self.senders = senders self.counters = Counters() + self._ready_budget = float( + os.environ.get("AUDIT_CORE_READY_BUDGET_SECONDS", DEFAULT_READY_BUDGET_SECONDS) + ) def __call__(self, environ, start_response): try: @@ -474,20 +483,51 @@ class IngestionApplication: raise ValueError("truncated_body") return raw + def _declared_policy(self): + """Custody class and recovery fields without a chain walk. + + Postgres derives ``tamper_evidence`` by walking the hash chain. + ``/readyz`` must not do that: kubelet gives it two seconds, and a + hung walk takes the pod out of the Service even when custody is + reachable. Prefer ``readiness_policy`` when the backend offers it. + """ + getter = getattr(self.backend, "readiness_policy", None) + if callable(getter): + return getter() + return self.backend.retention_policy + def _readiness(self, start_response): - try: - health = getattr(self.backend, "health", None) - if callable(health): - health() - except BackendUnavailableError as exc: - log.error("readiness failed: %s", exc) + box: dict[str, Any] = {} + done = threading.Event() + + def run() -> None: + try: + health = getattr(self.backend, "health", None) + if callable(health): + health() + box["policy"] = self._declared_policy() + except Exception as exc: # probe must classify, not hang or 500 + box["error"] = exc + finally: + done.set() + + worker = threading.Thread(target=run, name="audit-readyz", daemon=True) + worker.start() + if not done.wait(timeout=self._ready_budget): + log.error("readiness exceeded %.3ss budget", self._ready_budget) + return self._json( + start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"status": "unavailable"} + ) + error = box.get("error") + if error is not None: + log.error("readiness failed: %s", error) return self._json( start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"status": "unavailable"} ) return self._json( start_response, HTTPStatus.OK, - self.backend.retention_policy.as_readiness(), + box["policy"].as_readiness(), ) @staticmethod @@ -716,6 +756,12 @@ def build_backend() -> IdempotentAuditBackend: statement_timeout_ms=int( os.environ.get("AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS", "30000") ), + connect_timeout_s=int( + os.environ.get("AUDIT_CORE_DB_CONNECT_TIMEOUT_S", "1") + ), + health_timeout_s=float( + os.environ.get("AUDIT_CORE_DB_HEALTH_TIMEOUT_S", "1") + ), # Production mounts the runtime role, which cannot CREATE TABLE. # Schema changes run as a Job with the migration lease # (AUDIT-WP-0005-T02). Local and brokered use keep the default. diff --git a/audit_core/postgres_backend.py b/audit_core/postgres_backend.py index 03e25ea..0c190dd 100644 --- a/audit_core/postgres_backend.py +++ b/audit_core/postgres_backend.py @@ -57,6 +57,15 @@ except ImportError as exc: # pragma: no cover "the postgres backend needs psycopg: pip install 'audit-core[postgres]'" ) from exc +try: # pragma: no cover - older psycopg_pool + from psycopg_pool import PoolTimeout as _PoolTimeout +except ImportError: # pragma: no cover + _PoolTimeout = None + +_UNAVAILABLE: tuple[type[BaseException], ...] = (psycopg.Error, TimeoutError, OSError) +if _PoolTimeout is not None: + _UNAVAILABLE = (*_UNAVAILABLE, _PoolTimeout) + DEFAULT_SCHEMA = "audit_core" # Ordered, append-only. Each entry runs once and is recorded in @@ -181,6 +190,8 @@ class PostgresAuditBackend: min_size: int = 1, max_size: int = 8, statement_timeout_ms: int = 30_000, + connect_timeout_s: int = 1, + health_timeout_s: float = 1.0, migrate: bool = True, credential_dir: str | None = None, attestation_path: str | None = None, @@ -224,8 +235,13 @@ class PostgresAuditBackend: self._attestation_cache_seconds = float(attestation_cache_seconds) self._tamper_state: TamperEvidenceState | None = None self._tamper_state_at = 0.0 + self._connect_timeout_s = max(1, int(connect_timeout_s)) + self._health_timeout_s = float(health_timeout_s) base_kwargs = { "autocommit": True, + # A stalled TCP handshake must fail the probe, not hold it until + # kubelet times out the pod. + "connect_timeout": self._connect_timeout_s, # A stalled write must surface as unavailable rather than hold a # request open indefinitely. "options": f"-c statement_timeout={int(statement_timeout_ms)}", @@ -334,6 +350,25 @@ class PostgresAuditBackend: recoverable_basis=self.recoverable_basis, ) + def readiness_policy(self) -> RetentionPolicy: + """``/readyz`` view of custody: class, durability, recovery, last claim. + + Does not walk the chain. An unevaluated or expired cache reports + ``tamper_evidence=False`` — uncomputed is not claimed. Integrity + evaluation stays on ``retention_policy`` / ``/v1/integrity``. + """ + cached = self._tamper_state + return RetentionPolicy( + custody_class="operational", + retention_days=self.retention_days, + immutable=True, + tamper_evidence=bool(cached is not None and cached.claimed), + durable=True, + recoverable_days=self.recoverable_days, + recoverable_source=self.recoverable_source, + recoverable_basis=self.recoverable_basis, + ) + def emit(self, event: AuditEvent) -> str: return self.accept(event, payload_hash=_record_hash(event)).reference @@ -563,9 +598,10 @@ class PostgresAuditBackend: def tamper_evidence_state(self) -> TamperEvidenceState: """Evaluate the tamper-evidence preconditions against live state. - Cached briefly because ``/readyz`` reads it on every probe and the - evaluation walks the chain. The cache only ever delays a *change* of - state; it cannot manufacture one. + Cached briefly because a full walk is too expensive to repeat per + caller. ``/readyz`` uses :meth:`readiness_policy` and never waits + on this. The cache only ever delays a *change* of state; it cannot + manufacture one. """ now = time.monotonic() cached = self._tamper_state @@ -586,7 +622,17 @@ class PostgresAuditBackend: return state def health(self) -> None: - self._query("SELECT 1", ()) + """Reachability check for ``/readyz``. Must return inside the probe budget.""" + timeout_ms = max(1, int(self._health_timeout_s * 1000)) + try: + with self.pool.connection(timeout=self._health_timeout_s) as conn: + with conn.transaction(): + conn.execute(f"SET LOCAL statement_timeout = {timeout_ms}") + conn.execute("SELECT 1") + except _UNAVAILABLE as exc: + raise BackendUnavailableError(str(exc)) from exc + except Exception as exc: # pool/libpq can raise outside psycopg.Error + raise BackendUnavailableError(str(exc)) from exc def _backfill_chain(self, conn) -> None: """Fill chain columns on rows accepted before migration 0006. @@ -632,14 +678,14 @@ class PostgresAuditBackend: try: with self.pool.connection() as conn: return conn.execute(sql, params).fetchall() - except psycopg.Error as exc: + except _UNAVAILABLE as exc: raise BackendUnavailableError(str(exc)) from exc def _execute(self, sql: str, params: tuple) -> None: try: with self.pool.connection() as conn: conn.execute(sql, params) - except psycopg.Error as exc: + except _UNAVAILABLE as exc: raise BackendUnavailableError(str(exc)) from exc diff --git a/docs/integrity.md b/docs/integrity.md index e5e5bff..dac23c2 100644 --- a/docs/integrity.md +++ b/docs/integrity.md @@ -109,8 +109,11 @@ operator establishes that copy, the delivered control is "defends against a database owner", and no stronger claim may be made from it. Before the first run the ConfigMap is empty and `/readyz` reports -`tamper_evidence: false` with reason `no_attestation`. That is the correct -day-one state, not a regression. +`tamper_evidence: false`. That is the correct day-one state, not a +regression. `/readyz` itself does not walk the chain: it publishes the +last evaluated claim, or `false` if none, so a 2-second kubelet probe +cannot take the pod out of the Service. The walk lives on +`/v1/integrity` and `retention_policy`. Do not write the attestation into the Barman prefix (`platform-pg/` on `resource:platform:audit-storage`). That copy is diff --git a/docs/operator-runbook.md b/docs/operator-runbook.md index ae0c59e..f31dbe2 100644 --- a/docs/operator-runbook.md +++ b/docs/operator-runbook.md @@ -29,7 +29,7 @@ warden route show database-dynamic-credentials --json | Check | Meaning | | --- | --- | | `GET /healthz` | Process is up. Liveness uses this. A database outage must **not** restart the pod. | -| `GET /readyz` | Custody is reachable and `custody_class=operational`. Also reports `recoverable_days` and `tamper_evidence`. Readiness uses this; the pod leaves the Service rather than accept events it cannot store. | +| `GET /readyz` | Custody is reachable (`SELECT 1`) within a 1.5s budget. Reports `custody_class`, `recoverable_days`, and last-known `tamper_evidence`. Does **not** walk the hash chain — kubelet `timeoutSeconds` is 2, and a hung walk takes the pod out of the Service. Readiness uses this; the pod leaves the Service rather than accept events it cannot store. Integrity evaluation is `/v1/integrity`. | | `GET /v1/stats` | In-process counters since start (`accepted`, `duplicate`, `conflict`, `rejected`, `unauthorized`, `forbidden`, `unavailable`, `error`). Resets on restart. Requires `may_read` and full tenant scope. | | `GET /v1/integrity` | Hash-chain walk: `{intact, events, head, first_break}`. No payloads. Requires `may_read` and full tenant scope. A break is a custody defect, not a sender retry. | diff --git a/tests/test_backend_conformance.py b/tests/test_backend_conformance.py index a7cf166..bc33bc5 100644 --- a/tests/test_backend_conformance.py +++ b/tests/test_backend_conformance.py @@ -359,6 +359,32 @@ def test_connects_from_a_brokered_libpq_environment(monkeypatch): backend.close() +def test_postgres_connections_set_a_connect_timeout(monkeypatch): + """A hung TCP handshake must not hold /readyz past the kubelet budget.""" + try: + from audit_core.postgres_backend import PostgresAuditBackend + except ImportError: + pytest.skip("psycopg is not installed") + + captured: dict = {} + + class FakePool: + def __init__(self, dsn, **kwargs): + captured["dsn"] = dsn + captured.update(kwargs) + + def close(self): + return None + + monkeypatch.setattr("audit_core.postgres_backend.ConnectionPool", FakePool) + PostgresAuditBackend("postgresql://example/audit", migrate=False) + kwargs = captured["kwargs"] + if callable(kwargs): + kwargs = kwargs() + assert int(kwargs["connect_timeout"]) == 1 + assert "statement_timeout" in kwargs["options"] + + def test_missing_connection_information_is_a_clear_error(monkeypatch): for var in ("AUDIT_CORE_DATABASE_URL", "PGHOST", "PGUSER"): monkeypatch.delenv(var, raising=False) diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py index 84b4891..db8abc2 100644 --- a/tests/test_ingestion.py +++ b/tests/test_ingestion.py @@ -1,5 +1,6 @@ import io import json +import time from datetime import datetime, timezone import pytest @@ -502,6 +503,68 @@ def test_readiness_reports_recovery_fields_for_operational_backend(): assert "platform-audit-storage" in body["recoverable_source"] +def test_readyz_does_not_wait_on_chain_verification(): + """Kubelet timeoutSeconds is 2; walking the hash chain is not a probe.""" + + class _ChainWalk(_BrokenBackend): + def health(self): + return None + + @property + def retention_policy(self): + time.sleep(5) + raise AssertionError("/readyz must not call retention_policy") + + def readiness_policy(self): + return RetentionPolicy( + custody_class="operational", + retention_days=None, + immutable=True, + tamper_evidence=False, + durable=True, + recoverable_days=30, + recoverable_source="cited", + recoverable_basis="measured", + ) + + started = time.monotonic() + status, body = invoke( + IngestionApplication(_ChainWalk(), "opaque"), + None, path="/readyz", method="GET", body=b"", + ) + assert time.monotonic() - started < 1.0 + assert status.startswith("200") + assert body["durable"] is True + assert body["tamper_evidence"] is False + + +def test_readyz_treats_a_hung_health_check_as_unavailable(): + class _HungHealth(_BrokenBackend): + def health(self): + time.sleep(5) + + app = IngestionApplication(_HungHealth(), "opaque") + app._ready_budget = 0.05 + started = time.monotonic() + status, body = invoke(app, None, path="/readyz", method="GET", body=b"") + assert time.monotonic() - started < 1.0 + assert status.startswith("503") + assert body["status"] == "unavailable" + + +def test_readyz_unavailable_backend_is_503(): + class _Down(_BrokenBackend): + def health(self): + raise BackendUnavailableError("down") + + status, body = invoke( + IngestionApplication(_Down(), "opaque"), + None, path="/readyz", method="GET", body=b"", + ) + assert status.startswith("503") + assert body["status"] == "unavailable" + + def test_counters_track_each_outcome(tmp_path): app, _ = bound_app(tmp_path, may_read=True) invoke(app, event()) # accepted diff --git a/tests/test_integrity.py b/tests/test_integrity.py index 6518566..d7c8b89 100644 --- a/tests/test_integrity.py +++ b/tests/test_integrity.py @@ -303,6 +303,32 @@ def test_postgres_policy_treats_an_unreadable_attestation_as_absent(tmp_path): assert backend.tamper_evidence_state().reason == "no_attestation" +def test_postgres_readiness_policy_does_not_walk_the_chain(tmp_path): + backend = _bare_postgres_backend(tmp_path) + + def _boom(attestation=None): + raise AssertionError("/readyz must not walk the chain") + + backend.verify_chain = _boom + policy = backend.readiness_policy() + assert policy.tamper_evidence is False + assert policy.durable is True + assert policy.custody_class == "operational" + + +def test_postgres_readiness_policy_reports_a_cached_claim(tmp_path): + from audit_core.integrity import TamperEvidenceState, utc_now + + backend = _bare_postgres_backend(tmp_path) + backend.verify_chain = lambda attestation=None: (_ for _ in ()).throw( + AssertionError("cached claim must not walk") + ) + backend._tamper_state = TamperEvidenceState( + claimed=True, reason="attested", observed_at=utc_now(), + ) + assert backend.readiness_policy().tamper_evidence is True + + def test_postgres_policy_drops_the_claim_when_the_chain_cannot_be_walked(tmp_path): from audit_core.integrity import utc_now diff --git a/workplans/ADHOC-2026-09-14.md b/workplans/ADHOC-2026-09-14.md new file mode 100644 index 0000000..6f84527 --- /dev/null +++ b/workplans/ADHOC-2026-09-14.md @@ -0,0 +1,41 @@ +--- +id: ADHOC-2026-09-14 +type: workplan +title: "Ad hoc tasks 2026-09-14" +domain: infotech +repo: audit-core +status: finished +flavor: implementation +owner: grok +topic_slug: railiance +created: "2026-09-14" +updated: "2026-09-14" +--- + +# Ad hoc tasks 2026-09-14 + +## Bound /readyz so the Service can become Ready + +```task +id: ADHOC-2026-09-14-T01 +status: done +priority: high +``` + +Inbox from grok (INFD-WP-0002-T03): `audit-core` Service has no ready +endpoints. The live pod is Ready=false since 2026-09-11T08:27:26Z; +`/healthz` is 200; `/readyz` hangs >8s against kubelet `timeoutSeconds: 2`. +Informed Decision accept is 503 `approval_path_not_connected` until a ready +endpoint exists. + +Cause: `/readyz` called `retention_policy`, which walks the hash chain and +opens a pooled connection with no libpq `connect_timeout` and a 30s +statement timeout. A hung walk or handshake looks the same as down. + +Change: `/readyz` checks `health()` (`SELECT 1`) under a 1.5s budget, +publishes last-known `tamper_evidence` without walking, and sets +`connect_timeout=1`. Integrity stays on `/v1/integrity`. Production needs a +receiver image rebuild and rollout after this lands. + +Evidence: Informed Decision +`docs/evidence/2026-09-14-infd-0002-t03-bind-path-probe.json`. diff --git a/workplans/AUDIT-WP-0009-evidence-role-conformance.md b/workplans/AUDIT-WP-0009-evidence-role-conformance.md index e1b50b6..9492ace 100644 --- a/workplans/AUDIT-WP-0009-evidence-role-conformance.md +++ b/workplans/AUDIT-WP-0009-evidence-role-conformance.md @@ -9,7 +9,7 @@ flavor: implementation owner: claude topic_slug: railiance created: "2026-08-29" -updated: "2026-09-11" +updated: "2026-09-14" depends_on: - AUDIT-WP-0007 state_hub_workstream_id: "46a96b03-bc08-53b5-9c93-4071adabf734" @@ -458,6 +458,13 @@ Remaining: token custody, the protected registry entry, operator manifest application, and live ingestion evidence. No secret was created and no production manifest applied. +**2026-09-14 production blocker:** Informed Decision accept is closed because +the live receiver has been Ready=false since 2026-09-11. `/healthz` is 200; +`/readyz` hangs past kubelet `timeoutSeconds: 2` (observed >8s). The hang is +the chain walk T01 placed on the probe path. Bounded `/readyz` (health only, +cached claim, libpq `connect_timeout`) is ADHOC-2026-09-14-T01; T11 still +needs a rolled image before live accept can resume. + ### Factory sender deployment precondition — 2026-09-11