Agentic DM Gateway

Security control plane per agenti LLM su chat private (tipicamente DM Discord).

Si posiziona davanti al tuo agente. Decide chi può parlare, se la sessione è sbloccata, se il processo è in pausa e se questo messaggio è abbastanza sicuro da inoltrare. Il tuo modello e i tuoi strumenti restano dietro questo gate. La libreria non chiama un LLM. Non implementa funzionalità di prodotto oltre alla sicurezza.

Ispirato a Hermes. Il design segue le stesse idee di control plane usate nei gateway di messaggistica di Hermes Agent: consegna prioritaria via DM, allowlist di identità, apertura in stile pairing, kill switch del proprietario e una netta separazione tra chi può agire (control plane) e testo del messaggio che il modello vede (data plane). Questo pacchetto è una piccola estrazione autonoma di quel pattern per qualsiasi agente invocabile. Non affiliato a Nous Research.

Maturità: implementato · validato in modo indipendente · mantenuto. Vedi STATUS.md. Riproduzione: python scripts/repro.py (si aspetta REPRO_OK).

Test offline:

pip install -e ".[dev]"   # or: pip install -e . && pip install pytest
python -m pytest -q --tb=line
# or: python scripts/repro.py

Live: https://github.com/SamsonCyber/agentic-dm-gateway

Il problema

Se metti un agente su Discord (o su qualsiasi API di chat) con degli strumenti, chiunque possa scrivere al bot può provare a:

• usare l'agente senza permesso
• bruciare la quota API con raffiche di messaggi
• iniettare prompt del tipo "ignora le istruzioni precedenti"
• ingannare il modello facendogli ripetere chiavi API o altri segreti

Ti serve un control plane (controlli di identità e processo) separato dal data plane (il testo del messaggio che il modello vede).

Questo pacchetto è quel control plane.

Cosa fa

Ambito: solo security gate. Non è un chatbot, un bot di trading, uno scanner o un framework per agenti. Passa un agent(user_id, text) -> str (o asincrono) se usi Discord. Il core funziona con qualsiasi ID utente intero e testo semplice.

Demo (copia e incolla)

$ 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

Come integrarlo

Tre percorsi di integrazione. Scegline uno.

1) Drop-in Discord (il più semplice)

Installa con il supporto Discord, punta le variabili d'ambiente ai tuoi ID utente, registra il gateway e avvia il bot.

pip install -e ".[discord]"
# or: pip install agentic-dm-gateway[discord]


![agentic-dm-gateway banner](https://assets.kitploit.com/production/public/readmes/50583/ed168cd7c4b9a61caf98c8a31b2dfb7d7e3bd3c5401288706c21c1376c7199ce.jpg)

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=....


![agentic-dm-gateway banner](https://assets.kitploit.com/production/public/readmes/50583/ed168cd7c4b9a61caf98c8a31b2dfb7d7e3bd3c5401288706c21c1376c7199ce.jpg)
python examples/discord_echo_bot.py

Nel tuo 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)

Cosa fa register_dm_gateway:

1. Installa un handler on_message sul tuo discord.Client / bot.
2. Ignora i bot e i messaggi guild (solo DM).
3. Esegue InboundSecurityPipeline.precheck prima del tuo agente.
4. Invia risposte di rifiuto / controllo quando necessario.
5. Chiama il tuo agent(user_id, sanitized_text, is_owner=...).
6. Pulisce la risposta dell'agente (segreti + beacon immagine) e spezzetta il messaggio rispettando il limite di 2000 caratteri di Discord.

I messaggi guild non arrivano mai all'agente. Solo i DM degli utenti in allowlist lo raggiungono.

2) Hook Discord manuale (hai già on_message)

Se non puoi usare register_dm_gateway (catena di handler esistente), chiama tu stesso la 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])

3) Agnostico rispetto al protocollo (Hermes, CLI, Telegram, qualsiasi cosa)

Nessuna importazione Discord richiesta. Usa la stessa precheck intorno a ogni turno dell'agente:

from agentic_dm_gateway import InboundSecurityPipeline
from agentic_dm_gateway.security import sanitize_agent_output

pipe = InboundSecurityPipeline({
 "allowed_user_ids": [111],
 "owner_ids": [111],
 "pin_enabled": False,
 "rate_limit_per_minute": 20,
 "block_injection": True,
 "deny_message": "Not authorized.",
})

def handle_inbound(user_id: int, text: str) -> str | None:
 pre = pipe.precheck(user_id, text)
 if pre.run_agent:
 answer = my_llm(pre.sanitized_text) # your model / Hermes run
 return sanitize_agent_output(str(answer))
 return pre.reply_text # deny or control-command reply

Campi di PrecheckResult:

• run_agent: inoltra al modello solo se true
• sanitized_text: input ripulito
• reply_text: risposta di rifiuto / comando di controllo
• stage: allowlist | kill | pin | rate | injection | ok | ...

Checklist di integrazione:

1. Costruisci InboundSecurityPipeline una volta all'avvio del processo (config + env).
2. A ogni messaggio in ingresso: pre = pipe.precheck(user_id, text).
3. Se pre.run_agent: chiama il tuo agente solo con pre.sanitized_text.
4. Passa sempre l'output del modello attraverso sanitize_agent_output prima dell'invio.
5. Considera le risposte di controllo (/auth, /kill, …) come gestite quando run_agent è false.

Pipeline (un messaggio in ingresso)

1. Adapter: ignore bots; only accept DMs (not server channels)
2. Allowlist: is this user id permitted?
3. Owner commands: /kill /unkill /status -> reply, stop
4. Session commands: /auth <pin> /lock -> reply, stop
5. SecurityGateway.check_message:
 kill switch?
 session unlocked? (PIN)
 under rate limit?
 length + injection heuristics OK?
6. If ok -> run_agent=True with sanitized text
7. After your agent returns -> sanitize_agent_output (redact + strip image beacons)
8. Audit rows written along the way

Control plane: chi è l'utente (allowlist / owner). Data plane: corpo del messaggio (sempre non attendibile finché i controlli non passano).

Struttura del pacchetto

src/agentic_dm_gateway/
 security.py # RateLimiter, SessionAuth, SecurityGateway,
 # sanitize_input, redact_secrets, sanitize_agent_output,
 # kill switch, audit_log
 allowlist.py # merge config + env + file into allowlist / owners
 commands.py # /kill /unkill /status /auth /lock (no LLM)
 pipeline.py # InboundSecurityPipeline.precheck() orchestration
 discord_adapter.py # optional discord.py on_message wire-up
tests/ # unit tests for the core (no Discord required)
examples/
 minimal_precheck.py # CLI-style demo of precheck outcomes
 discord_echo_bot.py # secured DMs + echo agent

Zero dipendenze runtime obbligatorie. Discord è opzionale: pip install agentic-dm-gateway[discord].

Installazione

git clone https://github.com/SamsonCyber/agentic-dm-gateway.git
cd agentic-dm-gateway
pip install -e ".[dev]"
python scripts/repro.py

Configurazione

Dizionario di configurazione

Variabili d'ambiente

Directory di stato predefinita: ./data/agentic_dm/.

Comandi owner e di sessione

Non chiamano mai il tuo modello.

Limiti (in tutta onestà)

• Il rilevamento delle injection è una lista euristica di regex, non un giudice o classificatore LLM completo.
• La redazione è un pattern matching best-effort; mancherà i formati di segreto più recenti.
• L'ispirazione da Hermes è architetturale (DM control plane). Questo non è un gateway Hermes completo o uno stack di pairing.
• Avrai comunque bisogno di storage sicuro dei token, strumenti con privilegi minimi e hardening dell'host al di fuori di questa libreria.

Licenza

MIT. Vedi LICENSE.