qonto-assistant/src/qonto_assistant/audit.py

from __future__ import annotations

import json
import logging
from collections.abc import Callable, Mapping
from copy import deepcopy
from dataclasses import asdict, is_dataclass
from datetime import UTC, datetime
from threading import RLock
from typing import Any
from uuid import uuid4

from qonto_assistant.contracts import AuditEvent

REDACTED = "[redacted]"
SECRET_KEYS = {
    "api_key",
    "authorization",
    "authorization_header",
    "openbao_token",
    "secret",
    "token",
}
AUDIT_STREAM_ID = "qonto-assistant.audit"
AUDIT_EVENT_CLASSES = ("audit.allow", "audit.deny")


def utc_now_iso() -> str:
    return datetime.now(tz=UTC).replace(microsecond=0).isoformat()


def _sanitize(value: Any) -> Any:
    if is_dataclass(value):
        return _sanitize(asdict(value))
    if isinstance(value, Mapping):
        return {
            key: (REDACTED if key.lower() in SECRET_KEYS else _sanitize(item))
            for key, item in value.items()
        }
    if isinstance(value, list):
        return [_sanitize(item) for item in value]
    if isinstance(value, tuple):
        return [_sanitize(item) for item in value]
    return value


class AuditLogger:
    """Emit sequenced audit records and source-side reconciliation evidence.

    Counters represent source transitions, not observer delivery receipts. An
    observer compares these counters with received ``audit.allow`` and
    ``audit.deny`` records and also checks ``stream_sequence`` for gaps.
    """

    def __init__(
        self,
        *,
        logger: logging.Logger | None = None,
        sink: Callable[[dict[str, Any]], None] | None = None,
        instance_id: str | None = None,
        clock: Callable[[], str] = utc_now_iso,
    ) -> None:
        self.logger = logger or logging.getLogger("qonto_assistant.audit")
        self.logger.setLevel(logging.INFO)
        self.sink = sink
        self.instance_id = instance_id or str(uuid4())
        self.clock = clock
        self.started_at = clock()
        self._lock = RLock()
        self._sequence = 0
        self._transition_counts = {event_class: 0 for event_class in AUDIT_EVENT_CLASSES}
        self._heartbeat_counts = dict(self._transition_counts)
        self._heartbeat_at = self.started_at

    def emit(self, event: AuditEvent | Mapping[str, Any]) -> dict[str, Any]:
        payload = _sanitize(asdict(event) if is_dataclass(event) else dict(event))
        decision = payload.get("decision")
        event_class = f"audit.{decision}" if decision in {"allow", "deny"} else "audit.event"

        with self._lock:
            self._sequence += 1
            if event_class in self._transition_counts:
                self._transition_counts[event_class] += 1
            payload.update(self._stream_fields(event_class=event_class, sequence=self._sequence))
            self._publish(payload)

        return deepcopy(payload)

    def emit_heartbeat(self, *, reason: str = "periodic") -> dict[str, Any]:
        """Emit a positive liveness claim plus counts for observer reconciliation."""
        with self._lock:
            now = self.clock()
            self._sequence += 1
            window_counts = {
                event_class: self._transition_counts[event_class]
                - self._heartbeat_counts[event_class]
                for event_class in AUDIT_EVENT_CLASSES
            }
            payload: dict[str, Any] = {
                **self._stream_fields(event_class="audit.heartbeat", sequence=self._sequence),
                "timestamp": now,
                "reason": reason,
                "assertion": (
                    "nothing-to-report"
                    if window_counts["audit.deny"] == 0
                    else "transitions-reported"
                ),
                "window_started_at": self._heartbeat_at,
                "window_ended_at": now,
                "source_transition_counts": dict(self._transition_counts),
                "window_transition_counts": window_counts,
            }
            self._heartbeat_counts = dict(self._transition_counts)
            self._heartbeat_at = now
            self._publish(payload)

        return deepcopy(payload)

    def reconciliation_snapshot(self) -> dict[str, Any]:
        """Return non-secret source state without adding an audit-stream record."""
        with self._lock:
            now = self.clock()
            return {
                "stream_id": AUDIT_STREAM_ID,
                "stream_instance_id": self.instance_id,
                "instance_started_at": self.started_at,
                "snapshot_at": now,
                "last_stream_sequence": self._sequence,
                "last_heartbeat_at": self._heartbeat_at,
                "source_transition_counts": dict(self._transition_counts),
                "since_last_heartbeat_counts": {
                    event_class: self._transition_counts[event_class]
                    - self._heartbeat_counts[event_class]
                    for event_class in AUDIT_EVENT_CLASSES
                },
            }

    def _stream_fields(self, *, event_class: str, sequence: int) -> dict[str, Any]:
        return {
            "event_class": event_class,
            "stream_id": AUDIT_STREAM_ID,
            "stream_instance_id": self.instance_id,
            "stream_sequence": sequence,
        }

    def _publish(self, payload: dict[str, Any]) -> None:
        # Structured logging is the production hot-path sink. Publish it first
        # so an optional secondary sink cannot suppress the primary record.
        self.logger.info(json.dumps(payload, sort_keys=True))
        if self.sink is not None:
            self.sink(deepcopy(payload))