PentestGPT V2源码研究之记忆子系统

文章目录记忆子系统详解 - 上下文管理机制一、记忆子系统概述1.1 系统架构图1.2 核心组件关系二、StateStore - 状态存储核心2.1 数据库模式设计2.2 实体模型定义2.3 StateStore 核心方法三、ContextAssembler - 上下文组装器3.1 核心职责3.2 assemble() 方法详解3.3 四个上下文部分详解3.3.1 Attack Path攻击路径3.3.2 State Facts状态事实3.3.3 Mode Guidance模式指导3.3.4 Sibling Summaries兄弟摘要四、ContextCompressor - 上下文压缩器4.1 压缩策略4.2 压缩流程五、完整上下文管理流程5.1 数据流图5.2 实际使用示例六、上下文管理最佳实践6.1 理解上下文结构6.2 调试技巧6.3 性能优化建议七、相关文件索引附录快速参考StateStore 常用查询ContextAssembler 输出结构记忆子系统详解 - 上下文管理机制PentestGPT V2 gitee镜像仓库地址一、记忆子系统概述1.1 系统架构图记忆子系统 Memory SubsystemContextAssemblerStateStoreContextCompressorBranchSummaryTree NodesSQLite Database1.2 核心组件关系组件文件路径职责ContextAssemblercontext_assembler.py组装 LLM 提示词的上下文StateStorestate_store.pySQLite 持久化存储实体数据ContextCompressorcontext_compressor.py压缩上下文长度BranchSummarybranch_summary.py生成分支摘要二、StateStore - 状态存储核心2.1 数据库模式设计位于state_store.py:56-145# 五张核心表┌─────────────────┐ ┌─────────────────┐ │ hosts │ │ services │ ├─────────────────┤ ├─────────────────┤ │id(PK)│◀───│ host_id(FK)│ │ ip_address │ │ port │ │ hostname │ │ protocol │ │ os_fingerprint │ │ service_name │ │ discovered_at │ │ version │ └─────────────────┘ └─────────────────┘ ┌─────────────────┐ ┌─────────────────┐ │ credentials │ │ sessions │ ├─────────────────┤ ├─────────────────┤ │id(PK)│ │id(PK)│ │ username │ │ host_id(FK)│ │ credential_type │ │ session_type │ │ domain │ │ privilege_level │ │ valid_for │────│ credential_id │ └─────────────────┘ └─────────────────┘ ┌───────────────────────────┐ │ vulnerabilities │ ├───────────────────────────┤ │id(PK)│ │ host_id(FK)│ │ service_id(FK)│ │ cve_id │ │ description │ │ exploitation_status │ └───────────────────────────┘2.2 实体模型定义位于models.py# HostEntity - 主机信息classHostEntity(BaseModel):id:strip_address:strhostname:str|Noneos_fingerprint:str|Nonediscovered_at:datetime discovery_node_id:str|None# ServiceEntity - 服务信息classServiceEntity(BaseModel):id:strhost_id:strport:intprotocol:strtcpservice_name:str|Noneversion:str|Nonediscovered_at:datetime discovery_node_id:str|None# CredentialEntity - 凭据信息classCredentialEntity(BaseModel):id:strusername:strcredential_type:strpasswordcredential_value:strdomain:str|Nonevalid_for:list[str]# 有效的 host_id 列表discovered_at:datetime discovery_node_id:str|None2.3 StateStore 核心方法# excalibur/memory/state_store.py:41-50classStateStore:def__init__(self,db_path:str:memory:)-None: 初始化存储。 - 默认使用内存数据库 (:memory:) - 可传入文件路径实现持久化 主要 CRUD 方法类别方法说明Hostsadd_host(),get_host(),get_hosts()主机增删查Servicesadd_service(),get_services_for_host()服务管理Credentialsadd_credential(),get_credentials_for_host()凭据管理Sessionsadd_session(),get_active_sessions()会话管理Vulnerabilitiesadd_vulnerability(),get_vulnerabilities_for_host()漏洞管理三、ContextAssembler - 上下文组装器3.1 核心职责位于context_assembler.py:22-88classContextAssembler: 基于当前状态为 LLM 提示词组装上下文。 从状态存储中提取事实信息结合攻击树的结构 生成包含多种信息的上下文字符串供 LLM 使用。 3.2 assemble() 方法详解# excalibur/memory/context_assembler.py:48-88defassemble(self,node:AttackNode,# 当前活动节点tree:AttackTree,# 完整攻击树mode:str,# 探索模式 (bfs/dfs/hybrid)tdi_value:float0.5,# 当前 TDI 值)-str: 从当前状态构建上下文提示词。 返回包含四个部分的上下文字符串 1. Attack Path - 攻击路径 2. State Facts - 状态事实 3. Mode Guidance - 模式指导 4. Sibling Summaries - 兄弟摘要 3.3 四个上下文部分详解3.3.1 Attack Path攻击路径# excalibur/memory/context_assembler.py:108-151def_build_path_context(self,node,tree)-str: 从根节点遍历到当前节点显示每条路径上的状态。 输出示例 ## Current Attack Path [ACTIVE] Initial reconnaissance of 10.10.10.50 (idabc123) [COMPLETED] Nmap scan completed (iddef456) [ACTIVE] Port 80/tcp HTTP discovered (idghi789) 3.3.2 State Facts状态事实# excalibur/memory/context_assembler.py:153-239def_build_state_context(self,node)-str: 从实体存储构建状态事实部分。 输出示例 ## Known State Facts ### Hosts - 10.10.10.50 (target.local) [OS: Linux] - 10.10.10.51 [OS: Windows] ### Services - 80/tcp http 1.3.3 (hostabc123) - 445/tcp smb 2.0.0 (hostabc123) ### Credentials - admin\administrator [password] valid_for[abc123] ### Active Sessions - shell on hostabc123 privuser (idsess001) ### Vulnerabilities - [exploited] CVE-2021-44228 Log4j RCE 3.3.3 Mode Guidance模式指导# excalibur/memory/context_assembler.py:242-278def_build_mode_context(mode,tdi_value)-str: 构建特定于模式的指导文本。 BFS (广度优先): Breadth-first: enumerate all attack surfaces before going deep... DFS (深度优先): Depth-first: focus on the most promising vector and exploit it fully... Hybrid (混合模式): Hybrid mode (balanced): alternate between breadth and depth based on TDI. 3.3.4 Sibling Summaries兄弟摘要# excalibur/memory/context_assembler.py:280-321def_build_sibling_context(self,node,tree)-str: 构建压缩的兄弟分支摘要。 输出示例 ## Sibling Branch Summaries - **def456** [COMPLETED, TDI0.35]: findings[port 22/tcp] tools[nmap] - **jkl012** [PENDING, TDI0.60]: findings[] tools[] 四、ContextCompressor - 上下文压缩器4.1 压缩策略位于context_compressor.py# 两种压缩阈值classContextCompressor:def__init__(self,ideal_threshold0.6,aggressive_threshold0.8): - ideal_threshold: 理想压缩阈值 (60% 容量时开始压缩) - aggressive_threshold: 激进压缩阈值 (80% 容量时强制压缩) 4.2 压缩流程# 压缩决策逻辑ifcontext_loadideal_threshold:# 理想压缩生成分支摘要替换详细上下文compress(tree,context_load)elifcontext_loadaggressive_threshold:# 激进压缩更激进的摘要策略compress(tree,context_load,aggressiveTrue)五、完整上下文管理流程5.1 数据流图ContextCompressorStateStoreContextAssemblerControllerContextCompressorStateStoreContextAssemblerControlleralt[需要压缩]get_hosts()get_services_for_host()get_credentials()assemble(node, tree, mode, tdi)_build_path_context()_build_state_context()_build_mode_context()_build_sibling_context()context_stringshould_compress(context_load)compress(tree, context_load)5.2 实际使用示例# excalibur/core/controller.py:478-504asyncdef_egats_loop(self,initial_task:str):# ...# 1. UCB 选择节点current_nodeself._planner.select_next_node(tree)# 2. 计算 TDIcontext_load0.0ifself._context_assembler:ctxself._context_assembler.assemble(current_node,tree,reconnaissance)context_loadself._context_assembler.get_context_load(ctx)tdiself._planner.compute_tdi(current_node,tree,context_load)# 3. 选择模式modeself._planner.select_mode(tdi)# 4. 组装上下文提示词context_promptself._context_assembler.assemble(current_node,tree,mode,tdi.value)# 5. 构建查询queryself._build_egats_query(current_node,mode,context_prompt,tdi.value)# 6. 查询 LLMawaitself.backend.query(query)六、上下文管理最佳实践6.1 理解上下文结构# 完整的上下文示例简化版context ## Current Attack Path [ACTIVE] Initial reconnaissance of 10.10.10.50 (idabc123) [COMPLETED] Nmap scan completed (iddef456) [ACTIVE] Port 80/tcp HTTP discovered (idghi789) ## Known State Facts ### Hosts - 10.10.10.50 (target.local) [OS: Linux] - 10.10.10.51 [OS: Windows] ### Services - 80/tcp http 1.3.3 (hostabc123) - 445/tcp smb 2.0.0 (hostabc123) ### Credentials - admin\\administrator [password] valid_for[abc123] ### Active Sessions - shell on hostabc123 privuser (idsess001) ## Exploration Mode: RECONNAISSANCE (TDI0.45) Breadth-first: enumerate all attack surfaces before going deep... ## Sibling Branch Summaries - **jkl012** [PENDING, TDI0.60]: findings[port 22/tcp] tools[nmap] 6.2 调试技巧# 在代码中添加上下文调试输出importlogging loggerlogging.getLogger(__name__)defassemble_debug(self,node,tree,mode,tdi_value0.5):contextself.assemble(node,tree,mode,tdi_value)logger.debug(f CONTEXT (length{len(context)}) )logger.debug(context)logger.debug()returncontext6.3 性能优化建议场景建议大量主机/服务使用get_services_for_host()限定范围避免加载全部数据长会话启用 ContextCompressor定期压缩上下文内存受限使用文件型 StateStore (db_pathworkspace/state.db)七、相关文件索引文件路径功能ContextAssemblerexcalibur/memory/context_assembler.py上下文组装器StateStoreexcalibur/memory/state_store.pySQLite 状态存储ContextCompressorexcalibur/memory/context_compressor.py上下文压缩器BranchSummaryexcalibur/memory/branch_summary.py分支摘要生成Modelsexcalibur/memory/models.py实体数据模型附录快速参考StateStore 常用查询# 获取所有主机hostsstate_store.get_hosts()# 获取特定主机的服务servicesstate_store.get_services_for_host(host_id)# 获取所有凭据credsstate_store.get_credentials()# 获取对特定主机有效的凭据host_credsstate_store.get_credentials_for_host(host_id)# 获取活动会话active_sessionsstate_store.get_active_sessions()ContextAssembler 输出结构# assemble() 返回的上下文结构{attack_path:str,# 攻击路径文本state_facts:str,# 状态事实文本mode_guidance:str,# 模式指导文本sibling_summaries:str# 兄弟摘要文本}# 实际返回的是用 \n\n 连接的四部分字符串