44：L构建自动化响应：蓝队的快速防御

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import LabelEncoder

class AIDecisionEngine:
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100, random_state=42)
        self.label_encoder = LabelEncoder()
    
    def train_model(self, training_data):
        """训练决策模型"""
        # 特征提取
        X = training_data[['threat_type', 'severity', 'asset_value', 'attack_vector', 'target_industry']]
        y = training_data['response_strategy']
        
        # 编码分类特征
        for col in ['threat_type', 'attack_vector', 'target_industry']:
            X[col] = self.label_encoder.fit_transform(X[col])
        
        # 训练模型
        self.model.fit(X, y)
    
    def select_response_strategy(self, threat):
        """选择响应策略"""
        # 特征提取
        features = {
            'threat_type': threat['type'],
            'severity': threat['severity'],
            'asset_value': threat['asset_value'],
            'attack_vector': threat['attack_vector'],
            'target_industry': threat['target_industry']
        }
        
        # 编码特征
        for key in ['threat_type', 'attack_vector', 'target_industry']:
            if features[key] in self.label_encoder.classes_:
                features[key] = self.label_encoder.transform([features[key]])[0]
            else:
                features[key] = -1
        
        # 预测
        X = pd.DataFrame([features])
        prediction = self.model.predict(X)[0]
        confidence = self.model.predict_proba(X)[0][np.where(self.model.classes_ == prediction)[0][0]]
        
        return {
            'response_strategy': prediction,
            'confidence': confidence
        }

class预案Manager:
    def __init__(self):
        self.预案s = {}
        self.historical_data = []
    
    def add_预案(self, 预案_id, 预案):
        """添加预案"""
        self.预案s[预案_id] = 预案
    
    def update_预案(self, 预案_id, 预案):
        """更新预案"""
        self.预案s[预案_id] = 预案
    
    def get_预案(self, 预案_id):
        """获取预案"""
        return self.预案s.get(预案_id)
    
    def select_预案(self, threat):
        """选择合适的预案"""
        # 根据威胁类型选择预案
        threat_type = threat['type']
        
        # 查找匹配的预案
        for 预案_id, 预案 in self.预案s.items():
            if 预案['threat_type'] == threat_type:
                return 预案
        
        # 如果没有匹配的预案，返回默认预案
        return self.预案s.get('default')
    
    def optimize_预案s(self):
        """根据历史数据优化预案"""
        if not self.historical_data:
            return
        
        # 分析历史响应数据
        df = pd.DataFrame(self.historical_data)
        
        # 按威胁类型分组
        grouped = df.groupby('threat_type')
        
        for threat_type, group in grouped:
            # 计算每种响应策略的成功率
            success_rates = group.groupby('response_strategy')['success'].mean()
            
            # 选择成功率最高的响应策略
            best_strategy = success_rates.idxmax()
            
            # 更新对应威胁类型的预案
            for 预案_id, 预案 in self.预案s.items():
                if 预案['threat_type'] == threat_type:
                    预案['response_strategy'] = best_strategy
                    break
    
    def record_response(self, threat, response_strategy, success):
        """记录响应结果"""
        self.historical_data.append({
            'threat_type': threat['type'],
            'response_strategy': response_strategy,
            'success': success
        })
        
        # 每100条记录优化一次预案
        if len(self.historical_data) % 100 == 0:
            self.optimize_预案s()

class ResponseOrchestrator:
    def __init__(self):
        self.tools = {}
    
    def register_tool(self, tool_name, tool_function):
        """注册工具"""
        self.tools[tool_name] = tool_function
    
    def orchestrate_response(self, 预案, threat):
        """编排响应流程"""
        response_steps = 预案['response_steps']
        results = []
        
        for step in response_steps:
            tool_name = step['tool']
            parameters = step['parameters']
            
            # 替换参数中的变量
            for key, value in parameters.items():
                if isinstance(value, str) and value.startswith('$'):
                    var_name = value[1:]
                    if var_name in threat:
                        parameters[key] = threat[var_name]
            
            # 执行工具
            if tool_name in self.tools:
                result = self.tools[tool_name](**parameters)
                results.append({
                    'step': step,
                    'result': result
                })
            else:
                results.append({
                    'step': step,
                    'result': 'Tool not found'
                })
        
        return results
    
    def execute_manual_steps(self, steps, assignees):
        """执行需要人工干预的步骤"""
        tasks = []
        
        for i, step in enumerate(steps):
            assignee = assignees[i % len(assignees)]
            tasks.append({
                'task_id': f'task_{i}',
                'description': step['description'],
                'assignee': assignee,
                'status': 'pending',
                'created_at': pd.Timestamp.now().isoformat()
            })
        
        return tasks