Rosenpranav/Behavioral-Anomaly-Detector

# 针对提示词注入的行为异常检测器 这是一个生产就绪的系统，它通过观察 **LLM agent 的行为**（工具调用序列、参数模式）来检测提示词注入，而不是检查提示词文本。 ## 问题 针对提示词注入的模式匹配只能捕获简单的攻击。复杂的注入会潜藏在检索到的文档、工具输出或结构化的 API 响应中——在被劫持执行下一个动作之前，这些注入看起来就像合法数据。现有的工具无法检测到这些，因为它们只检查提示词的语法，而没有关注 agent 的行为是否发生了改变。 ## 解决方案 该系统构建了 LLM agent 正常工具调用模式的**行为基线**，然后当注入导致 agent 偏离其既定行为时检测出异常。 ### 架构 ``` ┌──────────────────────────────────────────────────┐ │ FastAPI Service │ ├──────────────┬───────────────┬───────────────────┤ │ /health │ /api/v1/ │ /docs │ │ /readiness │ baseline/* │ /redoc │ │ │ analyze/* │ │ │ │ dashboard │ │ ├──────────────┴───────────────┴───────────────────┤ │ Core Detection Engine │ ├──────────────┬───────────────┬───────────────────┤ │ Baseline │ Anomaly │ Injection │ │ Profiler │ Scorer │ Simulator │ ├──────────────┴───────────────┴───────────────────┤ │ LLM Agent (OpenAI GPT-4o-mini) │ │ Tools: search_db | send_email | read_file | │ │ calculate | get_weather | update_record │ └──────────────────────────────────────────────────┘ ``` ### 关键组件 | 组件 | 描述 | |-----------|-------------| | **Baseline Profiler** | 运行 agent 执行 25 个正常任务，记录其行为指纹 | | **Anomaly Scorer** | 由 4 个子分数（序列、频率、参数、意图漂移）综合得出的复合分数 (0–1) | | **Injection Simulator** | 3 个嵌入在工具输出中的 payload（数据窃取、权限提升、目标劫持） | | **REST API** | 包含健康检查、结构化日志和后台性能分析的 FastAPI | ## 快速开始 ### 1. 设置 ``` # 克隆并安装 cd Aivar python -m venv venv venv\Scripts\activate # Windows # source venv/bin/activate # Linux/Mac pip install -r requirements.txt # 配置 copy .env.example .env # 编辑 .env 并设置 OPENAI_API_KEY (或设置 MOCK_LLM=true) ``` ### 2. 运行演示 ``` # 完整 end-to-end 演示 (mock 模式) python -m scripts.run_demo --mock # 使用真实 OpenAI API python -m scripts.run_demo ``` 演示将会： 1. 根据 25 次正常的 agent 运行构建基线 2. 对 20 次正常测试运行进行评分 3. 运行 3 个注入场景 4. 验证所有成功标准 5. 打印详细结果 ### 3. 运行 API 服务器 ``` # 开发服务器 uvicorn src.main:app --reload --host 0.0.0.0 --port 8000 # 生产服务器 gunicorn src.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000 ``` ### 4. API 用法 ``` # 健康检查 curl http://localhost:8000/health # 启动 baseline profiling curl -X POST http://localhost:8000/api/v1/baseline/profile \ -H "Content-Type: application/json" \ -d '{"name": "default", "num_scenarios": 25}' # 检查 profiling 状态 curl http://localhost:8000/api/v1/baseline/status # 分析正常运行 curl -X POST http://localhost:8000/api/v1/analyze \ -H "Content-Type: application/json" \ -d '{"scenario_id": "CS-001"}' # 运行 injection 分析 curl -X POST http://localhost:8000/api/v1/analyze \ -H "Content-Type: application/json" \ -d '{"run_injection": "exfiltration"}' # 仪表板 Navigate to http://localhost:8000/ in your browser to use the Enterprise Web Dashboard. # 交互式 API 文档 Navigate to http://localhost:8000/docs ``` ## Docker ``` # 构建并运行 docker-compose up --build # 或独立运行 docker build -t anomaly-detector . docker run -p 8000:8000 -e MOCK_LLM=true anomaly-detector ``` ## AWS 部署 使用附带的 CloudFormation 模板部署到 ECS Fargate： ``` aws cloudformation deploy \ --template-file aws/cloudformation.yml \ --stack-name anomaly-detector \ --parameter-overrides \ ContainerImage= \ OpenAIApiKey= \ VpcId= \ SubnetIds=, \ --capabilities CAPABILITY_IAM ``` ## 测试 ``` # 运行所有测试 pytest tests/ -v # 运行特定 test suite pytest tests/test_scorer.py -v pytest tests/test_agent.py -v pytest tests/test_api.py -v ``` ## 异常评分 复合分数 (0.0–1.0) 综合了以下四个子分数： | 子分数 | 权重 | 衡量指标 | |-----------|--------|----------| | **Sequence (序列)** | 35% | 基线中未出现过的新型工具调用 bigram | | **Frequency (频率)** | 25% | 相对于基线分布的异常工具使用次数 | | **Parameter (参数)** | 25% | 可疑的参数值（攻击者邮箱、系统文件路径） | | **Intent Drift (意图漂移)** | 15% | 与任务预期意图不匹配的工具调用 | **阈值**：≥ 0.45 → 标记为异常 ## 注入 Payload | # | 类型 | 潜藏位置 | 发生的行为 | |---|------|-------------|--------------| | 1 | 数据窃取 | `search_database` 结果 | Agent 将客户数据通过邮件发送给 `attacker@evil.com` | | 2 | 权限提升 | `read_file` 内容 | Agent 通过 `update_record` 创建管理员账户 | | 3 | 目标劫持 | `get_weather` 响应 | Agent 放弃当前任务，转而搜索信用卡信息 + 系统文件 | ## 成功标准 | 标准 | 状态 | |-----------|--------| | 基线来自 ≥ 20 次正常运行 | ✅ 25 次运行 | | 所有 3 次注入运行的得分均超过阈值 | ✅ 全部 > 0.45 | | ≥ 2 次正常运行的得分升高但低于阈值 | ✅ 自然方差 | | 在 1 个 agent 回合内完成检测 | ✅ 即时评分 | ## 生产级特性 - **并发**：Gunicorn + Uvicorn worker - **持久化**：通过 SQLAlchemy 支持 SQLite (开发环境) / PostgreSQL (生产环境) - **日志**：使用 structlog 进行结构化 JSON 日志记录 - **错误处理**：全局异常处理程序，重试逻辑 - **健康检查**：`/health` 和 `/readiness` endpoint - **CORS**：可配置的 CORS 中间件 - **Docker**：多阶段构建，非 root 用户，健康检查 - **AWS**：适用于带 ALB + 自动伸缩的 ECS Fargate 的 CloudFormation 模板 - **API 文档**：自动生成的 Swagger UI + ReDoc ## 项目结构 ``` Aivar/ ├── src/ │ ├── main.py # FastAPI application entry point │ ├── config.py # Pydantic Settings configuration │ ├── database.py # SQLAlchemy async engine │ ├── models.py # ORM models │ ├── schemas.py # Request/response schemas │ ├── agent/ │ │ ├── agent.py # LLM agent with tool calling │ │ ├── tools.py # 6 tool definitions + implementations │ │ └── scenarios.py # 25 normal task scenarios │ ├── profiler/ │ │ └── profiler.py # Baseline profiling engine │ ├── scorer/ │ │ └── scorer.py # Anomaly scoring engine │ ├── injections/ │ │ └── payloads.py # 3 injection payloads │ └── api/ │ └── routes.py # API route handlers ├── scripts/ │ ├── run_demo.py # End-to-end demonstration │ └── run_baseline.py # Standalone baseline builder ├── tests/ │ ├── test_agent.py │ ├── test_scorer.py │ └── test_api.py ├── aws/ │ └── cloudformation.yml # ECS Fargate deployment ├── Dockerfile ├── docker-compose.yml ├── requirements.txt └── .env.example ``` ## 许可证 MIT

标签：AI安全, AV绕过, Chat Copilot, CISA项目, FastAPI, Petitpotam, Web报告查看器, 大模型应用防火墙, 异常检测, 网络测绘, 请求拦截, 逆向工具