Tamper-Evident Audit Trails and Compliance Evidence

from agentmesh.governance.audit import AuditLog

audit = AuditLog()

entry = audit.log(
    event_type="tool_invocation",
    agent_did="did:web:sales-assistant.example.com",
    action="allow",
    resource="/crm/contacts",
    data={"tool": "crm_lookup", "query": "acme corp"},
    outcome="success",
    trace_id="trace-7f3a",
)

print(entry.entry_id)     # unique UUID-based ID
print(entry.entry_hash)   # SHA-256 hash of the entry
print(entry.timestamp)    # UTC datetime

        Root Hash
       /         \
    H(AB)       H(CD)
   /    \      /    \
  H(A)  H(B) H(C)  H(D)   ← leaf = SHA-256 of entry

# Full integrity check — recomputes all hashes
is_valid, error = audit.verify_integrity()
print(f"Chain valid: {is_valid}")

# Get the Merkle root (publish this for external auditors)
root = audit._chain.get_root_hash()
print(f"Merkle root: {root}")

# Prove a specific entry is in the log
proof = audit.get_proof(entry.entry_id)
print(proof["entry"])          # the AuditEntry
print(proof["merkle_root"])    # current Merkle root hash
print(proof["merkle_proof"])   # list of (hash, position) tuples
print(proof["verified"])       # True if the proof checks out

from agentmesh.governance.audit import MerkleAuditChain

verified = MerkleAuditChain.verify_proof(
    entry_hash="abc123...",
    proof=[("def456...", "left"), ("789aaa...", "right")],
    root_hash="expected-root...",
)
print(f"Entry in log: {verified}")

from hypervisor.audit.commitment_engine import CommitmentEngine

engine = CommitmentEngine()

# Commit the current chain state
record = engine.commit(session_id="session-001", delta_engine=delta_engine)
print(record.hash_chain_root)     # root hash of the session's delta chain
print(record.delta_count)         # number of deltas committed
print(record.committed_at)        # UTC timestamp

# Later: verify the session hasn't been tampered with
is_valid = engine.verify(session_id="session-001", delta_engine=delta_engine)
print(is_valid)  # True if root hashes match

# Queue for batch external anchoring
engine.queue_for_batch()
engine.flush_batch()    # publishes hashes to configured external store

from agent_os.event_sink import GovernanceEventSink

class GovernanceEventSink(Protocol):
    def emit(self, events: Sequence[GovernanceEvent]) -> SinkExportResult:
        """Export a batch of governance events. MUST NOT raise exceptions."""
        ...

    def shutdown(self, timeout_ms: int = 5000) -> bool:
        """Gracefully shut down. Returns True if completed within timeout."""
        ...

    def force_flush(self, timeout_ms: int = 30000) -> bool:
        """Force immediate flush. Returns True if completed within timeout."""
        ...

from agent_os.event_sink import GovernanceEventProcessor

processor = GovernanceEventProcessor(
    max_queue_size=1024,         # AGT_GSP_MAX_QUEUE_SIZE
    schedule_delay_ms=2000,      # AGT_GSP_SCHEDULE_DELAY_MS
    max_batch_size=100,          # AGT_GSP_MAX_BATCH_SIZE
    export_timeout_ms=10000,     # AGT_GSP_EXPORT_TIMEOUT_MS
)

processor.start()
processor.emit(event)
processor.force_flush()
processor.shutdown(timeout_ms=5000)

import os
from agentmesh.governance.audit import AuditEntry, AuditLog

class PostgresSink:
    """Push audit entries to a PostgreSQL table.

    Implements the AuditSink protocol from agentmesh.governance.audit_backends.
    """

    def __init__(self, dsn: str):
        import psycopg2
        self._conn = psycopg2.connect(dsn)

    def write(self, entry: AuditEntry) -> None:
        with self._conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO audit_log
                    (entry_id, timestamp, event_type, agent_did,
                     action, resource, outcome, entry_hash, trace_id)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
                """,
                (
                    entry.entry_id,
                    entry.timestamp.isoformat(),
                    entry.event_type,
                    entry.agent_did,
                    entry.action,
                    entry.resource,
                    entry.outcome,
                    entry.entry_hash,
                    entry.trace_id,
                ),
            )
            self._conn.commit()

    def write_batch(self, entries: list[AuditEntry]) -> None:
        for entry in entries:
            self.write(entry)

    def verify_integrity(self) -> tuple[bool, str | None]:
        return True, None  # delegate to AuditLog.verify_integrity()

    def close(self) -> None:
        self._conn.close()

# Use it
sink = PostgresSink(dsn=os.environ["DATABASE_URL"])
audit = AuditLog(sink=sink)

from datetime import datetime, timezone, timedelta

now = datetime.now(timezone.utc)
one_hour_ago = now - timedelta(hours=1)

data = audit.export(start_time=one_hour_ago, end_time=now)
print(data["entries"])    # list of entry dicts
print(data["metadata"])   # chain metadata including Merkle root

events = audit.export_cloudevents(start_time=one_hour_ago)
for ce in events:
    print(ce["type"])    # e.g., "ai.agentmesh.tool.invoked"
    print(ce["source"])  # agent DID
    print(ce["data"])    # serialized entry content

# Human-readable summary
agt verify

# Machine-readable JSON for CI/CD
agt verify --json > asi_report.json

# Shields.io badge for your README
agt verify --badge

OWASP ASI 2026 Governance Verification
=======================================
✅ ASI-01  Prompt Injection         PolicyInterceptor
✅ ASI-02  Insecure Tool Use        ToolAliasRegistry
✅ ASI-03  Excessive Agency          GovernancePolicy
✅ ASI-04  Unauthorized Escalation   EscalationPolicy
✅ ASI-05  Trust Boundary Violation  CardRegistry
✅ ASI-06  Insufficient Logging      AuditChain
✅ ASI-07  Insecure Identity         AgentIdentity
✅ ASI-08  Policy Bypass             PolicyConflictResolver
✅ ASI-09  Supply Chain Integrity    IntegrityVerifier
✅ ASI-10  Behavioral Anomaly        ComplianceEngine

Coverage: 10/10 (100%)

from agent_compliance.verify import GovernanceVerifier
from agent_compliance.integrity import IntegrityVerifier

# OWASP ASI coverage
verifier = GovernanceVerifier()
attestation = verifier.verify()

print(f"Passed:   {attestation.passed}")
print(f"Coverage: {attestation.coverage_pct()}%")
print(f"Hash:     {attestation.attestation_hash}")
print(attestation.summary())

# Supply chain integrity — SHA-256 file hashes + critical function bytecode hashes
integrity = IntegrityVerifier(manifest_path="integrity.json")
report = integrity.verify()
print(report.summary())

import json
from datetime import datetime, timezone, timedelta
from pathlib import Path
from agentmesh.governance.audit import AuditLog
from agent_compliance.verify import GovernanceVerifier

def generate_compliance_report(audit: AuditLog, days: int = 30) -> dict:
    now = datetime.now(timezone.utc)
    start = now - timedelta(days=days)

    export = audit.export(start_time=start, end_time=now)
    entries = export["entries"]

    chain_valid, chain_error = audit.verify_integrity()
    attestation = GovernanceVerifier().verify()

    report = {
        "report_generated": now.isoformat(),
        "period_start": start.isoformat(),
        "period_end": now.isoformat(),
        "audit_trail": {
            "total_entries": len(entries),
            "chain_integrity_valid": chain_valid,
            "chain_integrity_error": chain_error,
            "merkle_root": audit._chain.get_root_hash(),
        },
        "owasp_asi_2026": {
            "passed": attestation.passed,
            "controls_passed": attestation.controls_passed,
            "controls_total": attestation.controls_total,
            "coverage_pct": attestation.coverage_pct(),
            "attestation_hash": attestation.attestation_hash,
        },
    }

    Path("compliance_report.json").write_text(
        json.dumps(report, indent=2, default=str), encoding="utf-8"
    )
    return report

# .github/workflows/compliance.yml
name: Governance Compliance

on:
  push:
    branches: [main]
  pull_request:

jobs:
  compliance:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install governance packages
        run: pip install agentmesh-platform agent-governance-toolkit

      - name: Generate integrity manifest
        run: agt integrity --generate integrity.json

      - name: Verify OWASP ASI 2026 coverage
        run: agt verify --json > asi_report.json

      - name: Verify supply-chain integrity
        run: agt integrity --manifest integrity.json --json > integrity_report.json

      - name: Upload compliance artifacts
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: compliance-reports
          path: |
            asi_report.json
            integrity_report.json
            integrity.json

from datetime import datetime, timezone, timedelta

now = datetime.now(timezone.utc)
yesterday = now - timedelta(days=1)

# Find all denied tool calls in the last 24 hours
denied = audit.query(
    event_type="tool_blocked",
    outcome="denied",
    start_time=yesterday,
    limit=200,
)
print(f"Blocked {len(denied)} tool calls in the last 24h")

# Investigate a specific agent
agent = "did:web:support-bot.example.com"
violations = audit.query(agent_did=agent, event_type="policy_violation")
alerts = audit.query(agent_did=agent, event_type="rogue_detection")

# Single-entry lookup
entry = audit.get_entry(entry_id="some-uuid-here")

# Convenience shortcuts
agent_entries = audit.get_entries_for_agent(agent, limit=200)
all_violations = audit.get_entries_by_type("policy_violation", limit=50)