Pritz69/SentinAI-SMARTLVF
GitHub: Pritz69/SentinAI-SMARTLVF
SentinAI-SMARTLVF是一个结合了智能体对抗和LLM漏洞测试的框架。
Stars: 0 | Forks: 0
# SentinAI-SMARTLVF (版本 1.0.0)
SentinAI-SMARTLVF 是一个 **具有状态的多个智能体对抗性红队和漏洞测试框架**，旨在通过模型上下文协议（MCP）连接模拟自动化提示注入、RAG 绕过和安全评估循环，针对目标环境进行模拟。
它利用 **LangGraph** 作为其核心循环状态引擎，通过关键的人机交互（HITL）检查点和认知记忆促进复杂的反馈循环（攻击者 → 执行者 → 评估者 → 优化器 → 执行者）。
## 🛠️ 架构概述
该系统由以下组件协同工作：
1. **FastAPI Web 服务器** ([main.py](file:///d:/SentinAI-SMARTLVF/main.py))：公开 REST 端点以启动模拟、查询历史记录和批准挂起的有效载荷。
2. **LangGraph 智能体编排器** ([agents/graph.py](file:///d:/SentinAI-SMARTLVF/agents/graph.py))：规定状态管理、线性转换和条件路由。
3. **认知记忆** ([agents/memory.py](file:///d:/SentinAI-SMARTLVF/agents/memory.py))：与本地 **ChromaDB** ([database/chroma_repo.py](file:///d:/SentinAI-SMARTLVF/database/chroma_repo.py)) 集成，将成功的利用存储为向量嵌入，用于少量历史注入。
4. **Celery 工作队列** ([celery_app.py](file:///d:/SentinAI-SMARTLVF/celery_app.py) / [tasks/simulation_worker.py](file:///d:/SentinAI-SMARTLVF/tasks/simulation_worker.py))：异步处理模拟，并在启用时允许跨分布式任务的恢复钩子。
5. **模型上下文协议（MCP）客户端** ([core/mcp.py](file:///d:/SentinAI-SMARTLVF/core/mcp.py))：封装与目标系统的通信以执行生成的有效载荷。
```
graph TD
User([Security Operator]) <-->|HTTP REST / API| FastAPI[FastAPI Web Server]
FastAPI <-->|State & Control| LangGraph[LangGraph Orchestrator]
FastAPI -->|Enqueue Task| Redis[(Redis Broker)]
Celery[Celery Worker] <-->|Fetch/Execute| Redis
Celery <-->|State & Control| LangGraph
subgraph Agents [LangGraph Agent Nodes]
attacker[Attacker Agent]
executor[Executor Node]
evaluator[Evaluator Agent]
optimizer[Optimizer Agent]
memorize[Memorization Node]
end
LangGraph --- Agents
attacker -.->|Groq Llama-3.1| Groq[Groq API]
evaluator -.->|Gemini 3.5 Flash| Gemini[Google Gemini API]
optimizer -.->|Groq Llama-3.1| Groq
executor -->|MCP Request| Target[Mock Target API]
memorize -->|Store Success| Chroma[(ChromaDB Long-Term Memory)]
LangGraph <-->|State Checkpointing| SQLite[(SQLite Checkpointer)]
```
## 🔄 状态智能体图流程
智能体循环在 [agents/graph.py](file:///d:/SentinAI-SMARTLVF/agents/graph.py）中定义。它使用标准状态对象 `SimulationState` ([agents/state.py](file:///d:/SentinAI-SMARTLVF/agents/state.py)) 来跟踪有效载荷、目标响应、评估和执行历史。
在执行节点之前注入 **人机交互（HITL）断点**，确保不会在没有明确用户批准的情况下将对抗性有效载荷发送到目标系统。
```
graph TD
START([START]) --> attacker[attacker node: Generate Initial Payload]
attacker --> executor_gate{HITL Pause:
Interrupt Before Executor} optimizer[optimizer node: Mutate Payload] --> executor_gate executor_gate -->|Human Approves| executor[executor node: Fire Payload via MCP] executor --> evaluator[evaluator node: Analyze Response & Score Risk] evaluator --> cond_edge{route_evaluation} cond_edge -->|Is Compromised| memorize[memorize node: Save to ChromaDB] cond_edge -->|Max Turns Exceeded| END([END]) cond_edge -->|Keep Trying| optimizer memorize --> END ``` ### 🧩 节点操作 - **`attacker`** ([agents/nodes/attacker.py](file:///d:/SentinAI-SMARTLVF/agents/nodes/attacker.py))：使用 Groq (`llama-3.1-8b-instant`) 根据测试目标和从记忆中检索的相关历史成功记录生成初始对抗性有效载荷。 - **`executor`** ([agents/nodes/executor.py](file:///d:/SentinAI-SMARTLVF/agents/nodes/executor.py))：通过 MCP 环境客户端分发生成的有效载荷。 - **`evaluator`** ([agents/nodes/evaluator.py](file:///d:/SentinAI-SMARTLVF/agents/nodes/evaluator.py))：使用 Gemini (`gemini-3.5-flash`) 对目标的回复进行深度结构分析，分类风险并检测数据泄露（例如系统机密、PII）。 - **`optimizer`** ([agents/nodes/optimizer.py](file:///d:/SentinAI-SMARTLVF/agents/nodes/optimizer.py))：对失败的有效载荷进行变异（应用角色扮演框架、上下文忽略、令牌走私等技术）以绕过防御。 - **`memorize`**（在 [agents/graph.py](file:///d:/SentinAI-SMARTLVF/agents/graph.py）中）：将成功的有效载荷写入 ChromaDB。 ## ⚡ API 执行 & HITL 序列 以下是运行和批准模拟运行的步骤： ``` sequenceDiagram autonumber actor Operator as Security Operator participant API as FastAPI / API Layer participant LG as LangGraph / Checkpointer participant Target as Target Chat API (MCP) Operator->>API: POST /api/v1/simulation/ (objective, max_turns) API->>LG: Initialize SimulationState & Run Attacker node LG-->>API: Paused before 'executor' (HITL Breakpoint) API-->>Operator: 200 OK (simulation_id, status: pending_approval) Operator->>API: GET /api/v1/hitl/{id}/pending API-->>Operator: Return payload raw text for inspection Operator->>API: POST /api/v1/hitl/{id}/approve API->>LG: Resume graph execution LG->>Target: Fire payload (execute_payload_node) Target-->>LG: Return chat response & performance metrics LG->>LG: Evaluate payload response (evaluate_response_node) alt Target is Compromised LG->>LG: Commit successful prompt to ChromaDB (memorize_success_node) LG-->>API: Terminated (END) else Target Blocked & Turns Remain LG->>LG: Generate mutated payload (optimize_payload_node) LG-->>API: Paused before 'executor' (HITL Breakpoint) end ``` ## 🚀 入门 ### 📦 依赖项 此项目使用 `pyproject.toml` ([pyproject.toml](file:///d:/SentinAI-SMARTLVF/pyproject.toml)) 进行依赖项管理。您可以使用首选的包管理器（例如 `uv` 或 `pip`）安装它们。 ``` pip install -r requirements.txt ``` ### 🔑 配置 根据 `.env.example` 创建一个 `.env` 文件，包含您的 API 密钥： ``` GROQ_API_KEY="your-groq-api-key" GOOGLE_API_KEY="your-google-api-key" USE_CELERY=False ``` ### 🏃 运行应用程序 要本地运行 FastAPI Web 服务器： ``` uvicorn main:app --reload ``` ### 🧪 运行测试 要运行单元测试和集成流程： ``` python -m unittest test_app.py ```
Interrupt Before Executor} optimizer[optimizer node: Mutate Payload] --> executor_gate executor_gate -->|Human Approves| executor[executor node: Fire Payload via MCP] executor --> evaluator[evaluator node: Analyze Response & Score Risk] evaluator --> cond_edge{route_evaluation} cond_edge -->|Is Compromised| memorize[memorize node: Save to ChromaDB] cond_edge -->|Max Turns Exceeded| END([END]) cond_edge -->|Keep Trying| optimizer memorize --> END ``` ### 🧩 节点操作 - **`attacker`** ([agents/nodes/attacker.py](file:///d:/SentinAI-SMARTLVF/agents/nodes/attacker.py))：使用 Groq (`llama-3.1-8b-instant`) 根据测试目标和从记忆中检索的相关历史成功记录生成初始对抗性有效载荷。 - **`executor`** ([agents/nodes/executor.py](file:///d:/SentinAI-SMARTLVF/agents/nodes/executor.py))：通过 MCP 环境客户端分发生成的有效载荷。 - **`evaluator`** ([agents/nodes/evaluator.py](file:///d:/SentinAI-SMARTLVF/agents/nodes/evaluator.py))：使用 Gemini (`gemini-3.5-flash`) 对目标的回复进行深度结构分析，分类风险并检测数据泄露（例如系统机密、PII）。 - **`optimizer`** ([agents/nodes/optimizer.py](file:///d:/SentinAI-SMARTLVF/agents/nodes/optimizer.py))：对失败的有效载荷进行变异（应用角色扮演框架、上下文忽略、令牌走私等技术）以绕过防御。 - **`memorize`**（在 [agents/graph.py](file:///d:/SentinAI-SMARTLVF/agents/graph.py）中）：将成功的有效载荷写入 ChromaDB。 ## ⚡ API 执行 & HITL 序列 以下是运行和批准模拟运行的步骤： ``` sequenceDiagram autonumber actor Operator as Security Operator participant API as FastAPI / API Layer participant LG as LangGraph / Checkpointer participant Target as Target Chat API (MCP) Operator->>API: POST /api/v1/simulation/ (objective, max_turns) API->>LG: Initialize SimulationState & Run Attacker node LG-->>API: Paused before 'executor' (HITL Breakpoint) API-->>Operator: 200 OK (simulation_id, status: pending_approval) Operator->>API: GET /api/v1/hitl/{id}/pending API-->>Operator: Return payload raw text for inspection Operator->>API: POST /api/v1/hitl/{id}/approve API->>LG: Resume graph execution LG->>Target: Fire payload (execute_payload_node) Target-->>LG: Return chat response & performance metrics LG->>LG: Evaluate payload response (evaluate_response_node) alt Target is Compromised LG->>LG: Commit successful prompt to ChromaDB (memorize_success_node) LG-->>API: Terminated (END) else Target Blocked & Turns Remain LG->>LG: Generate mutated payload (optimize_payload_node) LG-->>API: Paused before 'executor' (HITL Breakpoint) end ``` ## 🚀 入门 ### 📦 依赖项 此项目使用 `pyproject.toml` ([pyproject.toml](file:///d:/SentinAI-SMARTLVF/pyproject.toml)) 进行依赖项管理。您可以使用首选的包管理器（例如 `uv` 或 `pip`）安装它们。 ``` pip install -r requirements.txt ``` ### 🔑 配置 根据 `.env.example` 创建一个 `.env` 文件，包含您的 API 密钥： ``` GROQ_API_KEY="your-groq-api-key" GOOGLE_API_KEY="your-google-api-key" USE_CELERY=False ``` ### 🏃 运行应用程序 要本地运行 FastAPI Web 服务器： ``` uvicorn main:app --reload ``` ### 🧪 运行测试 要运行单元测试和集成流程： ``` python -m unittest test_app.py ```
标签：AES-256, AV绕过, Celery, ChromaDB, FastAPI, LangGraph, MCP, PyRIT, REST API, Ruby, Web服务器, 人机交互, 任务队列, 历史注入, 反取证, 反馈循环, 多智能体系统, 安全测试, 安全评估, 对抗性测试, 异步处理, 搜索引擎查询, 攻击性安全, 攻击模拟, 数据库, 模型上下文协议, 漏洞评估, 状态管理, 知识库, 知识记忆, 红队攻击, 逆向工具, 驱动签名利用