diff --git a/Containerfile b/Containerfile new file mode 100644 index 0000000..9df5742 --- /dev/null +++ b/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"] diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md new file mode 100644 index 0000000..647931f --- /dev/null +++ b/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 | diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py new file mode 100644 index 0000000..417a0a6 --- /dev/null +++ b/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() diff --git a/pyproject.toml b/pyproject.toml index 7692f8d..ae6eaa8 100644 --- a/pyproject.toml +++ b/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"] diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py new file mode 100644 index 0000000..949c81f --- /dev/null +++ b/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" diff --git a/workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md b/workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md new file mode 100644 index 0000000..c6ae87b --- /dev/null +++ b/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.