wattocyber/agentic-dm-gatewayPiano di controllo della sicurezza per agenti LLM: allowlist, kill switch del proprietario, sessioni PIN, limiti di frequenza, rilevamento delle iniezioni di prompt e pulizia dell'output per bloccare la fuga di segreti e l'esfiltrazione tramite beacon nelle immagini.
Piano di controllo di sicurezza per agenti LLM su chat private (tipicamente DM di Discord).
Si colloca davanti al tuo agente. Decide chi può parlare, se la sessione è sbloccata, se il processo è in pausa e se questo messaggio è sufficientemente sicuro da essere inoltrato. Il tuo modello e i tuoi strumenti restano dietro quel cancello. La libreria non chiama un LLM. Non implementa funzionalità di prodotto al di là della sicurezza.
Ispirato a Hermes. Il design segue le stesse idee di piano di controllo usate nei gateway di messaggistica di Hermes Agent: consegna DM-first, allowlist di identità, apertura in stile pairing, kill switch del proprietario e una netta separazione tra chi può agire (piano di controllo) e testo del messaggio che il modello vede (piano dati). Questo pacchetto è un piccolo estratto autonomo di quel pattern per qualsiasi agente richiamabile. Non affiliato a Nous Research.
Maturità: implementato · validato in modo indipendente · mantenuto. Vedi STATUS.md.
Riproduci: python scripts/repro.py (attende 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://gitlab.com/WattoCyber/agentic-dm-gateway
Se metti un agente su Discord (o su qualsiasi API di chat) dotato di strumenti, chiunque possa inviare messaggi al bot può tentare di:
Serve un control plane (controlli di identità e di processo) separato dal data plane (il testo dei messaggi che il modello vede).
Questo pacchetto è quel control plane.
| Controllo | Comportamento |
|---|---|
| Allowlist | Solo gli ID utente configurati possono procedere. Tutti gli altri vengono scartati (in modo silenzioso o con una breve stringa di rifiuto). |
| Owner vs friend | Gli owner saltano il PIN e possono mettere in pausa l'intero agente. I friend possono richiedere un PIN condiviso per un'apertura a tempo limitato (idea di pairing in stile Hermes, semplificata). |
| Kill switch | File di pausa globale o flag di ambiente. Nessun turno dell'agente mentre è attivo. |
| Rate limit | Finestra scorrevole per utente (al minuto e all'ora). |
| Controlli sull'input | Lunghezza massima, rimozione di caratteri di controllo anomali, euristiche regex per frasi comuni di injection / secret-fishing. |
| Pulizia dell'output | Oscuramento di token con forma di segreto (chiavi API, JWT, header Bearer) e rimozione di beacon immagine markdown/HTML che possono esfiltrare tramite auto-fetch. |
| Log di audit | JSONL append-only di eventi allow/deny/auth/kill per revisioni successive. |
| Comandi locali | /auth, /lock, /kill, /unkill, /status gestiti senza chiamare un modello. |
Ambito: solo gate di sicurezza. Non è un chatbot, un trading bot, uno scanner o un framework per agenti. Passa un agent(user_id, text) -> str (o async) se usi Discord. Il core funziona con qualsiasi user id intero e testo semplice.
$ 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
Tre percorsi di integrazione. Scegline uno.
Installa con il supporto Discord, punta le variabili d'ambiente ai tuoi user id, registra il gateway, avvia il 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
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:
on_message sul tuo discord.Client / bot.InboundSecurityPipeline.precheck prima del tuo agente.agent(user_id, sanitized_text, is_owner=...).I messaggi di guild non raggiungono mai l'agente. Solo i DM degli utenti in allowlist lo fanno.
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])
Nessun import di Discord richiesto. Utilizza lo stesso precheck attorno a qualsiasi 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.",
})