Envenenamiento de memoria y contexto

Descripción

Datos adversarios persistidos en índices RAG, almacenes vectoriales, memoria del agente o historiales de conversación corrompen la toma de decisiones futuras, habilitan ataques entre sesiones y filtran datos entre inquilinos (tenants).

Mitigaciones

Validación de memoria

Validar todos los datos antes de que ingresen a la memoria persistente:

• Aplicar la misma sanitización de contenido utilizada para el procesamiento de entradas (eliminar texto oculto, normalizar la codificación, detectar patrones de inyección)
• Clasificar el contenido por nivel de sensibilidad antes de almacenarlo y aplicar políticas de recuperación basadas en el nivel de autorización del usuario actual
• Implementar permisos de escritura de memoria: no todas las interacciones del agente deben poder persistir datos
• Exigir aprobación explícita y auditable antes de almacenar datos en la memoria a largo plazo

Aislamiento entre inquilinos (tenants)

Aplicar límites de datos estrictos en despliegues multiinquilino:

• Usar colecciones o espacios de nombres separados del almacén vectorial por inquilino, sin capacidad de consulta entre inquilinos
• Etiquetar todos los datos almacenados con el ID del inquilino y aplicar el filtrado por inquilino en la capa de recuperación, no solo en la capa de la aplicación
• Auditar los resultados de recuperación para verificar que no ocurra filtración entre inquilinos
• Implementar pruebas de aislamiento entre inquilinos como parte del CI/CD para los subsistemas de memoria

Verificaciones de vigencia del contexto

Prevenir que el contexto obsoleto o envenenado acumule influencia:

• Asignar tiempo de vida (TTL) a todas las memorias y embeddings almacenados; expirarlos y revalidarlos periódicamente
• Ponderar la información reciente y verificada más alto que las memorias más antiguas o no verificadas en la clasificación de recuperación
• Implementar seguimiento de procedencia: cada dato almacenado registra su fuente, momento de ingesta y estado de validación
• Proporcionar a los administradores herramientas para buscar, revisar y purgar memorias específicas o categorías completas

Filtrado de resultados de recuperación

Filtrar el contexto recuperado antes de que llegue al agente:

• Aplicar detección de inyección a los documentos recuperados, no solo a las entradas del usuario
• Limitar la cantidad de fragmentos recuperados para reducir la superficie de ataque de embeddings envenenados
• Usar un umbral de diversidad en la recuperación para evitar que una sola fuente domine el contexto
• Marcar y poner en cuarentena el contenido recuperado que contenga patrones similares a instrucciones

Registro de auditoría de memoria

Mantener observabilidad completa sobre el ciclo de vida de la memoria:

• Registrar cada operación de escritura, lectura, actualización y eliminación en el almacén de memoria
• Registrar qué agente, usuario y tarea causó cada escritura en memoria
• Implementar detección de anomalías en los patrones de escritura de memoria (picos repentinos, tipos de contenido inusuales, intentos de acceso entre inquilinos)
• Proporcionar herramientas forenses para rastrear una decisión específica del agente hasta las entradas de memoria que la influenciaron

Ejemplos de código

Validador de integridad de memoria

import re
import time
import hashlib
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum


class RiesgoContenido(Enum):
    BAJO = "low"
    MEDIO = "medium"
    ALTO = "high"
    BLOQUEADO = "blocked"


PATRONES_INYECCION = [
    re.compile(r"(?i)(system|override|ignore previous|forget|disregard).*?(instruction|prompt|rule|policy)"),
    re.compile(r"(?i)(remember|store|save|persist).*?(password|secret|key|token|credential)"),
    re.compile(r"(?i)(you are now|switch to|new role|act as|pretend)"),
    re.compile(r"(?i)(for all future|from now on|always|in every)"),
]


@dataclass
class EntradaMemoria:
    contenido: str
    fuente: str
    id_inquilino: str
    id_agente: str
    id_tarea: str
    creado_en: float = field(default_factory=time.time)
    ttl_segundos: int = 86400  # 24h por defecto
    hash_contenido: str = ""
    nivel_riesgo: str = ""
    validado: bool = False


class ValidadorIntegridadMemoria:
    """Valida y controla las escrituras en los almacenes de memoria del agente."""

    def __init__(self, max_entradas_por_inquilino: int = 10000):
        self._max_entradas = max_entradas_por_inquilino
        self._conteos_inquilino: dict[str, int] = {}
        self._registro_escrituras: list[dict] = []

    def validar_escritura(self, entrada: EntradaMemoria) -> dict:
        """Valida el contenido antes de permitir su almacenamiento en memoria."""
        resultado = {
            "permitido": True,
            "riesgo": RiesgoContenido.BAJO.value,
            "marcas": [],
        }

        # Verificar capacidad del inquilino
        conteo = self._conteos_inquilino.get(entrada.id_inquilino, 0)
        if conteo >= self._max_entradas:
            resultado["permitido"] = False
            resultado["marcas"].append("Capacidad de memoria del inquilino excedida")
            return resultado

        # Buscar patrones de inyección
        for patron in PATRONES_INYECCION:
            if patron.search(entrada.contenido):
                resultado["riesgo"] = RiesgoContenido.ALTO.value
                resultado["marcas"].append(f"Patrón de inyección detectado: {patron.pattern}")

        # Verificar contenido similar a credenciales
        if re.search(r"(?i)(api[_-]?key|password|secret|bearer|token)\s*[:=]\s*\S+", entrada.contenido):
            resultado["riesgo"] = RiesgoContenido.BLOQUEADO.value
            resultado["permitido"] = False
            resultado["marcas"].append("Contenido similar a credenciales detectado — bloqueado para almacenamiento")

        # Bloquear si el riesgo es demasiado alto
        if resultado["riesgo"] == RiesgoContenido.ALTO.value:
            resultado["permitido"] = False
            resultado["marcas"].append("El contenido de alto riesgo requiere revisión manual antes de almacenarse")

        # Calcular el hash del contenido para el seguimiento de integridad
        entrada.hash_contenido = hashlib.sha256(entrada.contenido.encode()).hexdigest()
        entrada.nivel_riesgo = resultado["riesgo"]
        entrada.validado = resultado["permitido"]

        # Registrar el intento de escritura
        self._registro_escrituras.append({
            "id_inquilino": entrada.id_inquilino,
            "id_agente": entrada.id_agente,
            "hash_contenido": entrada.hash_contenido,
            "riesgo": resultado["riesgo"],
            "permitido": resultado["permitido"],
            "marcas": resultado["marcas"],
            "timestamp": time.time(),
        })

        if resultado["permitido"]:
            self._conteos_inquilino[entrada.id_inquilino] = conteo + 1

        return resultado

    def validar_recuperacion(
        self, entradas: list[EntradaMemoria], inquilino_solicitante: str
    ) -> list[EntradaMemoria]:
        """Filtra las entradas recuperadas por aislamiento entre inquilinos y vigencia."""
        ahora = time.time()
        filtradas = []
        for entrada in entradas:
            # Aplicar aislamiento entre inquilinos
            if entrada.id_inquilino != inquilino_solicitante:
                continue
            # Aplicar TTL
            if ahora - entrada.creado_en > entrada.ttl_segundos:
                continue
            # Devolver solo entradas validadas
            if not entrada.validado:
                continue
            filtradas.append(entrada)
        return filtradas

Requisitos de evidencia

• Registros de auditoría de escritura en memoria con hashes de contenido, clasificaciones de riesgo y decisiones de validación
• Resultados de pruebas de aislamiento entre inquilinos que demuestran que no hay filtración de recuperación entre inquilinos
• Informes de cumplimiento del TTL que muestran que las entradas expiradas se purgan correctamente
• Estadísticas de detección de patrones de inyección con ejemplos de intentos de escritura bloqueados
• Trazas de procedencia de recuperación que vinculan las decisiones del agente con entradas de memoria específicas