tegwick
ded432a63f
Postgres now reports custody_class=operational with a cited 30-day recoverable window. Join ITC-CAP operations.audit at D4, publish the interface card, and overlay user-engine tenants [*] from Git so an ExternalSecret refresh cannot shrink it.
502 lines
20 KiB
Python
Executable file
502 lines
20 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Executable delivery/retry/replay failure matrix (AUDIT-WP-0005-T05).
|
|
|
|
The controlling assertion is one sentence: **one source outbox event produces
|
|
exactly one durable normalized event**, across retries, replay, and
|
|
infrastructure disruption. Every scenario below either supports that claim or
|
|
fails loudly.
|
|
|
|
Modes:
|
|
MODE=local (default) start PostgreSQL and the receiver locally in Docker
|
|
and run the full matrix, including infrastructure disruption
|
|
MODE=remote run against an already-deployed receiver at BASE_URL; the
|
|
disruption scenarios are skipped unless DISRUPT=1, because
|
|
restarting a production database is not this script's call
|
|
|
|
Local mode exists so the matrix is a rehearsed script by the time it runs
|
|
against Railiance, rather than something improvised during a deployment.
|
|
|
|
Evidence is written to evidence/failure-matrix-<timestamp>.json and contains no
|
|
secrets — it is intended to be handed back to NK-WP-0024.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
|
|
REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
MODE = os.environ.get("MODE", "local")
|
|
BASE_URL = os.environ.get("BASE_URL", "http://127.0.0.1:8091")
|
|
DISRUPT = os.environ.get("DISRUPT", "1" if MODE == "local" else "0") == "1"
|
|
|
|
PG_CONTAINER = os.environ.get("FM_PG_CONTAINER", "fm-audit-pg")
|
|
PG_PORT = int(os.environ.get("FM_PG_PORT", "55460"))
|
|
APP_PORT = int(os.environ.get("FM_APP_PORT", "8091"))
|
|
|
|
SENDER_TOKEN = os.environ.get("FM_SENDER_TOKEN", "sender-current")
|
|
SENDER_NEXT = os.environ.get("FM_SENDER_NEXT", "sender-next")
|
|
READER_TOKEN = os.environ.get("FM_READER_TOKEN", "reader-token")
|
|
TENANT = "tenant:friendly:binky"
|
|
|
|
# Retry policy modelled on a bounded transactional-outbox sender. Which codes
|
|
# are retried is not arbitrary — it is the response contract in
|
|
# audit_core/ingestion.py, and getting it wrong is how senders either lose
|
|
# events or duplicate them.
|
|
RETRYABLE = {503, 500}
|
|
TERMINAL = {400, 401, 403, 409}
|
|
|
|
|
|
@dataclass
|
|
class Result:
|
|
name: str
|
|
ok: bool
|
|
detail: str = ""
|
|
skipped: bool = False
|
|
|
|
|
|
@dataclass
|
|
class Matrix:
|
|
results: list[Result] = field(default_factory=list)
|
|
|
|
def record(self, name: str, ok: bool, detail: str = "", skipped: bool = False):
|
|
self.results.append(Result(name, ok, detail, skipped))
|
|
mark = "SKIP" if skipped else ("PASS" if ok else "FAIL")
|
|
colour = {"PASS": "\033[32m", "FAIL": "\033[31m", "SKIP": "\033[33m"}[mark]
|
|
print(f" {colour}{mark}\033[0m {name}" + (f" — {detail}" if detail else ""))
|
|
return ok
|
|
|
|
@property
|
|
def failed(self) -> list[Result]:
|
|
return [r for r in self.results if not r.ok and not r.skipped]
|
|
|
|
|
|
# --- transport --------------------------------------------------------------
|
|
|
|
def request(method: str, path: str, token: str | None = None,
|
|
body: dict | None = None, idempotency_key: str | None = None,
|
|
timeout: float = 10.0) -> tuple[int, dict]:
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(f"{BASE_URL}{path}", data=data, method=method)
|
|
if token:
|
|
req.add_header("Authorization", f"Bearer {token}")
|
|
if idempotency_key:
|
|
req.add_header("Idempotency-Key", idempotency_key)
|
|
if data:
|
|
req.add_header("Content-Type", "application/json")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as response:
|
|
return response.status, json.loads(response.read() or b"{}")
|
|
except urllib.error.HTTPError as exc:
|
|
raw = exc.read() or b"{}"
|
|
try:
|
|
return exc.code, json.loads(raw)
|
|
except ValueError:
|
|
return exc.code, {"raw": raw.decode("utf-8", "replace")}
|
|
except (urllib.error.URLError, TimeoutError, ConnectionError) as exc:
|
|
# No response at all — indistinguishable to a sender from a timeout,
|
|
# and therefore retryable.
|
|
return 0, {"error": f"unreachable: {exc}"}
|
|
|
|
|
|
def event(event_id: str, *, tenant: str = TENANT, correlation: str = "corr-matrix",
|
|
data: dict | None = None, subject: str = "membership-1") -> dict:
|
|
return {
|
|
"id": event_id, "type": "membership.added", "source": "user-engine",
|
|
"subject": subject, "tenant": tenant, "correlation_id": correlation,
|
|
"occurred_at": "2026-08-09T00:00:00Z", "data": data or {"membership_id": "m-1"},
|
|
}
|
|
|
|
|
|
def deliver(payload: dict, token: str = SENDER_TOKEN, *, attempts: int = 8,
|
|
backoff: float = 0.5) -> tuple[int, dict, int]:
|
|
"""Deliver like a bounded outbox sender. Returns (status, body, tries)."""
|
|
tries = 0
|
|
status, body = 0, {}
|
|
for attempt in range(attempts):
|
|
tries += 1
|
|
status, body = request("POST", "/v1/events", token, payload, payload["id"])
|
|
if status in TERMINAL or status in (200, 202):
|
|
return status, body, tries
|
|
if status in RETRYABLE or status == 0:
|
|
time.sleep(backoff * (attempt + 1))
|
|
continue
|
|
return status, body, tries
|
|
return status, body, tries
|
|
|
|
|
|
def stored_count(event_id: str) -> int:
|
|
"""Count custody records for an event id, read straight from the database.
|
|
|
|
Deliberately not via the API: the assertion is about what is *stored*, and
|
|
asking the service under test to vouch for itself is weaker evidence.
|
|
"""
|
|
if MODE != "local":
|
|
status, body = request("GET", f"/v1/events/{event_id}", READER_TOKEN)
|
|
return 1 if status == 200 else 0
|
|
out = subprocess.run(
|
|
["docker", "exec", PG_CONTAINER, "psql", "-At", "-U", "postgres",
|
|
"-d", "audit_core", "-c",
|
|
f"SELECT count(*) FROM audit_core.events WHERE event_id = '{event_id}'"],
|
|
capture_output=True, text=True,
|
|
)
|
|
try:
|
|
return int(out.stdout.strip())
|
|
except ValueError:
|
|
return -1
|
|
|
|
|
|
# --- local stack ------------------------------------------------------------
|
|
|
|
REPO = pathlib.Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def start_stack() -> subprocess.Popen:
|
|
subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)
|
|
subprocess.run(
|
|
["docker", "run", "-d", "--name", PG_CONTAINER,
|
|
"-e", "POSTGRES_PASSWORD=matrix", "-e", "POSTGRES_DB=audit_core",
|
|
"-p", f"127.0.0.1:{PG_PORT}:5432", "postgres:16-alpine"],
|
|
check=True, capture_output=True,
|
|
)
|
|
wait_for_postgres()
|
|
|
|
senders = json.dumps([
|
|
{"name": "user-engine", "tokens": [SENDER_TOKEN, SENDER_NEXT],
|
|
"sources": ["user-engine"], "tenants": [TENANT]},
|
|
{"name": "ops-reader", "tokens": [READER_TOKEN], "sources": ["*"],
|
|
"tenants": ["*"], "may_write": False, "may_read": True},
|
|
])
|
|
env = {
|
|
**os.environ,
|
|
"AUDIT_CORE_DATABASE_URL":
|
|
f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core",
|
|
"AUDIT_CORE_SENDERS": senders,
|
|
"AUDIT_CORE_HOST": "127.0.0.1",
|
|
"AUDIT_CORE_HTTP_PORT": str(APP_PORT),
|
|
"AUDIT_CORE_REQUIRE_CUSTODY_CLASS": "operational",
|
|
"AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS": "5000",
|
|
"AUDIT_CORE_LOG_LEVEL": "WARNING",
|
|
}
|
|
python = REPO / ".venv" / "bin" / "python"
|
|
app = subprocess.Popen(
|
|
[str(python if python.exists() else sys.executable), "-m", "audit_core.ingestion"],
|
|
cwd=REPO, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
wait_for_app()
|
|
return app
|
|
|
|
|
|
def wait_for_postgres(timeout: int = 60) -> None:
|
|
for _ in range(timeout):
|
|
probe = subprocess.run(
|
|
["docker", "exec", PG_CONTAINER, "pg_isready", "-U", "postgres"],
|
|
capture_output=True,
|
|
)
|
|
if probe.returncode == 0:
|
|
return
|
|
time.sleep(1)
|
|
raise SystemExit("postgres did not become ready")
|
|
|
|
|
|
def wait_for_app(timeout: int = 40) -> bool:
|
|
for _ in range(timeout):
|
|
status, _ = request("GET", "/healthz", timeout=2)
|
|
if status == 200:
|
|
return True
|
|
time.sleep(1)
|
|
return False
|
|
|
|
|
|
def stop_stack(app: subprocess.Popen | None) -> None:
|
|
if app and app.poll() is None:
|
|
app.terminate()
|
|
try:
|
|
app.wait(timeout=15)
|
|
except subprocess.TimeoutExpired:
|
|
app.kill()
|
|
subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)
|
|
|
|
|
|
# --- the matrix -------------------------------------------------------------
|
|
|
|
def run(matrix: Matrix, app: subprocess.Popen | None) -> None:
|
|
print("\n== delivery ==")
|
|
e1 = f"fm-deliver-{uuid.uuid4().hex[:8]}"
|
|
status, body, tries = deliver(event(e1))
|
|
matrix.record("S01 successful delivery is accepted once",
|
|
status == 202 and body.get("status") == "accepted" and stored_count(e1) == 1,
|
|
f"status={status} tries={tries} stored={stored_count(e1)}")
|
|
|
|
status, body, _ = deliver(event(e1))
|
|
matrix.record("S02 resubmission reconciles as duplicate, no second record",
|
|
status == 200 and body.get("status") == "duplicate" and stored_count(e1) == 1,
|
|
f"status={status} stored={stored_count(e1)}")
|
|
|
|
status, _, _ = deliver(event(e1, subject="changed"))
|
|
matrix.record("S03 same id, different payload is a conflict",
|
|
status == 409 and stored_count(e1) == 1, f"status={status}")
|
|
|
|
print("\n== rejection and dead-letter visibility ==")
|
|
e2 = f"fm-tenant-{uuid.uuid4().hex[:8]}"
|
|
status, body, tries = deliver(event(e2, tenant="tenant:coulomb"))
|
|
matrix.record("S04 cross-tenant claim refused, terminal for the sender",
|
|
status == 400 and body.get("error") == "tenant_not_allowed" and tries == 1,
|
|
f"status={status} tries={tries}")
|
|
|
|
status, dead = request("GET", "/v1/dead-letters", READER_TOKEN)
|
|
visible = any(d.get("event_id") == e2 for d in dead.get("dead_letters", []))
|
|
matrix.record("S05 rejected event is visible as a dead letter", status == 200 and visible)
|
|
|
|
status, _ = request("POST", "/v1/events", "not-a-token", event("fm-x"), "fm-x")
|
|
matrix.record("S06 bad credential is rejected and not retried", status == 401,
|
|
f"status={status}")
|
|
|
|
print("\n== redaction ==")
|
|
e3 = f"fm-redact-{uuid.uuid4().hex[:8]}"
|
|
status, _, _ = deliver(event(e3, data={"membership_id": "m-1", "auth_token": "s3cret"}))
|
|
_, record = request("GET", f"/v1/events/{e3}", READER_TOKEN)
|
|
data = record.get("details", {}).get("data", {})
|
|
redaction = record.get("details", {}).get("redaction", {})
|
|
matrix.record("S07 secret-shaped field is redacted, event still stored",
|
|
status == 202 and data.get("auth_token") == "[redacted]"
|
|
and data.get("membership_id") == "m-1"
|
|
and redaction.get("paths") == ["data.auth_token"],
|
|
f"stored auth_token={data.get('auth_token')!r}")
|
|
|
|
_, findings = request("GET", "/v1/secret-findings", READER_TOKEN)
|
|
counted = any(f["field_path"] == "data.auth_token" and f["outcome"] == "redacted"
|
|
for f in findings.get("secret_findings", []))
|
|
matrix.record("S08 redaction is counted by field path", counted)
|
|
|
|
print("\n== correlation lookup ==")
|
|
correlation = f"corr-{uuid.uuid4().hex[:8]}"
|
|
ids = [f"fm-corr-{i}-{uuid.uuid4().hex[:6]}" for i in range(3)]
|
|
for event_id in ids:
|
|
deliver(event(event_id, correlation=correlation))
|
|
status, found = request("GET", f"/v1/events?correlation_id={correlation}", READER_TOKEN)
|
|
matrix.record("S09 correlation lookup returns every related event",
|
|
status == 200 and {e["event_id"] for e in found.get("events", [])} == set(ids),
|
|
f"found={len(found.get('events', []))}/3")
|
|
|
|
print("\n== privilege separation ==")
|
|
status, _ = request("GET", f"/v1/events/{e1}", SENDER_TOKEN)
|
|
matrix.record("S10 sender credential cannot read the trail back", status == 403,
|
|
f"status={status}")
|
|
status, _ = request("POST", "/v1/events", READER_TOKEN, event("fm-ro"), "fm-ro")
|
|
matrix.record("S11 reader credential cannot write", status == 403, f"status={status}")
|
|
|
|
print("\n== credential rotation during active ingestion ==")
|
|
rotated = []
|
|
for i in range(3):
|
|
token = SENDER_TOKEN if i == 0 else SENDER_NEXT
|
|
event_id = f"fm-rot-{i}-{uuid.uuid4().hex[:6]}"
|
|
status, _, _ = deliver(event(event_id), token=token)
|
|
rotated.append((event_id, status))
|
|
matrix.record("S12 rotation to the next token has no delivery gap",
|
|
all(s == 202 for _, s in rotated)
|
|
and all(stored_count(i) == 1 for i, _ in rotated),
|
|
f"statuses={[s for _, s in rotated]}")
|
|
|
|
print("\n== replay ==")
|
|
if MODE == "local":
|
|
replay_ok, detail = replay_scenarios(e1)
|
|
matrix.record("S13 operator replay reconciles, no second record", replay_ok, detail)
|
|
else:
|
|
matrix.record("S13 operator replay reconciles, no second record", True,
|
|
"needs database access; run in local mode", skipped=True)
|
|
|
|
print("\n== infrastructure disruption ==")
|
|
if not DISRUPT:
|
|
for name in ("S14 receiver unavailable: sender retries, exactly one record",
|
|
"S15 database restart mid-ingestion: no acknowledged-but-absent event"):
|
|
matrix.record(name, True, "DISRUPT=0", skipped=True)
|
|
return
|
|
|
|
matrix.record(*receiver_unavailable(app))
|
|
matrix.record(*database_restart())
|
|
|
|
|
|
def replay_scenarios(event_id: str) -> tuple[bool, str]:
|
|
"""Replay, then replay again. Neither may create a second record."""
|
|
from audit_core.postgres_backend import PostgresAuditBackend
|
|
|
|
backend = PostgresAuditBackend(
|
|
f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core", migrate=False
|
|
)
|
|
try:
|
|
first = backend.replay(event_id)
|
|
second = backend.replay(event_id)
|
|
count = stored_count(event_id)
|
|
return (
|
|
first.duplicate and second.duplicate and count == 1,
|
|
f"first.duplicate={first.duplicate} second.duplicate={second.duplicate} stored={count}",
|
|
)
|
|
finally:
|
|
backend.close()
|
|
|
|
|
|
def receiver_unavailable(app: subprocess.Popen | None) -> tuple[str, bool, str]:
|
|
"""Stop the receiver mid-flight; the sender must retry to exactly one record."""
|
|
name = "S14 receiver unavailable: sender retries, exactly one record"
|
|
if app is None:
|
|
return (name, True, "local mode only")
|
|
app.terminate()
|
|
app.wait(timeout=15)
|
|
|
|
event_id = f"fm-down-{uuid.uuid4().hex[:8]}"
|
|
status, _ = request("POST", "/v1/events", SENDER_TOKEN, event(event_id), event_id, timeout=3)
|
|
if status != 0:
|
|
return (name, False, f"expected no response while down, got {status}")
|
|
|
|
restarted = restart_app()
|
|
if not restarted:
|
|
return (name, False, "receiver did not come back up")
|
|
status, _, tries = deliver(event(event_id))
|
|
count = stored_count(event_id)
|
|
return (name, status == 202 and count == 1,
|
|
f"after recovery status={status} tries={tries} stored={count}")
|
|
|
|
|
|
_app_handle: list[subprocess.Popen] = []
|
|
|
|
|
|
def restart_app() -> bool:
|
|
app = start_app_process()
|
|
_app_handle.append(app)
|
|
return wait_for_app()
|
|
|
|
|
|
def start_app_process() -> subprocess.Popen:
|
|
senders = json.dumps([
|
|
{"name": "user-engine", "tokens": [SENDER_TOKEN, SENDER_NEXT],
|
|
"sources": ["user-engine"], "tenants": [TENANT]},
|
|
{"name": "ops-reader", "tokens": [READER_TOKEN], "sources": ["*"],
|
|
"tenants": ["*"], "may_write": False, "may_read": True},
|
|
])
|
|
env = {
|
|
**os.environ,
|
|
"AUDIT_CORE_DATABASE_URL":
|
|
f"postgresql://postgres:matrix@127.0.0.1:{PG_PORT}/audit_core",
|
|
"AUDIT_CORE_SENDERS": senders,
|
|
"AUDIT_CORE_HOST": "127.0.0.1",
|
|
"AUDIT_CORE_HTTP_PORT": str(APP_PORT),
|
|
"AUDIT_CORE_REQUIRE_CUSTODY_CLASS": "operational",
|
|
"AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS": "5000",
|
|
"AUDIT_CORE_LOG_LEVEL": "WARNING",
|
|
}
|
|
python = REPO / ".venv" / "bin" / "python"
|
|
return subprocess.Popen(
|
|
[str(python if python.exists() else sys.executable), "-m", "audit_core.ingestion"],
|
|
cwd=REPO, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
|
|
|
|
def database_restart() -> tuple[str, bool, str]:
|
|
"""Restart PostgreSQL under load.
|
|
|
|
The property under test is not "no errors" — errors are expected and
|
|
correct. It is that no event is *acknowledged* without being stored. A 202
|
|
that does not survive the restart is the failure mode this whole service
|
|
exists to prevent.
|
|
"""
|
|
name = "S15 database restart mid-ingestion: no acknowledged-but-absent event"
|
|
acknowledged: list[str] = []
|
|
|
|
subprocess.run(["docker", "restart", "-t", "1", PG_CONTAINER],
|
|
capture_output=True, check=True)
|
|
|
|
deadline = time.time() + 45
|
|
attempts = 0
|
|
while time.time() < deadline and len(acknowledged) < 5:
|
|
event_id = f"fm-db-{attempts}-{uuid.uuid4().hex[:6]}"
|
|
attempts += 1
|
|
status, _ = request("POST", "/v1/events", SENDER_TOKEN,
|
|
event(event_id), event_id, timeout=8)
|
|
if status == 202:
|
|
acknowledged.append(event_id)
|
|
elif status not in (0, 503, 500):
|
|
return (name, False, f"unexpected status {status} during restart")
|
|
time.sleep(0.5)
|
|
|
|
wait_for_postgres()
|
|
if not acknowledged:
|
|
return (name, False, "nothing was acknowledged after restart; receiver did not recover")
|
|
|
|
missing = [e for e in acknowledged if stored_count(e) != 1]
|
|
return (name, not missing,
|
|
f"acknowledged={len(acknowledged)} attempts={attempts} missing={missing}")
|
|
|
|
|
|
# --- entry point ------------------------------------------------------------
|
|
|
|
def main() -> int:
|
|
matrix = Matrix()
|
|
app = None
|
|
print(f"== audit-core failure matrix (MODE={MODE} BASE_URL={BASE_URL} DISRUPT={DISRUPT}) ==")
|
|
try:
|
|
if MODE == "local":
|
|
app = start_stack()
|
|
_app_handle.append(app)
|
|
elif not wait_for_app(timeout=5):
|
|
print(f"receiver at {BASE_URL} is not reachable", file=sys.stderr)
|
|
return 2
|
|
run(matrix, app)
|
|
finally:
|
|
for handle in _app_handle:
|
|
if handle.poll() is None:
|
|
handle.terminate()
|
|
try:
|
|
handle.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
handle.kill()
|
|
if MODE == "local":
|
|
subprocess.run(["docker", "rm", "-f", PG_CONTAINER], capture_output=True)
|
|
|
|
passed = sum(1 for r in matrix.results if r.ok and not r.skipped)
|
|
skipped = sum(1 for r in matrix.results if r.skipped)
|
|
print(f"\n== {passed} passed, {len(matrix.failed)} failed, {skipped} skipped ==")
|
|
|
|
evidence = {
|
|
"workplan_task": "AUDIT-WP-0005-T05",
|
|
"mode": MODE,
|
|
"disruption_scenarios_run": DISRUPT,
|
|
"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"controlling_assertion":
|
|
"one source outbox event produces exactly one durable normalized event "
|
|
"across retries, replay and infrastructure disruption",
|
|
"passed": passed,
|
|
"failed": len(matrix.failed),
|
|
"skipped": skipped,
|
|
"scenarios": [
|
|
{"name": r.name, "outcome": "skip" if r.skipped else ("pass" if r.ok else "fail"),
|
|
"detail": r.detail}
|
|
for r in matrix.results
|
|
],
|
|
}
|
|
out_dir = REPO / "evidence"
|
|
out_dir.mkdir(exist_ok=True)
|
|
path = out_dir / f"failure-matrix-{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}.json"
|
|
path.write_text(json.dumps(evidence, indent=2) + "\n")
|
|
print(f"evidence: {path.relative_to(REPO)} (non-secret, for NK-WP-0024)")
|
|
|
|
if matrix.failed:
|
|
print("\nFailures:")
|
|
for result in matrix.failed:
|
|
print(f" - {result.name}: {result.detail}")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|