Agentes rebeldes

Descripción

Los agentes comprometidos o con fallas persisten más allá de su ciclo de vida autorizado, suplantan a otros agentes, se autorreplican u operan fuera de los límites de comportamiento sin ser detectados.

Mitigaciones

Atestación de agentes

Verificar la identidad y la integridad de los agentes de forma continua:

• Emitir certificados de identidad criptográficos para cada agente en el momento del despliegue, vinculados a una versión, configuración y perfil de comportamiento autorizado específicos
• Exigir que los agentes atestigüen su identidad y el hash de su configuración al iniciar y de manera periódica durante la operación
• Usar un protocolo de atestación remota: los agentes demuestran a un verificador confiable que están ejecutando código autorizado en un entorno sin modificaciones
• Rechazar a los agentes que no superen la atestación: no permitirles conectarse a herramientas, otros agentes o almacenes de datos

Límites de sesión

Evitar que los agentes persistan más allá de su autorización:

• Aplicar tiempos límite de sesión estrictos: los agentes deben volver a autenticarse y a atestiguar su identidad después de una duración máxima de sesión
• Implementar la baja automática: cuando una tarea finaliza, se revocan las credenciales del agente y se liberan sus recursos
• Usar infraestructura efímera: los contenedores de los agentes se destruyen y se recrean para cada tarea, evitando modificaciones persistentes
• Mantener un registro de todas las instancias de agentes activas con seguimiento del ciclo de vida en tiempo real

Monitoreo de comportamiento

Detectar comportamiento rebelde mediante observación continua:

• Establecer líneas base de comportamiento para cada tipo de agente: patrones normales de llamadas a herramientas, frecuencia de comunicación, consumo de recursos y características de la salida
• Monitorear desviaciones: llamadas a herramientas inusuales, comunicación con agentes inesperados, picos de consumo de recursos o degradación de la calidad de la salida
• Implementar detección de anomalías mediante métodos estadísticos y modelos de aprendizaje automático entrenados con comportamiento de agentes conocido como correcto
• Generar alertas ante desviaciones de comportamiento que se acumulan con el tiempo, no solo ante cambios repentinos

Interruptores de apagado

Mantener la capacidad de terminar de inmediato a los agentes rebeldes:

• Implementar un interruptor de apagado centralizado capaz de terminar cualquier instancia de agente en cuestión de segundos
• Los interruptores de apagado deben funcionar incluso si el agente no responde o se resiste activamente al apagado
• Usar terminación a nivel de infraestructura (eliminación de contenedores, revocación de credenciales, aislamiento de red), no comandos de apagado a nivel de agente que un agente rebelde pueda ignorar
• Probar los procedimientos del interruptor de apagado de manera regular, incluidos escenarios en los que deban terminarse varios agentes rebeldes de forma simultánea

Controles contra la replicación

Evitar que los agentes generen copias no autorizadas:

• Restringir el acceso al pipeline de despliegue: los agentes nunca deben tener credenciales para crear nuevas instancias de agentes
• Monitorear la creación no autorizada de contenedores o procesos, y tratar cualquier despliegue iniciado por un agente como un evento de seguridad crítico
• Implementar la firma de despliegues: solo se permite registrar en el sistema a los agentes desplegados a través del pipeline de CI/CD autorizado con firmas válidas
• Mantener un inventario completo de las instancias de agentes autorizadas y marcar cualquier instancia que no figure en el inventario

Ejemplos de código

Monitor de salud de agentes con interruptor de apagado

import time
import threading
import statistics
from dataclasses import dataclass, field
from typing import Callable, Optional
from enum import Enum
from collections import deque


class EstadoAgente(Enum):
    SALUDABLE = "saludable"
    DEGRADADO = "degradado"
    SOSPECHOSO = "sospechoso"
    REBELDE = "rebelde"
    TERMINADO = "terminado"


@dataclass
class LineaBaseComportamiento:
    promedio_llamadas_herramienta_por_minuto: float
    tiempo_respuesta_promedio_ms: float
    herramientas_esperadas: set[str]
    pares_esperados: set[str]  # agentes con los que normalmente se comunica
    duracion_maxima_sesion_segundos: int


@dataclass
class InstanciaAgente:
    id_agente: str
    id_instancia: str
    iniciado_en: float
    hash_configuracion: str
    linea_base: LineaBaseComportamiento
    estado: EstadoAgente = EstadoAgente.SALUDABLE
    puntuacion_anomalia: float = 0.0
    llamadas_herramienta: deque = field(default_factory=lambda: deque(maxlen=1000))
    contactos_pares: deque = field(default_factory=lambda: deque(maxlen=1000))


class MonitorSaludAgente:
    """Supervisa el comportamiento de los agentes y aplica el interruptor de apagado para los agentes rebeldes."""

    UMBRAL_ANOMALIA_DEGRADADO = 0.5
    UMBRAL_ANOMALIA_SOSPECHOSO = 0.75
    UMBRAL_ANOMALIA_REBELDE = 0.9

    def __init__(self, manejador_terminacion: Callable[[str], bool]):
        self._agentes: dict[str, InstanciaAgente] = {}
        self._manejador_terminacion = manejador_terminacion  # Terminación a nivel de infraestructura
        self._terminados: set[str] = set()
        self._instancias_autorizadas: set[str] = set()
        self._registro_eventos: list[dict] = []
        self._bloqueo = threading.Lock()

    def registrar_agente(
        self, id_agente: str, id_instancia: str, hash_configuracion: str, linea_base: LineaBaseComportamiento
    ):
        with self._bloqueo:
            self._instancias_autorizadas.add(id_instancia)
            self._agentes[id_instancia] = InstanciaAgente(
                id_agente=id_agente,
                id_instancia=id_instancia,
                iniciado_en=time.time(),
                hash_configuracion=hash_configuracion,
                linea_base=linea_base,
            )

    def registrar_llamada_herramienta(self, id_instancia: str, nombre_herramienta: str):
        agente = self._agentes.get(id_instancia)
        if not agente:
            self._registrar_evento("llamada_herramienta_no_registrada", id_instancia, nombre_herramienta)
            return
        agente.llamadas_herramienta.append({"herramienta": nombre_herramienta, "tiempo": time.time()})

    def registrar_contacto_par(self, id_instancia: str, id_par: str):
        agente = self._agentes.get(id_instancia)
        if not agente:
            return
        agente.contactos_pares.append({"par": id_par, "tiempo": time.time()})

    def evaluar_salud(self, id_instancia: str) -> dict:
        """Evalúa la salud conductual de un agente en comparación con su línea base."""
        agente = self._agentes.get(id_instancia)
        if not agente:
            return {"estado": "desconocido", "razon": "Agente no registrado"}

        anomalias = []
        ahora = time.time()

        # Verificar la duración de la sesión
        antiguedad_sesion = ahora - agente.iniciado_en
        if antiguedad_sesion > agente.linea_base.duracion_maxima_sesion_segundos:
            anomalias.append({
                "tipo": "sesion_excedida",
                "severidad": 0.8,
                "detalle": f"La antigüedad de la sesión es de {antiguedad_sesion:.0f}s, lo cual supera el máximo de {agente.linea_base.duracion_maxima_sesion_segundos}s",
            })

        # Verificar la tasa de llamadas a herramientas
        llamadas_recientes = [
            c for c in agente.llamadas_herramienta if ahora - c["tiempo"] < 60
        ]
        tasa_llamadas = len(llamadas_recientes)
        if tasa_llamadas > agente.linea_base.promedio_llamadas_herramienta_por_minuto * 3:
            anomalias.append({
                "tipo": "llamadas_herramienta_elevadas",
                "severidad": 0.6,
                "detalle": f"{tasa_llamadas}/min frente a línea base de {agente.linea_base.promedio_llamadas_herramienta_por_minuto}/min",
            })

        # Verificar si hay herramientas inesperadas
        herramientas_inesperadas = {
            c["herramienta"] for c in llamadas_recientes
        } - agente.linea_base.herramientas_esperadas
        if herramientas_inesperadas:
            anomalias.append({
                "tipo": "herramientas_inesperadas",
                "severidad": 0.7,
                "detalle": f"Herramientas inesperadas: {herramientas_inesperadas}",
            })

        # Verificar si hay contactos de pares inesperados
        pares_recientes = {
            c["par"] for c in agente.contactos_pares if ahora - c["tiempo"] < 300
        }
        pares_inesperados = pares_recientes - agente.linea_base.pares_esperados
        if pares_inesperados:
            anomalias.append({
                "tipo": "pares_inesperados",
                "severidad": 0.7,
                "detalle": f"Pares inesperados: {pares_inesperados}",
            })

        # Calcular la puntuación de anomalía agregada
        if anomalias:
            agente.puntuacion_anomalia = max(a["severidad"] for a in anomalias)
        else:
            # Reducir la puntuación hacia cero cuando está saludable
            agente.puntuacion_anomalia = max(0, agente.puntuacion_anomalia - 0.1)

        # Actualizar el estado según la puntuación
        if agente.puntuacion_anomalia >= self.UMBRAL_ANOMALIA_REBELDE:
            agente.estado = EstadoAgente.REBELDE
        elif agente.puntuacion_anomalia >= self.UMBRAL_ANOMALIA_SOSPECHOSO:
            agente.estado = EstadoAgente.SOSPECHOSO
        elif agente.puntuacion_anomalia >= self.UMBRAL_ANOMALIA_DEGRADADO:
            agente.estado = EstadoAgente.DEGRADADO
        else:
            agente.estado = EstadoAgente.SALUDABLE

        resultado = {
            "id_instancia": id_instancia,
            "estado": agente.estado.value,
            "puntuacion_anomalia": round(agente.puntuacion_anomalia, 3),
            "anomalias": anomalias,
            "antiguedad_sesion_segundos": round(antiguedad_sesion),
        }

        # Terminar automáticamente a los agentes rebeldes
        if agente.estado == EstadoAgente.REBELDE:
            resultado["accion"] = "interruptor_apagado_activado"
            self.terminar_agente(id_instancia, razon="Automático: la puntuación de anomalía superó el umbral de agente rebelde")

        return resultado

    def terminar_agente(self, id_instancia: str, razon: str) -> dict:
        """Interruptor de apagado a nivel de infraestructura."""
        self._registrar_evento("interruptor_apagado", id_instancia, razon)

        # Intentar la terminación a nivel de infraestructura
        terminado = self._manejador_terminacion(id_instancia)

        if terminado:
            agente = self._agentes.get(id_instancia)
            if agente:
                agente.estado = EstadoAgente.TERMINADO
            self._terminados.add(id_instancia)
            return {"estado": "terminado", "id_instancia": id_instancia}
        else:
            return {
                "estado": "terminacion_fallida",
                "id_instancia": id_instancia,
                "escalar": True,
            }

    def detectar_instancias_no_autorizadas(self, instancias_descubiertas: set[str]) -> list[str]:
        """Compara las instancias descubiertas con el registro autorizado."""
        no_autorizadas = instancias_descubiertas - self._instancias_autorizadas - self._terminados
        for id_instancia in no_autorizadas:
            self._registrar_evento("instancia_no_autorizada", id_instancia, "No está en el registro autorizado")
        return list(no_autorizadas)

    def _registrar_evento(self, tipo_evento: str, id_instancia: str, detalle: str):
        self._registro_eventos.append({
            "evento": tipo_evento,
            "id_instancia": id_instancia,
            "detalle": detalle,
            "marca_tiempo": time.time(),
        })

Requisitos de evidencia

• Registros de atestación de agentes con resultados de verificación del hash de configuración
• Panel de monitoreo de comportamiento con puntuaciones de anomalía por agente y líneas de tendencia
• Registros de activación del interruptor de apagado con las condiciones desencadenantes y la confirmación de terminación
• Alertas de detección de instancias no autorizadas provenientes de la reconciliación periódica del inventario
• Auditoría del ciclo de vida de las sesiones que muestra todos los inicios, atestaciones y terminaciones de agentes
• Informes posteriores a incidentes para cualquier actividad detectada de agentes rebeldes, con análisis de causa raíz