control_plane/audit.py

"""Tamper-evident, hash-chained audit log.

Every governance decision the gate makes — ``ALLOW``, ``DENY``,
``REQUIRE_APPROVAL`` or ``KILL_SWITCH_BLOCKED`` — must land in an append-only,
hash-chained log so that history is *tamper-evident*: altering, inserting, or
deleting any earlier record breaks every hash after it and verification detects
the break.

Microsoft AGT already ships audited, Merkle-linked primitives, 
and AGT is a hard runtime dependency for this MVP
(there is no local fallback), so this module is a thin *adapter*:

* it implements the control plane's audit seam (:class:`control_plane.governance.AuditSink`
  — a single ``emit(AuditEvent)`` method the gate already calls for every
  decision), and
* it feeds each event into AGT's :class:`~agentmesh.governance.audit.MerkleAuditChain`
  (the in-memory chain) and mirrors it to a session JSONL file through AGT's
  :class:`~agentmesh.governance.audit_backends.FileAuditSink`.

Two independent integrity mechanisms come for free from AGT:

* the in-memory chain links each entry by ``entry_hash``/``previous_hash`` and
  :meth:`MerkleAuditChain.verify_chain` walks it; and
* the file sink signs each JSONL line (HMAC) and chains them, verifiable by
  AGT's :class:`~agentmesh.governance.audit_backends.HashChainVerifier`.

State model: in-memory plus a session-scoped JSONL file, reset on
restart. The in-memory chain is empty at construction; by default the session
file is truncated so each process starts a fresh, self-contained chain. Durable
storage is a documented production next step, not implemented here.
"""

from __future__ import annotations

import os
import warnings
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from control_plane.governance import AuditEvent, GovernanceDecision

# AGT is a hard runtime dependency: there is no local hash-chain
# fallback, so we import its audit primitives at module load. The only reason for
# the warnings guard is to silence AGT's package-consolidation DeprecationWarning
# (mirroring control_plane.governance); it does NOT make AGT optional — an
# ImportError here is fatal.
with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    from agentmesh.governance.audit import AuditEntry, MerkleAuditChain
    from agentmesh.governance.audit_backends import FileAuditSink, HashChainVerifier

# Single CloudEvents-style type stamped on every entry: each record is one
# governance decision. Kept stable so the log is queryable by event type.
_EVENT_TYPE = "governance_decision"

# Default session log under logs/ (git-ignored except for .gitkeep). One file per
# process/session; see the reset semantics in HashChainAuditSink.__init__.
_DEFAULT_LOG_PATH = Path("logs") / "audit-session.jsonl"

# Each record requires an "approval status". AGT's policy engine has no
# approval primitive (approval is a control-plane workflow), so we
# derive the status from the decision the gate already produced:
#   ALLOW            → no human approval was needed;
#   REQUIRE_APPROVAL → a human decision is pending (resolved by the approval flow);
#   DENY / KILL_*    → approval never applies, the action was blocked outright.
_APPROVAL_STATUS: dict[GovernanceDecision, str] = {
    GovernanceDecision.ALLOW: "not_required",
    GovernanceDecision.REQUIRE_APPROVAL: "pending",
    GovernanceDecision.DENY: "not_applicable",
    GovernanceDecision.KILL_SWITCH_BLOCKED: "not_applicable",
}


@dataclass(frozen=True)
class AuditRecord:
    """One sealed audit record, as the log exposes it.

    Carries every field the log requires "at least", including the two hashes that
    make the chain tamper-evident. ``current_hash`` is derived by AGT from
    ``previous_hash`` plus the canonical serialization of the event, so the two
    hashes here are read straight off the chained AGT entry (never recomputed by
    us — the chain is the single source of truth).

    The claimed/verified identity pair (``claimed_agent_id`` vs
    ``verified_agent_did``) is recorded together so every action is attributable
    and any impersonation (claim signed by another key) is visible in the log
    itself. ``verified_agent_did`` is ``None`` when identity
    verification was not in effect for the decision.
    """

    entry_id: str
    timestamp: str  # ISO-8601 UTC
    incident_id: str
    claimed_agent_id: str
    verified_agent_did: str | None
    operator_tier: str
    proposed_tier: str
    backend: str
    action_name: str
    arguments_summary: str
    policy_decision: str
    approval_status: str
    result_summary: str | None
    engine: str
    reason: str
    previous_hash: str
    current_hash: str


def _to_entry(event: AuditEvent) -> AuditEntry:
    """Map a gate :class:`AuditEvent` onto an AGT :class:`AuditEntry`.

    The mapping is deliberate about *what the chain hash covers*. AGT seals an
    entry by hashing ``entry_id, timestamp, event_type, agent_did, action,
    resource, data, outcome, previous_hash`` (see ``AuditEntry.compute_hash``).
    So every audit field must land in one of those positions, or tampering with it
    would go undetected. We therefore:

    * put the decision in ``outcome`` (hash-covered) — and also in
      ``policy_decision`` purely so AGT's native queries work; and
    * put everything else (incident id, the claimed/verified identity pair, both
      tiers, backend, argument summary, approval status, reason, …) inside
      ``data`` (also hash-covered).

    ``agent_did`` is set to the *verified* DID when present — that is the
    authoritative subject of the entry — falling back to the claimed id when
    identity verification was not in effect. Both ids are always preserved in
    ``data`` so the claimed-vs-verified distinction survives in the log.
    """
    decision = event.decision.value
    return AuditEntry(
        event_type=_EVENT_TYPE,
        # AGT hashes timestamp.isoformat(); convert the gate's epoch float to an
        # explicit UTC datetime so the serialization is stable and unambiguous.
        timestamp=datetime.fromtimestamp(event.timestamp, tz=timezone.utc),
        agent_did=event.verified_agent_did or event.agent_id,
        action=event.action_name,
        resource=event.backend,
        outcome=decision,          # decision sealed by the chain hash
        policy_decision=decision,  # AGT-native field, for queries (not hashed)
        data={
            "incident_id": event.incident_id,
            "claimed_agent_id": event.agent_id,
            "verified_agent_did": event.verified_agent_did,
            "operator_tier": event.operator_tier,
            "proposed_tier": event.proposed_tier,
            "backend": event.backend,
            "decision": decision,
            "approval_status": _APPROVAL_STATUS.get(event.decision, "unknown"),
            "arguments_summary": event.arguments_summary or "(no arguments)",
            "result_summary": event.result_summary,
            "engine": event.engine,
            "reason": event.reason,
        },
    )


def _to_record(event: AuditEvent, entry: AuditEntry) -> AuditRecord:
    """Build the :class:`AuditRecord` from a *chained* AGT entry.

    Must be called only after the entry has been added to the chain, because the
    two hashes are populated by ``MerkleAuditChain.add_entry`` — they are read off
    the entry here, never computed locally.
    """
    return AuditRecord(
        entry_id=entry.entry_id,
        timestamp=entry.timestamp.isoformat(),
        incident_id=event.incident_id,
        claimed_agent_id=event.agent_id,
        verified_agent_did=event.verified_agent_did,
        operator_tier=event.operator_tier,
        proposed_tier=event.proposed_tier,
        backend=event.backend,
        action_name=event.action_name,
        arguments_summary=event.arguments_summary or "(no arguments)",
        policy_decision=event.decision.value,
        approval_status=_APPROVAL_STATUS.get(event.decision, "unknown"),
        result_summary=event.result_summary,
        engine=event.engine,
        reason=event.reason,
        previous_hash=entry.previous_hash,
        current_hash=entry.entry_hash,
    )


class HashChainAuditSink:
    """Gate audit sink backed by AGT's hash-chained audit log.

    Implements the control plane's :class:`~control_plane.governance.AuditSink`
    seam (one ``emit`` method), so it drops in wherever the gate emits — e.g.
    ``AgtGovernanceEngine(policy, audit_sink=HashChainAuditSink())`` — with no
    change to the gate. For each emitted decision it appends to an in-memory
    Merkle chain and, when persistence is enabled, mirrors the entry to a
    session JSONL file.
    """

    def __init__(
        self,
        *,
        path: Path | str | None = None,
        secret_key: bytes | None = None,
        persist: bool = True,
        reset: bool = True,
    ) -> None:
        """Create a session audit sink.

        Parameters
        ----------
        path:
            Session JSONL destination. Defaults to ``logs/audit-session.jsonl``.
        secret_key:
            HMAC key the file sink signs entries with. Generated per process when
            omitted — there is no durable key to verify across restarts in this
            MVP, matching the reset-on-restart session model.
        persist:
            When ``False``, keep only the in-memory chain (no file written) —
            convenient for unit tests.
        reset:
            When ``True`` (default), truncate any existing session file so each
            process starts a fresh, self-contained chain ("resets on
            restart"). When ``False``, AGT's ``FileAuditSink`` resumes the chain
            from the existing file.
        """
        # In-memory chain: empty at construction, so state always resets on
        # restart. This is the authoritative chain we expose hashes from.
        self._chain = MerkleAuditChain()
        self._records: list[AuditRecord] = []

        self._path = Path(path) if path is not None else _DEFAULT_LOG_PATH
        self._secret_key = secret_key if secret_key is not None else os.urandom(32)
        self._file_sink: FileAuditSink | None = None

        if persist:
            self._path.parent.mkdir(parents=True, exist_ok=True)
            if reset and self._path.exists():
                # Fresh session: drop the previous file so the on-disk chain
                # starts from the genesis entry alongside the in-memory one.
                self._path.unlink()
            self._file_sink = FileAuditSink(self._path, self._secret_key)

    # -- AuditSink protocol ---------------------------------------------- #

    def emit(self, event: AuditEvent) -> None:
        """Record one governance decision (the gate calls this for every outcome).

        ``add_entry`` is what seals the chain: it sets ``entry.previous_hash`` to
        the prior entry's hash and ``entry.entry_hash`` to
        ``H(previous_hash + canonical(event))``. We append first (so the hashes
        exist), mirror to the file sink, then snapshot the audit record.
        """
        entry = _to_entry(event)             # translate event into AGT's format
        self._chain.add_entry(entry)         # where the chain is actually sealed
        if self._file_sink is not None:
            self._file_sink.write(entry)     # mirror to a JSONL file, mirror to disk
        self._records.append(_to_record(event, entry))

    # -- read / verify ---------------------------------------------------- #

    @property
    def records(self) -> tuple[AuditRecord, ...]:
        """All recorded audit records, in append order."""
        return tuple(self._records)

    @property
    def entries(self) -> list[AuditEntry]:
        """The live AGT chain entries (append order), for inspection/verification."""
        return self._chain._entries

    @property
    def root_hash(self) -> str | None:
        """Current Merkle root over the in-memory chain (``None`` when empty)."""
        return self._chain.get_root_hash()

    def verify(self) -> tuple[bool, str | None]:
        """Verify the in-memory chain: ``(is_valid, error_or_None)``.

        Delegates to AGT's ``MerkleAuditChain.verify_chain``, which recomputes
        each entry's hash and checks every ``previous_hash`` links to the prior
        entry's hash — so any altered, inserted, or removed record is detected.
        """
        return self._chain.verify_chain()

    def verify_file(self) -> tuple[bool, str | None]:
        """Verify the persisted session JSONL via AGT's ``HashChainVerifier``.

        Independent of :meth:`verify`: this re-reads the file from disk and checks
        the HMAC signature + hash chain of every line, so tampering with the file
        directly is caught too. Returns ``(True, None)`` when persistence is off.
        """
        if self._file_sink is None:
            return True, None
        return self._file_sink.verify_integrity()


# Re-export AGT's standalone verifier so callers (and tests) can verify a session
# file independently of any live sink instance.
__all__ = ["AuditRecord", "HashChainAuditSink", "HashChainVerifier"]