12：L构建AI驱动的IDS/IPS：蓝队的威胁检测系统

# 威胁检测模型实现
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import precision_recall_fscore_support

# 加载网络流量数据
data = pd.read_csv('network_traffic.csv')

# 特征选择
features = ['packet_size', 'duration', 'protocol', 'src_ip', 'dst_ip', 'src_port', 'dst_port', 'flow_bytes']
target = 'label'

# 数据预处理
data = pd.get_dummies(data, columns=['protocol'])

# 训练测试集分割
X_train, X_test, y_train, y_test = train_test_split(
    data[features], data[target], test_size=0.2, random_state=42
)

# 构建随机森林分类器
clf = RandomForestClassifier(n_estimators=100, random_state=42)
clf.fit(X_train, y_train)

# 模型评估
y_pred = clf.predict(X_test)
precision, recall, f1, _ = precision_recall_fscore_support(y_test, y_pred, average='binary')
print(f"精确率: {precision:.4f}")
print(f"召回率: {recall:.4f}")
print(f"F1分数: {f1:.4f}")

# 误报管理算法实现
class FalseAlarmManager:
    def __init__(self, model):
        self.model = model
        self.threshold = 0.5
    
    def adjust_threshold(self, validation_data, validation_labels):
        # 计算不同阈值下的误报率和漏报率
        thresholds = np.linspace(0, 1, 100)
        false_alarms = []
        missed_detections = []
        
        for threshold in thresholds:
            y_prob = self.model.predict_proba(validation_data)[:, 1]
            y_pred = (y_prob > threshold).astype(int)
            
            # 计算误报率和漏报率
            false_alarm = sum((y_pred == 1) & (validation_labels == 0)) / len(validation_labels)
            missed_detection = sum((y_pred == 0) & (validation_labels == 1)) / len(validation_labels)
            
            false_alarms.append(false_alarm)
            missed_detections.append(missed_detection)
        
        # 找到误报率和漏报率的平衡点
        optimal_idx = np.argmin(np.array(false_alarms) + np.array(missed_detections))
        self.threshold = thresholds[optimal_idx]
        return self.threshold
    
    def predict_with_threshold(self, X):
        y_prob = self.model.predict_proba(X)[:, 1]
        return (y_prob > self.threshold).astype(int)

# 多维度威胁评估实现
class ThreatAssessor:
    def __init__(self):
        self.threat_weights = {
            'network': 0.3,
            'behavior': 0.3,
            'context': 0.4
        }
    
    def assess_threat(self, network_features, behavior_features, context_features):
        # 网络特征评估
        network_score = self._assess_network(network_features)
        
        # 行为特征评估
        behavior_score = self._assess_behavior(behavior_features)
        
        # 上下文特征评估
        context_score = self._assess_context(context_features)
        
        # 综合评估
        total_score = (
            network_score * self.threat_weights['network'] +
            behavior_score * self.threat_weights['behavior'] +
            context_score * self.threat_weights['context']
        )
        
        # 威胁等级划分
        if total_score >= 0.8:
            return 'critical'
        elif total_score >= 0.6:
            return 'high'
        elif total_score >= 0.4:
            return 'medium'
        else:
            return 'low'
    
    def _assess_network(self, features):
        # 实现网络特征评估
        pass
    
    def _assess_behavior(self, features):
        # 实现行为特征评估
        pass
    
    def _assess_context(self, features):
        # 实现上下文特征评估
        pass

# 自动响应机制实现
class AutoResponder:
    def __init__(self, threat_assessor):
        self.threat_assessor = threat_assessor
    
    def respond_to_threat(self, threat_info):
        # 评估威胁等级
        threat_level = self.threat_assessor.assess_threat(
            threat_info['network_features'],
            threat_info['behavior_features'],
            threat_info['context_features']
        )
        
        # 根据威胁等级采取相应措施
        if threat_level == 'critical':
            return self._critical_response(threat_info)
        elif threat_level == 'high':
            return self._high_response(threat_info)
        elif threat_level == 'medium':
            return self._medium_response(threat_info)
        else:
            return self._low_response(threat_info)
    
    def _critical_response(self, threat_info):
        # 关键威胁响应
        return {
            'action': 'block',
            'message': 'Critical threat detected, immediate action taken'
        }
    
    def _high_response(self, threat_info):
        # 高威胁响应
        return {
            'action': 'quarantine',
            'message': 'High threat detected, entity quarantined'
        }
    
    def _medium_response(self, threat_info):
        # 中等威胁响应
        return {
            'action': 'monitor',
            'message': 'Medium threat detected, enhanced monitoring initiated'
        }
    
    def _low_response(self, threat_info):
        # 低威胁响应
        return {
            'action': 'log',
            'message': 'Low threat detected, logged for review'
        }