wattocyber/agentic-dm-gatewayLLM 에이전트를 위한 보안 컨트롤 플레인: 허용 목록(allowlist), 소유자 킬 스위치, PIN 세션, 속도 제한, 프롬프트 인젝션 탐지, 그리고 비밀 유출과 이미지 비콘 외부 유출을 차단하는 출력 스크러빙.
프라이빗 채팅(일반적으로 Discord DM)을 통한 LLM 에이전트용 보안 제어 평면입니다.
이것은 에이전트 앞에 위치합니다. 누가 대화할 수 있는지, 세션이 잠금 해제되었는지, 프로세스가 일시 중지되었는지, 그리고 이 메시지가 전달하기에 충분히 안전한지를 결정합니다. 모델과 도구는 이 게이트 뒤에 남습니다. 이 라이브러리는 LLM을 호출하지 않습니다. 보안 이외의 제품 기능은 구현하지 않습니다.
Hermes에서 영감을 받았습니다. 설계는 Hermes Agent 메시징 게이트웨이에서 사용되는 것과 동일한 제어 평면 아이디어를 따릅니다: DM 우선 전달, 신원 허용 목록, 페어링 방식의 개방, 소유자 킬 스위치, 그리고 누가 행동할 수 있는가(제어 평면)와 모델이 보는 메시지 텍스트(데이터 평면) 사이의 엄격한 분리. 이 패키지는 모든 에이전트 호출 가능 대상에 대해 해당 패턴을 작게 독립적으로 추출한 것입니다. Nous Research와 제휴 관계가 없습니다.
성숙도: 구현됨 · 독립적으로 검증됨 · 유지 관리됨. STATUS.md를 참조하세요.
재현: python scripts/repro.py (REPRO_OK를 예상함).
오프라인 테스트:
pip install -e ".[dev]" # or: pip install -e . && pip install pytest
python -m pytest -q --tb=line
# or: python scripts/repro.py
라이브: https://gitlab.com/WattoCyber/agentic-dm-gateway
도구가 있는 에이전트를 Discord(또는 모든 채팅 API)에 배치하면, 봇에 메시지를 보낼 수 있는 누구나 다음을 시도할 수 있습니다:
데이터 플레인(모델이 보는 메시지 텍스트)과 분리된 컨트롤 플레인(신원 및 프로세스 제어)이 필요합니다.
이 패키지가 바로 그 컨트롤 플레인입니다.
| 제어 | 동작 |
|---|---|
| 허용 목록 | 설정된 사용자 ID만 진행할 수 있습니다. 그 외는 모두 차단됩니다(조용히 또는 짧은 거부 문자열과 함께). |
| 소유자 vs 친구 | 소유자는 PIN을 건너뛰고 전체 에이전트를 일시 중지할 수 있습니다. 친구는 시간 제한 개방을 위해 공유 PIN이 필요할 수 있습니다(Hermes 스타일 페어링 아이디어를 단순화한 것). |
| 킬 스위치 | 전역 일시 중지 파일 또는 환경 변수 플래그. 활성화 중에는 에이전트 턴이 없습니다. |
| 속도 제한 | 사용자별 슬라이딩 윈도우(분당 및 시간당). |
| 입력 검사 | 최대 길이, 이상한 제어 문자 제거, 일반적인 주입/비밀 정보 탈취 문구에 대한 정규식 휴리스틱. |
| 출력 정제 | 비밀 형태의 토큰(API 키, JWT, Bearer 헤더)을 마스킹하고, 자동 가져오기를 통해 유출될 수 있는 마크다운/HTML 이미지 비콘을 제거합니다. |
| 감사 로그 | 나중 검토를 위한 허용/거부/인증/킬 이벤트의 추가 전용 JSONL. |
| 로컬 명령 | /auth, /lock, /kill, /unkill, /status는 모델 호출 없이 처리됩니다. |
범위: 보안 게이트 전용. 챗봇, 트레이딩 봇, 스캐너, 에이전트 프레임워크가 아닙니다. Discord를 사용하는 경우 agent(user_id, text) -> str(또는 비동기)을 전달하세요. 코어는 모든 정수 사용자 ID와 일반 텍스트와 함께 작동합니다.
$ 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
세 가지 통합 경로가 있습니다. 하나를 선택하세요.
Discord 지원과 함께 설치하고, 환경 변수를 사용자 ID로 지정하고, 게이트웨이를 등록한 뒤 봇을 실행하세요.
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
자신의 봇에서:
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)
register_dm_gateway가 하는 일:
discord.Client / 봇에 on_message 핸들러를 설치합니다.InboundSecurityPipeline.precheck를 실행합니다.agent(user_id, sanitized_text, is_owner=...)를 호출합니다.길드 메시지는 에이전트에 도달하지 않습니다. 허용 목록에 있는 사용자의 DM만 도달합니다.
on_message가 있는 경우)register_dm_gateway를 사용할 수 없는 경우(기존 핸들러 체인), 파이프라인을 직접 호출하세요:
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])
Discord 가져오기는 필요하지 않습니다. 모든 에이전트 턴 주변에서 동일한 사전 검사를 사용하세요:
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
PrecheckResult 필드:
run_agent: true인 경우에만 모델로 전달sanitized_text: 정제된 입력reply_text: 거부 / 제어 명령 응답stage: allowlist | kill | pin | rate | injection | ok | ...훅 체크리스트:
InboundSecurityPipeline을 한 번 빌드합니다 (config + env).pre = pipe.precheck(user_id, text).pre.run_agent가 참이면: pre.sanitized_text만 사용하여 에이전트를 호출합니다.sanitize_agent_output을 통해 전달합니다./auth, /kill, …)은 run_agent가 false일 때 완료된 것으로 처리합니다.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
제어 평면: 사용자가 누구인지(허용 목록 / 소유자). 데이터 평면: 메시지 본문(검사를 통과할 때까지 항상 신뢰할 수 없음).