diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index 117955e..b9f98e0 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -10,12 +10,27 @@ | --- | --- | --- | --- | --- | | 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 | active | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md | +| workplan | AUDIT-WP-0003 | finished | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md | +| workplan | AUDIT-WP-0004 | proposed | — | workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md | +| workplan | AUDIT-WP-0005 | proposed | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.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 | done | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md | | task | AUDIT-WP-0003-T02 | done | — | workplans/AUDIT-WP-0003-user-engine-event-ingestion-service.md | -| task | AUDIT-WP-0003-T03 | progress | — | 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 | +| 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-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-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 | +| task | AUDIT-WP-0005-T04 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | +| task | AUDIT-WP-0005-T05 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | +| task | AUDIT-WP-0005-T06 | todo | — | workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md | diff --git a/audit_core/__init__.py b/audit_core/__init__.py index 788bb22..d331eda 100644 --- a/audit_core/__init__.py +++ b/audit_core/__init__.py @@ -4,21 +4,31 @@ Contract reference: ``docs/audit-backend-contract.md``. """ from audit_core.interface import ( + AcceptResult, AuditBackend, AuditEvent, + BackendUnavailableError, + EventConflictError, EventValidationError, + IdempotentAuditBackend, RetentionPolicy, SCHEMA_VERSION_V1ALPHA1, validate_event, ) from audit_core.mock_file_backend import MockFileAuditBackend +from audit_core.sqlite_backend import SQLiteAuditBackend __all__ = [ + "AcceptResult", "AuditBackend", "AuditEvent", + "BackendUnavailableError", + "EventConflictError", "EventValidationError", + "IdempotentAuditBackend", "MockFileAuditBackend", "RetentionPolicy", "SCHEMA_VERSION_V1ALPHA1", + "SQLiteAuditBackend", "validate_event", ] diff --git a/audit_core/ingestion.py b/audit_core/ingestion.py index 417a0a6..97441bf 100644 --- a/audit_core/ingestion.py +++ b/audit_core/ingestion.py @@ -1,126 +1,247 @@ -"""Authenticated, idempotent HTTP ingestion for user-engine outbox events.""" +"""Authenticated, idempotent HTTP ingestion for user-engine outbox events. + +Response contract (AUDIT-WP-0004-T02). Senders key their retry behaviour off +these, so they are part of the interface, not an implementation detail: + +=== ========== ========================================================= +202 accepted Event is durably in custody. Do not retry. +200 duplicate Exact resubmission of an event already in custody. Do not + retry; delivery already succeeded. +400 rejected Malformed or disallowed. Retrying will not help — dead + letter it. +401 unauthorized Credential missing or invalid. Do not retry without a + new credential. +409 conflict The event id is held with a different payload. Retrying + will not help; this indicates a sender bug or id reuse. +503 unavailable Not accepted, but retryable. Retry with backoff. +500 error Unexpected fault. Not accepted. Retryable with backoff. +=== ========== ========================================================= +""" from __future__ import annotations import hashlib import hmac import json +import logging import os -import sqlite3 -from datetime import datetime +import sys +from datetime import datetime, timezone from http import HTTPStatus from typing import Any from wsgiref.simple_server import make_server -from audit_core.interface import AuditEvent +from audit_core.interface import ( + AuditEvent, + BackendUnavailableError, + EventConflictError, + EventValidationError, + IdempotentAuditBackend, +) +from audit_core.sqlite_backend import SQLiteAuditBackend 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}" +log = logging.getLogger("audit_core.ingestion") class IngestionApplication: - def __init__(self, store: SQLiteEventStore, bearer_token: str) -> None: + """WSGI application accepting user-engine outbox events. + + Writes through the audit backend contract rather than to storage directly, + so a 202 means a backend with a declared retention policy acknowledged the + event. + """ + + def __init__(self, backend: IdempotentAuditBackend, bearer_token: str) -> None: if not bearer_token: raise ValueError("bearer token is required") - self.store = store + policy = backend.retention_policy + if not policy.durable: + # The mock file backend declares durable=False. Refusing it here is + # what stops a development sink from silently becoming the + # production one (AUDIT-WP-0004-T01). + raise ValueError( + f"backend custody_class={policy.custody_class!r} is not durable; " + "refusing to accept audit events against it" + ) + self.backend = backend self.token = bearer_token def __call__(self, environ, start_response): + try: + return self._handle(environ, start_response) + except Exception: + # Nothing may escape: an unhandled exception here means + # start_response is never called and the sender sees a dropped + # connection it cannot classify. + log.exception("unhandled error in ingestion request") + return self._json( + start_response, HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "internal_error"} + ) + + def _handle(self, environ, start_response): path = environ.get("PATH_INFO", "") - if path in ("/healthz", "/readyz"): + if path == "/healthz": 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"}) - 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() + + if not self._authorized(environ): + return self._json( + start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"} ) + + try: + raw = self._read_body(environ) + event = normalize(json.loads(raw), environ.get("HTTP_IDEMPOTENCY_KEY")) 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}) + + try: + result = self.backend.accept(event, hashlib.sha256(raw).hexdigest()) + except EventConflictError as exc: + log.warning("event conflict: %s", exc) + return self._json(start_response, HTTPStatus.CONFLICT, {"error": "event_id_conflict"}) + except EventValidationError as exc: + return self._json(start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)}) + except BackendUnavailableError as exc: + log.error("backend unavailable: %s", exc) + return self._json( + start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"error": "backend_unavailable"} + ) + + return self._json( + start_response, + HTTPStatus.OK if result.duplicate else HTTPStatus.ACCEPTED, + { + "status": "duplicate" if result.duplicate else "accepted", + "reference": result.reference, + }, + ) + + def _authorized(self, environ) -> bool: + supplied = str(environ.get("HTTP_AUTHORIZATION", "")) + expected = f"Bearer {self.token}" + 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 + + def _read_body(self, environ) -> bytes: + try: + length = int(environ.get("CONTENT_LENGTH") or 0) + except (TypeError, ValueError): + raise ValueError("invalid_content_length") from None + if length <= 0: + raise ValueError("empty_body") + if length > MAX_BODY_BYTES: + raise ValueError("payload_too_large") + raw = environ["wsgi.input"].read(length) + if len(raw) != length: + raise ValueError("truncated_body") + return raw + + 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) + return self._json( + start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"status": "unavailable"} + ) + policy = self.backend.retention_policy + return self._json( + start_response, + HTTPStatus.OK, + {"status": "ok", "custody_class": policy.custody_class, "durable": policy.durable}, + ) @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))) - ]) + 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") + 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")) + observed_at = _normalize_timestamp(payload["occurred_at"]) 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, + event_id=str(payload["id"]), + observed_at=observed_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 _normalize_timestamp(value: Any) -> str: + """Parse an event timestamp, requiring an explicit offset. + + A naive timestamp is ambiguous by up to a day, which is not good enough for + an audit trail — the sender must say which offset it meant. + """ + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except (TypeError, ValueError): + raise ValueError("invalid_timestamp") from None + if parsed.tzinfo is None: + raise ValueError("timestamp_missing_timezone") + return parsed.astimezone(timezone.utc).isoformat() + + 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()) + 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(), + logging.basicConfig( + level=os.environ.get("AUDIT_CORE_LOG_LEVEL", "INFO"), + format='{"ts":"%(asctime)s","level":"%(levelname)s","logger":"%(name)s","msg":"%(message)s"}', + stream=sys.stdout, ) - 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: + 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() diff --git a/audit_core/interface.py b/audit_core/interface.py index 2f38bdf..df5d9de 100644 --- a/audit_core/interface.py +++ b/audit_core/interface.py @@ -86,6 +86,31 @@ class EventValidationError(ValueError): """Raised when an event record fails contract validation.""" +class EventConflictError(Exception): + """Raised when an event id is resubmitted with a different payload. + + Distinct from :class:`EventValidationError`: the event is well-formed, but + it contradicts a record already in custody. Callers map this to a conflict + response rather than a rejection, because retrying it will never succeed. + """ + + +class BackendUnavailableError(Exception): + """Raised when a backend cannot currently accept writes. + + Signals a retryable condition — the event is not in custody and the caller + should try again. Never raised for events the backend has durably accepted. + """ + + +@dataclass(frozen=True) +class AcceptResult: + """Outcome of an idempotent accept.""" + + duplicate: bool + reference: str + + def validate_event(event: AuditEvent) -> None: """Validate an event against the v1alpha1 contract. @@ -127,4 +152,30 @@ class AuditBackend(Protocol): """Describe retention and custody guarantees for readiness checks.""" def emit(self, event: AuditEvent) -> str: - """Persist an event and return a backend-specific reference.""" \ No newline at end of file + """Persist an event and return a backend-specific reference.""" + + +@runtime_checkable +class IdempotentAuditBackend(AuditBackend, Protocol): + """An :class:`AuditBackend` that can absorb duplicate submissions. + + Ingestion requires this rather than the plain backend protocol. The reason + is atomicity: if duplicate detection lived in the ingestion layer and + custody in the backend, the two could diverge — an event recorded as seen + but never durably stored, which is precisely the loss this service exists + to prevent. Keeping both inside one backend call lets an implementation put + them in a single transaction. + """ + + def accept(self, event: AuditEvent, payload_hash: str) -> AcceptResult: + """Durably record an event, tolerating exact resubmission. + + Returns an :class:`AcceptResult` whose ``duplicate`` flag distinguishes + a first acceptance from a replay of an identical event. Both mean the + event is in custody. + + Raises :class:`EventConflictError` if ``event.event_id`` is already held + with a different ``payload_hash``, and :class:`BackendUnavailableError` + if the write could not be attempted. Returning normally must mean the + event is durable. + """ \ No newline at end of file diff --git a/audit_core/sqlite_backend.py b/audit_core/sqlite_backend.py new file mode 100644 index 0000000..6eeb6a0 --- /dev/null +++ b/audit_core/sqlite_backend.py @@ -0,0 +1,177 @@ +"""Durable SQLite audit backend. + +Implements the idempotent backend contract for single-node deployments and for +development. Production custody moves to PostgreSQL under AUDIT-WP-0005; this +backend stays the development and test implementation and defines the +behaviour the Postgres backend must match. +""" + +from __future__ import annotations + +import json +import sqlite3 +import threading +from datetime import datetime, timezone + +from audit_core.interface import ( + AcceptResult, + AuditEvent, + BackendUnavailableError, + EventConflictError, + EventValidationError, + RetentionPolicy, + validate_event, +) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS events ( + event_id TEXT PRIMARY KEY, + payload_hash TEXT NOT NULL, + accepted_at TEXT NOT NULL, + correlation_id TEXT, + tenant TEXT NOT NULL, + record TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS events_correlation_idx ON events (correlation_id); +CREATE INDEX IF NOT EXISTS events_tenant_idx ON events (tenant); +""" + + +class SQLiteAuditBackend: + """Store audit events in SQLite with idempotent accept semantics. + + Configured for durability rather than speed: WAL journalling, ``synchronous + = FULL`` so an acknowledged write has reached disk, and a busy timeout so + concurrent writers wait instead of raising immediately. + """ + + def __init__(self, path: str, retention_days: int | None = None, busy_timeout_ms: int = 5000) -> None: + self.path = path + self.retention_days = retention_days + self.busy_timeout_ms = int(busy_timeout_ms) + # One connection per thread. A shared connection lets concurrent + # statements interleave, which was observed to let two callers both + # believe they were the first to accept the same event. + self._local = threading.local() + with self._connect_raw() as setup: + setup.executescript(_SCHEMA) + + def _connect_raw(self) -> sqlite3.Connection: + try: + db = sqlite3.connect(self.path, isolation_level=None) + db.execute("PRAGMA journal_mode = WAL") + db.execute("PRAGMA synchronous = FULL") + db.execute(f"PRAGMA busy_timeout = {self.busy_timeout_ms}") + return db + except sqlite3.Error as exc: + raise BackendUnavailableError(f"cannot open audit store: {exc}") from exc + + @property + def db(self) -> sqlite3.Connection: + conn = getattr(self._local, "conn", None) + if conn is None: + conn = self._local.conn = self._connect_raw() + return conn + + @property + def retention_policy(self) -> RetentionPolicy: + return RetentionPolicy( + custody_class="development", + retention_days=self.retention_days, + immutable=False, + tamper_evidence=False, + durable=True, + ) + + def emit(self, event: AuditEvent) -> str: + """Persist an event, generating no idempotency guarantee.""" + return self.accept(event, payload_hash=_record_hash(event)).reference + + def accept(self, event: AuditEvent, payload_hash: str) -> AcceptResult: + try: + validate_event(event) + except EventValidationError: + raise + reference = f"audit:{event.event_id}" + details = event.details if isinstance(event.details, dict) else {} + db = self.db + try: + # BEGIN IMMEDIATE takes the write lock up front, so the insert and + # the follow-up read are one atomic pair. Without it, two callers + # racing on the same event id can both be told they were first. + db.execute("BEGIN IMMEDIATE") + except sqlite3.Error as exc: + raise BackendUnavailableError(str(exc)) from exc + + try: + inserted = db.execute( + """ + INSERT INTO events + (event_id, payload_hash, accepted_at, correlation_id, tenant, record) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(event_id) DO NOTHING + RETURNING event_id + """, + ( + event.event_id, + payload_hash, + datetime.now(timezone.utc).replace(microsecond=0).isoformat(), + str(details.get("correlation_id") or "") or None, + event.tenant, + json.dumps(event.as_record(), sort_keys=True), + ), + ).fetchone() + existing = None + if inserted is None: + existing = db.execute( + "SELECT payload_hash FROM events WHERE event_id = ?", (event.event_id,) + ).fetchone() + db.execute("COMMIT") + except sqlite3.Error as exc: + _rollback(db) + raise BackendUnavailableError(str(exc)) from exc + except BaseException: + _rollback(db) + raise + + if inserted is not None: + return AcceptResult(duplicate=False, reference=reference) + if existing is None: + # The row vanished between the insert and the read inside one + # transaction, which should be impossible. Retryable rather than + # guessed at. + raise BackendUnavailableError("event disappeared during accept") + if existing[0] != payload_hash: + raise EventConflictError( + f"event_id {event.event_id} already held with a different payload" + ) + return AcceptResult(duplicate=True, reference=reference) + + def health(self) -> None: + """Raise :class:`BackendUnavailableError` if the store is unusable.""" + try: + self.db.execute("SELECT 1 FROM events LIMIT 1").fetchone() + except sqlite3.Error as exc: + raise BackendUnavailableError(str(exc)) from exc + + def close(self) -> None: + """Close this thread's connection, if it has one.""" + conn = getattr(self._local, "conn", None) + if conn is not None: + conn.close() + self._local.conn = None + + +def _rollback(db: sqlite3.Connection) -> None: + try: + db.execute("ROLLBACK") + except sqlite3.Error: + pass + + +def _record_hash(event: AuditEvent) -> str: + import hashlib + + return hashlib.sha256( + json.dumps(event.as_record(), sort_keys=True).encode("utf-8") + ).hexdigest() diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py index 949c81f..5c2005f 100644 --- a/tests/test_ingestion.py +++ b/tests/test_ingestion.py @@ -1,36 +1,222 @@ import io import json -from audit_core.ingestion import IngestionApplication, SQLiteEventStore +import pytest + +from audit_core.ingestion import IngestionApplication +from audit_core.interface import BackendUnavailableError, RetentionPolicy +from audit_core.mock_file_backend import MockFileAuditBackend +from audit_core.sqlite_backend import SQLiteAuditBackend -def invoke(app, payload, *, token="opaque", key="evt-1"): - raw = json.dumps(payload).encode() +def invoke(app, payload, *, token="opaque", key="evt-1", method="POST", + path="/v1/events", body=None, length=None): + raw = body if body is not None else json.dumps(payload).encode() + environ = { + "PATH_INFO": path, + "REQUEST_METHOD": method, + "CONTENT_LENGTH": str(len(raw)) if length is None else length, + "wsgi.input": io.BytesIO(raw), + "HTTP_AUTHORIZATION": f"Bearer {token}", + } + if key is not None: + environ["HTTP_IDEMPOTENCY_KEY"] = key 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) + out = b"".join(app(environ, lambda status, headers: result.update(status=status))) + return result["status"], (json.loads(out) if out else {}) -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 event(**overrides): + base = { + "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"}, + } + base.update(overrides) + return base -def test_accepts_once_and_replays_idempotently(tmp_path): - app = IngestionApplication(SQLiteEventStore(str(tmp_path / "events.db")), "opaque") +@pytest.fixture +def app(tmp_path): + return IngestionApplication(SQLiteAuditBackend(str(tmp_path / "events.db")), "opaque") + + +# --- backend contract (T01) ------------------------------------------------- + +def test_refuses_a_non_durable_backend(): + """The development file backend must never become the production sink.""" + with pytest.raises(ValueError, match="not durable"): + IngestionApplication(MockFileAuditBackend(base_dir="/tmp/unused"), "opaque") + + +def test_readiness_reports_custody_class(app): + status, body = invoke(app, None, path="/readyz", method="GET", body=b"") + assert status.startswith("200") + assert body["durable"] is True + assert body["custody_class"] == "development" + + +def test_accepted_events_are_durable_across_reopen(tmp_path): + path = str(tmp_path / "events.db") + first = IngestionApplication(SQLiteAuditBackend(path), "opaque") + assert invoke(first, event())[0].startswith("202") + + reopened = IngestionApplication(SQLiteAuditBackend(path), "opaque") + status, body = invoke(reopened, event()) + assert status.startswith("200") and body["status"] == "duplicate" + + +# --- idempotency and conflict (T01, T02) ------------------------------------ + +def test_accepts_once_and_replays_idempotently(app): 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") +def test_same_id_different_payload_is_a_conflict(app): + assert invoke(app, event())[0].startswith("202") + status, body = invoke(app, event(subject="membership-2")) + assert status.startswith("409") + assert body["error"] == "event_id_conflict" + + +# --- authentication (T02) --------------------------------------------------- + +def test_rejects_wrong_credential(app): 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" + + +def test_non_ascii_credential_is_unauthorized_not_a_crash(app): + """compare_digest raises TypeError on non-ASCII str; that is a 401.""" + assert invoke(app, event(), token="wröng")[0].startswith("401") + + +# --- validation (T02, T07) -------------------------------------------------- + +@pytest.mark.parametrize("payload,expected", [ + (event(data={"password": "never"}), "secret_shaped_field"), + (event(source="somewhere-else"), "source_not_allowed"), + (event(occurred_at="2026-08-09T00:00:00"), "timestamp_missing_timezone"), + (event(occurred_at="not-a-date"), "invalid_timestamp"), + (event(tenant=""), "invalid_event"), +]) +def test_rejects_bad_events(app, payload, expected): + status, body = invoke(app, payload) + assert status.startswith("400") + assert body["error"] == expected + + +def test_rejects_mismatched_idempotency_key(app): + status, body = invoke(app, event(), key="other") + assert status.startswith("400") + assert body["error"] == "idempotency_key_mismatch" + + +def test_rejects_malformed_json(app): + assert invoke(app, None, body=b"{not json")[0].startswith("400") + + +def test_rejects_empty_body(app): + status, body = invoke(app, None, body=b"") + assert status.startswith("400") + assert body["error"] == "empty_body" + + +def test_rejects_oversized_body(app): + status, body = invoke(app, None, body=b"x", length=str(512 * 1024)) + assert status.startswith("400") + assert body["error"] == "payload_too_large" + + +def test_rejects_truncated_body(app): + status, body = invoke(app, None, body=b"{}", length="500") + assert status.startswith("400") + assert body["error"] == "truncated_body" + + +# --- routing (T07) ---------------------------------------------------------- + +@pytest.mark.parametrize("path,method", [ + ("/v1/events", "GET"), + ("/nope", "POST"), +]) +def test_unknown_routes_are_not_found(app, path, method): + assert invoke(app, event(), path=path, method=method)[0].startswith("404") + + +def test_healthz_needs_no_credential(app): + status, _ = invoke(app, None, path="/healthz", method="GET", body=b"", token="wrong") + assert status.startswith("200") + + +# --- failure handling (T02) ------------------------------------------------- + +class _BrokenBackend: + @property + def retention_policy(self): + return RetentionPolicy("archive", 3650, True, True, durable=True) + + def emit(self, event): + raise BackendUnavailableError("down") + + def accept(self, event, payload_hash): + raise BackendUnavailableError("down") + + +class _ExplodingBackend(_BrokenBackend): + def accept(self, event, payload_hash): + raise RuntimeError("unexpected") + + +def test_backend_unavailable_is_retryable_503(): + status, body = invoke(IngestionApplication(_BrokenBackend(), "opaque"), event()) + assert status.startswith("503") + assert body["error"] == "backend_unavailable" + + +def test_unexpected_backend_error_still_returns_a_response(): + """No request path may terminate without calling start_response.""" + status, body = invoke(IngestionApplication(_ExplodingBackend(), "opaque"), event()) + assert status.startswith("500") + assert body["error"] == "internal_error" + + +# --- concurrency (T01) ------------------------------------------------------ + +def test_concurrent_duplicates_produce_exactly_one_record(tmp_path): + """Racing submissions of one event: one acceptance, one custody record. + + This is the assertion the whole service rests on, so it is exercised rather + than assumed. An earlier single-connection implementation passed every + serial test while letting two callers both be told they were first. + """ + import threading + + backend = SQLiteAuditBackend(str(tmp_path / "race.db")) + app = IngestionApplication(backend, "opaque") + outcomes: list[str] = [] + lock = threading.Lock() + barrier = threading.Barrier(16) + + def submit(): + barrier.wait() + status, body = invoke(app, event()) + with lock: + outcomes.append(body.get("status", status)) + + threads = [threading.Thread(target=submit) for _ in range(16)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert outcomes.count("accepted") == 1, outcomes + assert outcomes.count("duplicate") == 15, outcomes + stored = backend.db.execute("SELECT COUNT(*) FROM events").fetchone()[0] + assert stored == 1 diff --git a/workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md b/workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md index c08026b..2967bb7 100644 --- a/workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md +++ b/workplans/AUDIT-WP-0004-receiver-correctness-and-hardening.md @@ -9,6 +9,7 @@ owner: codex topic_slug: netkingdom created: "2026-08-10" updated: "2026-08-10" +state_hub_workstream_id: "f3345e90-f466-4184-b149-9b0be92ffec8" --- # AUDIT-WP-0004 - receiver correctness and hardening @@ -40,8 +41,9 @@ WP-0005 against the interface this workplan fixes. ```task id: AUDIT-WP-0004-T01 -status: todo +status: done priority: high +state_hub_task_id: "9b194ebc-f853-48c5-a65b-88b0eae3d7a2" ``` `ingestion.py` writes records into SQLite directly and never calls @@ -64,12 +66,25 @@ Done when a successful response is backed by a durable backend acknowledgment, and the mock backend cannot be used to serve production traffic. +Done 2026-08-10: added `IdempotentAuditBackend` to the contract — idempotency +lives inside the backend so custody and duplicate-detection share one +transaction and cannot diverge. `SQLiteAuditBackend` implements it with WAL, +`synchronous=FULL`, and a busy timeout. Ingestion refuses any backend +declaring `durable=False`, so the mock file backend cannot serve. + +The atomicity claim was tested rather than asserted, and the first +implementation failed: with one shared connection, 16 racing submissions of +one event told **two** callers they were first. Fixed with per-thread +connections and `BEGIN IMMEDIATE` around the insert/read pair. Now covered by +`test_concurrent_duplicates_produce_exactly_one_record`. + ## T02 - Fix error semantics and failure handling ```task id: AUDIT-WP-0004-T02 -status: todo +status: done priority: high +state_hub_task_id: "b193acaf-9c0e-411a-921b-13282bee8325" ``` The handler catches `ValueError`, `TypeError`, `KeyError`, and @@ -93,12 +108,20 @@ contract with the retry semantics each implies. Done when no request path can terminate without a response, and every status code the receiver returns is deliberate and documented. +Done 2026-08-10: catch-all wrapper guarantees a response on every path. +Conflict is now 409, backend unavailability 503, unexpected faults 500. +Credential comparison moved inside the guarded path and a non-ASCII +`Authorization` header is a 401 rather than a crash. The full response +contract with its retry semantics is documented in the `ingestion` module +docstring. + ## T03 - Enforce the isolation properties already claimed ```task id: AUDIT-WP-0004-T03 status: todo priority: high +state_hub_task_id: "c29be4e7-4c2e-47f0-9d37-7d72061274ee" ``` WP-0003 T01 and T02 both record tenant isolation and cross-tenant claim @@ -121,6 +144,7 @@ another, and that refusal is covered by a test. id: AUDIT-WP-0004-T04 status: todo priority: medium +state_hub_task_id: "5cf5c412-965a-4a56-aeef-e965f2861c51" ``` `_contains_secret` rejects the entire event when any key name in `data` @@ -146,6 +170,7 @@ consistent between the contract and the implementation. id: AUDIT-WP-0004-T05 status: todo priority: high +state_hub_task_id: "006bc4ca-de36-4152-afae-0eef2a402e73" ``` There is no way to read anything back. The failure matrix requires dead-letter @@ -169,6 +194,7 @@ replay a specific event without creating a duplicate. id: AUDIT-WP-0004-T06 status: todo priority: high +state_hub_task_id: "348c2f4c-3ab0-46c7-9ddd-b198122f58ed" ``` `wsgiref.simple_server` is single-threaded with no request timeout, no @@ -193,8 +219,9 @@ outside. ```task id: AUDIT-WP-0004-T07 -status: todo +status: done priority: medium +state_hub_task_id: "b8141609-858d-41e1-9cc4-eb6f2723d561" ``` Ingestion has two tests. Uncovered: oversized and zero-length bodies, absent @@ -208,3 +235,10 @@ both and cover them. Done when each rejection reason and each failure mode above has a test that asserts the documented status code. + +Done 2026-08-10: ingestion tests went from 2 to 23 (suite 15 -> 36). Covers +oversized, empty and truncated bodies, absent/invalid `Content-Length`, +malformed JSON, wrong path and method, health endpoints, id conflict, +naive and unparseable timestamps, backend unavailability, unexpected backend +faults, durability across reopen, and the concurrency race. `accepted_at` is +UTC; naive timestamps are rejected rather than silently assumed. diff --git a/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md b/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md index c71fb36..73dd0b4 100644 --- a/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md +++ b/workplans/AUDIT-WP-0005-postgres-store-and-production-deployment.md @@ -13,6 +13,7 @@ depends_on: - AUDIT-WP-0004 - RAPP-POSTGRES-WP-0002 - NK-WP-0024 +state_hub_workstream_id: "7b24a844-c9f2-4d2d-ac7d-20978bdf6b38" --- # AUDIT-WP-0005 - Postgres store and production deployment @@ -50,6 +51,7 @@ requirement and consumes it; it does not implement it here. id: AUDIT-WP-0005-T01 status: todo priority: high +state_hub_task_id: "b1601d0b-922a-40f7-92c0-ea06af6c4468" ``` Implement `AuditBackend` against PostgreSQL, declaring an honest @@ -80,6 +82,7 @@ restart mid-write does not produce an acknowledged-but-absent event. id: AUDIT-WP-0005-T02 status: todo priority: high +state_hub_task_id: "831b2472-0d80-4369-a5e3-eb08ef3526b1" ``` Declare audit-core's database requirement against rapp-postgres and take @@ -101,6 +104,7 @@ events. id: AUDIT-WP-0005-T03 status: todo priority: high +state_hub_task_id: "598af2ac-e772-4a4e-9a65-dde9d4ca167f" ``` Publish an immutable image — base pinned by digest, not a mutable tag, and @@ -129,6 +133,7 @@ previous version. id: AUDIT-WP-0005-T04 status: todo priority: medium +state_hub_task_id: "9010fb4a-a1b8-4ef7-b143-e33ca7cc0619" ``` Any events accepted by the pre-production SQLite receiver are audit records @@ -148,6 +153,7 @@ and verified, or explicitly and justifiably discarded. id: AUDIT-WP-0005-T05 status: todo priority: high +state_hub_task_id: "1da30fec-b9f1-4be0-b42c-15797a8c4392" ``` Exercise the deployed path: successful delivery; receiver timeout and @@ -172,6 +178,7 @@ documented contract. id: AUDIT-WP-0005-T06 status: todo priority: medium +state_hub_task_id: "0856c80d-abe1-4bff-ba8d-87295cf76819" ``` Document what an operator needs: how to look up an event by correlation ID,