联邦学习隐私防御机制

联邦学习虽然通过数据最小化提供了基础隐私保护，但模型更新本身仍可能泄露敏感信息。本章系统介绍差分隐私、安全聚合等隐私保护机制的理论基础和实践方法。

1. 差分隐私基础

1.1 定义

差分隐私（Differential Privacy, DP）¹是隐私保护的标准数学框架：

P (M (D) \in S) \leq e^{ϵ} \cdot P (M (D^{'}) \in S) + δ

其中：

$M$ ：隐私机制
$D, D^{'}$ ：仅相差一个样本的相邻数据集
$ϵ$ ：隐私预算
$δ$ ：失败概率

1.2 隐私保证的性质

性质	描述
组合性	连续应用隐私机制，隐私预算累加
后处理不变性	对DP输出的任意处理不增加隐私风险
群体隐私	保护 $k$ 个个体的隐私等价于 $ϵ / k$ 的个体隐私

1.3 敏感度

全局敏感度定义为相邻数据集上函数输出的最大差异：

Δ f = D, D^{'} max ∥ f (D) - f (D^{'}) ∥_{p}

对于梯度下降，全局敏感度与梯度裁剪阈值相关：

Δ g = C (裁剪阈值)

2. 用户级差分隐私

2.1 从样本级到用户级

联邦学习中的隐私保护单元是用户（客户端）而非单个样本：

def user_level_dp(client_gradient, clip_norm=1.0, noise_scale=0.1):
    """
    用户级差分隐私
    """
    # 步骤1：梯度裁剪
    # 每个用户贡献的梯度被裁剪到固定范数
    grad_norm = torch.norm(client_gradient)
    if grad_norm > clip_norm:
        clipped_grad = client_gradient * (clip_norm / grad_norm)
    else:
        clipped_grad = client_gradient
    
    # 步骤2：添加高斯噪声
    # 噪声标准差与敏感度和隐私参数相关
    noise = torch.randn_like(clipped_grad) * (noise_scale * clip_norm)
    private_grad = clipped_grad + noise
    
    return private_grad

2.2 隐私会计

RDP（Renyi差分隐私） 用于精确计算隐私损失：

def compute_epsilon_rdp(q, sigma, steps, alpha=20):
    """
    计算RDP隐私参数
    q: 采样率
    sigma: 噪声标准差
    steps: 迭代步数
    alpha: Renyi阶数
    """
    # 高斯机制的RDP
    eps_alpha = (alpha * q**2) / (2 * sigma**2)
    
    # 组合多步
    total_eps = steps * eps_alpha
    
    # 转换为 (ε, δ)-DP
    delta = 1e-5
    epsilon = total_eps + np.sqrt(2 * np.log(1/delta) * total_eps)
    
    return epsilon, delta

2.3 DP-FTRL算法

Google在Gboard中部署的DP-FTRL算法²：

class DP_FTRL:
    """
    差分隐私Follow-the-Regularized-Leader
    用于联邦学习的隐私保护训练
    """
    def __init__(self, epsilon=1.0, delta=1e-5, 
                 clip_norm=1.0, sensitivity=1.0):
        self.epsilon = epsilon
        self.delta = delta
        self.clip_norm = clip_norm
        self.sensitivity = sensitivity
        self.noise_scale = self._compute_noise_scale()
    
    def _compute_noise_scale(self):
        """计算噪声规模"""
        # 基于隐私会计计算
        c = 128 / np.sqrt(self.epsilon * np.log(1/self.delta))
        return c * self.sensitivity / self.clip_norm
    
    def local_update(self, model, local_data, learning_rate):
        """本地更新"""
        optimizer = torch.optim.SGD(model.parameters(), lr=learning_rate)
        
        for batch in local_data:
            optimizer.zero_grad()
            loss = compute_loss(model, batch)
            loss.backward()
            
            # 裁剪梯度
            self._clip_gradients(model)
            
            # 聚合后添加噪声（在服务器端）
            optimizer.step()
        
        return self._extract_gradients(model)
    
    def _clip_gradients(self, model):
        """梯度裁剪"""
        total_norm = 0.0
        for p in model.parameters():
            if p.grad is not None:
                total_norm += p.grad.data.norm(2).item() ** 2
        total_norm = total_norm ** 0.5
        
        clip_coef = self.clip_norm / (total_norm + 1e-6)
        if clip_coef < 1:
            for p in model.parameters():
                if p.grad is not None:
                    p.grad.data.mul_(clip_coef)

3. 安全聚合协议（SecAgg）

3.1 协议概述

安全聚合（Secure Aggregation, SecAgg）³确保服务器只能看到聚合结果，无法获取个体更新：

客户端1 ──┐                      ┌── 聚合结果
客户端2 ──┼── 加密通信 ── 服务器 ──┘
  ...    │                     
客户端K ──┘

3.2 密钥分享机制

class SecureAggregation:
    """
    安全聚合协议
    """
    def __init__(self, num_clients, threshold=None):
        self.num_clients = num_clients
        self.threshold = threshold or (num_clients // 2 + 1)
        self.masks = {}  # 客户端掩码
    
    def setup_phase(self):
        """
        设置阶段：生成掩码
        """
        # 每个客户端生成随机掩码
        for client_id in range(self.num_clients):
            self.masks[client_id] = self._generate_mask()
    
    def _generate_mask(self):
        """生成掩码份额"""
        # 使用Shamir秘密分享
        mask = torch.randn(1000)  # 模型参数维度
        shares = self._shamir_share(mask, self.threshold, self.num_clients)
        return shares
    
    def _shamir_share(self, secret, threshold, num_shares):
        """
        Shamir (t, n) 秘密分享
        secret = sum_{i=1}^{t} a_i * x^{i-1} 在 x=0 处取值
        """
        coeffs = [secret] + [torch.randn_like(secret) for _ in range(threshold - 1)]
        shares = {}
        for x in range(1, num_shares + 1):
            share = sum(coeffs[i] * (x ** i) for i in range(len(coeffs)))
            shares[x] = share
        return shares
    
    def client_send_masked_update(self, client_id, gradient):
        """
        客户端：发送掩码更新
        """
        masked_update = gradient.clone()
        
        # 添加与其他客户端的掩码差值
        for other_id in range(self.num_clients):
            if other_id != client_id:
                # 只添加自己知道的掩码部分
                masked_update += self._get_mask_diff(client_id, other_id)
        
        return masked_update
    
    def _get_mask_diff(self, client_a, client_b):
        """获取掩码差值"""
        # 两个客户端协商的掩码
        mask_ab = self._derive_pairwise_mask(client_a, client_b)
        return mask_ab
    
    def server_aggregate(self, masked_updates):
        """
        服务器：聚合掩码更新
        """
        # 服务器只能看到聚合结果
        aggregated = torch.zeros_like(list(masked_updates.values())[0])
        
        for update in masked_updates.values():
            aggregated += update
        
        # 掩码在聚合后相互抵消
        # 服务器恢复真实聚合梯度
        
        return aggregated

3.3 丢包处理

实际部署中必须处理客户端丢包：

class RobustSecAgg:
    """
    鲁棒安全聚合：处理丢包
    """
    def __init__(self, expected_clients, min_clients_required):
        self.expected_clients = expected_clients
        self.min_required = min_clients_required
    
    def handle_dropout(self, participated_clients, dropped_clients):
        """
        处理客户端掉线
        """
        # 确保仍有足够客户端参与
        if len(participated_clients) < self.min_required:
            raise ValueError("Too many clients dropped")
        
        # 对于掉线客户端，恢复其掩码份额
        recovered_masks = {}
        for dropped_id in dropped_clients:
            recovered_masks[dropped_id] = self._recover_dropped_mask(
                dropped_id, participated_clients
            )
        
        return recovered_masks
    
    def _recover_dropped_mask(self, dropped_id, available_clients):
        """
        恢复掉线客户端的掩码
        """
        # 使用可用客户端的掩码份额重构
        shares = []
        for client_id in available_clients[:self.threshold]:
            shares.append(self.mask_shares[client_id][dropped_id])
        
        # 拉格朗日插值恢复
        recovered = self._lagrange_interpolate(shares, dropped_id)
        return recovered

4. 分布式差分隐私

4.1 核心思想

分布式差分隐私⁴在客户端本地添加噪声，然后通过安全聚合消除噪声的同时保护个体隐私：

class DistributedDifferentialPrivacy:
    """
    分布式差分隐私实现
    """
    def __init__(self, epsilon=1.0, num_clients=100):
        self.epsilon = epsilon
        self.num_clients = num_clients
        self.noise_scale = 1.0 / (epsilon * np.sqrt(num_clients))
    
    def local_phase(self, client_gradient):
        """
        本地阶段：添加本地噪声
        """
        # 裁剪
        clipped = self._clip(client_gradient)
        
        # 添加本地噪声
        local_noise = torch.randn_like(clipped) * self.noise_scale
        private_update = clipped + local_noise
        
        return private_update
    
    def secure_aggregation_phase(self, private_updates):
        """
        安全聚合阶段：噪声在聚合时抵消
        """
        # 安全聚合所有客户端的私有更新
        aggregated = self._secure_sum(private_updates)
        
        # 归一化（噪声在聚合后被平均化）
        normalized = aggregated / self.num_clients
        
        return normalized

4.2 隐私分析

分布式DP的隐私保证：

P (观察到 o) \leq e^{ϵ} \cdot P (观察到 o^{'})

关键性质：

单个客户端的噪声贡献被 $n$ 个客户端平均
服务器无法区分诚实客户端和任何其他客户端的贡献
满足信息论意义上的隐私

5. ByITFL与LoByITFL

5.1 ByITFL：完美信息论隐私

ByITFL⁵使用Lagrange编码计算实现完美信息论隐私：

class ByITFL:
    """
    ByITFL: 拜占庭鲁棒 + 完美信息论隐私
    """
    def __init__(self, epsilon, num_clients, malicious_threshold):
        self.epsilon = epsilon
        self.num_clients = num_clients
        self.malicious_threshold = malicious_threshold
    
    def lagrange_coded_update(self, gradients):
        """
        Lagrange编码更新
        """
        # 将更新编码为多项式
        poly_coeffs = self._create_polynomial(gradients)
        
        # 在不同点求值
        encoded_updates = {}
        for client_id in range(1, self.num_clients + 1):
            encoded_updates[client_id] = self._evaluate_poly(
                poly_coeffs, client_id
            )
        
        return encoded_updates
    
    def aggregate_with_privacy(self, received_updates, client_ids):
        """
        私密聚合
        """
        # 使用Lagrange系数进行私密恢复
        lambda_0 = self._compute_lagrange_coeff(0, client_ids)
        
        # 聚合
        aggregated = torch.zeros_like(list(received_updates.values())[0])
        for client_id, update in received_updates.items():
            aggregated += lambda_0[client_id] * update
        
        return aggregated

5.2 LoByITFL：降低通信开销

class LoByITFL:
    """
    LoByITFL: 低通信开销的ByITFL
    需要可信第三方（仅在初始化阶段）
    """
    def __init__(self, epsilon, ttp_available=True):
        self.ttp_available = ttp_available
        self.epsilon = epsilon
    
    def low_communication_encode(self, gradient):
        """
        低通信编码
        """
        # 量化梯度
        quantized = self._quantize(gradient, bits=8)
        
        # 添加隐私噪声
        noise = self._generate_noise(gradient.shape, self.epsilon)
        private_gradient = quantized + noise
        
        return private_gradient
    
    def decode_and_aggregate(self, private_gradients):
        """
        解码并聚合
        """
        # 简单平均（由于量化和噪声）
        aggregated = torch.mean(
            torch.stack(private_gradients), dim=0
        )
        return aggregated

6. 隐私-效用权衡

6.1 权衡分析

隐私保护与模型效用之间存在基本权衡：

效用 ↑\Rightarrow 隐私 ↓

def privacy_utility_tradeoff(epsilon, delta, dataset_size, model_complexity):
    """
    分析隐私-效用权衡
    """
    # 噪声标准差与隐私参数的关系
    noise_std = np.sqrt(model_complexity / (epsilon * np.sqrt(dataset_size)))
    
    # 估计效用损失
    utility_loss = noise_std * np.sqrt(np.log(1/delta))
    
    # 最优隐私预算
    optimal_epsilon = np.sqrt(model_complexity / (dataset_size * utility_loss**2))
    
    return {
        'noise_std': noise_std,
        'utility_loss': utility_loss,
        'optimal_epsilon': optimal_epsilon
    }

6.2 实际部署指南

应用场景	推荐隐私预算	模型效用
研究/探索	ε=8-10	高
工业应用	ε=2-5	中
医疗/金融	ε=0.5-1	低-中
极高隐私需求	ε<0.5	较低

6.3 隐私放大技术

class AmplifiedPrivacy:
    """
    隐私放大技术
    """
    def __init__(self, base_epsilon, amplification_method='sampling'):
        self.base_epsilon = base_epsilon
        self.amplification_method = amplification_method
    
    def amplify_by_sampling(self, sampling_probability, num_rounds):
        """
        通过随机采样放大隐私
        采样率 p → 隐私放大因子 √(1/p)
        """
        amplified_epsilon = self.base_epsilon * np.sqrt(sampling_probability)
        total_epsilon = amplified_epsilon * np.sqrt(num_rounds)
        return total_epsilon
    
    def amplify_by_shuffling(self, min_group_size):
        """
        通过随机洗牌放大隐私
        """
        # 洗牌提供匿名保证
        amplification_factor = 1.0 / np.sqrt(min_group_size)
        return self.base_epsilon * amplification_factor

7. 实践建议

7.1 部署清单

class PrivacyPreservingFLDeployment:
    """
    隐私保护联邦学习部署清单
    """
    def __init__(self):
        self.checks = {
            'differential_privacy': False,
            'secure_aggregation': False,
            'gradient_clipping': False,
            'noise_calibration': False,
            'privacy_accounting': False
        }
    
    def deploy(self, config):
        """
        部署配置
        """
        # 1. 差分隐私配置
        if config.get('enable_dp', True):
            self._setup_differential_privacy(
                epsilon=config.get('epsilon', 2.0),
                delta=config.get('delta', 1e-5),
                clip_norm=config.get('clip_norm', 1.0)
            )
        
        # 2. 安全聚合配置
        if config.get('enable_secagg', True):
            self._setup_secure_aggregation(
                protocol=config.get('secagg_protocol', 'standard'),
                threshold=config.get('threshold', None)
            )
        
        # 3. 隐私会计
        self._setup_privacy_accounting(
            mechanism=config.get('accounting', 'rdp')
        )
        
        return self._verify_deployment()
    
    def _verify_deployment(self):
        """验证部署安全性"""
        all_passed = all(self.checks.values())
        return {
            'deployment_ready': all_passed,
            'checks': self.checks
        }

7.2 性能基准

配置	隐私预算(ε, δ)	CIFAR-10准确率	隐私保证
无DP	-	85.2%	无
DP (ε=10)	(10, 1e-5)	82.1%	弱
DP (ε=5)	(5, 1e-5)	78.5%	中
DP (ε=2)	(2, 1e-5)	72.3%	强
DP+SecAgg	(2, 1e-5)	72.1%	最强

8. 总结

联邦学习的隐私保护需要多层次防御：

差分隐私：数学可证明的隐私保证
安全聚合：防止服务器窥探个体更新
分布式DP：结合两者的优势
隐私会计：精确追踪隐私损失
隐私放大：通过采样/洗牌增强隐私

关键考虑因素：

隐私强度与模型效用的权衡
计算与通信开销
威胁模型假设
实际部署约束

参考资料

Dwork et al. “The Algorithmic Foundations of Differential Privacy” (Foundations and Trends in TCSS 2014) ↩
Google: “Federated Learning with Formal Differential Privacy Guarantees” (2022) ↩
Bonawitz et al. “Practical Secure Aggregation for Privacy-Preserving Machine Learning” (CCS 2017) ↩
Google: “Distributed Differential Privacy for Federated Learning” (2023) ↩
ByITFL: “Perfect Privacy for Discriminator-Based Byzantine-Resilient Federated Learning” (arXiv:2506.13561) ↩

Metaphor

探索

Federated Learning Privacy Defense

联邦学习隐私防御机制

1. 差分隐私基础

1.1 定义

1.2 隐私保证的性质

1.3 敏感度

2. 用户级差分隐私

2.1 从样本级到用户级

2.2 隐私会计

2.3 DP-FTRL算法

3. 安全聚合协议（SecAgg）

3.1 协议概述

3.2 密钥分享机制

3.3 丢包处理

4. 分布式差分隐私

4.1 核心思想

4.2 隐私分析

5. ByITFL与LoByITFL

5.1 ByITFL：完美信息论隐私

5.2 LoByITFL：降低通信开销

6. 隐私-效用权衡

6.1 权衡分析

6.2 实际部署指南

6.3 隐私放大技术

7. 实践建议

7.1 部署清单

7.2 性能基准

8. 总结

参考资料

关系图谱

目录

Metaphor

探索

Federated Learning Privacy Defense

联邦学习隐私防御机制

1. 差分隐私基础

1.1 定义

1.2 隐私保证的性质

1.3 敏感度

2. 用户级差分隐私

2.1 从样本级到用户级

2.2 隐私会计

2.3 DP-FTRL算法

3. 安全聚合协议（SecAgg）

3.1 协议概述

3.2 密钥分享机制

3.3 丢包处理

4. 分布式差分隐私

4.1 核心思想

4.2 隐私分析

5. ByITFL与LoByITFL

5.1 ByITFL：完美信息论隐私

5.2 LoByITFL：降低通信开销

6. 隐私-效用权衡

6.1 权衡分析

6.2 实际部署指南

6.3 隐私放大技术

7. 实践建议

7.1 部署清单

7.2 性能基准

8. 总结

参考资料

Footnotes

关系图谱

目录