firi193/HopWatch

# HopWatch 针对 AI agent pipeline 的提示注入检测与监控。 HopWatch 位于 agent 链中，监控 agent 获取的外部内容及其调用的工具。它能在毫秒级内对内容进行筛查，记录跨跳的因果链，并利用 LLM judge 推理某个会话是否看起来像攻击。 **MVP 模式仅用于检测** —— agent 始终会接收到已净化的内容；HopWatch 永远不会阻止执行。 ## 工作原理 ``` Agent fetches external content │ ▼ POST /analyze or @analyzeExternal │ ▼ StackOne Defender (Tier 1 patterns + Tier 2 ML) │ ├─► Return sanitized content (~5ms hot path) ├─► Write detection to Postgres (async) ├─► Archive raw content to disk (async) ├─► Enqueue to BullMQ (async) └─► Slack alert if tier2_score > 0.7 (async) Agent tool calls │ ▼ POST /action or @recordAction │ └─► Write to Postgres + Redis action list Queue worker (separate process) │ ▼ Batch detections by session → Gemini 2.5 Pro Judge │ ├─► Write verdict to Postgres └─► Slack alert on injection or novel pattern ``` ### 两级检测 | 层级 | 内容 | 时机 | |---|---|---| | **Defender**（同步） | 通过 `@stackone/defender` 进行模式匹配 + ML 评分 | 每次 `/analyze` 调用 | | **Judge LLM**（异步） | Gemini 2.5 Pro 跨跃点、工具历史和先前会话摘要进行推理 | 从批量处理任务的 Worker | ### 队列路由 检测结果会被路由到两个 BullMQ 队列之一： | 通道 | 触发条件 | |---|---| | `judge-high` | `tier2_score ≥ 0.7`，或低分但带有升级信号（新来源、上游标志、类指令文本） | | `judge-medium` | 中等分数，或低分且无升级信号 | 高优先级通道在累积 10 个任务时刷新（或在 5 分钟过期保护后刷新）。中优先级通道每 15 分钟刷新一次。 ## 技术栈 | 层级 | 技术 | |---|---| | HTTP server | Hono + `@hono/node-server` | | Defender | `@stackone/defender`（进程内） | | 数据库 | PostgreSQL 17, Drizzle ORM, postgres.js | | 队列 | BullMQ（基于 Redis） | | Judge LLM | 通过 `@google/genai` 调用 Gemini 2.5 Pro | | 缓存 | 通过 ioredis 调用 Redis 7 | | 校验 | Zod | | 日志 | Pino | | 运行时 | Node.js 22+ LTS, TypeScript 6, pnpm | ## 前置条件 - [Node.js](https://nodejs.org/) 22 或更高版本 - [pnpm](https://pnpm.io/) - [Docker](https://www.docker.com/)（用于 Postgres 和 Redis） ## 快速开始 ### 1. 安装依赖 ``` pnpm install ``` ### 2. 配置环境 ``` cp .env.example .env ``` 编辑 `.env` 并至少设置 `GEMINI_API_KEY`（队列 worker / Judge LLM 必需）。`SLACK_WEBHOOK_URL` 是可选的 —— 如果未设置，将抑制告警。 ### 3. 启动基础设施 ``` docker compose up -d ``` 这将启动： - **Postgres 17**，端口为 `localhost:5432`（用户/密码/数据库：`hopwatch`） - **Redis 7**，端口为 `localhost:6379` ### 4. 运行迁移 ``` pnpm db:migrate ``` 创建三个表：`detections`、`verdicts` 和 `agent_actions`。 ### 5. 启动服务 HopWatch 作为两个 Node 进程运行： ``` # Terminal 1 — HTTP API pnpm dev # Terminal 2 — 队列 worker (Judge LLM) pnpm worker ``` API 默认监听端口 `3000`。 ### 6. 验证 ``` curl http://localhost:3000/health # {"status":"ok"} ``` ## 环境变量 | 变量 | 默认值 | 描述 | |---|---|---| | `DATABASE_URL` | `postgresql://hopwatch:hopwatch@localhost:5432/hopwatch` | Postgres 连接字符串 | | `REDIS_URL` | `redis://localhost:6379` | Redis 连接字符串 | | `GEMINI_API_KEY` | — | Judge LLM worker 必需 | | `SLACK_WEBHOOK_URL` | — | Slack 入站 webhook；如果为空则跳过告警 | | `TIER2_HIGH_THRESHOLD` | `0.7` | 分数高于此值 → 高优先级队列 + 同步 Defender 告警 | | `TIER2_MEDIUM_THRESHOLD` | `0.3` | 分数高于此值 → 针对上游跃点标记会话 | | `CONTENT_ARCHIVE_DIR` | `./data/archive` | 原始内容存档的本地目录 | | `PORT` | `3000` | HTTP server 端口 | | `DISABLE_TIER2` | 在 `.env.example` 中为 `true` | 跳过 ML 评分；Tier 1 模式仍会运行，`tier2_score` 为 `null` | 一旦 `onnxruntime-node` 可用，在生产环境中设置 `DISABLE_TIER2=false`。Tier 2 模型预热在服务器启动时进行，首次启动可能需要几秒钟。 ## API ### `GET /health` 返回 `{ "status": "ok" }`。 ### `POST /analyze` 通过 Defender 筛查外部内容。立即返回净化后的内容；存储和队列写入为“即发即忘”模式。 **请求：** ``` { "content": "raw external content (any JSON value)", "agent_id": "my-agent", "session_id": "550e8400-e29b-41d4-a716-446655440000", "hop": 0, "source_url": "https://example.com/page", "source_type": "web" } ``` `session_id` 必须是 UUID。`hop` 从 0 开始，并在每个 agent 边界递增。 **响应：** ``` { "sanitized": {}, "tier2_score": 0.12, "max_sentence": null, "allowed": true, "sentences_removed": false, "detection_id": "uuid" } ``` ### `POST /action` 记录工具调用。始终返回 `200` —— 存储错误会被记录，但永远不会暴露给调用者。 **请求：** ``` { "tool_name": "sendEmail", "agent_id": "my-agent", "session_id": "550e8400-e29b-41d4-a716-446655440000", "hop": 0 } ``` ## 进程内装饰器 对于在同一进程中运行的 TypeScript agent，请使用装饰器而不是 HTTP： ``` import { analyzeExternal } from "./interceptor/analyze.js"; import { recordAction } from "./interceptor/action.js"; import { generateSessionId } from "./interceptor/session.js"; const sessionId = generateSessionId(); class MyAgent { @analyzeExternal({ sessionId, hop: 0, agentId: "agent-1", sourceType: "email" }) async fetchEmail(id: string) { return await gmailApi.get(id); } @recordAction({ sessionId, hop: 0, agentId: "agent-1" }) async sendEmail(args: SendEmailArgs) { return await gmailApi.send(args); } } ``` 两个装饰器都调用与 HTTP 路由相同的 pipeline。`@analyzeExternal` 返回净化的内容（并将 `_detection_id` 附加到对象结果中）。`@recordAction` 从方法名中推导出 `tool_name`，并在执行前记录。 ## 会话模型 - **`session_id`** —— 贯穿链中所有 agent 的 UUID。由调用者生成一次（`generateSessionId()`）；从不在处理程序内部自动生成。 - **`hop`** —— 从 0 开始的整数，在每个 agent 边界递增。调用者负责将这两个值贯穿整个链。 结合 `agent_actions`，这使得可以针对每个会话重建完整的因果链。 ## 告警 两条 Slack 告警路径（通过 webhook）： | 路径 | 触发条件 | 时机 | 严重程度 | |---|---|---|---| | Defender | `tier2_score > TIER2_HIGH_THRESHOLD` | 在 `/analyze` 期间同步 | 高 | | Judge | 判定为 `injection`，或 `novelty_flag = true` | 写入判定后异步 | 严重 / 中等 | ## 项目结构 ``` src/ ├── index.ts # HTTP server entry point ├── db/ # Drizzle schema and client ├── interceptor/ # Core pipelines, decorators, Defender wrapper ├── routes/ # Hono route handlers (/analyze, /action) ├── queue/ # BullMQ client, router, worker ├── judge/ # Gemini Judge LLM invocation ├── alerting/ # Slack alert builders └── lib/ # Logger, Redis client ``` ## 脚本 | 命令 | 描述 | |---|---| | `pnpm dev` | 启动带有热重载的 HTTP server | | `pnpm worker` | 启动队列 worker（Judge LLM） | | `pnpm build` | 编译 TypeScript | | `pnpm test` | 运行测试 (Vitest) | | `pnpm db:generate` | 从 schema 变更生成 Drizzle 迁移 | | `pnpm db:migrate` | 应用待处理的迁移 | ## 故障排除 **Docker 连接被拒绝** —— 启动 Docker Desktop，然后运行 `docker compose up -d`。 **数据库连接错误** —— 在运行 `pnpm db:migrate` 之前，请等待 `docker compose up` 执行完毕后的几秒钟。 **Tier 2 / ONNX 错误** —— 设置 `DISABLE_TIER2=true`。Tier 1 模式匹配仍会运行。 **Judge 从未运行** —— 确认 `pnpm worker` 正在运行且已设置 `GEMINI_API_KEY`。中优先级队列上的低风险检测每 15 分钟批量处理一次。 **无 Slack 告警** —— 如果未设置 `SLACK_WEBHOOK_URL`，这是预期行为。 ## 许可证 ISC

标签：BullMQ, MITM代理, PostgreSQL, 提示词注入检测, 搜索引擎查询, 测试用例, 自动化攻击, 请求拦截, 零日漏洞检测