10：L应用联邦学习：蓝队的分布式安全协作

import numpy as np
from sklearn.linear_model import LogisticRegression

class FederalLearning:
    def __init__(self, num_clients):
        self.num_clients = num_clients
        self.global_model = LogisticRegression()
        self.client_models = [LogisticRegression() for _ in range(num_clients)]
    
    def train_local(self, client_id, X, y):
        """在本地训练模型"""
        self.client_models[client_id].fit(X, y)
        return self.client_models[client_id].coef_, self.client_models[client_id].intercept_
    
    def aggregate(self, client_params):
        """聚合客户端模型参数"""
        # 平均系数
        avg_coef = np.mean([params[0] for params in client_params], axis=0)
        # 平均截距
        avg_intercept = np.mean([params[1] for params in client_params], axis=0)
        # 更新全局模型
        self.global_model.coef_ = avg_coef
        self.global_model.intercept_ = avg_intercept
        return avg_coef, avg_intercept
    
    def distribute(self):
        """分发全局模型参数"""
        for i in range(self.num_clients):
            self.client_models[i].coef_ = self.global_model.coef_.copy()
            self.client_models[i].intercept_ = self.global_model.intercept_.copy()

# 模拟数据
np.random.seed(42)
# 生成三个客户端的数据
X1 = np.random.normal(0, 1, (100, 5))
y1 = np.random.randint(0, 2, 100)

X2 = np.random.normal(0, 1, (100, 5))
y2 = np.random.randint(0, 2, 100)

X3 = np.random.normal(0, 1, (100, 5))
y3 = np.random.randint(0, 2, 100)

# 创建联邦学习实例
fl = FederalLearning(3)

# 训练轮数
num_rounds = 5

for round in range(num_rounds):
    print(f"训练轮次: {round+1}")
    # 本地训练
    params1 = fl.train_local(0, X1, y1)
    params2 = fl.train_local(1, X2, y2)
    params3 = fl.train_local(2, X3, y3)
    # 聚合
    fl.aggregate([params1, params2, params3])
    # 分发
    fl.distribute()

# 测试全局模型
X_test = np.random.normal(0, 1, (50, 5))
y_test = np.random.randint(0, 2, 50)
accuracy = fl.global_model.score(X_test, y_test)
print(f"全局模型准确率: {accuracy}")

import numpy as np
from phe import paillier

class SecureAggregation:
    def __init__(self):
        # 生成密钥对
        self.public_key, self.private_key = paillier.generate_paillier_keypair()
    
    def encrypt_params(self, params):
        """加密模型参数"""
        encrypted_params = []
        for param in params:
            if isinstance(param, np.ndarray):
                encrypted_param = np.vectorize(self.public_key.encrypt)(param)
            else:
                encrypted_param = self.public_key.encrypt(param)
            encrypted_params.append(encrypted_param)
        return encrypted_params
    
    def aggregate_encrypted(self, encrypted_params_list):
        """聚合加密的模型参数"""
        # 初始化聚合结果
        aggregated = []
        for i in range(len(encrypted_params_list[0])):
            if isinstance(encrypted_params_list[0][i], np.ndarray):
                # 对每个元素进行聚合
                agg = np.zeros_like(encrypted_params_list[0][i])
                for j in range(len(encrypted_params_list)):
                    agg += encrypted_params_list[j][i]
            else:
                # 标量聚合
                agg = encrypted_params_list[0][i]
                for j in range(1, len(encrypted_params_list)):
                    agg += encrypted_params_list[j][i]
            aggregated.append(agg)
        return aggregated
    
    def decrypt_params(self, encrypted_params):
        """解密模型参数"""
        decrypted_params = []
        for param in encrypted_params:
            if isinstance(param, np.ndarray):
                decrypted_param = np.vectorize(self.private_key.decrypt)(param)
            else:
                decrypted_param = self.private_key.decrypt(param)
            decrypted_params.append(decrypted_param)
        return decrypted_params

# 测试安全聚合
sa = SecureAggregation()

# 模拟三个客户端的模型参数
params1 = [np.array([1.0, 2.0, 3.0]), 0.5]
params2 = [np.array([2.0, 3.0, 4.0]), 1.0]
params3 = [np.array([3.0, 4.0, 5.0]), 1.5]

# 加密参数
encrypted1 = sa.encrypt_params(params1)
encrypted2 = sa.encrypt_params(params2)
encrypted3 = sa.encrypt_params(params3)

# 聚合
aggregated = sa.aggregate_encrypted([encrypted1, encrypted2, encrypted3])

# 解密
decrypted = sa.decrypt_params(aggregated)

print(f"聚合后的系数: {decrypted[0]}")
print(f"聚合后的截距: {decrypted[1]}")

import numpy as np
from sklearn.ensemble import RandomForestClassifier

class FederalThreatDetection:
    def __init__(self, num_organizations):
        self.num_organizations = num_organizations
        self.global_model = RandomForestClassifier(n_estimators=100)
        self.org_models = [RandomForestClassifier(n_estimators=50) for _ in range(num_organizations)]
    
    def train_local(self, org_id, X, y):
        """在本地训练威胁检测模型"""
        self.org_models[org_id].fit(X, y)
        # 提取模型特征重要性作为本地知识
        return self.org_models[org_id].feature_importances_
    
    def aggregate_knowledge(self, local_knowledge):
        """聚合本地知识"""
        # 平均特征重要性
        avg_importance = np.mean(local_knowledge, axis=0)
        return avg_importance
    
    def update_global_model(self, avg_importance):
        """根据聚合的知识更新全局模型"""
        # 这里简化处理，实际中可能需要更复杂的更新策略
        print(f"更新全局模型，基于特征重要性: {avg_importance}")

# 模拟数据
np.random.seed(42)

# 生成三个组织的威胁检测数据
# 特征：网络流量特征、系统日志特征等
X1 = np.random.normal(0, 1, (100, 10))
y1 = np.random.randint(0, 2, 100)  # 0表示正常，1表示威胁

X2 = np.random.normal(0, 1, (100, 10))
y2 = np.random.randint(0, 2, 100)

X3 = np.random.normal(0, 1, (100, 10))
y3 = np.random.randint(0, 2, 100)

# 创建联邦威胁检测实例
ftd = FederalThreatDetection(3)

# 训练轮数
num_rounds = 3

for round in range(num_rounds):
    print(f"训练轮次: {round+1}")
    # 本地训练
    knowledge1 = ftd.train_local(0, X1, y1)
    knowledge2 = ftd.train_local(1, X2, y2)
    knowledge3 = ftd.train_local(2, X3, y3)
    # 聚合知识
    avg_importance = ftd.aggregate_knowledge([knowledge1, knowledge2, knowledge3])
    # 更新全局模型
    ftd.update_global_model(avg_importance)

print("联合威胁检测模型训练完成")