简介：基于同态加密的联邦学习安全聚合系统源码，是一份适合毕业设计、课程设计与期末大作业的完整工程，面向有一定Python基础但希望快速上手联邦学习与隐私计算的学生。压缩包内共55个文件，整体大小仅2.08MB，核心包括31个模型及模块文件（model/modle）、9个Python源文件、4个批处理启动脚本，以及XML配置、密钥证书和README说明文档；结构清晰，注释覆盖关键模块，新手也能轻松部署。系统实现了同态加密下的模型安全聚合流程，功能完善、界面美观、操作简便，附带的启动脚本与说明可直接运行或二次扩展，已有97人学习使用。通过这份源码，读者既能获得一套可演示的课程设计成果，也能深入理解同态加密与联邦学习结合的关键技术细节，是一份实践价值较高的高分参考项目。
1. 为什么同态加密能让联邦学习服务器变成“可安全打工人”
想象一下，三家医院用自己的病例数据训练同一个诊断模型，签署了不能交原始数据的协议。于是它们用联邦学习只交换梯度，但服务器拿到这些梯度之后，依然有办法通过“梯度反推”恢复出接近原始样本的图像。这不是危言耸听，近几年的攻击论文已经把这类攻击做成了开源工具，同样也是联邦学习综述里反复讨论的安全聚合背景。基于同态加密的联邦学习安全聚合系统源码，要做的就是让服务器只看到密文梯度，在密文上完成加权平均，最后才由不参与聚合的一方解密。这样一来，服务器从“能看到所有人更新”变成一个“只能做密文加法的打工人”。这个项目既能当作密码学课程设计，也能作为毕业设计里高分离不开的模块，技术栈覆盖面足够深，也足够贴近工业级的隐私保护预期。下文从原理到实现，把这条路走一遍。
2. 同态加密与联邦学习安全聚合：原理与选型
在动手写代码前，先要明确两件事：威胁模型是谁，以及聚合过程需要什么运算。这两件事直接决定了选型。
2.1 梯度泄露威胁与安全聚合要解决的问题
联邦学习通常假设客户端数据不离开本地，但服务端确实会收到每个客户端的模型更新。大量针对梯度泄漏攻击（Gradient Leakage）的实验表明，恶意服务器可以上传一个随机假梯度，然后通过不断调整输入，使其和目标梯度非常接近，从而重建原始训练数据。即使只有几十维的梯度，也能恢复出清晰的文本或图片，这就是“梯度不是隐私”的直接证据。
安全聚合（Secure Aggregation）的定义由此产生：N个客户端将自己的更新上传后，服务器只能得到这N个更新的“和”或“平均”，得不到任何单个客户端的更新。从数学上看，联邦平均是一个线性运算：全局模型等于各个本地模型乘以样本权重再求和。线性运算天然适合加法同态加密——我在密文上做加法，你解密后得到的就是明文和。因此，本标题里的“同态加密”并不是为了炫技，而是为了满足“服务器不可信”这一更严格的隐私前提。
2.2 同态加密核心机制与选型：Paillier 够用吗
公钥同态加密允许没有私钥的人直接在密文上进行某些运算。最常见的是Paillier方案：E(m1) * E(m2) = E(m1+m2)，同时密文有一个性质：E(m) ^ c = E(c*m)。这意味着服务器可以用客户端传上来的密文，做“密文相加”和“密文乘以一个整数”两种操作，恰好覆盖了加权平均的两个步骤。如果把每个模型的参数看成一个向量，那么整个更新过程就是对向量逐元素做同样的密文运算。
那为什么不用全同态？全同态加密（比如CKKS、BFV）支持乘法和加法，理论上可以做更复杂的聚合甚至密文推理，但工程开销明显更大。在横向联邦学习中，聚合服务器不需要对梯度做任何非线性操作，用Paillier这种加法同态就够了。很多开源的联邦学习安全聚合方案也都基于Paillier。下面用phe库跑一个最小密文聚合：
from phe import paillier # 生成密钥对，n_length 为模数位数，1024只用于演示 pub_key, priv_key = paillier.generate_paillier_keypair(n_length=1024) # 浮点梯度需要先放大成整数，Paillier只能处理群上的整数 scale = 1000 g1 = int(0.75 * scale) # 客户端1的梯度 -> 750 g2 = int(-0.25 * scale) # 客户端2的梯度 -> -250 # 每个客户端用同一个公钥加密自己的梯度 e1 = pub_key.encrypt(g1) e2 = pub_key.encrypt(g2) # 服务器端做密文加法，对应明文 750 + (-250) = 500 e_sum = e1 + e2 # 密文乘以整数权重 w=2，对应明文乘以2 w = 2 e_weighted = e1 * w # 只有持有私钥的协调者才能解密 decrypted_sum = priv_key.decrypt(e_sum) / scale decrypted_weighted = priv_key.decrypt(e_weighted) / scale print(decrypted_sum, decrypted_weighted) # 0.5 1.5这里有几个参数要解释。n_length=1024是安全参数，实际项目建议至少2048；scale是把浮点梯度放大成整数的精度因子，太大会让加密整数超过模数从而发生溢出，太小则放大噪声，聚合精度受到影响。此外，phe库的encrypt接口是逐元素加密的，如果你手里的梯度是一个向量，需要先把numpy数组转换成list再用列表推导逐值加密，不能直接把list传给encrypt。
2.3 系统角色与整体数据流：私钥到底放在哪里
有了密文运算基础，现在可以画出本系统的基本角色。这里的核心设计决定是：谁有权解密。如果服务器既负责聚合又持有私钥，那它完全可以不聚合、先解密、再偷看单个更新，同态加密就完全失效。所以常见做法是引入一个独立的“解密协调者”，或者用秘密共享把私钥拆成碎片分发给客户端，聚合完成后再联合解密。在高分源码里，推荐至少把聚合服务器和解密者拆成两个进程，并在文档中说明这个威胁模型。
| 角色 | 能看到的数据 | 可执行操作 | 推荐实现 |
|---|---|---|---|
| 客户端 | 本地明文梯度 | 用公钥加密、上传密文 | 每个客户端保存同一份公钥 |
| 聚合服务器 | 所有客户端密文 | 密文相加、密文乘以整数权重 | 不持有私钥，只暴露聚合接口 |
| 解密协调者 | 聚合后的密文 | 解密并广播平均梯度 | 独立服务，或由参与方共同解密的密钥分享 |
整体的数据流也非常清晰：每个客户端在本地用同样的随机种子初始化模型，训练若干轮后得到梯度；然后对梯度做scale量化，用协调者下发的公钥逐元素加密；服务器收到所有密文后，把相同参数名的密文列表逐位相加；聚合完成后，解密协调者拿到求和密文，用私钥解密再除以scale和参与客户端数的乘积，得到平均梯度。这个流程把“聚合”和“权限”分离，是整个源码可靠性的关键。
3. 搭一个最小可用的同态加密联邦学习安全聚合系统源码：客户端、服务器与解密协调者
在PySyft或Flower里加一个安全聚合插件当然更快，但作为高分项目，最好把加密和聚合的代码直接放在自己工程里，方便演示和答辩。下面我们从零搭一个最小可运行的Python版本。
3.1 环境准备与目录结构
我推荐使用PyTorch做本地训练，phe做Paillier加密，因为二者都有成熟的社区和文档。创建虚拟环境并安装依赖：
mkdir secure-fedavg && cd secure-fedavg python -m venv venv && source venv/bin/activate pip install torch torchvision phe numpy这里没有锁定具体版本，因为PyTorch版本与CUDA强相关，phe用1.4.0即可。如果你的机器没有GPU，把模型的device固定为CPU就行；如果你用GPU，记得在客户端训练前把模型和batch都.to(device)。
项目源码建议按职责拆成四个模块，而不是把所有逻辑塞进一个文件：
secure-fedavg/ ├── crypto.py # 密钥生成、加密/解密封装，含SCALE常量 ├── client.py # 本地训练，输入模型和数据，输出加密更新 ├── server.py # 聚合服务器，只处理密文 ├── evaluator.py # 解密协调者，解密平均更新 └── config.yaml # 全局配置：batch_size, lr, n_length, scaleconfig.yaml的价值在于让实验可复现。答辩时你需要跑多组对比实验，把scale、学习率、参与客户端数都写进配置，比直接改源码更有说服力。
3.2 客户端本地训练与加密上传
客户端的职责是完成几步本地梯度下降，然后返回加密后的梯度。下面代码是核心逻辑，省去了MNIST数据集加载的部分：
# client.py import torch import torch.nn as nn import numpy as np from phe import PaillierPublicKey SCALE = 1000 # 全局精度因子，实际应从config读取 def local_train_and_encrypt(model, train_loader, pub_key: PaillierPublicKey, device='cpu'): model.to(device) model.train() optimizer = torch.optim.SGD(model.parameters(), lr=0.01) criterion = nn.CrossEntropyLoss() encrypted_update = {} n_samples = 0 for images, labels in train_loader: images, labels = images.to(device), labels.to(device) optimizer.zero_grad() loss = criterion(model(images), labels) loss.backward() n_samples += labels.size(0) with torch.no_grad(): for name, param in model.named_parameters(): if param.grad is None: continue # 放大并取整，逐元素加密 grad_int = (param.grad.cpu().numpy() * SCALE).astype(int).reshape(-1).tolist() encrypted_update[name] = [pub_key.encrypt(int(v)) for v in grad_int] break # 注意：只处理一个batch，实际要多batch累计梯度 return encrypted_update, n_samples逻辑说明：这里先把梯度乘以SCALE再取整，原因是Paillier密文的明文必须是整数。encrypt是逐元素操作，所以用列表推导遍历整个梯度向量。break只是为了让代码在demo里快速返回，真实项目中应该完整跑完一个epoch或若干step后，将梯度累计在本地，再一次性加密上传。
参数说明：lr=0.01是本地学习率，它的量级会影响梯度大小，进而影响SCALE是否溢出。假如梯度均值是0.1，SCALE=1000后成为100，没问题；但如果梯度均值是100，加密后会得到100000，仍远小于2048位密钥能表示的范围，倒不至于出错。真正要留意的是模型大时梯度向量长度数万，逐个加密会非常慢，所以实际工程里常用“只加密顶层梯度”或“稀疏化后加密”。
3.3 服务器端密文聚合与解密协调者解密
服务器不知道公钥以外的任何信息，它只负责把同名的参数向量逐个相加：
# server.py def secure_aggregate(encrypted_updates): """encrypted_updates: list, 每个元素是客户端返回的 {name: [密文列表]}""" aggregated = {} for update in encrypted_updates: for name, cipher_list in update.items(): if name not in aggregated: aggregated[name] = list(cipher_list) else: aggregated[name] = [c1 + c2 for c1, c2 in zip(aggregated[name], cipher_list)] return aggregated解密协调者拿到加密和之后，用私钥解密再除以SCALE * num_clients，得到平均梯度：
# evaluator.py def decrypt_average(aggregated, priv_key, num_clients): plain_avg = {} for name, cipher_list in aggregated.items(): plain = [priv_key.decrypt(c) / (SCALE * num_clients) for c in cipher_list] plain_avg[name] = plain return plain_avg这里有一个容易踩的坑：num_clients必须是实际参与本轮聚合的客户端数，而不是你配置文件里的总客户端数。假设一开始有5个客户端，中途掉线1个，如果仍然用5做分母，解密出来的平均梯度比真实值小20%。更稳妥的做法是服务器在启动每轮聚合前先清点存活客户端，把存活列表和密文一起发给解密协调者，协调者用存活数量做分母。
客户端、服务器、协调者的参数也需要统一管理。下面的表格是几个最关键的参数：
| 参数 | 推荐值 | 调参影响 |
|---|---|---|
| n_length | 2048 | 越大越安全，但加解密耗时明显上升 |
| SCALE | 1000 | 控制浮点精度；过大会溢出，过小误差大 |
| batch_size | 32 | 影响本地梯度估计的噪声 |
| lr | 0.01 | 控制梯度量级，进而影响SCALE取值 |
| local_epochs | 1 | 本地迭代次数，过大可能加剧客户端漂移 |
提示：在联邦学习综述和公开实现中，常用“样本数加权平均”而不是简单平均。加权平均在Paillier下要小心处理：如果某个客户端有500个样本、另一个有1000个样本，服务器希望给后者两倍权重。因为Paillier只允许密文乘以整数，所以可以将权重先放大为整数（比如权重分别为500和1000），再做乘法后在解密时统一除以样本总数。直接传浮点数权重是不行的。
4. 联邦学习安全聚合的通信压缩、掉线容错与抗恶意攻击
完成上述最小系统后，评委大概率会问：同态加密把梯度从32位浮点变成2048位整数，通信开销是不是大得无法接受？如果有客户端中途掉线怎么办？如果服务器恶意提交伪造密文怎么办？这一章处理这三个典型问题。
4.1 偏置压缩与量化：用更少的密文传同样多的信息
不少人会忽略：phe加密一个整数得到的密文长度远大于明文。以2048位密钥为例，一个密文是两个模数大小的整数，序列化后通常有数百字节，而原始float32只有4字节。对一个百万参数模型，全量加密上传会直接让网络瘫痪。最常见的缓解手段是量化加稀疏化。
量化就是上面代码里的SCALE缩放取整，但还可以进一步用更少的bit来表示梯度值。bits=8时，把梯度裁剪到[-1,1]并映射到[-127,127]，能将梯度范围减小两个数量级，再配合压缩算法，整体通信量大幅下降。稀疏化则是只保留绝对值最大的k%元素，剩下置零。下面是top-k稀疏化的具体实现：
def topk_sparsify(grad_int, k=0.1): """只保留绝对值最大的 k% 梯度，返回索引和值""" grad_np = np.array(grad_int) threshold = np.quantile(np.abs(grad_np), 1 - k) mask = np.abs(grad_np) >= threshold indices = np.where(mask)[0].tolist() values = grad_np[mask].astype(int).tolist() return indices, values之后在服务器端，需要根据indices把稀疏梯度还原成原长度再相加。如果不做还原，后面的解密协调者会拿不到完整的参数向量。
这里要特别提一下“偏置压缩”。联邦学习领域已有大量研究表明，在联邦学习中采用偏置压缩技术可通过传输经过压缩的本地更新数据来减少通信开销，但压缩会引入有偏估计，导致模型不收敛或收敛到错误最小值。工程里通常加一个误差反馈（Error Feedback）：客户端把本轮没传出去的残差缓存下来，下轮合并后一起压缩。这一块代码虽然不难，但一旦评委问到“压缩之后精度掉了多少”，你需要给出带误差反馈和不带误差反馈的对比实验。
4.2 掩码与掉线容错：防止客户端更新在多次聚合中被还原
Paillier加密可以防止服务器偷看单轮梯度，但如果服务器发起多次查询，比如先让客户端A单独加密上传再让所有客户端全部上传，它就能通过两次密文相减得到单个客户端的梯度。要解决这个问题，需要在加密前给每个客户端加上随机的掩码（Mask），多客户端掩码之和为0，服务器只能得到和而不能逐个拆分。
最简单的实现是为每一对客户端生成一个成对掩码：客户端A对B生成随机掩码，客户端B对A生成同一个掩码但取负。比如下面代码：
# 假设两个客户端共享同一个随机种子 seed = b'shared_seed_2024' rng_a = np.random.default_rng(seed) mask_a = rng_a.integers(0, 2**32, size=4).astype(np.int64) mask_b = -mask_a # 相反掩码 # 客户端A在加密明文前加上mask_a，客户端B在加密明文前加上mask_b # 服务器聚合密文后，明文里的掩码相互抵消： # gradA + maskA + gradB + maskB = gradA + gradB这个简化版本的关键假设是：两个客户端安全地共享了掩码种子，而服务器不知道。真实场景中，这种种子分发通常需要用Diffie-Hellman密钥交换或可信协调者来完成，否则掩码形同虚设。完整的Secure Aggregation协议还会为每个客户端保存一份掩码的秘密共享碎片，当某个客户端掉线时，其他客户端无法自行算出它的掩码，但可以通过碎片恢复它应该贡献的掩码，从而把掉线者的更新从聚合结果中剔除。我们的项目可以先做简化版，再在文档里标注扩展方向，已经足以体现对威胁模型的理解。
4.3 验证脚本：一键确认聚合结果是正确的
开发过程中最好的习惯是写一个独立测试，加密的聚合结果能和明文聚合结果对上。下面是推荐的最小验证：
from phe import paillier def test_secure_aggregate(): pub, priv = paillier.generate_paillier_keypair(n_length=1024) raw = [[1, 2, 3], [4, 5, 6]] enc = [[pub.encrypt(x) for x in vec] for vec in raw] agg = [enc[0][i] + enc[1][i] for i in range(len(raw[0]))] dec = [priv.decrypt(x) for x in agg] assert dec == [5, 7, 9] if __name__ == "__main__": test_secure_aggregate() print("PASS")如果你在源码里加入了掩码、压缩和掉线处理，测试脚本也应该相应增加分支。常见的调试问题整理成表，方便快速定位：
| 现象 | 可能原因 | 定位方法 |
|---|---|---|
| 解密结果巨大或乱码 | 梯度溢出，SCALE太大，学习率过大 | 打印明文梯度分布并调低SCALE |
| 聚合结果与明文平均有细微误差 | 取整引入的量化误差 | 增大SCALE，或使用随机舍入 |
| 有客户端掉线后解密结果偏小 | 分母仍然用了总客户端数 | 将存活客户端数传给协调者 |
| 掩码生效后聚合结果错误 | 掩码没有被正确取反，或掉线客户端的掩码未抵消 | 打印掩码和，检查种子是否一致 |
5. 高分项目验收点：把安全聚合源码做成答辩中的亮点
你的源码如果只是把梯度加密、聚合、解密这三个流程串起来，拿到及格分没问题，但“高分”需要额外考虑可复现性、安全设计和实验对比。下表是自检清单，每一项都可以在答辩时展开讲：
| 验收维度 | 检查方法 | 常见扣分点 |
|---|---|---|
| 加密正确性 | 运行上面test_secure_aggregate，确认密文聚合与明文聚合一致 | 只有训练时能跑通，没有独立测试 |
| 隐私保护 | 看私钥是否保存到了服务器进程 | 把私钥直接放在server.py里等于没加密 |
| 鲁棒性 | 模拟客户端掉线、延迟返回、损坏密文 | 掉线直接抛异常，没有兜底逻辑 |
| 通信优化 | 对比全量加密与top-k压缩后的通信量 | 忽略通信开销，加密全部参数 |
| 可复现性 | requirements.txt、固定随机种子、config.yaml完整 | 缺少运行说明或随机种子不稳定 |
在这些维度里，我最建议你增加一个“故障注入”机制，因为大多数人的项目里不会写到，而它恰恰能证明你能处理真实环境问题。比如通过环境变量模拟某个客户端掉线：
CHURN_CLIENT_ID=2 python run_experiment.py --config config.yaml在客户端发送完密文后，服务器检测到CHURN_CLIENT_ID就自动把它从本轮参与列表中剔除，并通知解密协调者用新的客户端数做分母。这样既不需要改代码，又能快速跑出“掉线一个客户端”和“不掉线”两组实验，对比模型精度曲线。答辩时现场执行一次这个命令，评委看到日志中先打印“drop client 2”，随后正常聚合，说服力会明显比口述更强。
另一个容易出彩的小技巧是：在论文或文档里附一张两行折线图，一行是明文FedAvg（明文聚合），另一行是同态加密后的FedAvg，两条线几乎重合。这能直观证明同态加密对模型收敛的影响小到可以忽略。你在实验报告中补上这种图，比任何公式都更能让评委立刻理解“同态加密没有破坏联邦学习的目标”。把这套脚本写进README的一键复现命令里，你的安全聚合系统源码就不只是能跑——还能在答辩现场被评审亲手复现，这是高分项目最直观的加分项。
本文还有配套的精品资源，点击获取
