第四章 - 重生之我是AI人：解剖主线一次任务从输入到报告
Khan安全团队
2026年6月21日 12:09
广东
在小说阅读器读本章
去阅读
以下文章来源于威胁情报Z分析
，作者Gachong
威胁情报Z分析
.
国际网络安全威胁情报，地缘政治事件分析。
- 一个故事
用户在 Web UI 输入框敲下：
对 http：//testphp.vulnweb.com 进行渗透测试
接下来 6 分钟里，这套系统里发生了什么？这一章我们按时间顺序走一遍，每一步都给出精确到行号的代码锚点。看完你能复现整个流程，也能定位到任意环节去 debug。
- 步骤 1：API 接收
文件：api/main.py：87-150
Pydantic 模型定义入参：
class PentestStartRequest(BaseModel)： target： str = Field(...， description=\"目标 URL 或 IP\") scope： list[str] = Field(default=[]， description=\"授权范围\") authorized_by： str = Field(...， description=\"授权人\") task_name： Optional[str] = Field(None， description=\"任务名称\") user_intent： Optional[str] = Field(None， description=\"用户原始意图\") options： Optional[dict] = Field(default={}， description=\"额外选项\")
POST /api/pentest/start 收到请求，做三件事：
-
生成 UUID 当 task\_id
-
把任务扔进 ThreadPoolExecutor(Windows 友好，见后面降级设计)
-
立即返回 {task\_id， status： \"started\"} 给前端
为什么不阻塞 HTTP？因为渗透测试跑几分钟，HTTP 早就超时了。
为什么不阻塞 WebSocket？因为任务执行可能在另一台机器、另一个进程。
- 步骤 2：Orchestrator 启动
文件：agents/orchestrator.py：45-82
Orchestrator 类加载，做延迟初始化：
class Orchestrator： def __init__(self)： self.llm = get_llm() # LLM 客户端 self.skill_loader = None # 技能加载器(延迟) self.poc_generator = None # POC 生成器(延迟) self.hexstrike_client = None # MCP 客户端(延迟)
def _init_components(self)： if self.skill_loader is None： from agents.skill_loader import get_skill_loader self.skill_loader = get_skill_loader() self.all_skills = self.skill_loader.list_skills()
关键点：
● LLM 在 \\_init\\_ 就加载(每次都要)
● 技能加载、POC 生成、MCP 客户端是延迟的(用时再加载，启动快)
● 全局单例模式：get\_skill\_loader() 整个进程只调一次
延迟初始化的好处：单元测试时不必真的连 LLM、连 ChromaDB。Mock 一下就行。
- 步骤 3：Recon 阶段
文件：agents/recon\_agent.py
进入 RECON 阶段，调 agents/recon\_agent.py 的 HexStrikeClient：
class HexStrikeClient： def __init__(self， base_url=settings.hexstrike_server_url)： self.base_url = base_url
def execute_command(self， command， category=\"general\")： return requests.post( f\"{self.base_url}/api/command\"， json={\"command\"： command， \"category\"： category}， timeout=180 ).json()
Hati 发 3 个命令到 Kali VM：
# 命令 1：nmap 扫端口nmap -sV -p- http：//testphp.vulnweb.com# 返回： {\"ports\"： [{\"port\"： 80， \"service\"： \"http\"}， {\"port\"： 3306， \"service\"： \"mysql\"}]}
# 命令 2：httpx 识别技术栈httpx -tech-detect -url http：//testphp.vulnweb.com# 返回： {\"tech\"： [\"PHP\"， \"Apache\"， \"MySQL\"]}
# 命令 3：requests 抓首页(主机侧，工具做不了)GET http：//testphp.vulnweb.com HTTP/1.1# 返回： HTML 文本
结果写进 state：
state[\"page_info\"] = { \"url\"： \"http：//testphp.vulnweb.com\"， \"is_login_page\"： False， \"title\"： \"Test Page\"， \"status\"： 200， \"headers\"： {...}}state[\"page_content\"] = \"...\" # HTML 文本前 5000 字符state[\"open_ports\"] = [80， 3306]state[\"tech\"] = [\"php\"， \"apache\"， \"mysql\"]
主机侧的 requests：注意 requests 是 Python 内置的，没走 MCP。简单 GET 不值得绕一圈 VM。复杂的工具调用(sqlmap、nuclei)才走 MCP。
- 步骤 4：ReAct 循环
文件：agents/orchestrator.py：82-400
这是整个项目的大脑。它分两段：
4.1 Think 阶段：LLM 决策
文件：agents/orchestrator.py：82-200
think() 函数被调，做这些事：
-
把目标信息、技术栈、当前阶段、用户意图打包成 prompt
-
调 LLM
-
LLM 输出 JSON 格式决策
核心 prompt 结构：
user_prompt = f\"\"\"## 任务信息目标： {target}当前阶段： {current_phase}用户意图： {user_intent}
## 页面分析信息- URL： {page_info.url}- 技术栈： {tech}- 开放端口： {open_ports}
## 决策选项1. execute_tool： 调用 MCP 工具2. execute_skill： 执行攻击技能3. query_rag： 查询 RAG 知识库4. generate_poc： 生成 POC 验证5. complete： 任务完成
请输出 JSON： {{\"action\"： \"...\"， \"reasoning\"： \"...\"， ...}}\"\"\"
LLM 实际输出：
{ \"action\"： \"execute_skill\"， \"skill_name\"： \"sqli-sql-injection\"， \"target_info\"： { \"url\"： \"http：//testphp.vulnweb.com/artists.php\"， \"parameter\"： \"artist\" }， \"reasoning\"： \"目标开了 3306 端口，且是 PHP+MySQL，优先测 SQL 注入\"}
真实 LLM 输出的 JSON 片段(从 log 里抄出来的)
4.2 Act 阶段：执行决策
文件：agents/orchestrator.py：333-400
act() 函数拿到决策，根据 action 字段分发：
def act(self， state， decision)： action = decision[\"action\"]
if action == \"execute_tool\"： state = self._do_attack_surfaces_discovery(state， decision) elif action == \"execute_skill\"： state = self._do_skill_based_attack(state， decision) elif action == \"query_rag\"： state = self._do_rag_poc_attack(state) elif action == \"generate_poc\"： state = self._do_generate_and_test_poc(state) elif action == \"complete\"： state = advance_phase(state， PentestPhase.COMPLETE)
有意思的细节：解析决策那里有容错。
文件：agents/orchestrator.py：312-331
def _parse_decision(self， response)： # 优先解析 JSON json_match = re.search(r'\\{.*\\}'， response， re.DOTALL) if json_match： try： return json.loads(json_match.group()) except json.JSONDecodeError： pass
# JSON 失败 → 关键词回退 response_lower = response.lower() if \"auth_bypass\" in response_lower： return {\"action\"： \"auth_bypass\"， \"reasoning\"： \"认证测试\"} elif \"skill_based\" in response_lower： return {\"action\"： \"skill_based_attack\"， \"reasoning\"： \"基于技能攻击\"} # ...
为啥要容错？LLM 偶尔会输出\"我建议你用 X\" 这种自然语言，而不是严格 JSON。回退到关键词匹配至少能跑通，不至于一次失败整个任务崩了。
4.3 三种行动路径展开
路径 A：execute\_skill
# Orchestrator 把 LLM 决策交给 SkillLoader 匹配matched = self.skill_loader.match_skills(state[\"page_info\"])# 返回： [{\"name\"： \"sqli-sql-injection\"， \"score\"： 5}， ...]
# 然后 LLM 把技能模板适配成具体 HTTP 请求adapted = self._llm_adapt_skill_to_target(skill_content， target)# 模板： \"id 参数后加 ' OR '1'='1\"# 改写： \"GET /artists.php？artist=1' OR '1'='1\"
# 实际发包response = requests.get(adapted)# 看响应里有没有 SQL 错误关键词if \"SQL syntax\" in response.text or \"MySQL\" in response.text： vuln = {\"type\"： \"SQL Injection\"， \"severity\"： \"high\"， ...} state[\"vulnerabilities\"].append(vuln)
路径 B：execute\_tool
# 直接调 MCP 工具result = self.hexstrike.execute_command( \"nuclei -u http：//target.com -t cves/\"， category=\"scanner\")# MCP 在 VM 里跑 nuclei，把结果返回
路径 C：query\_rag
# 从 RAG 知识库查相关 POCresults = self.rag.query(\"Apache Struts2 RCE\"， n_results=3)# 返回： 3 个最相似的历史 POC
# LLM 改写 POC 适配目标poc = self._llm_adapt_and_test_poc(results， state[\"target\"])
# 实际发包验证response = requests.post(poc[\"url\"]， data=poc[\"payload\"])if poc[\"success_indicator\"] in response.text： vuln = {...}
三种路径都会汇聚到同一个点：把发现的漏洞塞进 state[\"vulnerabilities\"]。
- 步骤 5：漏洞入栈 + 进度推送(贯穿整个 ReAct)
文件：state/progress\_tracker.py + api/websocket.py
每发现一个漏洞、每切换一个阶段，都会：
-
写一条进度到 Redis(progress\_tracker.py)
-
WebSocket 推送给前端(websocket.py)
WebSocket 消息类型：
{type： \"progress\"， phase： \"vuln_scan\"， step： \"正在执行技能 sqli\"}{type： \"vuln_found\"， vulnerability： {...}}{type： \"ai_token\"， content： \"我决定\"}{type： \"complete\"， report_url： \"/reports/xxx.md\"}
WebSocket 真实消息流(从浏览器 devtools 网络面板抓的)
- 步骤 6：报告生成(ReAct 结束后)
文件：agents/report\_agent.py(512 行)
ReAct 跑满 8 轮或 LLM 输出 \"complete\" → 进入 REPORT 阶段。
state[\"vulnerabilities\"] 此时长这样：
{ \"vulnerabilities\"： [ { \"id\"： \"vuln_001\"， \"name\"： \"SQL Injection in artist parameter\"， \"severity\"： \"high\"， \"cvss_score\"： 8.5， \"target\"： \"http：//testphp.vulnweb.com\"， \"url\"： \"/artists.php？artist=1\"， \"parameter\"： \"artist\"， \"evidence\"： { \"request\"： \"GET ...？artist=1' OR '1'='1\"， \"response_snippet\"： \"You have an error in your SQL syntax\" }， \"poc_path\"： \"reports/20260601_xxx_poc_001.txt\"， \"status\"： \"confirmed\" } ]}
ReportAgent 调一次 LLM，把结构化数据写成 Markdown。
输出：reports/{date}\{time}\ 双格式。}\_{task\_id}.{md，json
- 步骤 7：WebSocket 推完成
最后一帧 WebSocket 消息：
{ type： \"complete\"， report_md_url： \"/reports/20260601_143022_target_a1b2c3d4.md\"， report_json_url： \"/reports/20260601_143022_target_a1b2c3d4.json\"， duration_seconds： 372， vulnerabilities_count： 3}
Web UI 收到后，显示报告链接，任务结束。
- 流程时序图
- 一次任务的完整时间线
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
![[渗透测试]一次从XSS到任意文件读取"](/images/random/titlepic/2.jpg)
评论