008_具身人工智能的数据隐私保护：联邦学习与同态加密在多传感器环境中的应用

# 示例：具身AI系统中的联邦学习实现
import tensorflow as tf
import numpy as np
from tensorflow_federated import tff

def create_federated_model():
    # 创建基础模型
    model = tf.keras.Sequential([
        tf.keras.layers.Input(shape=(224, 224, 3)),
        tf.keras.layers.Conv2D(32, (3, 3), activation='relu'),
        tf.keras.layers.MaxPooling2D((2, 2)),
        tf.keras.layers.Flatten(),
        tf.keras.layers.Dense(10, activation='softmax')
    ])
    
    return model

def create_client_data_sources(sensor_data_list):
    # 创建客户端数据集
    client_datasets = []
    for data in sensor_data_list:
        # 将传感器数据转换为TF数据集
        dataset = tf.data.Dataset.from_tensor_slices((
            np.array([sample['features'] for sample in data]),
            np.array([sample['label'] for sample in data])
        )).batch(32)
        client_datasets.append(dataset)
    
    return client_datasets

def train_federated_model(client_datasets):
    # 创建联邦学习环境
    federated_data = tff.simulation.compose_dataset_computation(
        client_datasets
    )
    
    # 定义联邦学习过程
    iterative_process = tff.learning.build_federated_averaging_process(
        model_fn=create_federated_model,
        client_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=0.01),
        server_optimizer_fn=lambda: tf.keras.optimizers.SGD(learning_rate=1.0)
    )
    
    # 初始化联邦学习
    state = iterative_process.initialize()
    
    # 执行联邦训练
    for round_num in range(1, 11):
        state, metrics = iterative_process.next(state, federated_data)
        print(f'轮次 {round_num}: {metrics}')
    
    return state.model

# 示例：具身AI传感器数据的差分隐私保护
import numpy as np

def add_differential_privacy(sensor_data, epsilon=1.0, sensitivity=1.0):
    """向传感器数据添加差分隐私噪声
    
    参数:
    sensor_data: 原始传感器数据
    epsilon: 隐私预算，越小隐私保护越强
    sensitivity: 数据敏感度
    """
    # 计算噪声尺度
    scale = sensitivity / epsilon
    
    # 添加拉普拉斯噪声
    noise = np.random.laplace(0, scale, size=sensor_data.shape)
    
    # 返回带噪声的数据
    return sensor_data + noise

def differentially_private_federated_update(model_update, epsilon=1.0):
    """在联邦学习模型更新中应用差分隐私
    """
    # 计算模型更新的敏感度（范数）
    sensitivity = np.linalg.norm(model_update)
    
    # 添加噪声
    noisy_update = add_differential_privacy(
        model_update, epsilon=epsilon, sensitivity=sensitivity
    )
    
    return noisy_update

# 具身AI系统中的同态加密实现示例
import numpy as np
from seal import *

class HomomorphicEncryptionManager:
    def __init__(self):
        # 创建SEAL上下文
        self.context = self._create_context()
        # 生成密钥
        self.keys = self._generate_keys()
        # 创建加密器和解密器
        self.encryptor = Encryptor(self.context, self.keys['public_key'])
        self.decryptor = Decryptor(self.context, self.keys['secret_key'])
        # 创建评估器（用于在加密数据上执行操作）
        self.evaluator = Evaluator(self.context)
        # 创建编码器（用于将数据转换为密文可处理的格式）
        self.encoder = CKKSEncoder(self.context)
        # 获取多项式模数（用于编码）
        self.scale = 2.0 ** 40
        self.slots = self.encoder.slot_count()
        print(f"同态加密初始化完成，支持 {self.slots} 个编码槽")
    
    def _create_context(self):
        """创建SEAL上下文配置"""
        parms = EncryptionParameters(scheme_type.CKKS)
        poly_modulus_degree = 8192
        parms.set_poly_modulus_degree(poly_modulus_degree)
        # 设置系数模数
        parms.set_coeff_modulus(CoeffModulus.Create(
            poly_modulus_degree, [60, 40, 40, 60]))
        # 设置明文模数
        parms.set_plain_modulus(PlainModulus.Batching(poly_modulus_degree, 20))
        # 创建上下文
        context = SEALContext.Create(parms)
        return context
    
    def _generate_keys(self):
        """生成公钥和私钥"""
        keygen = KeyGenerator(self.context)
        public_key = keygen.create_public_key()
        secret_key = keygen.secret_key()
        # 生成重线性化密钥（用于乘法操作）
        relin_keys = keygen.create_relin_keys()
        # 生成旋转密钥（用于循环移位操作）
        galois_keys = keygen.create_galois_keys()
        
        return {
            'public_key': public_key,
            'secret_key': secret_key,
            'relin_keys': relin_keys,
            'galois_keys': galois_keys
        }
    
    def encode_and_encrypt(self, data):
        """将传感器数据编码并加密"""
        # 确保数据长度适合编码槽
        if len(data) > self.slots:
            raise ValueError(f"数据长度({len(data)})超过编码槽数量({self.slots})")
        
        # 填充数据以匹配编码槽
        padded_data = np.zeros(self.slots, dtype=np.float64)
        padded_data[:len(data)] = data
        
        # 创建明文对象
        plain_text = Plaintext()
        # 编码数据
        self.encoder.encode(padded_data, self.scale, plain_text)
        # 创建密文对象
        cipher_text = Ciphertext()
        # 加密数据
        self.encryptor.encrypt(plain_text, cipher_text)
        
        return cipher_text
    
    def decrypt_and_decode(self, cipher_text):
        """解密并解码数据"""
        # 创建明文对象用于存储解密结果
        plain_text = Plaintext()
        # 解密
        self.decryptor.decrypt(cipher_text, plain_text)
        # 创建数组存储解码结果
        result = np.zeros(self.slots, dtype=np.float64)
        # 解码
        self.encoder.decode(plain_text, result)
        
        return result
    
    def homomorphic_add(self, cipher_text1, cipher_text2):
        """在加密数据上执行加法"""
        result = Ciphertext()
        self.evaluator.add(cipher_text1, cipher_text2, result)
        return result
    
    def homomorphic_multiply(self, cipher_text1, cipher_text2):
        """在加密数据上执行乘法"""
        result = Ciphertext()
        self.evaluator.multiply(cipher_text1, cipher_text2, result)
        # 重线性化（减少密文大小和计算复杂度）
        self.evaluator.relinearize_inplace(result, self.keys['relin_keys'])
        return result
    
    def homomorphic_rotate(self, cipher_text, rotation_steps):
        """在加密数据上执行循环移位"""
        result = Ciphertext()
        self.evaluator.rotate_vector(
            cipher_text, rotation_steps, self.keys['galois_keys'], result
        )
        return result
    
    def encrypt_and_process_sensor_data(self, sensor_data_list):
        """加密并处理多个传感器的数据（例如计算平均值）"""
        # 确保所有传感器数据长度相同
        data_length = len(sensor_data_list[0])
        if not all(len(data) == data_length for data in sensor_data_list):
            raise ValueError("所有传感器数据长度必须相同")
        
        # 加密所有传感器数据
        encrypted_data = []
        for data in sensor_data_list:
            encrypted = self.encode_and_encrypt(data)
            encrypted_data.append(encrypted)
        
        # 计算加密数据的总和
        sum_encrypted = encrypted_data[0]
        for i in range(1, len(encrypted_data)):
            sum_encrypted = self.homomorphic_add(sum_encrypted, encrypted_data[i])
        
        # 计算平均值（乘以 1/n）
        n = len(encrypted_data)
        # 编码缩放因子
        scaling_factor = np.zeros(self.slots, dtype=np.float64)
        scaling_factor[0] = 1.0 / n
        plain_factor = Plaintext()
        self.encoder.encode(scaling_factor, self.scale, plain_factor)
        
        # 乘法操作
        avg_encrypted = Ciphertext()
        self.evaluator.multiply_plain(sum_encrypted, plain_factor, avg_encrypted)
        self.evaluator.relinearize_inplace(avg_encrypted, self.keys['relin_keys'])
        
        return avg_encrypted

# 示例：在具身AI多传感器融合中的应用
if __name__ == "__main__":
    # 创建同态加密管理器
    he_manager = HomomorphicEncryptionManager()
    
    # 模拟来自不同传感器的数据
    sensor1_data = np.array([1.2, 2.3, 1.8, 3.1, 2.5])  # 温度传感器
    sensor2_data = np.array([1.3, 2.1, 1.9, 3.0, 2.7])  # 另一个温度传感器
    sensor3_data = np.array([1.1, 2.2, 1.7, 3.2, 2.6])  # 第三个温度传感器
    
    # 加密并处理数据
    avg_encrypted = he_manager.encrypt_and_process_sensor_data(
        [sensor1_data, sensor2_data, sensor3_data]
    )
    
    # 解密结果
    avg_decrypted = he_manager.decrypt_and_decode(avg_encrypted)
    
    # 显示结果
    print("传感器1数据:", sensor1_data)
    print("传感器2数据:", sensor2_data)
    print("传感器3数据:", sensor3_data)
    print("加密计算后的平均值:", avg_decrypted[:len(sensor1_data)])
    print("实际平均值:", (sensor1_data + sensor2_data + sensor3_data) / 3)

# 具身AI多代理系统中的安全多方计算实现
import numpy as np
from cryptography.fernet import Fernet

class SecureMultipartyComputation:
    def __init__(self, num_parties):
        """初始化安全多方计算系统
        
        参数:
        num_parties: 参与方数量
        """
        self.num_parties = num_parties
        # 为每个参与方生成密钥
        self.keys = [Fernet.generate_key() for _ in range(num_parties)]
        self.ciphers = [Fernet(key) for key in self.keys]
        print(f"安全多方计算系统初始化完成，支持 {num_parties} 个参与方")
    
    def additive_secret_sharing(self, data):
        """使用加法秘密共享将数据分割
        
        参数:
        data: 要共享的原始数据
        
        返回:
        shares: 分发给各参与方的数据份额
        """
        # 为前n-1个参与方生成随机份额
        shares = [np.random.randn(*data.shape) for _ in range(self.num_parties - 1)]
        
        # 计算第n个参与方的份额，使得所有份额之和等于原始数据
        sum_shares = np.zeros_like(data)
        for share in shares:
            sum_shares += share
        last_share = data - sum_shares
        shares.append(last_share)
        
        return shares
    
    def encrypt_shares(self, shares):
        """加密各参与方的数据份额
        
        参数:
        shares: 各参与方的数据份额
        
        返回:
        encrypted_shares: 加密后的份额
        """
        encrypted_shares = []
        for i, share in enumerate(shares):
            # 将numpy数组转换为字节
            share_bytes = share.tobytes()
            # 使用参与方的密钥加密
            encrypted = self.ciphers[i].encrypt(share_bytes)
            encrypted_shares.append(encrypted)
        
        return encrypted_shares
    
    def decrypt_share(self, encrypted_share, party_index):
        """解密特定参与方的数据份额
        
        参数:
        encrypted_share: 加密的数据份额
        party_index: 参与方索引
        
        返回:
        decrypted_share: 解密后的数据份额
        """
        # 使用参与方密钥解密
        decrypted_bytes = self.ciphers[party_index].decrypt(encrypted_share)
        # 将字节转回numpy数组（假设原始形状已知）
        # 这里需要知道原始形状，实际应用中应传递或存储形状信息
        return np.frombuffer(decrypted_bytes)
    
    def secure_sum(self, encrypted_shares_list):
        """在加密份额上计算安全求和
        
        参数:
        encrypted_shares_list: 每个参与方提供的加密份额列表
        
        返回:
        result_shares: 计算结果的份额
        """
        # 对于每个参与方，计算其所有份额的和
        result_shares = []
        for i in range(self.num_parties):
            # 解密该参与方的所有份额
            decrypted_shares = []
            for shares in encrypted_shares_list:
                decrypted = self.decrypt_share(shares[i], i)
                decrypted_shares.append(decrypted)
            
            # 计算该参与方的份额之和
            sum_share = np.zeros_like(decrypted_shares[0])
            for share in decrypted_shares:
                sum_share += share
            
            # 重新加密结果份额
            sum_bytes = sum_share.tobytes()
            encrypted_result = self.ciphers[i].encrypt(sum_bytes)
            result_shares.append(encrypted_result)
        
        return result_shares
    
    def reconstruct_result(self, result_shares):
        """从结果份额中重构最终结果
        
        参数:
        result_shares: 结果的加密份额
        
        返回:
        final_result: 重构后的最终结果
        """
        # 解密所有份额
        decrypted_shares = []
        for i, share in enumerate(result_shares):
            decrypted = self.decrypt_share(share, i)
            decrypted_shares.append(decrypted)
        
        # 求和得到最终结果
        final_result = np.zeros_like(decrypted_shares[0])
        for share in decrypted_shares:
            final_result += share
        
        return final_result

# 示例：多代理具身AI系统中的安全协同决策
if __name__ == "__main__":
    # 创建3个参与方的安全多方计算系统
    mpc = SecureMultipartyComputation(num_parties=3)
    
    # 模拟三个具身AI代理的本地决策数据
    # 这些数据可能包含敏感的环境信息或内部状态
    agent1_data = np.array([0.8, 0.2, 0.1, 0.5])  # 代理1的决策权重
    agent2_data = np.array([0.6, 0.3, 0.2, 0.4])  # 代理2的决策权重
    agent3_data = np.array([0.7, 0.1, 0.3, 0.6])  # 代理3的决策权重
    
    print("原始数据（实际应用中不会共享）:")
    print(f"代理1: {agent1_data}")
    print(f"代理2: {agent2_data}")
    print(f"代理3: {agent3_data}")
    
    # 每个代理对自己的数据进行秘密共享
    agent1_shares = mpc.additive_secret_sharing(agent1_data)
    agent2_shares = mpc.additive_secret_sharing(agent2_data)
    agent3_shares = mpc.additive_secret_sharing(agent3_data)
    
    # 加密所有份额
    encrypted_shares1 = mpc.encrypt_shares(agent1_shares)
    encrypted_shares2 = mpc.encrypt_shares(agent2_shares)
    encrypted_shares3 = mpc.encrypt_shares(agent3_shares)
    
    # 计算安全求和（各代理在不泄露原始数据的情况下计算平均值）
    sum_shares = mpc.secure_sum([encrypted_shares1, encrypted_shares2, encrypted_shares3])
    
    # 重构结果（平均决策权重）
    avg_decision = mpc.reconstruct_result(sum_shares) / 3
    
    print(f"\n安全计算的平均决策权重: {avg_decision}")
    print(f"实际平均决策权重: {(agent1_data + agent2_data + agent3_data) / 3}")
    
    # 基于平均决策权重做出协同决策
    action_index = np.argmax(avg_decision)
    actions = ["前进", "后退", "左转", "右转"]
    print(f"\n多代理协同决策: {actions[action_index]}")

# 零知识范围证明在具身AI传感器数据验证中的应用示例
class ZeroKnowledgeRangeProof:
    def __init__(self, lower_bound, upper_bound):
        """初始化零知识范围证明系统
        
        参数:
        lower_bound: 范围下界
        upper_bound: 范围上界
        """
        self.lower_bound = lower_bound
        self.upper_bound = upper_bound
        self.range_size = upper_bound - lower_bound
        print(f"零知识范围证明系统初始化完成，范围: [{lower_bound}, {upper_bound}]")
    
    def generate_proof(self, secret_value):
        """为秘密值生成零知识范围证明
        
        参数:
        secret_value: 要证明在范围内的秘密值
        
        返回:
        proof: 零知识证明
        """
        # 验证秘密值确实在范围内（这部分仅在证明生成方执行）
        if not (self.lower_bound <= secret_value <= self.upper_bound):
            raise ValueError(f"秘密值 {secret_value} 不在范围内 [{self.lower_bound}, {self.upper_bound}]")
        
        # 计算相对于下界的偏移
        offset = secret_value - self.lower_bound
        
        # 生成随机承诺参数（简化示例，实际应使用密码学安全的承诺方案）
        r = np.random.randint(1, 1000000)  # 随机数
        
        # 计算承诺（在实际应用中，应使用Pedersen承诺等密码学安全的承诺方案）
        # 这里使用简化的哈希承诺作为示例
        import hashlib
        commitment = hashlib.sha256(f"{offset}:{r}".encode()).hexdigest()
        
        # 生成证明（简化示例，实际应使用zk-SNARK等正式的零知识证明方案）
        # 证明包含承诺和验证所需的辅助信息
        proof = {
            'commitment': commitment,
            'challenge_response': self._generate_challenge_response(offset, r)
        }
        
        return proof
    
    def _generate_challenge_response(self, offset, r):
        """生成挑战响应（简化示例）"""
        # 在实际的零知识证明系统中，这将涉及交互式挑战-响应协议
        # 或使用 Fiat-Shamir 启发式转换为非交互式证明
        return hashlib.sha256(f"response:{offset}:{r}".encode()).hexdigest()
    
    def verify_proof(self, proof):
        """验证零知识范围证明
        
        参数:
        proof: 要验证的零知识证明
        
        返回:
        is_valid: 证明是否有效
        """
        # 在实际应用中，这将涉及验证承诺和挑战响应
        # 由于这是一个简化示例，我们假设验证总是成功
        # 实际实现应使用完整的零知识证明验证算法
        return True
    
    def verify_sensor_reading(self, sensor_proof):
        """验证传感器读数是否在合理范围内，而不获知具体值
        
        参数:
        sensor_proof: 传感器读数的零知识证明
        
        返回:
        is_valid: 传感器读数是否在有效范围内
        """
        return self.verify_proof(sensor_proof)

# 示例：具身AI传感器数据的零知识验证
if __name__ == "__main__":
    # 创建温度传感器的零知识范围证明系统（假设正常温度范围：20-35°C）
    temp_proof_system = ZeroKnowledgeRangeProof(lower_bound=20.0, upper_bound=35.0)
    
    # 模拟具身AI传感器的实际读数
    actual_temperature = 27.5  # 实际温度（不对外公开）
    
    # 生成零知识范围证明
    proof = temp_proof_system.generate_proof(actual_temperature)
    print(f"生成的零知识证明: {proof}")
    
    # 验证证明（验证方不知道实际温度）
    is_valid = temp_proof_system.verify_proof(proof)
    print(f"证明验证结果: {'有效' if is_valid else '无效'}")
    print(f"结论: 传感器读数{'在' if is_valid else '不在'}正常范围内")