164_隐私计算与联邦学习安全：从同态加密到差分隐私的实战指南

import random
import math
from Crypto.Util import number

class Paillier:
    def __init__(self, key_size=1024):
        # 密钥生成
        self.p = number.getPrime(key_size // 2)
        self.q = number.getPrime(key_size // 2)
        self.n = self.p * self.q
        self.nsq = self.n * self.n
        self.g = self.n + 1  # 常见选择
        self.lam = math.lcm(self.p-1, self.q-1)  # Carmichael函数
        self.mu = pow(self.lam, -1, self.n)  # 模逆元
    
    def encrypt(self, plaintext):
        # 加密过程
        if not 0 <= plaintext < self.n:
            raise ValueError("明文必须在0到n-1之间")
        r = random.randint(1, self.n-1)
        while math.gcd(r, self.n) != 1:
            r = random.randint(1, self.n-1)
        
        # c = g^m * r^n mod n^2
        c1 = pow(self.g, plaintext, self.nsq)
        c2 = pow(r, self.n, self.nsq)
        ciphertext = (c1 * c2) % self.nsq
        return ciphertext
    
    def decrypt(self, ciphertext):
        # 解密过程
        if not 0 <= ciphertext < self.nsq:
            raise ValueError("密文必须在0到n^2-1之间")
        
        # m = L(c^λ mod n^2) * μ mod n
        # L(x) = (x-1)/n
        c_lam = pow(ciphertext, self.lam, self.nsq)
        l_c = (c_lam - 1) // self.n
        plaintext = (l_c * self.mu) % self.n
        return plaintext
    
    def add(self, c1, c2):
        # 同态加法：E(a) * E(b) mod n^2 = E(a+b mod n)
        return (c1 * c2) % self.nsq
    
    def scalar_mult(self, c, k):
        # 同态标量乘法：E(a)^k mod n^2 = E(a*k mod n)
        return pow(c, k, self.nsq)

# 使用示例
if __name__ == "__main__":
    paillier = Paillier()
    a = 42
    b = 23
    
    # 加密
    c_a = paillier.encrypt(a)
    c_b = paillier.encrypt(b)
    
    # 密文上计算
    c_sum = paillier.add(c_a, c_b)
    c_scalar = paillier.scalar_mult(c_a, 3)
    
    # 解密结果
    print(f"原始明文: {a}, {b}")
    print(f"解密后和: {paillier.decrypt(c_sum)}")  # 应输出65
    print(f"解密后标量乘积: {paillier.decrypt(c_scalar)}")  # 应输出126

import numpy as np

class DifferentialPrivacy:
    def __init__(self, epsilon=1.0, sensitivity=1.0):
        self.epsilon = epsilon  # 隐私预算
        self.sensitivity = sensitivity  # 敏感度
    
    def laplace_mechanism(self, true_value):
        """使用Laplace机制添加噪声"""
        # 计算噪声尺度参数
        scale = self.sensitivity / self.epsilon
        # 从Laplace分布采样噪声
        noise = np.random.laplace(0, scale)
        # 返回添加噪声后的结果
        return true_value + noise
    
    def gaussian_mechanism(self, true_value):
        """使用高斯机制添加噪声"""
        # 计算噪声标准差
        # 这里使用常见的参数设置，δ为失效概率
        delta = 1e-5
        sigma = self.sensitivity * np.sqrt(2 * np.log(1.25 / delta)) / self.epsilon
        # 从高斯分布采样噪声
        noise = np.random.normal(0, sigma)
        # 返回添加噪声后的结果
        return true_value + noise
    
    def composition(self, queries, epsilon_per_query, delta=1e-5):
        """处理组合查询的隐私预算分配"""
        # 简单组合：预算相加
        total_epsilon = len(queries) * epsilon_per_query
        print(f"简单组合后的总隐私预算: {total_epsilon}")
        
        # 高级组合：使用高级组合定理
        advanced_epsilon = np.sqrt(2 * len(queries) * np.log(1/delta)) * epsilon_per_query
        print(f"高级组合后的总隐私预算: {advanced_epsilon}")
        
        return total_epsilon, advanced_epsilon

# 使用示例
if __name__ == "__main__":
    dp = DifferentialPrivacy(epsilon=0.5, sensitivity=1.0)
    
    # 原始统计查询结果
    true_count = 100
    
    # 添加噪声
    noisy_count_laplace = dp.laplace_mechanism(true_count)
    noisy_count_gaussian = dp.gaussian_mechanism(true_count)
    
    print(f"原始计数: {true_count}")
    print(f"Laplace机制添加噪声后: {noisy_count_laplace}")
    print(f"Gaussian机制添加噪声后: {noisy_count_gaussian}")
    
    # 组合查询示例
    dp.composition(["查询1", "查询2", "查询3"], 0.1)

import numpy as np
from sklearn.linear_model import LogisticRegression

# 模拟联邦学习场景
class ReconstructionAttack:
    def __init__(self, model_weights):
        self.model_weights = model_weights
    
    def simple_reconstruction(self, feature_dim, num_samples=10):
        """简单的权重重构攻击"""
        # 假设特征是二值的
        reconstructed_data = []
        
        for _ in range(num_samples):
            # 根据权重绝对值大小推测特征值
            sample = np.zeros(feature_dim)
            # 为权重大的特征分配值
            top_indices = np.argsort(np.abs(self.model_weights))[-3:]
            sample[top_indices] = np.sign(self.model_weights[top_indices])
            reconstructed_data.append(sample)
        
        return np.array(reconstructed_data)

# 使用示例
if __name__ == "__main__":
    # 假设这是从联邦学习中获取的模型权重
    model_weights = np.array([0.5, -0.3, 0.8, -0.2, 0.4, -0.6, 0.7, -0.1, 0.3, -0.4])
    
    attacker = ReconstructionAttack(model_weights)
    reconstructed_data = attacker.simple_reconstruction(feature_dim=10)
    
    print("重构的数据样本:")
    print(reconstructed_data)

import numpy as np
from sklearn.ensemble import RandomForestClassifier

class MembershipInferenceAttack:
    def __init__(self, target_model, shadow_models=5):
        self.target_model = target_model
        self.shadow_models = shadow_models
        self.attack_model = RandomForestClassifier()
    
    def train_shadow_models(self, shadow_data, shadow_labels):
        """训练影子模型模拟目标模型行为"""
        shadow_preds = []
        shadow_membership = []
        
        # 为每个影子模型划分成员和非成员数据
        n_samples = len(shadow_data)
        for i in range(self.shadow_models):
            # 随机划分成员和非成员数据
            member_idx = np.random.choice(n_samples, size=n_samples//2, replace=False)
            non_member_idx = np.setdiff1d(np.arange(n_samples), member_idx)
            
            # 训练影子模型
            shadow_model = RandomForestClassifier()
            shadow_model.fit(shadow_data[member_idx], shadow_labels[member_idx])
            
            # 获取成员和非成员的预测概率
            member_probs = shadow_model.predict_proba(shadow_data[member_idx])
            non_member_probs = shadow_model.predict_proba(shadow_data[non_member_idx])
            
            # 收集预测结果和成员信息
            shadow_preds.extend(member_probs)
            shadow_preds.extend(non_member_probs)
            shadow_membership.extend([1]*len(member_probs))
            shadow_membership.extend([0]*len(non_member_probs))
        
        # 训练攻击模型
        self.attack_model.fit(shadow_preds, shadow_membership)
    
    def attack(self, test_data, test_labels):
        """执行成员推断攻击"""
        # 获取目标模型对测试样本的预测概率
        target_probs = self.target_model.predict_proba(test_data)
        
        # 使用攻击模型判断样本是否为成员
        membership_preds = self.attack_model.predict(target_probs)
        membership_probs = self.attack_model.predict_proba(target_probs)
        
        return membership_preds, membership_probs

import numpy as np
from cryptography.hazmat.primitives.asymmetric import paillier

class SecureAggregation:
    def __init__(self, num_clients):
        self.num_clients = num_clients
        self.public_keys = []
        self.private_keys = []
        
        # 为每个客户端生成密钥对
        for _ in range(num_clients):
            private_key, public_key = paillier.generate_paillier_keypair()
            self.private_keys.append(private_key)
            self.public_keys.append(public_key)
    
    def client_encrypt(self, model_update, client_id):
        """客户端加密模型更新"""
        encrypted_update = []
        for param in model_update:
            # 使用Pailier加密每个参数
            encrypted_param = self.public_keys[client_id].encrypt(param)
            encrypted_update.append(encrypted_param)
        return encrypted_update
    
    def server_aggregate(self, encrypted_updates):
        """服务器聚合加密的模型更新"""
        # 假设所有客户端的模型结构相同
        aggregated_update = []
        
        # 聚合每个参数位置
        num_params = len(encrypted_updates[0])
        for i in range(num_params):
            # 初始化聚合值
            agg_param = None
            for client_update in encrypted_updates:
                if agg_param is None:
                    agg_param = client_update[i]
                else:
                    # Paillier加法同态性质
                    agg_param += client_update[i]
            
            # 除以客户端数量进行平均
            aggregated_update.append(agg_param)
        
        return aggregated_update
    
    def decrypt_aggregated(self, aggregated_update):
        """解密聚合结果（通常由可信第三方完成）"""
        decrypted_update = []
        for param in aggregated_update:
            # 使用私钥解密
            decrypted_param = self.private_keys[0].decrypt(param)
            # 除以客户端数量得到平均值
            decrypted_param = decrypted_param / self.num_clients
            decrypted_update.append(decrypted_param)
        return decrypted_update

import numpy as np

class DP_FederatedLearning:
    def __init__(self, epsilon_global=1.0, delta=1e-5):
        self.epsilon_global = epsilon_global  # 全局隐私预算
        self.delta = delta  # 失效概率
    
    def add_noise_to_update(self, model_update, clip_norm, num_clients, round_num):
        """向模型更新添加差分隐私噪声"""
        # 计算本轮使用的隐私预算（按轮次分配）
        epsilon_round = self.epsilon_global / round_num
        
        # 计算噪声标准差
        sigma = clip_norm * np.sqrt(2 * np.log(1.25 / self.delta)) / (epsilon_round / num_clients)
        
        # 对模型更新进行裁剪
        update_norm = np.linalg.norm(model_update)
        if update_norm > clip_norm:
            model_update = model_update * (clip_norm / update_norm)
        
        # 添加高斯噪声
        noisy_update = model_update + np.random.normal(0, sigma, size=model_update.shape)
        
        return noisy_update
    
    def aggregate_updates(self, updates, clip_norm, round_num):
        """聚合带差分隐私保护的模型更新"""
        num_clients = len(updates)
        aggregated_update = np.zeros_like(updates[0])
        
        for update in updates:
            # 对每个更新添加噪声
            noisy_update = self.add_noise_to_update(update, clip_norm, num_clients, round_num)
            aggregated_update += noisy_update
        
        # 计算平均值
        aggregated_update = aggregated_update / num_clients
        
        return aggregated_update

table
  技术类型 | 密码学安全 | 协议安全 | 实现安全 | 性能安全 | 法规合规
  同态加密 | ■■■■ | ■■■ | ■■■■ | ■■ | ■■■■
  差分隐私 | ■■■■ | ■■■■ | ■■■ | ■■■ | ■■■■
  联邦学习 | ■■■ | ■■ | ■■■ | ■■■■ | ■■■

### 5.2 隐私泄露风险评估

**风险评估工具实现**：

```python
import numpy as np

def calculate_privacy_loss(epsilon, delta):
    """计算隐私损失度量"""
    # 使用(ε,δ)-差分隐私作为隐私损失度量
    return {
        "epsilon": epsilon,
        "delta": delta,
        "privacy_loss_parameter": epsilon,
        "recommended_threshold": epsilon <= 1.0 and delta <= 1e-5
    }

def evaluate_reconstruction_risk(model_params, data_dim, attack_strength=0.8):
    """评估模型参数的重构风险"""
    # 模型复杂度影响重构风险
    param_count = np.prod(model_params.shape)
    complexity_ratio = param_count / data_dim
    
    # 重构风险评分（0-10）
    risk_score = min(10, complexity_ratio * attack_strength * 10)
    
    return {
        "risk_score": risk_score,
        "risk_level": "高" if risk_score > 7 else ("中" if risk_score > 4 else "低"),
        "recommendation": "实施模型压缩或添加差分隐私" if risk_score > 7 else 
                          "考虑添加防护措施" if risk_score > 4 else "风险可接受"
    }

def membership_inference_risk_estimate(model_accuracy, dataset_size):
    """估计成员推断攻击风险"""
    # 准确率越高，成员推断风险越大
    # 数据集越小，成员推断风险越大
    base_risk = model_accuracy * (1000 / max(dataset_size, 1000))
    
    # 风险评分（0-10）
    risk_score = min(10, base_risk * 10)
    
    return {
        "risk_score": risk_score,
        "risk_level": "高" if risk_score > 7 else ("中" if risk_score > 4 else "低"),
        "factors": {
            "model_accuracy": model_accuracy,
            "dataset_size": dataset_size
        }
    }