隐私计算与数据安全

import numpy as np

def laplace_mechanism(query_result, sensitivity, epsilon):
    """
    拉普拉斯机制：添加服从拉普拉斯分布的噪声
    
    query_result: 真实查询结果
    sensitivity: 查询的敏感度（单个记录变化对结果的最大影响）
    epsilon: 隐私预算（越小隐私保护越强，但噪声越大）
    """
    scale = sensitivity / epsilon
    noise = np.random.laplace(0, scale)
    return query_result + noise

# 示例：Count 查询
sensitivity = 1  # 添加/删除一个人最多改变 count 1
epsilon = 0.1    # 较强的隐私保护

real_count = 1000
private_count = laplace_mechanism(real_count, sensitivity, epsilon)
print(f"真实值: {real_count}, 差分隐私值: {private_count:.2f}")

# 隐私预算累积
epsilon_total = 0.0
queries = [
    ("avg_salary", 1000),
    ("count_employees", 1),
    ("max_age", 100),
]

for query_name, sensitivity in queries:
    epsilon = 0.1
    if epsilon_total + epsilon > 1.0:  # 总预算上限
        print(f"隐私预算耗尽，无法执行 {query_name}")
        break
    result = laplace_mechanism(get_result(query_name), sensitivity, epsilon)
    epsilon_total += epsilon

import torch
import torch.nn as nn
from torch.utils.data import DataLoader

class FederatedLearning:
    def __init__(self, global_model, clients):
        self.global_model = global_model
        self.clients = clients  # 各参与方的本地数据
    
    def train_round(self):
        """
        一轮联邦学习：
        1. 服务端下发全局模型
        2. 各客户端本地训练
        3. 客户端上传梯度/模型更新
        4. 服务端聚合（FedAvg）
        """
        local_updates = []
        
        for client in self.clients:
            # 客户端本地训练（数据不出本地）
            local_model = copy.deepcopy(self.global_model)
            local_update = client.local_train(local_model)
            local_updates.append(local_update)
        
        # 服务端聚合（仅聚合梯度，不接触原始数据）
        self.global_model = self.federated_avg(local_updates)
        
    def federated_avg(self, local_updates):
        """FedAvg 算法"""
        global_dict = self.global_model.state_dict()
        
        for k in global_dict.keys():
            global_dict[k] = torch.stack([
                update[k] * client.weight 
                for update, client in zip(local_updates, self.clients)
            ]).sum(dim=0)
        
        self.global_model.load_state_dict(global_dict)
        return self.global_model

// Intel SGX 应用示例
#include "sgx_urts.h"
#include "enclave_u.h"

sgx_enclave_id_t eid;
sgx_status_t ret = sgx_create_enclave("enclave.signed.so", SGX_DEBUG_FLAG, &token, &updated, &eid, NULL);

// 在 Enclave 内部执行敏感计算
int result;
ecall_secure_computation(eid, &ret, encrypted_input, &encrypted_output);

// Enclave 内存对外部不可见，即使 root 权限也无法读取