wattocyber/agentic-dm-gatewayPlan de contrôle de sécurité pour agents LLM : listes d'autorisation, interrupteur d'arrêt d'urgence du propriétaire, sessions PIN, limites de débit, détection d'injection de prompt et nettoyage des sorties pour bloquer les fuites de secrets et l'exfiltration par balises image.
Plan de contrôle de sécurité pour les agents LLM sur messagerie privée (généralement les DM Discord).
Il se place devant votre agent. Il décide qui peut parler, si la session est déverrouillée, si le processus est en pause, et si ce message est suffisamment sûr pour être transmis. Votre modèle et vos outils restent derrière cette barrière. La bibliothèque n'appelle pas de LLM. Elle n'implémente aucune fonctionnalité produit au-delà de la sécurité.
Inspiré de Hermes. La conception suit les mêmes idées de plan de contrôle que celles utilisées dans les passerelles de messagerie de Hermes Agent : livraison prioritaire aux DM, listes d'autorisation d'identité, ouverture de type appairage, interrupteur d'arrêt du propriétaire, et séparation stricte entre qui peut agir (plan de contrôle) et le texte du message que voit le modèle (plan de données). Ce paquet est un extrait autonome et compact de ce modèle pour tout agent appelable. Non affilié à Nous Research.
Maturité : implémenté · validé indépendamment · maintenu. Voir STATUS.md.
Reproduire : python scripts/repro.py (attend REPRO_OK).
Tests hors ligne :
pip install -e ".[dev]" # or: pip install -e . && pip install pytest
python -m pytest -q --tb=line
# or: python scripts/repro.py
En ligne : https://gitlab.com/WattoCyber/agentic-dm-gateway
Si vous placez un agent sur Discord (ou toute API de messagerie) avec des outils, quiconque peut envoyer un message au bot peut tenter de :
Vous avez besoin d'un plan de contrôle (contrôles d'identité et de processus) distinct du plan de données (le texte des messages que voit le modèle).
Ce paquet est ce plan de contrôle.
| Contrôle | Comportement |
|---|---|
| Liste d'autorisation | Seuls les identifiants d'utilisateurs configurés peuvent poursuivre. Tous les autres sont rejetés (silencieusement ou avec une courte chaîne de refus). |
| Propriétaire vs ami | Les propriétaires ignorent le PIN et peuvent mettre en pause l'agent entier. Les amis peuvent avoir besoin d'un PIN partagé pour une ouverture limitée dans le temps (idée d'appairage à la Hermes, simplifiée). |
| Interrupteur d'arrêt | Fichier de pause global ou variable d'environnement. Aucun tour d'agent tant qu'il est actif. |
| Limites de débit | Fenêtre glissante par utilisateur (par minute et par heure). |
| Vérifications d'entrée | Longueur maximale, suppression des caractères de contrôle étranges, heuristiques regex pour les phrases courantes d'injection / de pêche aux secrets. |
| Nettoyage de sortie | Masquage des jetons ayant l'apparence de secrets (clés d'API, JWT, en-têtes Bearer) et suppression des balises image markdown/HTML pouvant exfiltrer via un chargement automatique. |
| Journal d'audit | JSONL en ajout seul des événements allow/deny/auth/kill pour examen ultérieur. |
| Commandes locales | /auth, /lock, /kill, /unkill, /status traitées sans appeler de modèle. |
Portée : porte de sécurité uniquement. Ni chatbot, ni bot de trading, ni scanner, ni framework d'agent. Passez un agent(user_id, text) -> str (ou asynchrone) si vous utilisez Discord. Le cœur fonctionne avec n'importe quel identifiant utilisateur entier et du texte brut.
$ 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
Trois voies d'intégration. Choisissez-en une.
Installez avec le support Discord, configurez les variables d'environnement avec vos identifiants utilisateur, enregistrez la passerelle, lancez le 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
Dans votre propre 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)
Ce que fait register_dm_gateway :
on_message sur votre discord.Client / bot.InboundSecurityPipeline.precheck avant votre agent.agent(user_id, sanitized_text, is_owner=...).Les messages de guild n'atteignent jamais l'agent. Seuls les DMs provenant d'utilisateurs figurant sur la liste d'autorisation le font.
on_message)Si vous ne pouvez pas utiliser register_dm_gateway (chaîne de gestionnaires existante), appelez le pipeline vous-même :
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])
Aucun import Discord requis. Utilisez le même précontrôle autour de n'importe quel tour d'agent :