Production Deployment Guide

import uuid
import asyncpg
from attesta import AuditLogger, ActionContext, ApprovalResult
from attesta.core.audit import build_entry

class PostgresAuditLogger:
    """Persists audit entries to PostgreSQL."""

    def __init__(self, dsn: str):
        self.dsn = dsn
        self._pool = None

    async def _get_pool(self):
        if self._pool is None:
            self._pool = await asyncpg.create_pool(self.dsn, min_size=2, max_size=10)
        return self._pool

    async def log(self, ctx: ActionContext, result: ApprovalResult) -> str:
        entry = build_entry(ctx, result)
        pool = await self._get_pool()

        async with pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO attesta_audit (
                    entry_id, chain_hash, action_name, agent_id,
                    risk_score, risk_level, challenge_type,
                    challenge_passed, verdict, review_duration_seconds,
                    environment, intercepted_at, decided_at, metadata
                ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
                """,
                entry.entry_id, entry.chain_hash,
                entry.action_name, entry.agent_id,
                entry.risk_score, entry.risk_level,
                entry.challenge_type, entry.challenge_passed,
                entry.verdict, entry.review_duration_seconds,
                entry.environment, entry.intercepted_at,
                entry.decided_at, entry.metadata,
            )
        return entry.entry_id

CREATE TABLE attesta_audit (
    entry_id            TEXT PRIMARY KEY,
    chain_hash          TEXT NOT NULL,
    action_name         TEXT NOT NULL,
    agent_id            TEXT DEFAULT '',
    risk_score          REAL NOT NULL,
    risk_level          TEXT NOT NULL,
    challenge_type      TEXT DEFAULT '',
    challenge_passed    BOOLEAN,
    verdict             TEXT NOT NULL,
    review_duration_seconds REAL DEFAULT 0,
    environment         TEXT DEFAULT '',
    intercepted_at      TIMESTAMPTZ,
    decided_at          TIMESTAMPTZ,
    metadata            JSONB DEFAULT '{}',
    created_at          TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_audit_agent ON attesta_audit(agent_id);
CREATE INDEX idx_audit_verdict ON attesta_audit(verdict);
CREATE INDEX idx_audit_risk_level ON attesta_audit(risk_level);
CREATE INDEX idx_audit_created ON attesta_audit(created_at);

import json
import uuid
import aioboto3
from attesta import AuditLogger, ActionContext, ApprovalResult
from attesta.core.audit import build_entry

class S3AuditLogger:
    """Writes each audit entry as an individual S3 object."""

    def __init__(self, bucket: str, prefix: str = "attesta/audit/"):
        self.bucket = bucket
        self.prefix = prefix

    async def log(self, ctx: ActionContext, result: ApprovalResult) -> str:
        entry = build_entry(ctx, result)
        key = (
            f"{self.prefix}"
            f"{entry.intercepted_at[:10]}/"  # partition by date
            f"{entry.entry_id}.json"
        )

        session = aioboto3.Session()
        async with session.client("s3") as s3:
            await s3.put_object(
                Bucket=self.bucket,
                Key=key,
                Body=entry.to_json().encode("utf-8"),
                ContentType="application/json",
                ServerSideEncryption="aws:kms",
            )
        return entry.entry_id

# View trust profile for an agent
attesta trust show --agent deploy-bot

# List all agent trust profiles
attesta trust list

# Revoke trust after an incident
attesta trust revoke --agent compromised-bot

from attesta.core.trust import TrustEngine
from pathlib import Path

engine = TrustEngine(storage_path=Path(".attesta/trust.json"))

# Check trust
score = engine.compute_trust("deploy-bot", domain="my-infra")
print(f"Trust score: {score:.2f}")

# Record an incident (drops trust significantly)
engine.record_incident(
    agent_id="deploy-bot",
    action_name="unauthorized_access",
    severity="high",
)

# Emergency revocation
engine.revoke("compromised-bot")

# Use a buffered writer for high throughput
import asyncio
from attesta.core.audit import AuditLogger

logger = AuditLogger(path="/var/log/attesta/audit.jsonl")

# The built-in logger is already async and non-blocking
# For extreme throughput, batch writes with an async queue:
class BufferedAuditLogger:
    def __init__(self, inner: AuditLogger, batch_size: int = 100):
        self._inner = inner
        self._queue = asyncio.Queue(maxsize=batch_size * 2)
        self._task = None

    async def log(self, ctx, result) -> str:
        entry_id = await self._inner.log(ctx, result)
        return entry_id

from attesta.core.audit import AuditLogger

logger = AuditLogger(path="/var/log/attesta/audit.jsonl")

# Find approvals that were too fast for their risk level
stamps = logger.find_rubber_stamps(
    max_review_seconds=5.0,
    min_risk="high",
)

if stamps:
    print(f"Found {len(stamps)} potential rubber stamps:")
    for entry in stamps:
        print(
            f"  {entry.action_name} by {entry.agent_id} "
            f"({entry.review_duration_seconds:.1f}s, "
            f"risk={entry.risk_level})"
        )

from attesta.core.audit import AuditLogger

logger = AuditLogger(path="/var/log/attesta/audit.jsonl")
intact, total, broken = logger.verify_chain()

if not intact:
    # ALERT: Audit trail has been tampered with
    print(f"CRITICAL: {len(broken)} broken links in {total} entries")
    print(f"Broken at indices: {broken}")
else:
    print(f"Audit chain intact: {total} entries verified")

from prometheus_client import Counter, Histogram, Gauge
from attesta import AuditLogger, ActionContext, ApprovalResult

# Define metrics
approval_total = Counter(
    "attesta_approvals_total",
    "Total approval decisions",
    ["verdict", "risk_level", "environment"],
)
review_duration = Histogram(
    "attesta_review_duration_seconds",
    "Time spent on human review",
    ["risk_level"],
)
trust_score = Gauge(
    "attesta_trust_score",
    "Current trust score per agent",
    ["agent_id"],
)

class MetricsAuditLogger:
    """Wraps an audit logger and emits Prometheus metrics."""

    def __init__(self, inner):
        self._inner = inner

    async def log(self, ctx: ActionContext, result: ApprovalResult) -> str:
        # Record metrics
        approval_total.labels(
            verdict=result.verdict.value,
            risk_level=result.risk_assessment.level.value,
            environment=ctx.environment,
        ).inc()

        review_duration.labels(
            risk_level=result.risk_assessment.level.value,
        ).observe(result.review_time_seconds)

        # Delegate to the real logger
        return await self._inner.log(ctx, result)