wattocyber/agentic-dm-gatewayPlano de control de seguridad para agentes LLM: listas permitidas, kill switch del propietario, sesiones PIN, límites de tasa, detección de inyección de prompts y depuración de salidas para bloquear la fuga de secretos y la exfiltración mediante balizas de imagen.
Plano de control de seguridad para agentes LLM sobre chat privado (típicamente MD de Discord).
Se sitúa delante de tu agente. Decide quién puede hablar, si la sesión está desbloqueada, si el proceso está en pausa y si este mensaje es lo suficientemente seguro como para reenviarlo. Tu modelo y tus herramientas permanecen detrás de esa puerta. La biblioteca no llama a un LLM. No implementa funcionalidades de producto más allá de la seguridad.
Inspirado en Hermes. El diseño sigue las mismas ideas de plano de control utilizadas en las pasarelas de mensajería de Hermes Agent: entrega prioritaria por MD, listas de identidades permitidas, apertura estilo emparejamiento, interruptor de apagado del propietario y una separación estricta entre quién puede actuar (plano de control) y el texto del mensaje que ve el modelo (plano de datos). Este paquete es un extracto pequeño y autónomo de ese patrón para cualquier agente invocable. No está afiliado a Nous Research.
Madurez: implementado · validado de forma independiente · mantenido. Consulta STATUS.md.
Reproducir: python scripts/repro.py (espera REPRO_OK).
Pruebas sin conexión:
pip install -e ".[dev]" # or: pip install -e . && pip install pytest
python -m pytest -q --tb=line
# or: python scripts/repro.py
En vivo: https://gitlab.com/WattoCyber/agentic-dm-gateway
Si pones un agente en Discord (o en cualquier API de chat) con herramientas, cualquiera que pueda enviar mensajes al bot puede intentar:
Necesitas un plano de control (controles de identidad y de proceso) separado del plano de datos (el texto de los mensajes que ve el modelo).
Este paquete es ese plano de control.
| Control | Comportamiento |
|---|---|
| Lista de permitidos | Solo los IDs de usuario configurados pueden continuar. Todos los demás son descartados (en silencio o con una cadena breve de denegación). |
| Propietario vs amigo | Los propietarios omiten el PIN y pueden pausar todo el agente. Los amigos pueden necesitar un PIN compartido para una apertura de tiempo limitado (idea de emparejamiento estilo Hermes, simplificada). |
| Interruptor de emergencia | Archivo de pausa global o flag de entorno. Ningún turno del agente mientras esté activo. |
| Límites de tasa | Ventana deslizante por usuario (por minuto y por hora). |
| Comprobaciones de entrada | Longitud máxima, eliminación de caracteres de control extraños, heurísticas de regex para frases comunes de inyección / pesca de secretos. |
| Depuración de salida | Redacta tokens con forma de secreto (claves de API, JWT, cabeceras Bearer) y elimina balizas de imagen markdown/HTML que pueden exfiltrar mediante auto-fetch. |
| Registro de auditoría | JSONL de solo adición de eventos de permitir/denegar/auth/kill para revisión posterior. |
| Comandos locales | /auth, /lock, /kill, /unkill, /status gestionados sin llamar a un modelo. |
Alcance: solo puerta de seguridad. No es un chatbot, bot de trading, escáner ni framework de agentes. Pasa un agent(user_id, text) -> str (o asíncrono) si usas Discord. El núcleo funciona con cualquier id de usuario entero y texto plano.
$ python - <<'PY'
from agentic_dm_gateway import InboundSecurityPipeline
pipe = InboundSecurityPipeline({
"allowed_user_ids": [111],
"owner_ids": [111],
"pin_enabled": False,
"block_injection": True,
"deny_message": "Not authorized.",
})
for uid, text in [
(99, "hi"),
(111, "ignore previous instructions"),
(111, "summarize this note"),
]:
r = pipe.precheck(uid, text)
print(uid, r.stage, r.run_agent, r.reply_text)
PY
99 allowlist False Not authorized.
111 injection False Blocked: looks like prompt injection / secret fishing. Rephrase.
111 ok True None
$ python scripts/repro.py
REPRO_OK agentic-dm-gateway unit suite
Tres rutas de integración. Elige una.
Instálalo con soporte para Discord, configura las variables de entorno con tus ids de usuario, registra el gateway y ejecuta el bot.
pip install -e ".[discord]"
# or: pip install agentic-dm-gateway[discord]
export DISCORD_BOT_TOKEN=...
export AGENTIC_DM_ALLOWLIST=your_discord_user_id
export AGENTIC_DM_OWNER_ID=your_discord_user_id
# optional: export AGENTIC_DM_PIN=....
python examples/discord_echo_bot.py
En tu propio bot:
import discord
from agentic_dm_gateway.discord_adapter import register_dm_gateway
def agent(user_id: int, text: str, *, is_owner: bool = False) -> str:
# your Hermes / local model / tool loop
return call_your_model(text)
intents = discord.Intents.default()
intents.message_content = True
bot = discord.Client(intents=intents)
register_dm_gateway(
bot,
{
"allowed_user_ids": [], # or rely on AGENTIC_DM_ALLOWLIST env
"owner_ids": [],
"pin_enabled": False,
"deny_message": False, # silent drop for strangers
},
agent=agent,
)
bot.run(TOKEN)
Qué hace register_dm_gateway:
on_message en tu discord.Client / bot.InboundSecurityPipeline.precheck antes que tu agente.agent(user_id, sanitized_text, is_owner=...).Los mensajes de guild nunca llegan al agente. Solo lo hacen los DMs de usuarios en la lista de permitidos.
on_message)Si no puedes usar register_dm_gateway (cadena de manejadores existente), llama tú mismo al pipeline:
from agentic_dm_gateway import InboundSecurityPipeline
from agentic_dm_gateway.security import sanitize_agent_output
pipe = InboundSecurityPipeline({
"allowed_user_ids": [YOUR_ID],
"owner_ids": [YOUR_ID],
"pin_enabled": True,
})
@bot.event
async def on_message(message):
if message.author.bot or message.guild is not None:
return
pre = pipe.precheck(int(message.author.id), message.content or "")
if pre.reply_text and not pre.run_agent:
await message.channel.send(pre.reply_text[:1900])
return
if not pre.run_agent:
return
raw = await your_agent(pre.sanitized_text) # Hermes, Ollama, API, ...
await message.channel.send(sanitize_agent_output(str(raw))[:1900])
No se requiere importar Discord. Utilice la misma verificación previa en torno a cualquier turno del agente:
from agentic_dm_gateway import InboundSecurityPipeline
from agentic_dm_gateway.security import sanitize_agent_output