generated from coulomb/repo-seed
130 lines
3.8 KiB
Python
130 lines
3.8 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", "archive", "hot_search"]
|
|
|
|
_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
|
|
|
|
|
|
@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."""
|
|
|
|
|
|
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 archive or hot-search 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.""" |