Comunicación insegura entre agentes

Descripción

Los sistemas multiagente intercambian mensajes, delegan tareas y comparten contexto sin autenticación, cifrado o verificación de integridad, lo que habilita la suplantación, la escucha no autorizada y la delegación no autorizada.

Mitigaciones

TLS mutuo (mTLS)

Autenticar y cifrar toda la comunicación entre agentes:

• Emitir certificados X.509 únicos a cada instancia de agente desde una CA controlada por la organización
• Exigir verificación de TLS mutuo: tanto el agente emisor como el receptor deben presentar certificados válidos
• Aplicar el fijado de certificados (certificate pinning) para prevenir ataques de intermediario incluso si una CA se ve comprometida
• Rotar los certificados según un calendario regular y revocar los certificados de agentes dados de baja

Firma de mensajes

Garantizar la integridad y el no repudio de los mensajes:

• Firmar cada mensaje entre agentes usando la clave privada del agente emisor
• Incluir un timestamp, un nonce y un número de secuencia de mensaje en el payload firmado para prevenir ataques de repetición
• Los agentes receptores deben verificar la firma y rechazar mensajes con timestamps expirados o nonces reutilizados
• Mantener un registro de auditoría de mensajes con firmas para análisis forense posterior a incidentes

Validación de esquema

Aplicar una estructura de mensaje estricta:

• Definir esquemas JSON Schema o Protocol Buffer para cada tipo de mensaje intercambiado entre agentes
• Rechazar mensajes que no cumplan con el esquema esperado — sin tolerancia para campos inesperados
• Versionar los esquemas de mensajes y exigir que tanto el emisor como el receptor acuerden la versión
• Tratar las violaciones de esquema como eventos de seguridad y alertar ante violaciones repetidas de un agente específico

Controles de delegación

Gobernar la delegación de tareas entre agentes:

• Implementar una política de delegación explícita que defina qué agentes pueden delegar a qué otros agentes
• Las solicitudes de delegación deben incluir la identidad del delegador, el alcance de la tarea y un token de autorización
• Los agentes receptores deben verificar que están autorizados a aceptar delegaciones del agente solicitante
• Rastrear las cadenas de delegación para prevenir la delegación circular y detectar patrones de delegación anómalos

Aislamiento de canales

Separar los canales de comunicación por sensibilidad:

• Usar canales dedicados y cifrados para diferentes categorías de mensajes (plano de control frente a plano de datos)
• Aplicar segmentación de red para que los agentes solo puedan alcanzar los agentes específicos con los que necesitan comunicarse
• Implementar controles de acceso a nivel de mensaje: los agentes solo pueden leer mensajes dirigidos a ellos

Ejemplos de código

Sobre de mensaje firmado

import json
import time
import uuid
import hmac
import hashlib
from dataclasses import dataclass, asdict
from typing import Any, Optional


@dataclass
class MensajeAgente:
    id_emisor: str
    id_receptor: str
    tipo_mensaje: str
    payload: dict[str, Any]
    timestamp: float
    nonce: str
    secuencia: int
    firma: str = ""


class CanalMensajeSeguro:
    """Mensajería autenticada y con integridad verificada entre agentes."""

    VENTANA_REPETICION_SEGUNDOS = 300  # ventana de repetición de 5 minutos

    def __init__(self, id_agente: str, clave_firma: bytes):
        self._id_agente = id_agente
        self._clave_firma = clave_firma
        self._secuencia = 0
        self._nonces_vistos: dict[str, float] = {}  # nonce -> timestamp
        self._emisores_permitidos: set[str] = set()
        self._esquemas_mensaje: dict[str, dict] = {}

    def permitir_emisor(self, id_emisor: str):
        self._emisores_permitidos.add(id_emisor)

    def registrar_esquema(self, tipo_mensaje: str, esquema: dict):
        self._esquemas_mensaje[tipo_mensaje] = esquema

    def crear_mensaje(
        self, id_receptor: str, tipo_mensaje: str, payload: dict
    ) -> MensajeAgente:
        """Crea un mensaje firmado para su transmisión."""
        self._secuencia += 1
        msj = MensajeAgente(
            id_emisor=self._id_agente,
            id_receptor=id_receptor,
            tipo_mensaje=tipo_mensaje,
            payload=payload,
            timestamp=time.time(),
            nonce=uuid.uuid4().hex,
            secuencia=self._secuencia,
        )
        msj.firma = self._firmar(msj)
        return msj

    def verificar_mensaje(self, msj: MensajeAgente, clave_emisor: bytes) -> dict:
        """Verifica autenticidad, integridad, vigencia y cumplimiento de esquema."""
        verificaciones = {
            "emisor_permitido": False,
            "firma_valida": False,
            "no_repetido": False,
            "no_expirado": False,
            "esquema_valido": False,
            "dirigido_a_nosotros": False,
        }

        # Verificar receptor
        verificaciones["dirigido_a_nosotros"] = msj.id_receptor == self._id_agente
        if not verificaciones["dirigido_a_nosotros"]:
            return {"valido": False, "verificaciones": verificaciones, "razon": "El mensaje no está dirigido a este agente"}

        # Verificar que el emisor esté permitido
        verificaciones["emisor_permitido"] = msj.id_emisor in self._emisores_permitidos
        if not verificaciones["emisor_permitido"]:
            return {"valido": False, "verificaciones": verificaciones, "razon": f"El emisor '{msj.id_emisor}' no está en la lista permitida"}

        # Verificar la firma
        firma_esperada = self._calcular_firma(msj, clave_emisor)
        verificaciones["firma_valida"] = hmac.compare_digest(msj.firma, firma_esperada)
        if not verificaciones["firma_valida"]:
            return {"valido": False, "verificaciones": verificaciones, "razon": "Falló la verificación de la firma"}

        # Verificar repetición (unicidad del nonce)
        if msj.nonce in self._nonces_vistos:
            return {"valido": False, "verificaciones": verificaciones, "razon": "Se detectó un nonce repetido"}
        verificaciones["no_repetido"] = True

        # Verificar vigencia
        antiguedad = time.time() - msj.timestamp
        verificaciones["no_expirado"] = antiguedad <= self.VENTANA_REPETICION_SEGUNDOS
        if not verificaciones["no_expirado"]:
            return {"valido": False, "verificaciones": verificaciones, "razon": f"Mensaje expirado (antigüedad de {antiguedad:.0f}s)"}

        # Validación de esquema
        esquema = self._esquemas_mensaje.get(msj.tipo_mensaje)
        if esquema:
            verificaciones["esquema_valido"] = self._validar_esquema(msj.payload, esquema)
        else:
            verificaciones["esquema_valido"] = False
            return {"valido": False, "verificaciones": verificaciones, "razon": f"No hay esquema registrado para '{msj.tipo_mensaje}'"}

        # Registrar el nonce para prevenir repeticiones
        self._nonces_vistos[msj.nonce] = msj.timestamp
        self._limpiar_nonces_antiguos()

        return {"valido": True, "verificaciones": verificaciones}

    def _firmar(self, msj: MensajeAgente) -> str:
        return self._calcular_firma(msj, self._clave_firma)

    def _calcular_firma(self, msj: MensajeAgente, clave: bytes) -> str:
        firmable = f"{msj.id_emisor}|{msj.id_receptor}|{msj.tipo_mensaje}|{json.dumps(msj.payload, sort_keys=True)}|{msj.timestamp}|{msj.nonce}|{msj.secuencia}"
        return hmac.new(clave, firmable.encode(), hashlib.sha256).hexdigest()

    def _validar_esquema(self, payload: dict, esquema: dict) -> bool:
        # Simplificado: verifica que existan los campos requeridos y que no haya campos inesperados
        requeridos = set(esquema.get("required", []))
        permitidos = set(esquema.get("properties", {}).keys())
        presentes = set(payload.keys())
        return requeridos.issubset(presentes) and presentes.issubset(permitidos)

    def _limpiar_nonces_antiguos(self):
        corte = time.time() - self.VENTANA_REPETICION_SEGUNDOS * 2
        self._nonces_vistos = {
            n: t for n, t in self._nonces_vistos.items() if t > corte
        }

Requisitos de evidencia

• Inventario de certificados mTLS con registros de emisión, rotación y revocación
• Registros de verificación de firma de mensajes que muestran los mensajes aceptados y rechazados
• Registros de detección de repetición con colisiones de nonces capturadas
• Alertas de violación de esquema con atribución al emisor
• Trazas de auditoría de cadenas de delegación que mapean el flujo de tareas entre agentes