MCP 在攻防模拟中的应用

# mcp_attack_manager.py
from typing import Dict, List, Optional
import logging
from datetime import datetime

class AttackScenario:
    def __init__(self, scenario_id: str, name: str, description: str, 
                 attack_type: str, target: str, payload: Dict, 
                 expected_result: str, severity: str):
        self.scenario_id = scenario_id
        self.name = name
        self.description = description
        self.attack_type = attack_type
        self.target = target
        self.payload = payload
        self.expected_result = expected_result
        self.severity = severity
        self.created_at = datetime.now()
        self.updated_at = datetime.now()
        
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "scenario_id": self.scenario_id,
            "name": self.name,
            "description": self.description,
            "attack_type": self.attack_type,
            "target": self.target,
            "payload": self.payload,
            "expected_result": self.expected_result,
            "severity": self.severity,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat()
        }

class AttackResult:
    def __init__(self, result_id: str, scenario_id: str, 
                 start_time: datetime, end_time: datetime, 
                 status: str, actual_result: str, 
                 is_success: bool, details: Dict):
        self.result_id = result_id
        self.scenario_id = scenario_id
        self.start_time = start_time
        self.end_time = end_time
        self.status = status
        self.actual_result = actual_result
        self.is_success = is_success
        self.details = details
        self.duration = (end_time - start_time).total_seconds()
        
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "result_id": self.result_id,
            "scenario_id": self.scenario_id,
            "start_time": self.start_time.isoformat(),
            "end_time": self.end_time.isoformat(),
            "status": self.status,
            "actual_result": self.actual_result,
            "is_success": self.is_success,
            "details": self.details,
            "duration": self.duration
        }

class MCPAttackManager:
    def __init__(self):
        self.scenarios = {}
        self.results = {}
        self.logger = logging.getLogger("mcp_attack_manager")
    
    def add_scenario(self, scenario: AttackScenario):
        """添加攻击场景"""
        self.scenarios[scenario.scenario_id] = scenario
        self.logger.info(f"Added attack scenario: {scenario.scenario_id}")
    
    def remove_scenario(self, scenario_id: str):
        """移除攻击场景"""
        if scenario_id in self.scenarios:
            del self.scenarios[scenario_id]
            self.logger.info(f"Removed attack scenario: {scenario_id}")
    
    def update_scenario(self, scenario: AttackScenario):
        """更新攻击场景"""
        scenario.updated_at = datetime.now()
        self.scenarios[scenario.scenario_id] = scenario
        self.logger.info(f"Updated attack scenario: {scenario.scenario_id}")
    
    def get_scenario(self, scenario_id: str) -> Optional[AttackScenario]:
        """获取攻击场景"""
        return self.scenarios.get(scenario_id)
    
    def get_all_scenarios(self) -> List[Dict]:
        """获取所有攻击场景"""
        return [scenario.to_dict() for scenario in self.scenarios.values()]
    
    def add_result(self, result: AttackResult):
        """添加攻击结果"""
        self.results[result.result_id] = result
        self.logger.info(f"Added attack result: {result.result_id}")
    
    def get_result(self, result_id: str) -> Optional[AttackResult]:
        """获取攻击结果"""
        return self.results.get(result_id)
    
    def get_scenario_results(self, scenario_id: str) -> List[Dict]:
        """获取特定场景的攻击结果"""
        return [result.to_dict() for result in self.results.values() 
                if result.scenario_id == scenario_id]
    
    def get_all_results(self) -> List[Dict]:
        """获取所有攻击结果"""
        return [result.to_dict() for result in self.results.values()]
    
    def run_scenario(self, scenario_id: str) -> str:
        """运行攻击场景"""
        self.logger.info(f"Running attack scenario: {scenario_id}")
        
        # 这里简化处理，实际应调用攻击执行模块
        # 生成结果 ID
        result_id = f"result_{scenario_id}_{datetime.now().timestamp()}"
        
        # 创建攻击结果
        result = AttackResult(
            result_id=result_id,
            scenario_id=scenario_id,
            start_time=datetime.now(),
            end_time=datetime.now(),
            status="completed",
            actual_result="攻击执行成功",
            is_success=True,
            details={"message": "攻击执行完成"}
        )
        
        # 添加攻击结果
        self.add_result(result)
        
        self.logger.info(f"Attack scenario {scenario_id} completed, result_id: {result_id}")
        return result_id

# mcp_scenario_generator.py
from typing import Dict, List, Optional
import logging
from datetime import datetime
import random
import json

class ThreatIntelligence:
    def __init__(self, threat_id: str, threat_type: str, 
                 description: str, severity: str, 
                 attack_vector: str, mitigation: str):
        self.threat_id = threat_id
        self.threat_type = threat_type
        self.description = description
        self.severity = severity
        self.attack_vector = attack_vector
        self.mitigation = mitigation
        self.created_at = datetime.now()
    
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "threat_id": self.threat_id,
            "threat_type": self.threat_type,
            "description": self.description,
            "severity": self.severity,
            "attack_vector": self.attack_vector,
            "mitigation": self.mitigation,
            "created_at": self.created_at.isoformat()
        }

class AIScenarioGenerator:
    def __init__(self):
        self.threat_intelligence = []
        self.logger = logging.getLogger("mcp_ai_scenario_generator")
    
    def add_threat_intelligence(self, ti: ThreatIntelligence):
        """添加威胁情报"""
        self.threat_intelligence.append(ti)
        self.logger.info(f"Added threat intelligence: {ti.threat_id}")
    
    def generate_scenario(self, mcp_system_info: Dict) -> Dict:
        """生成攻击场景"""
        self.logger.info("Generating attack scenario")
        
        # 从威胁情报中随机选择一个威胁
        if not self.threat_intelligence:
            self.logger.error("No threat intelligence available")
            return None
        
        selected_ti = random.choice(self.threat_intelligence)
        
        # 基于威胁情报生成攻击场景
        scenario = {
            "scenario_id": f"scenario_{datetime.now().timestamp()}",
            "name": f"{selected_ti.threat_type} 攻击场景",
            "description": f"基于威胁情报 {selected_ti.threat_id} 生成的 {selected_ti.threat_type} 攻击场景",
            "attack_type": selected_ti.threat_type,
            "target": self._select_target(mcp_system_info),
            "payload": self._generate_payload(selected_ti, mcp_system_info),
            "expected_result": f"成功执行 {selected_ti.threat_type} 攻击",
            "severity": selected_ti.severity
        }
        
        self.logger.info(f"Generated attack scenario: {scenario['scenario_id']}, type: {scenario['attack_type']}")
        return scenario
    
    def _select_target(self, mcp_system_info: Dict) -> str:
        """选择攻击目标"""
        # 基于 MCP 系统信息选择攻击目标
        components = mcp_system_info.get("components", ["client", "server", "host"])
        return random.choice(components)
    
    def _generate_payload(self, ti: ThreatIntelligence, mcp_system_info: Dict) -> Dict:
        """生成攻击 payload"""
        # 基于威胁情报和 MCP 系统信息生成攻击 payload
        payload = {
            "attack_vector": ti.attack_vector,
            "target_component": self._select_target(mcp_system_info),
            "parameters": self._generate_parameters(ti.threat_type)
        }
        
        return payload
    
    def _generate_parameters(self, attack_type: str) -> Dict:
        """生成攻击参数"""
        # 根据攻击类型生成不同的攻击参数
        parameters = {}
        
        if attack_type == "身份伪造":
            parameters = {
                "client_id": f"fake_client_{random.randint(1000, 9999)}",
                "user_id": f"fake_user_{random.randint(1000, 9999)}",
                "token": "fake_token_1234567890"
            }
        elif attack_type == "工具越权":
            parameters = {
                "tool_id": "restricted_tool",
                "parameters": {"file_path": "/etc/passwd"}
            }
        elif attack_type == "沙箱逃逸":
            parameters = {
                "tool_id": "sandboxed_tool",
                "parameters": {"command": "bash -c 'echo hello > /host/tmp/escape.txt'"}
            }
        elif attack_type == "拒绝服务":
            parameters = {
                "tool_id": "resource_intensive_tool",
                "parameters": {"iterations": 1000000}
            }
        else:
            # 默认参数
            parameters = {
                "tool_id": "test_tool",
                "parameters": {"param1": "value1", "param2": "value2"}
            }
        
        return parameters
    
    def generate_batch_scenarios(self, mcp_system_info: Dict, count: int = 5) -> List[Dict]:
        """批量生成攻击场景"""
        self.logger.info(f"Generating {count} attack scenarios")
        
        scenarios = []
        for i in range(count):
            scenario = self.generate_scenario(mcp_system_info)
            if scenario:
                scenarios.append(scenario)
        
        self.logger.info(f"Generated {len(scenarios)} attack scenarios")
        return scenarios

# 示例使用
if __name__ == "__main__":
    # 配置日志
    logging.basicConfig(level=logging.INFO)
    
    # 创建 AI 场景生成器
    generator = AIScenarioGenerator()
    
    # 添加威胁情报
    ti1 = ThreatIntelligence(
        threat_id="ti_001",
        threat_type="身份伪造",
        description="攻击者伪造合法用户身份，获取未授权访问权限",
        severity="high",
        attack_vector="API 调用",
        mitigation="加强身份验证，使用 MFA，定期轮换凭证"
    )
    generator.add_threat_intelligence(ti1)
    
    ti2 = ThreatIntelligence(
        threat_id="ti_002",
        threat_type="工具越权",
        description="攻击者利用工具漏洞，执行超出权限范围的操作",
        severity="critical",
        attack_vector="工具调用",
        mitigation="实施最小权限原则，加强工具权限管理"
    )
    generator.add_threat_intelligence(ti2)
    
    ti3 = ThreatIntelligence(
        threat_id="ti_003",
        threat_type="沙箱逃逸",
        description="攻击者从沙箱环境中逃逸，执行恶意代码",
        severity="critical",
        attack_vector="沙箱环境",
        mitigation="加强沙箱隔离，定期更新沙箱环境"
    )
    generator.add_threat_intelligence(ti3)
    
    # MCP 系统信息
    mcp_system_info = {
        "components": ["client", "server", "host"],
        "tools": ["file_reader", "command_executor", "network_scanner"],
        "version": "v2.0"
    }
    
    # 生成攻击场景
    scenario = generator.generate_scenario(mcp_system_info)
    print(f"Generated scenario: {json.dumps(scenario, indent=2, ensure_ascii=False)}")
    
    # 批量生成攻击场景
    scenarios = generator.generate_batch_scenarios(mcp_system_info, count=3)
    print(f"\nGenerated {len(scenarios)} scenarios:")
    for s in scenarios:
        print(f"- {s['name']} (ID: {s['scenario_id']})")

# mcp_exploit_chain_analyzer.py
from typing import Dict, List, Optional
import logging
from datetime import datetime

class Vulnerability:
    def __init__(self, vuln_id: str, name: str, description: str, 
                 severity: str, component: str, cvss_score: float,
                 attack_vector: str):
        self.vuln_id = vuln_id
        self.name = name
        self.description = description
        self.severity = severity
        self.component = component
        self.cvss_score = cvss_score
        self.attack_vector = attack_vector
        self.related_vulns = []
    
    def add_related_vuln(self, vuln_id: str):
        """添加相关漏洞"""
        if vuln_id not in self.related_vulns:
            self.related_vulns.append(vuln_id)
    
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "vuln_id": self.vuln_id,
            "name": self.name,
            "description": self.description,
            "severity": self.severity,
            "component": self.component,
            "cvss_score": self.cvss_score,
            "attack_vector": self.attack_vector,
            "related_vulns": self.related_vulns
        }

class ExploitChain:
    def __init__(self, chain_id: str, vulns: List[Vulnerability], 
                 attack_path: List[str], total_risk: float):
        self.chain_id = chain_id
        self.vulns = vulns
        self.attack_path = attack_path
        self.total_risk = total_risk
        self.created_at = datetime.now()
    
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "chain_id": self.chain_id,
            "vulns": [vuln.to_dict() for vuln in self.vulns],
            "attack_path": self.attack_path,
            "total_risk": self.total_risk,
            "created_at": self.created_at.isoformat()
        }

class MCPExploitChainAnalyzer:
    def __init__(self):
        self.vulnerabilities = {}
        self.exploit_chains = []
        self.logger = logging.getLogger("mcp_exploit_chain_analyzer")
    
    def add_vulnerability(self, vuln: Vulnerability):
        """添加漏洞"""
        self.vulnerabilities[vuln.vuln_id] = vuln
        self.logger.info(f"Added vulnerability: {vuln.vuln_id}, component: {vuln.component}")
    
    def analyze_exploit_chains(self) -> List[ExploitChain]:
        """分析漏洞利用链"""
        self.logger.info("Starting exploit chain analysis")
        
        # 构建漏洞关联图
        vuln_graph = self._build_vuln_graph()
        
        # 寻找所有可能的漏洞利用链
        chains = self._find_all_chains(vuln_graph)
        
        # 评估每个漏洞利用链的风险
        exploit_chains = []
        for i, chain in enumerate(chains):
            # 获取链中的漏洞
            chain_vulns = [self.vulnerabilities[vuln_id] for vuln_id in chain if vuln_id in self.vulnerabilities]
            if not chain_vulns:
                continue
            
            # 计算总风险
            total_risk = sum(vuln.cvss_score for vuln in chain_vulns)
            
            # 创建漏洞利用链
            exploit_chain = ExploitChain(
                chain_id=f"chain_{datetime.now().timestamp()}_{i}",
                vulns=chain_vulns,
                attack_path=chain,
                total_risk=total_risk
            )
            
            exploit_chains.append(exploit_chain)
            self.exploit_chains.append(exploit_chain)
        
        # 按风险排序
        exploit_chains.sort(key=lambda x: x.total_risk, reverse=True)
        
        self.logger.info(f"Found {len(exploit_chains)} exploit chains")
        return exploit_chains
    
    def _build_vuln_graph(self) -> Dict[str, List[str]]:
        """构建漏洞关联图"""
        graph = {}
        
        for vuln_id, vuln in self.vulnerabilities.items():
            graph[vuln_id] = vuln.related_vulns
        
        return graph
    
    def _find_all_chains(self, graph: Dict[str, List[str]], max_length: int = 5) -> List[List[str]]:
        """寻找所有可能的漏洞利用链"""
        chains = []
        
        # 深度优先搜索寻找所有可能的链
        def dfs(node, path, visited):
            if len(path) > max_length:
                return
            
            if path not in chains:
                chains.append(path.copy())
            
            visited.add(node)
            
            for neighbor in graph.get(node, []):
                if neighbor not in visited:
                    path.append(neighbor)
                    dfs(neighbor, path, visited.copy())
                    path.pop()
        
        # 从每个漏洞开始搜索
        for node in graph.keys():
            dfs(node, [node], set())
        
        # 过滤掉长度为 1 的链（单个漏洞）
        chains = [chain for chain in chains if len(chain) > 1]
        
        return chains
    
    def get_exploit_chains_by_risk(self, min_risk: float = 0.0) -> List[ExploitChain]:
        """按风险获取漏洞利用链"""
        return [chain for chain in self.exploit_chains if chain.total_risk >= min_risk]
    
    def generate_fix_recommendations(self, chain: ExploitChain) -> List[str]:
        """生成修复建议"""
        recommendations = []
        
        for vuln in chain.vulns:
            recommendations.append(f"修复漏洞 {vuln.vuln_id}（{vuln.name}），CVSS 评分：{vuln.cvss_score}")
        
        recommendations.append(f"总体修复建议：优先修复 CVSS 评分高的漏洞，打破漏洞利用链")
        
        return recommendations

# 示例使用
if __name__ == "__main__":
    # 配置日志
    logging.basicConfig(level=logging.INFO)
    
    # 创建漏洞利用链分析器
    analyzer = MCPExploitChainAnalyzer()
    
    # 添加漏洞
    vuln1 = Vulnerability(
        vuln_id="vuln_001",
        name="身份认证绕过",
        description="攻击者可以绕过 MCP Server 的身份认证机制",
        severity="critical",
        component="server",
        cvss_score=9.8,
        attack_vector="API 调用"
    )
    analyzer.add_vulnerability(vuln1)
    
    vuln2 = Vulnerability(
        vuln_id="vuln_002",
        name="工具越权",
        description="攻击者可以通过特定工具执行超出权限的操作",
        severity="high",
        component="host",
        cvss_score=8.5,
        attack_vector="工具调用"
    )
    analyzer.add_vulnerability(vuln2)
    
    vuln3 = Vulnerability(
        vuln_id="vuln_003",
        name="沙箱逃逸",
        description="攻击者可以从 Docker 沙箱中逃逸",
        severity="critical",
        component="host",
        cvss_score=9.0,
        attack_vector="沙箱环境"
    )
    analyzer.add_vulnerability(vuln3)
    
    vuln4 = Vulnerability(
        vuln_id="vuln_004",
        name="命令注入",
        description="攻击者可以通过工具参数注入恶意命令",
        severity="high",
        component="tool",
        cvss_score=8.0,
        attack_vector="参数注入"
    )
    analyzer.add_vulnerability(vuln4)
    
    # 建立漏洞关联
    vuln1.add_related_vuln("vuln_002")
    vuln2.add_related_vuln("vuln_003")
    vuln3.add_related_vuln("vuln_004")
    vuln1.add_related_vuln("vuln_004")
    
    # 分析漏洞利用链
    chains = analyzer.analyze_exploit_chains()
    
    # 输出结果
    print(f"\nFound {len(chains)} exploit chains:")
    for chain in chains:
        print(f"\nExploit Chain {chain.chain_id}:")
        print(f"  Total Risk: {chain.total_risk}")
        print(f"  Attack Path: {' -> '.join(chain.attack_path)}")
        print("  Vulnerabilities:")
        for vuln in chain.vulns:
            print(f"    - {vuln.name} (ID: {vuln.vuln_id}, CVSS: {vuln.cvss_score})")
        
        # 生成修复建议
        recommendations = analyzer.generate_fix_recommendations(chain)
        print("  Fix Recommendations:")
        for rec in recommendations:
            print(f"    - {rec}")
    
    # 按风险获取漏洞利用链
    high_risk_chains = analyzer.get_exploit_chains_by_risk(min_risk=15.0)
    print(f"\nHigh risk chains (>= 15.0): {len(high_risk_chains)}")

# mcp_attack_simulation.py
from typing import Dict, List, Optional
import logging
import json
from datetime import datetime

# 导入前面实现的模块
from mcp_attack_manager import MCPAttackManager, AttackScenario
from mcp_scenario_generator import AIScenarioGenerator, ThreatIntelligence
from mcp_exploit_chain_analyzer import MCPExploitChainAnalyzer, Vulnerability

class MCPAttackSimulationSystem:
    def __init__(self):
        # 初始化各个模块
        self.attack_manager = MCPAttackManager()
        self.scenario_generator = AIScenarioGenerator()
        self.exploit_analyzer = MCPExploitChainAnalyzer()
        
        self.logger = logging.getLogger("mcp_attack_simulation")
        self.logger.info("MCP Attack Simulation System initialized")
    
    def initialize(self):
        """初始化系统"""
        self.logger.info("Initializing MCP Attack Simulation System")
        
        # 添加威胁情报
        self._add_threat_intelligence()
        
        # 添加示例漏洞
        self._add_sample_vulnerabilities()
        
        self.logger.info("MCP Attack Simulation System initialized successfully")
    
    def _add_threat_intelligence(self):
        """添加威胁情报"""
        # 身份伪造威胁
        ti1 = ThreatIntelligence(
            threat_id="ti_001",
            threat_type="身份伪造",
            description="攻击者伪造合法用户身份，获取未授权访问权限",
            severity="high",
            attack_vector="API 调用",
            mitigation="加强身份验证，使用 MFA，定期轮换凭证"
        )
        self.scenario_generator.add_threat_intelligence(ti1)
        
        # 工具越权威胁
        ti2 = ThreatIntelligence(
            threat_id="ti_002",
            threat_type="工具越权",
            description="攻击者利用工具漏洞，执行超出权限范围的操作",
            severity="critical",
            attack_vector="工具调用",
            mitigation="实施最小权限原则，加强工具权限管理"
        )
        self.scenario_generator.add_threat_intelligence(ti2)
        
        # 沙箱逃逸威胁
        ti3 = ThreatIntelligence(
            threat_id="ti_003",
            threat_type="沙箱逃逸",
            description="攻击者从沙箱环境中逃逸，执行恶意代码",
            severity="critical",
            attack_vector="沙箱环境",
            mitigation="加强沙箱隔离，定期更新沙箱环境"
        )
        self.scenario_generator.add_threat_intelligence(ti3)
        
        # 命令注入威胁
        ti4 = ThreatIntelligence(
            threat_id="ti_004",
            threat_type="命令注入",
            description="攻击者通过工具参数注入恶意命令",
            severity="high",
            attack_vector="参数注入",
            mitigation="加强参数验证和过滤"
        )
        self.scenario_generator.add_threat_intelligence(ti4)
        
        # 拒绝服务威胁
        ti5 = ThreatIntelligence(
            threat_id="ti_005",
            threat_type="拒绝服务",
            description="攻击者通过大量请求导致 MCP 系统崩溃",
            severity="medium",
            attack_vector="资源耗尽",
            mitigation="实施速率限制，优化系统性能"
        )
        self.scenario_generator.add_threat_intelligence(ti5)
    
    def _add_sample_vulnerabilities(self):
        """添加示例漏洞"""
        # 添加示例漏洞，用于漏洞利用链分析
        vuln1 = Vulnerability(
            vuln_id="vuln_001",
            name="身份认证绕过",
            description="攻击者可以绕过 MCP Server 的身份认证机制",
            severity="critical",
            component="server",
            cvss_score=9.8,
            attack_vector="API 调用"
        )
        self.exploit_analyzer.add_vulnerability(vuln1)
        
        vuln2 = Vulnerability(
            vuln_id="vuln_002",
            name="工具越权",
            description="攻击者可以通过特定工具执行超出权限的操作",
            severity="high",
            component="host",
            cvss_score=8.5,
            attack_vector="工具调用"
        )
        self.exploit_analyzer.add_vulnerability(vuln2)
        
        vuln3 = Vulnerability(
            vuln_id="vuln_003",
            name="沙箱逃逸",
            description="攻击者可以从 Docker 沙箱中逃逸",
            severity="critical",
            component="host",
            cvss_score=9.0,
            attack_vector="沙箱环境"
        )
        self.exploit_analyzer.add_vulnerability(vuln3)
        
        vuln4 = Vulnerability(
            vuln_id="vuln_004",
            name="命令注入",
            description="攻击者可以通过工具参数注入恶意命令",
            severity="high",
            component="tool",
            cvss_score=8.0,
            attack_vector="参数注入"
        )
        self.exploit_analyzer.add_vulnerability(vuln4)
        
        # 建立漏洞关联
        vuln1.add_related_vuln("vuln_002")
        vuln2.add_related_vuln("vuln_003")
        vuln3.add_related_vuln("vuln_004")
        vuln1.add_related_vuln("vuln_004")
    
    def generate_attack_scenarios(self, mcp_system_info: Dict, count: int = 5) -> List[Dict]:
        """生成攻击场景"""
        self.logger.info(f"Generating {count} attack scenarios")
        
        scenarios = []
        for i in range(count):
            scenario_dict = self.scenario_generator.generate_scenario(mcp_system_info)
            if scenario_dict:
                # 转换为 AttackScenario 对象并添加到攻击管理器
                scenario = AttackScenario(
                    scenario_id=scenario_dict["scenario_id"],
                    name=scenario_dict["name"],
                    description=scenario_dict["description"],
                    attack_type=scenario_dict["attack_type"],
                    target=scenario_dict["target"],
                    payload=scenario_dict["payload"],
                    expected_result=scenario_dict["expected_result"],
                    severity=scenario_dict["severity"]
                )
                self.attack_manager.add_scenario(scenario)
                scenarios.append(scenario_dict)
        
        self.logger.info(f"Generated {len(scenarios)} attack scenarios")
        return scenarios
    
    def run_attack_simulation(self, scenario_ids: List[str] = None) -> List[str]:
        """运行攻防模拟"""
        self.logger.info("Running attack simulation")
        
        # 如果没有指定场景 ID，运行所有场景
        if not scenario_ids:
            scenario_ids = [scenario.scenario_id for scenario in self.attack_manager.scenarios.values()]
        
        result_ids = []
        for scenario_id in scenario_ids:
            if scenario_id in self.attack_manager.scenarios:
                result_id = self.attack_manager.run_scenario(scenario_id)
                result_ids.append(result_id)
            else:
                self.logger.warning(f"Scenario {scenario_id} not found")
        
        self.logger.info(f"Attack simulation completed, {len(result_ids)} scenarios executed")
        return result_ids
    
    def analyze_exploit_chains(self) -> List[Dict]:
        """分析漏洞利用链"""
        self.logger.info("Analyzing exploit chains")
        
        chains = self.exploit_analyzer.analyze_exploit_chains()
        
        # 转换为字典格式
        chain_dicts = [chain.to_dict() for chain in chains]
        
        self.logger.info(f"Exploit chain analysis completed, found {len(chain_dicts)} chains")
        return chain_dicts
    
    def generate_report(self) -> Dict:
        """生成攻防模拟报告"""
        self.logger.info("Generating attack simulation report")
        
        # 获取所有攻击场景
        scenarios = self.attack_manager.get_all_scenarios()
        
        # 获取所有攻击结果
        results = self.attack_manager.get_all_results()
        
        # 分析漏洞利用链
        exploit_chains = self.analyze_exploit_chains()
        
        # 生成报告
        report = {
            "report_id": f"report_{datetime.now().timestamp()}",
            "generated_at": datetime.now().isoformat(),
            "scenarios": {
                "total": len(scenarios),
                "list": scenarios
            },
            "results": {
                "total": len(results),
                "list": results,
                "success_rate": self._calculate_success_rate(results)
            },
            "exploit_chains": {
                "total": len(exploit_chains),
                "list": exploit_chains,
                "high_risk_chains": len([chain for chain in exploit_chains if chain["total_risk"] >= 15.0])
            },
            "recommendations": self._generate_recommendations(scenarios, results, exploit_chains)
        }
        
        self.logger.info(f"Attack simulation report generated: {report['report_id']}")
        return report
    
    def _calculate_success_rate(self, results: List[Dict]) -> float:
        """计算攻击成功率"""
        if not results:
            return 0.0
        
        success_count = sum(1 for result in results if result["is_success"])
        return (success_count / len(results)) * 100
    
    def _generate_recommendations(self, scenarios: List[Dict], results: List[Dict], 
                                 exploit_chains: List[Dict]) -> List[str]:
        """生成修复建议"""
        recommendations = []
        
        # 基于攻击结果生成建议
        for result in results:
            if result["is_success"]:
                scenario = next((s for s in scenarios if s["scenario_id"] == result["scenario_id"]), None)
                if scenario:
                    recommendations.append(f"修复 {scenario['attack_type']} 攻击漏洞，该攻击已成功执行")
        
        # 基于漏洞利用链生成建议
        for chain in exploit_chains:
            if chain["total_risk"] >= 15.0:
                recommendations.append(f"优先修复风险值为 {chain['total_risk']} 的漏洞利用链：{' -> '.join(chain['attack_path'])}")
        
        # 通用建议
        recommendations.append("加强 MCP 系统的身份认证机制，使用 MFA")
        recommendations.append("实施最小权限原则，限制工具的访问权限")
        recommendations.append("加强沙箱隔离，定期更新沙箱环境")
        recommendations.append("对工具参数进行严格验证和过滤，防止注入攻击")
        recommendations.append("定期进行攻防模拟测试，及时发现和修复漏洞")
        
        # 去重
        recommendations = list(dict.fromkeys(recommendations))
        
        return recommendations

# 示例使用
if __name__ == "__main__":
    # 配置日志
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    
    # 创建 MCP 攻防模拟系统
    simulation_system = MCPAttackSimulationSystem()
    
    # 初始化系统
    simulation_system.initialize()
    
    # MCP 系统信息
    mcp_system_info = {
        "components": ["client", "server", "host"],
        "tools": ["file_reader", "command_executor", "network_scanner"],
        "version": "v2.0"
    }
    
    print("1. 生成攻击场景...")
    scenarios = simulation_system.generate_attack_scenarios(mcp_system_info, count=3)
    print(f"生成了 {len(scenarios)} 个攻击场景：")
    for i, scenario in enumerate(scenarios):
        print(f"  {i+1}. {scenario['name']} (ID: {scenario['scenario_id']})")
    
    print("\n2. 运行攻防模拟...")
    result_ids = simulation_system.run_attack_simulation()
    print(f"运行了 {len(result_ids)} 个攻击场景，结果 ID：{result_ids}")
    
    print("\n3. 分析漏洞利用链...")
    exploit_chains = simulation_system.analyze_exploit_chains()
    print(f"发现了 {len(exploit_chains)} 条漏洞利用链")
    
    print("\n4. 生成攻防模拟报告...")
    report = simulation_system.generate_report()
    print(f"报告生成成功，ID：{report['report_id']}")
    
    # 输出报告摘要
    print("\n报告摘要：")
    print(f"  生成时间：{report['generated_at']}")
    print(f"  攻击场景总数：{report['scenarios']['total']}")
    print(f"  攻击结果总数：{report['results']['total']}")
    print(f"  攻击成功率：{report['results']['success_rate']:.2f}%")
    print(f"  漏洞利用链总数：{report['exploit_chains']['total']}")
    print(f"  高风险漏洞利用链：{report['exploit_chains']['high_risk_chains']}")
    
    print("\n修复建议：")
    for i, rec in enumerate(report['recommendations']):
        print(f"  {i+1}. {rec}")
    
    # 保存报告到文件
    report_file = f"attack_simulation_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    with open(report_file, 'w', encoding='utf-8') as f:
        json.dump(report, f, indent=2, ensure_ascii=False)
    print(f"\n报告已保存到文件：{report_file}")

# 安装依赖
pip install fastapi uvicorn sqlalchemy pydantic redis psycopg2-binary python-dotenv

# 配置攻防模拟系统
cp config.example.yaml config.yaml
# 编辑配置文件
vim config.yaml

# 启动基础设施服务
docker-compose up -d postgresql redis grafana

# 初始化数据库
python init_db.py --config config.yaml

# 启动攻防模拟服务
python mcp_attack_simulation_service.py --config config.yaml

# 启动监测代理
python mcp_monitoring_agent.py --config config.yaml