audit-core/audit_core/ingestion.py

"""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 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,
    BackendUnavailableError,
    EventConflictError,
    EventValidationError,
    IdempotentAuditBackend,
)
from audit_core.sqlite_backend import SQLiteAuditBackend

MAX_BODY_BYTES = 256 * 1024
_SECRET_FRAGMENTS = ("password", "secret", "token", "credential", "private_key")

log = logging.getLogger("audit_core.ingestion")


class IngestionApplication:
    """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")
        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 == "/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"})

        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)})

        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)))],
        )
        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")
    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=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()
        )
    if isinstance(value, list):
        return any(_contains_secret(item) for item in value)
    return False


def main() -> None:
    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,
    )
    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()