📋 Camada 5a — Governança: Logs e Permissões

Implementação: ImmutableAuditTrail

import hashlib
import asyncpg

class ImmutableAuditTrail:
    """Audit trail com hash chain para garantia de integridade."""

    def __init__(self, db_pool: asyncpg.Pool):
        self.db = db_pool
        self._last_hash: dict[str, str] = {}  # session_id → último hash

    def _compute_hash(self, event: dict, prev_hash: str) -> str:
        """Computa hash do evento encadeado com o anterior."""
        content = json.dumps({**event, "prev_hash": prev_hash},
                              sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(content.encode()).hexdigest()

    async def append(self,
                     session_id: str,
                     event_type: str,
                     actor: str,
                     action: str,
                     details: dict = None) -> str:
        """
        Adiciona entrada imutável ao audit trail.
        Retorna hash da entrada para referência.
        """
        event = {
            "session_id": session_id,
            "event_type": event_type,
            "actor": actor,
            "action": action,
            "details": details or {},
            "timestamp": datetime.now(timezone.utc).isoformat()
        }

        # Hash encadeado com o registro anterior
        prev_hash = self._last_hash.get(session_id, "GENESIS")
        event_hash = self._compute_hash(event, prev_hash)
        self._last_hash[session_id] = event_hash

        # Inserir no Postgres (NUNCA deletar esta tabela)
        await self.db.execute("""
            INSERT INTO audit_log
            (session_id, event_type, actor, action, details, event_hash, prev_hash)
            VALUES ($1, $2, $3, $4, $5, $6, $7)
        """, session_id, event_type, actor, action,
            json.dumps(details or {}), event_hash, prev_hash)

        return event_hash

    async def verify_integrity(self, session_id: str) -> bool:
        """Verifica se a cadeia de hashes está íntegra (não foi adulterada)."""
        rows = await self.db.fetch("""
            SELECT event_hash, prev_hash, event_type, actor, action, details, created_at
            FROM audit_log WHERE session_id = $1
            ORDER BY id ASC
        """, session_id)

        prev_hash = "GENESIS"
        for row in rows:
            event = {
                "session_id": session_id,
                "event_type": row["event_type"],
                "actor": row["actor"],
                "action": row["action"],
                "details": json.loads(row["details"]),
                "timestamp": row["created_at"].isoformat()
            }
            expected_hash = self._compute_hash(event, prev_hash)
            if expected_hash != row["event_hash"]:
                print(f"[Audit] VIOLACAO DE INTEGRIDADE detectada! Hash inválido.")
                return False
            prev_hash = row["event_hash"]

        return True

# Uso típico:
# audit = ImmutableAuditTrail(db_pool)
# await audit.append(
#     session_id="sess_123",
#     event_type="tool_execution",
#     actor="finance_agent_02",
#     action="read_database",
#     details={"table": "financial_reports", "query": "SELECT..."}
# )

from enum import Enum
from functools import wraps

class AgentRole(Enum):
    READ_ONLY  = "read_only"
    WRITER     = "writer"
    EXECUTOR   = "executor"
    PRIVILEGED = "privileged"

class RBACManager:
    """Gerencia roles e permissões de agentes."""

    # Permissões por role (hierárquica: cada role inclui o anterior)
    PERMISSIONS = {
        AgentRole.READ_ONLY: {
            "web_search", "read_database", "fetch_api_data",
            "read_file", "query_elasticsearch"
        },
        AgentRole.WRITER: {
            "web_search", "read_database", "fetch_api_data",
            "read_file", "query_elasticsearch",
            "write_file", "insert_database", "generate_document"
        },
        AgentRole.EXECUTOR: {
            "web_search", "read_database", "fetch_api_data",
            "read_file", "query_elasticsearch",
            "write_file", "insert_database", "generate_document",
            "execute_code", "call_external_api", "update_database"
        },
        AgentRole.PRIVILEGED: None  # None = tudo permitido
    }

    # Roles por agent_type padrão
    DEFAULT_ROLE_ASSIGNMENTS = {
        "research_agent":  AgentRole.READ_ONLY,
        "analysis_agent":  AgentRole.READ_ONLY,
        "writing_agent":   AgentRole.WRITER,
        "tech_agent":      AgentRole.EXECUTOR,
        "finance_agent":   AgentRole.READ_ONLY,
        "legal_agent":     AgentRole.READ_ONLY,
        "orchestrator":    AgentRole.PRIVILEGED,
    }

    def __init__(self, custom_assignments: dict = None):
        self.assignments = {
            **self.DEFAULT_ROLE_ASSIGNMENTS,
            **(custom_assignments or {})
        }

    def get_role(self, agent_id: str) -> AgentRole:
        """Obtém role de um agente pelo seu ID."""
        # Extrai tipo do agent_id (ex: "finance_agent_02" → "finance_agent")
        agent_type = "_".join(agent_id.split("_")[:-1]) if "_" in agent_id else agent_id
        return self.assignments.get(agent_type, AgentRole.READ_ONLY)

    def can(self, agent_id: str, tool_name: str) -> bool:
        """Verifica se agente tem permissão para usar a tool."""
        role = self.get_role(agent_id)
        if role == AgentRole.PRIVILEGED:
            return True
        allowed = self.PERMISSIONS.get(role, set())
        return tool_name in allowed

    def check_or_raise(self, agent_id: str, tool_name: str):
        """Verifica permissão ou lança PermissionError."""
        if not self.can(agent_id, tool_name):
            role = self.get_role(agent_id)
            raise PermissionError(
                f"Agente '{agent_id}' (role={role.value}) não tem permissão "
                f"para usar tool '{tool_name}'"
            )

Middleware de interceptação de tool

class ToolPermissionMiddleware:
    """Intercepta chamadas de tool e aplica controles de segurança."""

    HIGH_RISK_TOOLS = {
        "delete_database", "send_email", "deploy_service",
        "transfer_funds", "delete_file"
    }

    def __init__(self, rbac: RBACManager, audit: ImmutableAuditTrail,
                 escalation: HumanEscalationPipeline, logger: StructuredLogger):
        self.rbac = rbac
        self.audit = audit
        self.escalation = escalation
        self.logger = logger

    async def execute_tool(self, agent_id: str, tool_name: str,
                           tool_args: dict, session_id: str,
                           actual_tool_fn) -> dict:
        """
        Executa tool com pipeline completo de segurança:
        1. Verifica RBAC
        2. Para tools de alto risco: solicita aprovação humana
        3. Loga no audit trail
        4. Executa a tool
        5. Loga resultado
        """
        # 1. Verificação de RBAC
        try:
            self.rbac.check_or_raise(agent_id, tool_name)
        except PermissionError as e:
            await self.audit.append(session_id, "permission_denied",
                                     agent_id, str(e))
            raise

        # 2. Aprovação humana para tools de alto risco
        if tool_name in self.HIGH_RISK_TOOLS:
            approved = await self.escalation.escalate(
                session_id=session_id,
                trigger=f"Tool de alto risco: {tool_name}",
                context={"agent": agent_id, "tool": tool_name, "args": tool_args},
                default_on_timeout=False
            )
            if not approved:
                await self.audit.append(session_id, "tool_rejected_by_human",
                                         "human", f"Rejeitou {tool_name}")
                raise PermissionError(f"Humano rejeitou execução de {tool_name}")

        # 3. Log pré-execução no audit trail
        await self.audit.append(session_id, "tool_execution_start", agent_id,
                                 tool_name, {"args": tool_args})

        # 4. Executar tool
        start = time.time()
        try:
            result = await actual_tool_fn(**tool_args)
            duration_ms = (time.time() - start) * 1000
            await self.audit.append(session_id, "tool_execution_success",
                                     agent_id, tool_name,
                                     {"duration_ms": duration_ms})
            return result
        except Exception as e:
            await self.audit.append(session_id, "tool_execution_error",
                                     agent_id, tool_name, {"error": str(e)})
            raise

Configuração: envio de logs para Loki

# docker-compose.yml — Loki + Promtail + Grafana
services:
  loki:
    image: grafana/loki:2.9.4
    ports: ["3100:3100"]
    command: -config.file=/etc/loki/local-config.yaml

  promtail:
    image: grafana/promtail:2.9.4
    volumes:
      - /var/log:/var/log
      - ./promtail-config.yaml:/etc/promtail/config.yaml

  grafana:
    image: grafana/grafana:10.3.1
    ports: ["3000:3000"]
    environment:
      - GF_AUTH_ANONYMOUS_ENABLED=true

# Python: envio direto para Loki via API
import aiohttp

class LokiLogShipper:
    def __init__(self, loki_url: str = "http://localhost:3100"):
        self.url = f"{loki_url}/loki/api/v1/push"

    async def ship(self, labels: dict, log_entry: dict):
        """Envia log para Loki com labels para indexação."""
        payload = {
            "streams": [{
                "stream": {
                    "service": "orchestrator",
                    "session_id": labels.get("session_id", ""),
                    "agent_id": labels.get("agent_id", ""),
                    "level": log_entry.get("level", "INFO")
                },
                "values": [[
                    str(int(time.time() * 1e9)),  # nanoseconds
                    json.dumps(log_entry)
                ]]
            }]
        }
        async with aiohttp.ClientSession() as session:
            await session.post(self.url, json=payload)

# LogQL query para encontrar erros de um session_id:
# {service="orchestrator", session_id="sess_abc123"} |= "ERROR"
# {service="orchestrator", agent_id="finance_agent_02"} | json | cost_usd > 0.01