Rosenpranav/PROMPT-INJECTION-BEHAVIORIAL-DETECTECTOR

# Prompt 注入行为异常检测器 一个生产就绪的系统，通过观察 **LLM agent 做了什么**（工具调用序列、参数模式）来检测 prompt 注入，而不是检查 prompt 文本。 ## 问题 针对 prompt 注入的模式匹配只能捕获简单的攻击。复杂的注入会潜藏在检索到的文档、工具输出或结构化 API 响应中——在进行劫持 agent 的下一步动作之前，这些注入看起来就像是合法的数据。现有的工具之所以无法检测到这些，是因为它们只检查 prompt 的语法，而没有关注 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** | 通过 25 个正常任务运行 agent，记录行为指纹 | | **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. 运行 Demo ``` # 完整的端到端演示 (mock 模式) python -m scripts.run_demo --mock # 使用真实 OpenAI API python -m scripts.run_demo ``` 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 # 运行特定测试套件 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 middleware - **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

标签：AV绕过, FastAPI, 人工智能, 提示词注入检测, 用户模式Hook绕过, 行为异常检测, 请求拦截, 逆向工具