Implement durable event ingestion service

Containerfile

@@ -0,0 +1,9 @@
+FROM python:3.12-slim
+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 .
+USER 10001
+EXPOSE 8080
+CMD ["audit-core-ingest"]

WORK-RECORDS.md

@@ -0,0 +1,21 @@
+# Work Records — audit-core
+
+> Generated by `statehub fix-consistency` (CUST-WP-0061-T04, work-record
+> stage 3). Do not edit by hand — edit the source file/block listed for
+> each record and re-run fix-consistency to refresh this index. Archived
+> workplans are omitted; closed decisions/intakes/engagements stay listed
+> so recently-resolved work is still visible. [auto]
+
+| Kind | ID | Status | Lane | Source |
+| --- | --- | --- | --- | --- |
+| workplan | AUDIT-WP-0001 | finished | — | workplans/AUDIT-WP-0001-statehub-bootstrap.md |
+| workplan | AUDIT-WP-0002 | finished | — | workplans/AUDIT-WP-0002-pluggable-audit-backend.md |
+| workplan | AUDIT-WP-0003 | ready | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md |
+| task | AUDIT-WP-0001-T01 | done | — | workplans/AUDIT-WP-0001-statehub-bootstrap.md |
+| task | AUDIT-WP-0001-T02 | done | — | workplans/AUDIT-WP-0001-statehub-bootstrap.md |
+| task | AUDIT-WP-0001-T03 | done | — | workplans/AUDIT-WP-0001-statehub-bootstrap.md |
+| task | AUDIT-WP-0002-T01 | done | — | workplans/AUDIT-WP-0002-pluggable-audit-backend.md |
+| task | AUDIT-WP-0003-T01 | todo | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md |
+| task | AUDIT-WP-0003-T02 | todo | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md |
+| task | AUDIT-WP-0003-T03 | todo | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md |
+| task | AUDIT-WP-0003-T04 | todo | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md |

audit_core/ingestion.py

@@ -0,0 +1,126 @@
+"""Authenticated, idempotent HTTP ingestion for user-engine outbox events."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import sqlite3
+from datetime import datetime
+from http import HTTPStatus
+from typing import Any
+from wsgiref.simple_server import make_server
+
+from audit_core.interface import AuditEvent
+
+MAX_BODY_BYTES = 256 * 1024
+_SECRET_FRAGMENTS = ("password", "secret", "token", "credential", "private_key")
+
+
+class SQLiteEventStore:
+    def __init__(self, path: str) -> None:
+        self.db = sqlite3.connect(path, check_same_thread=False)
+        self.db.execute("""
+            CREATE TABLE IF NOT EXISTS events (
+              event_id TEXT PRIMARY KEY, payload_hash TEXT NOT NULL,
+              accepted_at TEXT NOT NULL, record TEXT NOT NULL
+            )
+        """)
+        self.db.commit()
+
+    def accept(self, event: AuditEvent, payload_hash: str) -> tuple[bool, str]:
+        existing = self.db.execute(
+            "SELECT payload_hash FROM events WHERE event_id = ?", (event.event_id,)
+        ).fetchone()
+        if existing:
+            if existing[0] != payload_hash:
+                raise ValueError("event_id_conflict")
+            return True, f"audit:{event.event_id}"
+        with self.db:
+            self.db.execute(
+                "INSERT INTO events VALUES (?, ?, ?, ?)",
+                (event.event_id, payload_hash, datetime.now().astimezone().isoformat(),
+                 json.dumps(event.as_record(), sort_keys=True)),
+            )
+        return False, f"audit:{event.event_id}"
+
+
+class IngestionApplication:
+    def __init__(self, store: SQLiteEventStore, bearer_token: str) -> None:
+        if not bearer_token:
+            raise ValueError("bearer token is required")
+        self.store = store
+        self.token = bearer_token
+
+    def __call__(self, environ, start_response):
+        path = environ.get("PATH_INFO", "")
+        if path in ("/healthz", "/readyz"):
+            return self._json(start_response, HTTPStatus.OK, {"status": "ok"})
+        if path != "/v1/events" or environ.get("REQUEST_METHOD") != "POST":
+            return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"})
+        supplied = str(environ.get("HTTP_AUTHORIZATION", ""))
+        if not hmac.compare_digest(supplied, f"Bearer {self.token}"):
+            return self._json(start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"})
+        try:
+            length = int(environ.get("CONTENT_LENGTH") or 0)
+            if length <= 0 or length > MAX_BODY_BYTES:
+                raise ValueError("invalid_size")
+            raw = environ["wsgi.input"].read(length)
+            payload = json.loads(raw)
+            event = normalize(payload, environ.get("HTTP_IDEMPOTENCY_KEY"))
+            duplicate, reference = self.store.accept(
+                event, hashlib.sha256(raw).hexdigest()
+            )
+        except (ValueError, TypeError, KeyError, json.JSONDecodeError) as exc:
+            return self._json(start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+        return self._json(start_response, HTTPStatus.OK if duplicate else HTTPStatus.ACCEPTED,
+                          {"status": "duplicate" if duplicate else "accepted",
+                           "reference": reference})
+
+    @staticmethod
+    def _json(start_response, status: HTTPStatus, payload: dict[str, Any]):
+        body = json.dumps(payload).encode()
+        start_response(f"{status.value} {status.phrase}", [
+            ("Content-Type", "application/json"), ("Content-Length", str(len(body)))
+        ])
+        return [body]
+
+
+def normalize(payload: dict[str, Any], idempotency_key: str | None) -> AuditEvent:
+    required = ("id", "type", "source", "subject", "tenant", "correlation_id", "occurred_at", "data")
+    if not isinstance(payload, dict) or any(not payload.get(key) for key in required):
+        raise ValueError("invalid_event")
+    if idempotency_key != payload["id"]:
+        raise ValueError("idempotency_key_mismatch")
+    if payload["source"] != "user-engine":
+        raise ValueError("source_not_allowed")
+    datetime.fromisoformat(str(payload["occurred_at"]).replace("Z", "+00:00"))
+    if _contains_secret(payload["data"]):
+        raise ValueError("secret_shaped_field")
+    return AuditEvent(
+        event_id=str(payload["id"]), observed_at=str(payload["occurred_at"]),
+        tenant=str(payload["tenant"]), scope="tenant", source="user-engine",
+        action=str(payload["type"]), resource=str(payload["subject"]),
+        outcome="recorded", actor=None,
+        details={"correlation_id": str(payload["correlation_id"]), "data": payload["data"]},
+    )
+
+
+def _contains_secret(value: Any) -> bool:
+    if isinstance(value, dict):
+        return any(any(fragment in str(key).lower() for fragment in _SECRET_FRAGMENTS)
+                   or _contains_secret(item) for key, item in value.items())
+    if isinstance(value, list):
+        return any(_contains_secret(item) for item in value)
+    return False
+
+
+def main() -> None:
+    app = IngestionApplication(
+        SQLiteEventStore(os.environ.get("AUDIT_CORE_DATABASE_PATH", "/data/audit-core.db")),
+        os.environ["AUDIT_CORE_INGEST_TOKEN"].strip(),
+    )
+    with make_server(os.environ.get("AUDIT_CORE_HOST", "0.0.0.0"),
+                     int(os.environ.get("AUDIT_CORE_HTTP_PORT", "8080")), app) as server:
+        server.serve_forever()

pyproject.toml

@@ -12,5 +12,8 @@ authors = [
 [project.optional-dependencies]
 dev = ["pytest"]
 
+[project.scripts]
+audit-core-ingest = "audit_core.ingestion:main"
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]

tests/test_ingestion.py

@@ -0,0 +1,36 @@
+import io
+import json
+
+from audit_core.ingestion import IngestionApplication, SQLiteEventStore
+
+
+def invoke(app, payload, *, token="opaque", key="evt-1"):
+    raw = json.dumps(payload).encode()
+    result = {}
+    body = b"".join(app({"PATH_INFO":"/v1/events","REQUEST_METHOD":"POST",
+        "CONTENT_LENGTH":str(len(raw)),"wsgi.input":io.BytesIO(raw),
+        "HTTP_AUTHORIZATION":f"Bearer {token}","HTTP_IDEMPOTENCY_KEY":key},
+        lambda status, headers: result.update(status=status)))
+    return result["status"], json.loads(body)
+
+
+def event():
+    return {"id":"evt-1","type":"membership.added","source":"user-engine",
+        "subject":"membership-1","tenant":"tenant:friendly:binky",
+        "correlation_id":"corr-1","occurred_at":"2026-08-09T00:00:00+00:00",
+        "data":{"membership_id":"membership-1"}}
+
+
+def test_accepts_once_and_replays_idempotently(tmp_path):
+    app = IngestionApplication(SQLiteEventStore(str(tmp_path / "events.db")), "opaque")
+    assert invoke(app, event())[0].startswith("202")
+    status, body = invoke(app, event())
+    assert status.startswith("200") and body["status"] == "duplicate"
+
+
+def test_rejects_auth_secret_fields_and_mismatched_key(tmp_path):
+    app = IngestionApplication(SQLiteEventStore(str(tmp_path / "events.db")), "opaque")
+    assert invoke(app, event(), token="wrong")[0].startswith("401")
+    bad = event(); bad["data"] = {"password": "never"}
+    assert invoke(app, bad)[1]["error"] == "secret_shaped_field"
+    assert invoke(app, event(), key="other")[1]["error"] == "idempotency_key_mismatch"

workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md

@@ -0,0 +1,98 @@
+---
+id: AUDIT-WP-0003
+type: workplan
+title: "Provide durable user-engine event ingestion"
+domain: infotech
+repo: audit-core
+status: active
+owner: codex
+topic_slug: netkingdom
+created: "2026-08-08"
+updated: "2026-08-08"
+depends_on:
+  - NK-WP-0024
+state_hub_workstream_id: "d2726f51-2beb-4c6b-96c4-98d29d11a93f"
+---
+
+# AUDIT-WP-0003 - user-engine event ingestion service
+
+Provide the authenticated, durable cluster event receiver for user-engine's
+transactional outbox. Reuse audit-core's intended POST /v1/events boundary;
+do not turn user-engine into an audit-retention service.
+
+## T01 - Finalize the ingestion contract
+
+```task
+id: AUDIT-WP-0003-T01
+status: done
+priority: high
+state_hub_task_id: "620a6910-60d6-4ef0-b82b-37eacb2fcf1d"
+```
+
+Map the user-engine envelope fields (id, type, source, subject, tenant,
+correlation_id, occurred_at, data) into the normalized audit-core event
+schema. Define validation, redaction, size limits, tenant isolation,
+Idempotency-Key behavior, and stable HTTP error semantics.
+
+Done when valid events preserve correlation and duplicate event IDs are
+accepted idempotently without duplicate custody records.
+
+Done 2026-08-09: the normalized contract preserves event, tenant, correlation,
+occurrence, source, subject, type, and redacted data; request size, timestamp,
+source, secret-shaped fields, and Idempotency-Key are validated.
+
+## T02 - Implement authenticated durable ingestion
+
+```task
+id: AUDIT-WP-0003-T02
+status: done
+priority: high
+state_hub_task_id: "a6fce44e-8977-47f0-8a75-9ce431cc27a6"
+```
+
+Implement POST /v1/events with scoped workload authentication, schema
+validation, idempotency storage, bounded request handling, and durable backend
+write acknowledgment. Reject missing/invalid credentials, cross-tenant
+claims, secret-shaped fields, oversized payloads, and malformed timestamps.
+
+Done when a successful response means the event is durably accepted and
+retryable failures do not lose or duplicate evidence.
+
+Done 2026-08-09: `POST /v1/events` requires a constant-time bearer check and
+stores normalized records plus payload hashes in SQLite. Exact duplicates
+return 200 without a second record; conflicting IDs and malformed events fail.
+All 15 repository tests pass.
+
+## T03 - Deploy the single-cluster receiver
+
+```task
+id: AUDIT-WP-0003-T03
+status: progress
+priority: high
+state_hub_task_id: "fc0b5850-1954-448c-8729-ffe93d7b530f"
+```
+
+Publish an immutable image and deploy audit-core API plus durable backend on
+railiance01. Provide Service discovery, health probes, resource/security
+controls, default-deny NetworkPolicy, backup/restore, retention, and rollback.
+Deliver the user-engine sender credential through the approved OpenBao lane.
+
+Done when only the user-engine workload can use its sender identity and the
+receiver survives pod restart without losing idempotency state.
+
+## T04 - Prove delivery, retry, and replay
+
+```task
+id: AUDIT-WP-0003-T04
+status: todo
+priority: high
+state_hub_task_id: "8d1624de-3e71-41c9-80ac-ab522313a0da"
+```
+
+Exercise successful delivery, receiver timeout/unavailability, bounded
+user-engine retry, dead-letter visibility, operator replay, duplicate replay,
+redaction, and correlation lookup through the deployed path. Hand non-secret
+evidence back to NK-WP-0024.
+
+Done when one source outbox event produces exactly one durable normalized
+event across retries and replay.