audit-core/audit_core/interface.py

"""Pluggable audit backend contract.

See ``docs/audit-backend-contract.md`` for the full protocol, event schema,
retention policy, and migration path from the mock file backend.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Literal, Protocol, runtime_checkable
from uuid import uuid4

SCHEMA_VERSION_V1ALPHA1 = "audit-core.event.v1alpha1"

CustodyClass = Literal["development", "operational", "archive", "hot_search"]

# Production Postgres reports ``operational``. Manifests written before
# AUDIT-WP-0006 required ``archive``. The two are aliases for one deploy so
# a mixed rollout cannot refuse to start. ``development`` is never an alias.
_PRODUCTION_CUSTODY_CLASSES = frozenset({"operational", "archive"})


def custody_class_satisfies(actual: str, required: str) -> bool:
    """Whether a backend's class meets a startup requirement.

    ``operational`` and ``archive`` satisfy each other. A development
    backend satisfies only ``development``.
    """
    if actual == required:
        return True
    return {actual, required} <= _PRODUCTION_CUSTODY_CLASSES

_REQUIRED_STRING_FIELDS = (
    "schema_version",
    "event_id",
    "observed_at",
    "tenant",
    "scope",
    "source",
    "action",
    "resource",
    "outcome",
)


def utc_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


@dataclass(frozen=True)
class RetentionPolicy:
    """Declarative retention guarantees exposed by an audit backend."""

    custody_class: CustodyClass
    retention_days: int | None
    immutable: bool
    tamper_evidence: bool
    durable: bool
    # Recoverable history is the platform backup window, not a deletion
    # policy. ``None`` means not declared (development backends).
    recoverable_days: int | None = None
    recoverable_source: str | None = None
    recoverable_basis: str | None = None

    def as_readiness(self) -> dict[str, Any]:
        """Sender-visible /readyz body. Keeps ``custody_class`` and adds recovery."""
        payload: dict[str, Any] = {
            "status": "ok",
            "custody_class": self.custody_class,
            "durable": self.durable,
            "tamper_evidence": self.tamper_evidence,
        }
        if self.recoverable_days is not None or self.recoverable_source:
            payload["recoverable_days"] = self.recoverable_days
            payload["recoverable_source"] = self.recoverable_source
            payload["recoverable_basis"] = self.recoverable_basis
        return payload


@dataclass(frozen=True)
class AuditEvent:
    """Normalized audit event for the v1alpha1 module-level contract.

    The flat record maps to JSON via :meth:`as_record`. The nested
    ``audit-core.event.v1`` envelope in the product requirements is the
    long-term HTTP ingestion target; adapters will translate between them.
    """

    source: str
    action: str
    resource: str
    outcome: str
    tenant: str = "platform"
    scope: str = "platform-control-plane"
    actor: str | None = None
    reason: str | None = None
    details: dict[str, Any] = field(default_factory=dict)
    event_id: str = field(default_factory=lambda: str(uuid4()))
    observed_at: str = field(default_factory=utc_now)
    schema_version: str = SCHEMA_VERSION_V1ALPHA1

    def as_record(self) -> dict[str, Any]:
        return {
            "schema_version": self.schema_version,
            "event_id": self.event_id,
            "observed_at": self.observed_at,
            "tenant": self.tenant,
            "scope": self.scope,
            "source": self.source,
            "actor": self.actor,
            "action": self.action,
            "resource": self.resource,
            "outcome": self.outcome,
            "reason": self.reason,
            "details": self.details,
        }


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.

    Raises :class:`EventValidationError` when required fields are missing,
    empty, or use an unsupported schema version.
    """
    record = event.as_record()
    errors: list[str] = []

    if record["schema_version"] != SCHEMA_VERSION_V1ALPHA1:
        errors.append(
            f"unsupported schema_version: {record['schema_version']!r} "
            f"(expected {SCHEMA_VERSION_V1ALPHA1!r})"
        )

    for name in _REQUIRED_STRING_FIELDS:
        value = record.get(name)
        if not isinstance(value, str) or not value.strip():
            errors.append(f"{name} must be a non-empty string")

    if not isinstance(record.get("details"), dict):
        errors.append("details must be a mapping")

    if errors:
        raise EventValidationError("; ".join(errors))


@runtime_checkable
class AuditBackend(Protocol):
    """Protocol implemented by replaceable audit sinks.

    Production backends provide durable operational or archive custody.
    Development backends (such as :class:`~audit_core.mock_file_backend.MockFileAuditBackend`)
    are for wiring only and must not be treated as audit custody.
    """

    @property
    def retention_policy(self) -> RetentionPolicy:
        """Describe retention and custody guarantees for readiness checks."""

    def emit(self, event: AuditEvent) -> str:
        """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.
        """