深入拆解RedAmon：AI驱动自动化红队框架架构解析

// 文件位置: graph_db/schema.py
// RedAmon Neo4j 图数据库核心约束定义（租户隔离的多租户架构）

// 域名唯一性约束（租户隔离：同一域名可在不同项目中存在）
CREATE CONSTRAINT domain_unique IF NOT EXISTS FOR (d:Domain) REQUIRE (d.name, d.user_id, d.project_id) IS UNIQUE;

// 子域名唯一性约束
CREATE CONSTRAINT subdomain_unique IF NOT EXISTS FOR (s:Subdomain) REQUIRE (s.name, s.user_id, s.project_id) IS UNIQUE;

// IP 地址唯一性约束  
CREATE CONSTRAINT ip_unique IF NOT EXISTS FOR (i:IP) REQUIRE (i.address, i.user_id, i.project_id) IS UNIQUE;

// 端口唯一性约束（同一 IP 的同一端口在不同租户间可重复）
CREATE CONSTRAINT port_unique IF NOT EXISTS FOR (p:Port) REQUIRE (p.number, p.protocol, p.ip_address, p.user_id, p.project_id) IS UNIQUE;

// 服务唯一性约束
CREATE CONSTRAINT service_unique IF NOT EXISTS FOR (svc:Service) REQUIRE (svc.name, svc.port_number, svc.ip_address, svc.user_id, svc.project_id) IS UNIQUE;

// 端点唯一性约束（用于 Web 应用攻击面映射）
CREATE CONSTRAINT endpoint_unique IF NOT EXISTS FOR (e:Endpoint) REQUIRE (e.path, e.method, e.baseurl, e.user_id, e.project_id) IS UNIQUE;

// 漏洞唯一性约束
CREATE CONSTRAINT vulnerability_unique IF NOT EXISTS FOR (v:Vulnerability) REQUIRE v.id IS UNIQUE;

// CVE 唯一性约束（全局共享参考数据）
CREATE CONSTRAINT cve_unique IF NOT EXISTS FOR (c:CVE) REQUIRE c.id IS UNIQUE;

// 漏洞节点与关系
MERGE (v:Vulnerability {
    cve: $cve_id, 
    severity: $severity, 
    cvss: $cvss_score,
    description: $description
})
MERGE (srv)-[:HAS_VULNERABILITY]->(v);

// 端点节点（AI 接口）
MERGE (ep:Endpoint {
    path: $endpoint_path,
    method: $http_method,
    ai_interface_type: $ai_type,  // REST / GraphQL / WebSocket
    auth_required: $needs_auth
})
MERGE (srv)-[:EXPOSES_ENDPOINT]->(ep);

// 凭据节点与关系
MERGE (cred:Credential {
    username: $user,
    hash_type: $hash_type,
    crack_time: $time_to_crack
})
MERGE (h)-[:HAS_CREDENTIAL]->(cred);

// 攻击链关系
MERGE (a1:Asset {name: $asset1})
MERGE (a2:Asset {name: $asset2})
MERGE (a1)-[:CAN_COMPROMISE {technique: $mitre_technique, difficulty: $diff_level}]->(a2);

// 元数据标注
MERGE (t:Tag {name: $tag_name, category: $tag_category})
MERGE (v)-[:TAGGED_WITH]->(t);

graph TB
    subgraph "侦察阶段"
        A[输入: 域名/IP] --> B[侦察编排器]
        B --> C[40+ 并行工具]
        C --> D[原始数据]
    end

    subgraph "数据处理"
        D --> E[数据归一化]
        E --> F[Neo4j 图数据库]
        F --> G[攻击面可视化]
    end

    subgraph "AI 决策"
        G --> H[Root Agent]
        H --> I{SG-ReAct 散射}
        I --> J[Fireteam 子智能体]
        J --> K[并行执行]
        K --> L[结果聚集]
    end

    subgraph "执行阶段"
        L --> M[利用阶段]
        M --> N[MCP 工具调用]
        N --> O[Kali 沙箱执行]
        O --> P[后渗透阶段]
        P --> Q[漏洞修复]
    end

    subgraph "反馈循环"
        Q --> R[图数据库更新]
        R --> H
    end

    style H fill:#ff6b6b,stroke:#c92a2a,stroke-width:3px
    style F fill:#4ecdc4,stroke:#0f9d8a,stroke-width:2px
    style O fill:#ffe66d,stroke:#f4a127,stroke-width:2px
    style R fill:#a8e6cf,stroke:#38a169,stroke-width:2px

# 文件位置: agentic/orchestrator.py
# RedAmon Agent Orchestrator - ReAct 风格编排器实现

from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

from state import AgentState, InvokeResponse
from tools import MCPToolsManager, Neo4jToolManager, PhaseAwareToolExecutor

class AgentOrchestrator:
    """
    ReAct 风格智能体编排器 - 渗透测试核心引擎
    
    实现 Thought-Tool-Output 模式，支持：
    - 阶段跟踪（Informational → Exploitation → Post-Exploitation）
    - LLM 管理的待办事项列表
    - 基于检查点的阶段转换审批
    - 完整执行追踪（内存中）
    """
    
    def __init__(self):
        """初始化编排器及配置"""
        # 基础设施环境变量（保持在 docker-compose 中）
        self.neo4j_uri = os.getenv("NEO4J_URI", "bolt://localhost:7687")
        self.neo4j_user = os.getenv("NEO4J_USER", "neo4j") 
        self.neo4j_password = os.getenv("NEO4J_PASSWORD")
        
        # 模型和工具管理
        self.model_name: Optional[str] = None
        self.llm: Optional[BaseChatModel] = None
        self.tool_executor: Optional[PhaseAwareToolExecutor] = None
        self.neo4j_manager: Optional[Neo4jToolManager] = None
        self.graph = None
        
        # 会话管理（支持多用户并发）
        self._streaming_callbacks: dict[str, object] = {}
        self._guidance_queues: dict[str, asyncio.Queue] = {}
        
    asyncdef initialize(self):
        """异步初始化 - 设置 LLM 和工具执行器"""
        if self._initialized:
            return
            
        # 设置 LLM 客户端（根据项目设置配置）
        self.llm = await setup_llm()
        
        # 初始化工具管理器
        self.neo4j_manager = Neo4jToolManager(
            self.neo4j_uri, self.neo4j_user, self.neo4j_password
        )
        
        mcp_manager = MCPToolsManager()  # MCP 工具服务器管理
        self.tool_executor = PhaseAwareToolExecutor(
            self.neo4j_manager, mcp_manager
        )
        
        # 构建 LangGraph 状态机
        self.graph = self._build_graph()
        self._initialized = True
        
    def _build_graph(self) -> StateGraph:
        """构建 LangGraph 状态机 - 定义节点和边"""
        workflow = StateGraph(AgentState)
        
        # 添加核心节点
        workflow.add_node("initialize", initialize_node)
        workflow.add_node("think", think_node)
        workflow.add_node("execute_tool", execute_tool_node)
        workflow.add_node("fireteam_deploy", fireteam_deploy_node)
        workflow.add_node("fireteam_collect", fireteam_collect_node)
        workflow.add_node("await_approval", await_approval_node)
        workflow.add_node("generate_response", generate_response_node)
        
        # 定义状态流转
        workflow.set_entry_point("initialize")
        workflow.add_edge("initialize", "think")
        workflow.add_conditional_edges(
            "think",
            should_deploy_fireteam,
            {
                "execute": "execute_tool",
                "deploy": "fireteam_deploy",
                "end": END
            }
        )
        workflow.add_edge("execute_tool", "think")
        workflow.add_edge("fireteam_deploy", "fireteam_collect")
        workflow.add_edge("fireteam_collect", "think")
        
        # 编译状态机（带检查点支持）
        return workflow.compile(checkpointer=checkpointer)

# 文件位置: agentic/fireteam.py
# Fireteam 专家子智能体定义

from typing import List, Dict, Callable
from enum import Enum

class SpecialistType(Enum):
    """专家类型枚举"""
    NETWORK_AGENT = "network_agent"           # 网络侦察专家
    WEB_AGENT = "web_agent"                   # Web 应用专家
    CREDENTIAL_AGENT = "credential_agent"     # 凭据破解专家
    EXPLOITATION_AGENT = "exploitation_agent"# 漏洞利用专家
    POST_EXPLOIT_AGENT = "post_exploit_agent"# 后渗透专家
    REPORTING_AGENT = "reporting_agent"       # 报告生成专家

class SpecialistAgent:
    """专家子智能体基类"""
    
    def __init__(self, specialist_type: SpecialistType, llm_provider, tool_registry):
        self.type = specialist_type
        self.llm = llm_provider
        self.tools = tool_registry.get_tools_for_specialist(specialist_type)
        self.system_prompt = self._build_system_prompt()
    
    def _build_system_prompt(self) -> str:
        """构建专家专属系统提示词"""
        prompts = {
            SpecialistType.NETWORK_AGENT: """
            你是网络侦察专家。你的职责：
            1. 扫描目标网络段，发现所有活跃主机
            2. 识别开放端口和运行服务
            3. 检测服务版本和潜在漏洞
            4. 绘制网络拓扑图
            
            你可以使用的工具：
            - nmap_scan: 端口扫描和服务识别
            - masscan: 快速大范围端口发现
            - nuclei_scan: 基于 Nuclei 模板的漏洞扫描
            - subfinder_discover: 子域名发现
            
            工作原则：
            - 先做快速扫描（masscan），再做精细扫描（nmap）
            - 优先关注高危端口（21, 22, 80, 443, 3306, 3389, 5432, 8080）
            - 将发现的所有资产结构化记录
            """,
            
            SpecialistType.WEB_AGENT: """
            你是 Web 应用安全专家。你的职责：
            1. 爬取和枚举 Web 应用端点
            2. 检测常见 Web 漏洞（SQL 注入、XSS、RCE 等）
            3. 分析认证和授权机制
            4. 测试 GraphQL 端点安全
            
            你可以使用的工具：
            - katana_crawl: Web 爬虫
            - nuclei_http: HTTP 漏洞扫描
            - graphql_audit: GraphQL 安全测试
            - ffuf_fuzz: 目录和参数模糊测试
            
            工作原则：
            - 优先测试登录页面、文件上传、API 端点
            - 注意 GraphQL 端点的 introspection 暴露
            - 记录所有发现的敏感端点和漏洞
            """,
            
            SpecialistType.CREDENTIAL_AGENT: """
            你是凭据破解专家。你的职责：
            1. 收集目标系统的用户名和哈希
            2. 使用字典和规则进行密码破解
            3. 尝试常见默认凭据
            4. 利用凭据进行横向移动
            
            你可以使用的工具：
            - hydra_crack: 多协议暴力破解
            - hashcat_crack: 哈希破解
            - ssh_login: SSH 登录测试
            - smb_login: SMB 认证测试
            
            工作原则：
            - 优先尝试常见默认凭据（admin:admin, root:123456）
            - 使用行业专用密码字典（如 IoT 设备）
            - 破解成功后立即验证可用性
            - 记录破解时间和破解难度
            """,
            
            SpecialistType.EXPLOITATION_AGENT: """
            你是漏洞利用专家。你的职责：
            1. 分析漏洞可利用性
            2. 选择合适的 Exploit 模块
            3. 执行漏洞利用并获取 Shell
            4. 验证利用成功并建立持久化
            
            你可以使用的工具：
            - metasploit_exploit: Metasploit 框架利用
            - searchsploit: 搜索公开 Exploit
            - exploitdb_poc: Exploit-DB POC 执行
            - sqlmap_inject: SQL 注入利用
            
            工作原则：
            - 优先选择 Metasploit 成熟模块
            - 利用前先验证漏洞存在性
            - 注意利用失败时的清理工作
            - 获取 Shell 后立即建立持久化
            """,
            
            SpecialistType.POST_EXPLOIT_AGENT: """
            你是后渗透专家。你的职责：
            1. 在已攻陷主机上收集信息
            2. 提升权限到 SYSTEM/root
            3. 收集敏感数据和凭据
            4. 建立持久化后门
            
            你可以使用的工具：
            - linpeas_enum: Linux 枚举脚本
            - winpeas_enum: Windows 枚举脚本
            - mimikatz_dump: 凭据转储
            - bloodhound_audit: BloodHound 分析
            
            工作原则：
            - 先做信息收集再做提权
            - 优先寻找内核漏洞和配置错误
            - 收集所有哈希和票据
            - 建立隐蔽的持久化机制
            """,
            
            SpecialistType.REPORTING_AGENT: """
            你是报告生成专家。你的职责：
            1. 汇总所有发现资产和漏洞
            2. 按严重性排序漏洞
            3. 生成修复建议
            4. 生成专业渗透测试报告
            
            你可以使用的工具：
            - graph_query_attack_chain: 查询攻击链
            - vuln_priority_sort: 漏洞优先级排序
            - report_generate_markdown: 生成 Markdown 报告
            - cypherfix_create_pr: 创建修复 PR
            
            工作原则：
            - 漏洞按 CVSS 评分排序
            - 攻击链按 MITRE ATT&CK 映射
            - 修复建议要具体可执行
            - 报告要符合行业标准（PTES、OSSTMM）
            """
        }
        
        return prompts.get(self.type, "你是一个渗透测试专家。")
    
    asyncdef execute(self, task: Dict) -> Dict:
        """执行专家任务"""
        prompt = f"""
        {self.system_prompt}
        
        当前任务：{task['objective']}
        目标上下文：{task['context']}
        
        请制定执行计划并执行工具调用。输出 JSON 格式结果。
        """
        
        # 使用 LLM 生成工具调用序列
        tool_calls = await self.llm.generate_with_tools(
            prompt=prompt,
            tools=self.tools
        )
        
        # 执行工具调用
        results = []
        for call in tool_calls:
            tool_result = await call['tool'].execute(**call['parameters'])
            results.append({
                "tool": call['tool'].name,
                "result": tool_result
            })
        
        return {
            "specialist": self.type.value,
            "task": task['objective'],
            "results": results
        }

class FireteamManager:
    """Fireteam 管理器"""
    
    def __init__(self, llm_provider, tool_registry):
        self.specialists = {}
        for specialist_type in SpecialistType:
            self.specialists[specialist_type] = SpecialistAgent(
                specialist_type, 
                llm_provider, 
                tool_registry
            )
    
    def get_specialist(self, specialist_type: SpecialistType) -> SpecialistAgent:
        """获取指定专家"""
        return self.specialists.get(specialist_type)
    
    asyncdef execute_parallel(self, allocations: Dict[SpecialistType, List[Dict]]) -> List[Dict]:
        """并行执行多个专家任务"""
        tasks = []
        for specialist_type, task_list in allocations.items():
            specialist = self.get_specialist(specialist_type)
            for task in task_list:
                tasks.append(specialist.execute(task))
        
        returnawait asyncio.gather(*tasks)

# 文件位置: mcp/servers/nmap_server.py
# Nmap MCP 工具服务器 - FastMCP 实现

from fastmcp import FastMCP
import subprocess
import shlex
import re
import os

# 去除 ANSI 转义码（终端颜色）的正则表达式
ANSI_ESCAPE = re.compile(r'\x1b\[[0-9;]*[a-zA-Z]')

# 服务器配置
SERVER_NAME = "nmap"
SERVER_HOST = os.getenv("MCP_HOST", "0.0.0.0")
SERVER_PORT = int(os.getenv("NMAP_PORT", "8004"))

mcp = FastMCP(SERVER_NAME)

@mcp.tool()
def execute_nmap(args: str) -> str:
    """
    执行 nmap 网络扫描器（支持任意有效 CLI 参数）
    
    Nmap 是标准的网络发现和安全审计工具。
    用于服务版本检测、操作系统指纹识别和 NSE 脚本执行。
    
    参数：
        args: nmap 的命令行参数（不包含 'nmap' 命令本身）
    
    返回：
        命令输出（stdout + stderr 合并）
    
    示例：
        服务版本检测：
        - "-sV 10.0.0.5 -p 80,443"
        
        激进扫描（版本 + OS + 脚本 + 路由跟踪）：
        - "-A 10.0.0.5 -p 22,80"
        
        默认脚本 + 版本检测：
        - "-sV -sC 10.0.0.5 -p 80,443,8080"
        
        使用 NSE 进行漏洞扫描：
        - "-sV --script vuln 10.0.0.5"
        
        操作系统指纹识别：
        - "-O 10.0.0.5"
        
        特定 NSE 脚本：
        - "--script http-enum 10.0.0.5 -p 80"
        
        UDP 扫描：
        - "-sU 10.0.0.5 --top-ports 20"
    """
    try:
        cmd_args = shlex.split(args)
        result = subprocess.run(
            ["nmap"] + cmd_args,
            capture_output=True,
            text=True,
            timeout=600
        )
        # 去除 ANSI 转义码
        output = ANSI_ESCAPE.sub('', result.stdout)
        if result.stderr:
            clean_stderr = ANSI_ESCAPE.sub('', result.stderr)
            if clean_stderr.strip():
                output += f"\n[STDERR]: {clean_stderr}"
        return output if output.strip() else"[INFO] No results returned"
    except subprocess.TimeoutExpired:
        return"[ERROR] 命令超时（600秒）。考虑减少扫描端口或使用 --host-timeout。"
    except FileNotFoundError:
        return"[ERROR] nmap 命令未找到。请确保在 Kali 沙箱环境中运行。"
    except Exception as e:
        returnf"[ERROR] 执行失败: {str(e)}"

graph LR
    subgraph "Agent 层"
        A[Root Agent]
        B[Network Specialist]
        C[Web Specialist]
    end

    subgraph "MCP 工具层"
        D[Nmap Server]
        E[Nuclei Server]
        F[Metasploit Server]
        G[Hydra Server]
    end

    subgraph "Kali 沙箱"
        H[/workspace 目录]
        I[工具执行环境]
        J[结果输出]
    end

    A --> B
    A --> C
    B --> D
    B --> E
    C --> F
    C --> G
    D --> H
    E --> H
    F --> H
    G --> H
    H --> I
    I --> J
    J --> D
    J --> E
    J --> F
    J --> G

    style A fill:#ff6b6b,stroke:#c92a2a,stroke-width:3px
    style H fill:#ffe66d,stroke:#f4a127,stroke-width:2px
    style D fill:#4ecdc4,stroke:#0f9d8a,stroke-width:2px
    style E fill:#4ecdc4,stroke:#0f9d8a,stroke-width:2px
    style F fill:#4ecdc4,stroke:#0f9d8a,stroke-width:2px
    style G fill:#4ecdc4,stroke:#0f9d8a,stroke-width:2px

# 文件位置: graph_db/neo4j_interface.py
# Neo4j 图数据库操作接口

from neo4j import AsyncGraphDatabase
from typing import Dict, List, Optional

class Neo4jInterface:
    """Neo4j 图数据库操作接口"""
    
    def __init__(self, uri: str, user: str, password: str):
        self.driver = AsyncGraphDatabase.driver(uri, auth=(user, password))
    
    asyncdef close(self):
        """关闭数据库连接"""
        await self.driver.close()
    
    asyncdef insert_recon_data(self, data: Dict) -> None:
        """
        插入侦察数据到图数据库
        
        参数说明：
        - data: 侦察结果，包含子域名、主机、端口、服务等信息
        """
        asyncwith self.driver.session() as session:
            # 处理域名和子域名
            for subdomain in data.get('subdomains', []):
                await session.run("""
                    MERGE (d:Domain {name: $domain})
                    MERGE (s:Subdomain {name: $subdomain})
                    MERGE (s)-[:BELONGS_TO]->(d)
                    """, 
                    domain=data['target'],
                    subdomain=subdomain
                )
            
            # 处理主机和端口
            for host in data.get('hosts', []):
                await session.run("""
                    MERGE (h:Host {ip: $ip})
                    SET h.os = $os, h.mac = $mac
                    """,
                    ip=host['ip'],
                    os=host.get('os'),
                    mac=host.get('mac')
                )
                
                for port in host.get('ports', []):
                    await session.run("""
                        MERGE (h:Host {ip: $ip})
                        MERGE (p:Port {number: $port, protocol: $protocol})
                        MERGE (h)-[:HAS_OPEN_PORT]->(p)
                        SET p.status = $status, p.service = $service
                        """,
                        ip=host['ip'],
                        port=port['port'],
                        protocol=port.get('protocol', 'tcp'),
                        status=port.get('status', 'open'),
                        service=port.get('service')
                    )
            
            # 处理漏洞
            for vuln in data.get('vulnerabilities', []):
                await session.run("""
                    MERGE (v:Vulnerability {cve: $cve})
                    SET v.severity = $severity, v.cvss = $cvss,
                        v.description = $description, v.references = $refs
                    """,
                    cve=vuln.get('cve'),
                    severity=vuln.get('severity'),
                    cvss=vuln.get('cvss'),
                    description=vuln.get('description'),
                    refs=vuln.get('references', [])
                )
    
    asyncdef query_attack_surface(self, target: str) -> Dict:
        """
        查询目标的完整攻击面
        
        返回值：
        - 攻击面图的结构化快照
        """
        asyncwith self.driver.session() as session:
            # 查询所有子域名
            subdomains_result = await session.run("""
                MATCH (d:Domain {name: $domain})<-[:BELONGS_TO]-(s:Subdomain)
                RETURN s.name as subdomain
                """,
                domain=target
            )
            subdomains = [record['subdomain'] asyncfor record in subdomains_result]
            
            # 查询所有主机和端口
            hosts_result = await session.run("""
                MATCH (h:Host)<-[:RESOLVES_TO]-(:Subdomain)-[:BELONGS_TO]->(:Domain {name: $domain})
                OPTIONAL MATCH (h)-[:HAS_OPEN_PORT]->(p:Port)
                RETURN h.ip as ip, h.os as os, 
                       collect({port: p.number, protocol: p.protocol, service: p.service}) as ports
                """,
                domain=target
            )
            hosts = [record asyncfor record in hosts_result]
            
            # 查询所有漏洞
            vulns_result = await session.run("""
                MATCH (d:Domain {name: $domain})<-[:BELONGS_TO]-(s:Subdomain)-[:RESOLVES_TO]->(h:Host)
                MATCH (h)-[:HAS_OPEN_PORT]->(p:Port)-[:RUNS_SERVICE]->(srv:Service)
                MATCH (srv)-[:HAS_VULNERABILITY]->(v:Vulnerability)
                RETURN DISTINCT v.cve as cve, v.severity as severity, 
                              v.cvss as cvss, v.description as description
                """,
                domain=target
            )
            vulns = [record asyncfor record in vulns_result]
            
            return {
                "target": target,
                "subdomains": subdomains,
                "hosts": hosts,
                "vulnerabilities": vulns
            }
    
    asyncdef find_attack_chains(self, target: str, max_depth: int = 3) -> List[Dict]:
        """
        查询所有可能的攻击链
        
        参数说明：
        - target: 目标域名
        - max_depth: 攻击链最大深度（跳数）
        
        返回值：
        - 所有可能的攻击路径列表，每条路径包含：
          - 起点（可利用的漏洞或凭据）
          - 中间节点（可攻陷的主机）
          - 终点（高价值目标）
        """
        asyncwith self.driver.session() as session:
            result = await session.run("""
                MATCH path = (start:Asset {target: $target})-[:CAN_COMPROMISE*1..{max_depth}]->(end:Asset)
                WHERE start.value > 0 AND end.value > 0
                RETURN path,
                       reduce(score = 1.0, r IN relationships(path) | score * r.difficulty) as success_prob,
                       length(path) as depth
                ORDER BY success_prob DESC, depth ASC
                LIMIT 10
                """,
                target=target,
                max_depth=max_depth
            )
            
            attack_chains = []
            asyncfor record in result:
                path = record['path']
                chain = {
                    "success_probability": record['success_prob'],
                    "depth": record['depth'],
                    "nodes": [],
                    "relationships": []
                }
                
                for node in path.nodes:
                    chain['nodes'].append({
                        "label": list(node.labels)[0],
                        "properties": dict(node)
                    })
                
                for rel in path.relationships:
                    chain['relationships'].append({
                        "type": rel.type,
                        "properties": dict(rel)
                    })
                
                attack_chains.append(chain)
            
            return attack_chains
    
    asyncdef track_compromised_host(self, host_ip: str, credential: Dict, technique: str) -> None:
        """
        记录已攻陷主机和获取的凭据
        
        参数说明：
        - host_ip: 已攻陷主机的 IP
        - credential: 获取的凭据（用户名、密码、哈希等）
        - technique: 利用技术（MITRE ATT&CK ID）
        """
        asyncwith self.driver.session() as session:
            # 创建已攻陷主机节点
            await session.run("""
                MERGE (h:Host {ip: $ip})
                SET h.compromised = true,
                    h.compromised_at = datetime(),
                    h.compromised_via = $technique
                """,
                ip=host_ip,
                technique=technique
            )
            
            # 创建凭据节点并关联到主机
            await session.run("""
                MERGE (h:Host {ip: $ip})
                MERGE (c:Credential {username: $user, hash: $hash})
                MERGE (h)-[:HAS_CREDENTIAL]->(c)
                SET c.crack_time = $crack_time, c.type = $type
                """,
                ip=host_ip,
                user=credential.get('username'),
                hash=credential.get('hash'),
                crack_time=credential.get('crack_time'),
                type=credential.get('type', 'password')
            )
    
    asyncdef get_next_priority_targets(self, target: str, limit: int = 5) -> List[Dict]:
        """
        获取下一阶段优先目标列表
        
        参数说明：
        - target: 原始目标域名
        - limit: 返回目标数量
        
        返回值：
        - 按优先级排序的目标列表，每个目标包含：
          - 主机 IP
          - 优先级分数
          - 推荐理由
        """
        asyncwith self.driver.session() as session:
            result = await session.run("""
                MATCH (d:Domain {name: $domain})<-[:BELONGS_TO]-(s:Subdomain)-[:RESOLVES_TO]->(h:Host)
                WHERE NOT h.compromised = true
                OPTIONAL MATCH (h)-[:HAS_OPEN_PORT]->(p:Port)-[:RUNS_SERVICE]->(srv:Service)
                OPTIONAL MATCH (srv)-[:HAS_VULNERABILITY]->(v:Vulnerability)
                
                WITH h, COUNT(DISTINCT srv) as service_count,
                     COUNT(DISTINCT v) as vuln_count,
                     SUM(CASE WHEN v.severity = 'CRITICAL' THEN 1 ELSE 0 END) as critical_count
                
                RETURN h.ip as ip,
                       service_count as exposed_services,
                       vuln_count as total_vulnerabilities,
                       critical_count as critical_vulns,
                       (service_count * 0.3 + vuln_count * 0.5 + critical_count * 0.2) as priority_score
                ORDER BY priority_score DESC
                LIMIT $limit
                """,
                domain=target,
                limit=limit
            )
            
            targets = []
            asyncfor record in result:
                targets.append({
                    "ip": record['ip'],
                    "priority_score": record['priority_score'],
                    "exposed_services": record['exposed_services'],
                    "total_vulnerabilities": record['total_vulnerabilities'],
                    "critical_vulnerabilities": record['critical_vulns'],
                    "recommendation": f"主机暴露 {record['exposed_services']} 个服务，"
                                    f"发现 {record['total_vulnerabilities']} 个漏洞，"
                                    f"其中 {record['critical_vulns']} 个高危漏洞"
                })
            
            return targets

# 文件位置: agentic/cypherfix_codefix/orchestrator.py
# CypherFix 自动漏洞修复编排器 - 纯 ReAct 循环

import asyncio
import json
import logging
import os
from typing import Optional

import httpx

from .state import CodeFixState, CodeFixSettings
from .tools import CODEFIX_TOOLS
from .tools.github_repo import GitHubRepoManager
from .prompts.system import build_codefix_system_prompt
from .project_settings import load_cypherfix_settings

logger = logging.getLogger(__name__)

TOOL_HANDLERS = {
    "github_glob": github_glob,        # 模式匹配文件搜索
    "github_grep": github_grep,        # 内容搜索文件
    "github_read": github_read,        # 读取文件内容
    "github_edit": github_edit,        # 编辑文件
    "github_write": github_write,      # 写入文件
    "github_bash": github_bash,        # 执行 bash 命令
    "github_list_dir": github_list_dir,  # 列出目录
    "github_symbols": github_symbols,  # 符号搜索
    "github_find_definition": github_find_definition,   # 查找定义
    "github_find_references": github_find_references,   # 查找引用
    "github_repo_map": github_repo_map, # 仓库映射
}

# 需要顺序执行的工具（不能并行）
SEQUENTIAL_TOOLS = {"github_edit", "github_write", "github_bash"}

class CodeFixOrchestrator:
    """纯 ReAct 循环编排器 - LLM 是唯一控制器"""
    
    def __init__(self, state: CodeFixState, callback):
        self.state = state
        self.callback = callback
        self.repo_manager: Optional[GitHubRepoManager] = None
        self.llm_client = None
        self.approval_future: Optional[asyncio.Future] = None
        self.guidance_messages: list[str] = []
        self.max_output_chars = 20000
        self.context_compact_threshold = 0.80
        self.last_llm_text: str = ""
        
    asyncdef run(self, remediation_id: str):
        """主入口 - CypherFix 会话启动"""
        SEP = "=" * 80
        logger.info(f"\n{SEP}\n  CODEFIX SESSION START — {remediation_id}\n{SEP}")
        
        # 加载设置
        settings_dict = await load_cypherfix_settings(self.state.project_id)
        if settings_dict:
            self.state.settings = CodeFixSettings(**{
                k: v for k, v in settings_dict.items()
                if k in CodeFixSettings.model_fields
            })
            
        # 构建系统提示词（包含工具描述）
        system_prompt = build_codefix_system_prompt(self.state.settings)
        
        # 初始化 GitHub 仓库管理器
        if self.state.settings.github_repo:
            self.repo_manager = GitHubRepoManager(
                self.state.settings.github_repo,
                self.state.settings.default_branch
            )
            await self.repo_manager.clone_repo()
            
        # ReAct 循环：思考 → 工具调用 → 观察 → 迭代
        iteration = 0
        while iteration < self.state.settings.max_iterations:
            iteration += 1
            
            # 思考阶段：LLM 分析当前状态并决定下一步
            thought = await self._think_phase(system_prompt, iteration)
            
            # 工具执行阶段：执行 LLM 决定的工具调用
            tool_result = await self._tool_execution_phase(thought)
            
            # 判断是否完成
            if self._is_fix_complete(tool_result):
                break
                
        # 会话结束
        logger.info(f"\n{SEP}\n  CODEFIX SESSION END — {remediation_id}\n{SEP}")
        
    asyncdef _think_phase(self, system_prompt: str, iteration: int) -> dict:
        """思考阶段 - LLM 分析并决策"""
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"第 {iteration} 轮迭代"}
        ]
        
        # 调用 LLM 生成决策（包含工具调用）
        response = await self.llm_client.chat.completions.create(
            model=self.state.settings.model,
            messages=messages,
            tools=self._get_tool_schemas(),
            tool_choice="auto"
        )
        
        return response
        
    asyncdef _tool_execution_phase(self, thought: dict) -> dict:
        """工具执行阶段 - 执行 LLM 决定的工具调用"""
        ifnot thought.tool_calls:
            return {"status": "no_tools"}
            
        results = []
        for tool_call in thought.tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)
            
            # 执行工具
            handler = TOOL_HANDLERS.get(tool_name)
            if handler:
                result = await handler(**tool_args)
                results.append({
                    "tool": tool_name,
                    "result": result
                })
                
        return {"results": results}

# 文件位置: docker-compose.yml
# RedAmon 统一 Docker Compose 配置
# 启动: docker compose up -d
# 停止: docker compose down  
# 日志: docker compose logs -f
# 构建: docker compose --profile tools build

services:
# ===========================================================================
# 数据库服务
# ===========================================================================

postgres:
    image:postgres:16-alpine
    container_name:redamon-postgres
    environment:
      POSTGRES_USER:${POSTGRES_USER:-redamon}
      POSTGRES_PASSWORD:${POSTGRES_PASSWORD:-redamon_secret}
      POSTGRES_DB:${POSTGRES_DB:-redamon}
    ports:
      -"${POSTGRES_PORT:-5432}:5432"
    volumes:
      -postgres_data:/var/lib/postgresql/data
    restart:unless-stopped
    healthcheck:
      test:["CMD-SHELL","pg_isready -U ${POSTGRES_USER:-redamon} -d ${POSTGRES_DB:-redamon}"]
      interval:10s
      timeout:5s
      retries:5
    networks:
      -redamon

neo4j:
    image:neo4j:5.26-community
    container_name:redamon-neo4j
    environment:
      NEO4J_AUTH:neo4j/${NEO4J_PASSWORD:-changeme123}
      NEO4J_PLUGINS:'["apoc"]'
      NEO4J_dbms_security_procedures_unrestricted:apoc.*
      NEO4J_dbms_security_procedures_allowlist:apoc.*
    ports:
      -"${NEO4J_HTTP_PORT:-7474}:7474"# HTTP
      -"${NEO4J_BOLT_PORT:-7687}:7687"   # Bolt
    volumes:
      -neo4j_data:/data
      -neo4j_logs:/logs
    restart:unless-stopped
    healthcheck:
      test:["CMD","wget","-q","--spider","http://localhost:7474"]
      interval:15s
      timeout:10s
      retries:10
      start_period:30s
    networks:
      -redamon

# ===========================================================================
# 后端服务
# ===========================================================================

recon-orchestrator:
    build:./recon_orchestrator
    container_name:redamon-recon-orchestrator
    ports:
      -"${RECON_ORCH_PORT:-8010}:8010"
    volumes:
      -/var/run/docker.sock:/var/run/docker.sock# Docker-in-Docker 支持
      -./recon_orchestrator:/app
      -./recon:/app/recon:ro
    environment:
      RECON_IMAGE:redamon-recon:latest
      NEO4J_URI:bolt://localhost:7687
      NEO4J_USER:neo4j
      NEO4J_PASSWORD:${NEO4J_PASSWORD:-changeme123}
      AGENT_API_URL:http://localhost:${AGENT_PORT:-8090}
      INTERNAL_API_KEY:${INTERNAL_API_KEY:-changeme}
    restart:unless-stopped
    networks:
      -redamon

kali-sandbox:
    build:
      context:./mcp
      dockerfile:kali-sandbox/Dockerfile
      network:host
    container_name:redamon-kali
    cap_add:
      -NET_ADMIN    # 允许网络扫描
      -NET_RAW
      -SYS_PTRACE
    ports:
      -"${MCP_NETWORK_RECON_PORT:-8000}:8000"
      -"${MCP_NUCLEI_PORT:-8002}:8002"
      -"${MCP_METASPLOIT_PORT:-8003}:8003"
      -"${MCP_NMAP_PORT:-8004}:8004"
      -"${MCP_PLAYWRIGHT_PORT:-8005}:8005"
      -"4444:4444"     # Metasploit 监听器
    volumes:
      -./mcp/servers:/opt/mcp_servers
      -./agentic/agent-workspace:/workspace# 共享工作空间
    environment:
      PYTHONUNBUFFERED:"1"
      MCP_TRANSPORT:sse
      NEO4J_URI:bolt://neo4j:7687
      NEO4J_USER:neo4j
      NEO4J_PASSWORD:${NEO4J_PASSWORD:-changeme123}
      REDAMON_AGENT_URL:http://agent:8080
    restart:unless-stopped
    networks:
      redamon:
      pentest-net:

agent:
    build:
      context:.
      dockerfile:agentic/Dockerfile
      args:
        SKIP_KB:"${SKIP_KB:-false}"
    container_name:redamon-agent
    ports:
      -"${AGENT_PORT:-8090}:8080"
    environment:
      # Neo4j 连接（内部 Docker 网络）
      NEO4J_URI:bolt://neo4j:7687
      NEO4J_USER:neo4j
      NEO4J_PASSWORD:${NEO4J_PASSWORD:-changeme123}
      
      # MCP 工具服务器（内部 Docker 网络）
      MCP_NETWORK_RECON_URL:http://kali-sandbox:8000/sse
      MCP_NUCLEI_URL:http://kali-sandbox:8002/sse
      MCP_NMAP_URL:http://kali-sandbox:8004/sse
      MCP_METASPLOIT_URL:http://kali-sandbox:8003/sse
      
      # PostgreSQL（检查点持久化）
      DATABASE_URL:"postgresql://${POSTGRES_USER:-redamon}:${POSTGRES_PASSWORD:-redamon_secret}@postgres:5432/${POSTGRES_DB:-redamon}"
      PERSISTENT_CHECKPOINTER:"true"
      
      # Webapp API（内部 Docker 网络）
      WEBAPP_API_URL:http://webapp:3000
      INTERNAL_API_KEY:${INTERNAL_API_KEY:-changeme}
    volumes:
      -./agentic/logs:/app/logs
      -./agentic/skills:/app/skills:ro
      -./agentic/agent-workspace:/workspace# 每项目工作空间
    depends_on:
      neo4j:
        condition:service_healthy
      kali-sandbox:
        condition:service_healthy
      postgres:
        condition:service_healthy
    restart:unless-stopped
    networks:
      -redamon

# ===========================================================================
# 前端服务
# ===========================================================================

webapp:
    build:
      context:./webapp
      args:
        REDAMON_VERSION:"${REDAMON_VERSION:-0.0.0}"
    container_name:redamon-webapp
    ports:
      -"${WEBAPP_PORT:-3000}:3000"
    environment:
      # PostgreSQL (Prisma)
      DATABASE_URL:"postgresql://${POSTGRES_USER:-redamon}:${POSTGRES_PASSWORD:-redamon_secret}@postgres:5432/${POSTGRES_DB:-redamon}"
      # Neo4j (服务端图查询)
      NEO4J_URI:bolt://neo4j:7687
      NEO4J_USER:neo4j
      NEO4J_PASSWORD:${NEO4J_PASSWORD:-changeme123}
      # 后端服务（服务端 API 路由 → 内部 Docker 网络）
      RECON_ORCHESTRATOR_URL:http://recon-orchestrator:8010
      AGENT_API_URL:http://agent:8080
      AGENT_WS_URL:ws://agent:8080/ws/agent
    volumes:
      -./recon/output:/data/recon-output:ro
      -./mcp/nuclei-templates:/data/nuclei-templates:rw
      -report_data:/data/reports
    depends_on:
      postgres:
        condition:service_healthy
      neo4j:
        condition:service_healthy
    restart:unless-stopped
    networks:
      -redamon

networks:
redamon:
    name:redamon-network
    driver:bridge
pentest-net:
    driver:bridge
    ipam:
      config:
        -subnet:172.28.0.0/16

volumes:
postgres_data:
neo4j_data:
report_data:

# 文件位置: agentic/llm_providers.py
# 多 LLM 提供商统一接口

from abc import ABC, abstractmethod
from typing import List, Dict, Optional
import asyncio

class LLMProvider(ABC):
    """LLM 提供商抽象基类"""
    
    @abstractmethod
    asyncdef generate(self, prompt: str, **kwargs) -> str:
        """生成文本"""
        pass
    
    @abstractmethod
    asyncdef generate_with_tools(self, prompt: str, tools: List[Dict], **kwargs) -> Dict:
        """生成文本并调用工具"""
        pass
    
    @abstractmethod
    asyncdef stream_generate(self, prompt: str, **kwargs):
        """流式生成文本"""
        pass

class AnthropicProvider(LLMProvider):
    """Anthropic Claude 提供商"""
    
    def __init__(self, api_key: str, model: str = "claude-opus-4-20250514"):
        self.api_key = api_key
        self.model = model
        self.client = self._initialize_client()
    
    def _initialize_client(self):
        import anthropic
        return anthropic.AsyncAnthropic(api_key=self.api_key)
    
    asyncdef generate(self, prompt: str, **kwargs) -> str:
        """生成文本"""
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=kwargs.get('max_tokens', 4096),
            temperature=kwargs.get('temperature', 0.7),
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        
        return response.content[0].text
    
    asyncdef generate_with_tools(self, prompt: str, tools: List[Dict], **kwargs) -> Dict:
        """生成文本并调用工具"""
        # 转换工具定义为 Anthropic 格式
        anthropic_tools = []
        for tool in tools:
            anthropic_tools.append({
                "name": tool['name'],
                "description": tool['description'],
                "input_schema": {
                    "type": "object",
                    "properties": tool['parameters'],
                    "required": tool.get('required', [])
                }
            })
        
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=kwargs.get('max_tokens', 4096),
            temperature=kwargs.get('temperature', 0.7),
            messages=[
                {"role": "user", "content": prompt}
            ],
            tools=anthropic_tools
        )
        
        # 解析响应
        result = {
            "content": response.content[0].text if response.content else"",
            "tool_calls": []
        }
        
        for block in response.content:
            if block.type == "tool_use":
                result["tool_calls"].append({
                    "id": block.id,
                    "name": block.name,
                    "parameters": block.input
                })
        
        return result
    
    asyncdef stream_generate(self, prompt: str, **kwargs):
        """流式生成文本"""
        stream = await self.client.messages.create(
            model=self.model,
            max_tokens=kwargs.get('max_tokens', 4096),
            temperature=kwargs.get('temperature', 0.7),
            messages=[
                {"role": "user", "content": prompt}
            ],
            stream=True
        )
        
        asyncfor event in stream:
            if event.type == "content_block_delta":
                yield event.delta.text

class OpenAIProvider(LLMProvider):
    """OpenAI GPT 提供商"""
    
    def __init__(self, api_key: str, model: str = "gpt-4-turbo"):
        self.api_key = api_key
        self.model = model
        self.client = self._initialize_client()
    
    def _initialize_client(self):
        import openai
        return openai.AsyncOpenAI(api_key=self.api_key)
    
    asyncdef generate(self, prompt: str, **kwargs) -> str:
        """生成文本"""
        response = await self.client.chat.completions.create(
            model=self.model,
            max_tokens=kwargs.get('max_tokens', 4096),
            temperature=kwargs.get('temperature', 0.7),
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        
        return response.choices[0].message.content
    
    asyncdef generate_with_tools(self, prompt: str, tools: List[Dict], **kwargs) -> Dict:
        """生成文本并调用工具"""
        # 转换工具定义为 OpenAI 格式
        openai_tools = []
        for tool in tools:
            openai_tools.append({
                "type": "function",
                "function": {
                    "name": tool['name'],
                    "description": tool['description'],
                    "parameters": {
                        "type": "object",
                        "properties": tool['parameters'],
                        "required": tool.get('required', [])
                    }
                }
            })
        
        response = await self.client.chat.completions.create(
            model=self.model,
            max_tokens=kwargs.get('max_tokens', 4096),
            temperature=kwargs.get('temperature', 0.7),
            messages=[
                {"role": "user", "content": prompt}
            ],
            tools=openai_tools
        )
        
        # 解析响应
        message = response.choices[0].message
        result = {
            "content": message.content or"",
            "tool_calls": []
        }
        
        if message.tool_calls:
            for call in message.tool_calls:
                result["tool_calls"].append({
                    "id": call.id,
                    "name": call.function.name,
                    "parameters": json.loads(call.function.arguments)
                })
        
        return result
    
    asyncdef stream_generate(self, prompt: str, **kwargs):
        """流式生成文本"""
        stream = await self.client.chat.completions.create(
            model=self.model,
            max_tokens=kwargs.get('max_tokens', 4096),
            temperature=kwargs.get('temperature', 0.7),
            messages=[
                {"role": "user", "content": prompt}
            ],
            stream=True
        )
        
        asyncfor chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

class OpenRouterProvider(LLMProvider):
    """OpenRouter 多模型提供商"""
    
    def __init__(self, api_key: str, model: str = "anthropic/claude-opus-4"):
        self.api_key = api_key
        self.model = model
        self.base_url = "https://openrouter.ai/api/v1"
        self.client = self._initialize_client()
    
    def _initialize_client(self):
        import openai
        return openai.AsyncOpenAI(
            api_key=self.api_key,
            base_url=self.base_url
        )
    
    # OpenRouter 使用与 OpenAI 兼容的接口
    asyncdef generate(self, prompt: str, **kwargs) -> str:
        """生成文本"""
        response = await self.client.chat.completions.create(
            model=self.model,
            max_tokens=kwargs.get('max_tokens', 4096),
            temperature=kwargs.get('temperature', 0.7),
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        
        return response.choices[0].message.content
    
    asyncdef generate_with_tools(self, prompt: str, tools: List[Dict], **kwargs) -> Dict:
        """生成文本并调用工具（OpenRouter 部分模型支持）"""
        # 类似 OpenAI 的实现
        pass

class LLMFactory:
    """LLM 提供商工厂"""
    
    @staticmethod
    def create(provider: str, config: Dict) -> LLMProvider:
        """创建 LLM 提供商实例"""
        if provider == "anthropic":
            return AnthropicProvider(
                api_key=config['api_key'],
                model=config.get('model', 'claude-opus-4-20250514')
            )
        elif provider == "openai":
            return OpenAIProvider(
                api_key=config['api_key'],
                model=config.get('model', 'gpt-4-turbo')
            )
        elif provider == "openrouter":
            return OpenRouterProvider(
                api_key=config['api_key'],
                model=config.get('model', 'anthropic/claude-opus-4')
            )
        else:
            raise ValueError(f"不支持的 LLM 提供商: {provider}")

// 查询所有可从边界主机横向移动到数据库服务器的攻击路径
MATCH path = (boundary:IP {exposed:true})-[:RESOLVES_TO|HAS_OPEN_PORT|RUNS_SERVICE*1..3]->(db:IP)
WHERE db.role = 'database'
RETURN path, 
       [n IN nodes(path) | n.address] as attack_path,
       [r IN relationships(path) | r.type] as techniques
ORDER BY length(path) ASC
LIMIT 5