"""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"]