tegwick
5fd04e2095
Accept now extends a single-schema chain. Verify walks it; a rewritten payload_hash is a break. Tamper evidence is that detector plus an external chain-head attestation, not WORM.
216 lines
No EOL
7.3 KiB
Python
216 lines
No EOL
7.3 KiB
Python
"""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.
|
|
""" |