用Python和libOTe库5分钟实现不经意传输(OT)协议Demo

想象一下这样的场景：你需要从服务器获取某个特定数据，但不想让服务器知道你选择了哪个数据；同时服务器也希望只提供你选择的数据，而不是全部泄露。这种看似矛盾的需求，正是**不经意传输(Oblivious Transfer, OT)**协议要解决的经典问题。

OT协议作为现代密码学的基石之一，在隐私计算、安全多方计算等领域有着广泛应用。今天我们将使用Python的libOTe库，快速实现一个1-out-of-2 OT协议的Demo，让你在5分钟内看到这个神奇协议的实际运行效果。

1. 环境准备与库安装

在开始编码前，我们需要准备好Python环境和必要的依赖库。libOTe是一个高效的OT协议实现库，支持多种OT变体。

首先确保你已安装Python 3.7或更高版本。然后通过pip安装libOTe：

pip install libote

此外，我们还需要一些辅助库：

pip install numpy cryptography

注意：libOTe在某些平台上可能需要额外的编译工具。如果安装遇到问题，可以尝试先安装CMake和g++/clang等编译器。

安装完成后，可以通过以下命令验证是否安装成功：

import libote
print(libote.__version__)

如果输出版本号而没有报错，说明环境已准备就绪。

2. 理解1-out-of-2 OT协议

在编写代码前，让我们快速回顾1-out-of-2 OT协议的基本原理：

• 发送方(Sender) ：持有两个消息m₀和m₁
• 接收方(Receiver) ：希望获取其中一个消息m_c（c∈{0,1}），但不透露c的选择
• 协议保证 ：
  • 接收方只能获取m_c，无法得知另一个消息
  • 发送方无法确定接收方选择了哪个消息

这种协议在隐私保护数据查询、安全机器学习等场景中非常有用。下面我们将用libOTe实现这一协议。

3. 实现基础OT协议

libOTe提供了多种OT实现，我们先从最基本的Naor-Pinkas OT开始。创建一个新的Python文件 ot_demo.py ，添加以下代码：

import numpy as np
from libote import NaorPinkasOT, OTReceiver, OTSender

def run_naor_pinkas_ot():
    # 初始化OT协议
    ot = NaorPinkasOT()
    
    # 发送方的两个消息
    messages = [b"这是消息0", b"这是消息1"]
    
    # 接收方的选择 (0或1)
    choice = 1
    
    # 模拟网络通信的简单缓冲区
    send_buf = bytearray(1024)
    recv_buf = bytearray(1024)
    
    # 发送方和接收方初始化
    sender = OTSender(ot)
    receiver = OTReceiver(ot)
    
    # 协议执行流程
    # 第一步：接收方发送初始消息
    receiver.send(send_buf)
    recv_buf[:len(send_buf)] = send_buf
    
    # 第二步：发送方处理接收方的消息并回复
    sender.recv(recv_buf)
    sender.send(send_buf, messages)
    recv_buf[:len(send_buf)] = send_buf
    
    # 第三步：接收方获取最终结果
    result = receiver.recv(recv_buf, choice)
    
    print(f"接收方选择获取消息{choice}, 实际获得: {result.decode()}")

if __name__ == "__main__":
    run_naor_pinkas_ot()

运行这个脚本，你会看到接收方成功获取了选择的消息，而发送方并不知道接收方选择了哪个消息。

4. 使用更高效的OT扩展

基础OT协议虽然简单，但在实际应用中性能不够理想。libOTe提供了更高效的OT扩展实现，让我们看看如何改进：

from libote import IKNPOTExt

def run_ot_extension():
    # 初始化OT扩展协议
    ot_ext = IKNPOTExt()
    
    # 批量处理128个OT
    batch_size = 128
    
    # 发送方准备128对消息
    messages = [(f"消息{i}_0".encode(), f"消息{i}_1".encode()) 
               for i in range(batch_size)]
    
    # 接收方的128个选择 (0或1)
    choices = np.random.randint(0, 2, batch_size, dtype=bool)
    
    # 初始化发送方和接收方
    sender = OTSender(ot_ext)
    receiver = OTReceiver(ot_ext)
    
    # 执行协议
    # 1. 基础OT阶段
    send_buf = bytearray(1024)
    receiver.send(send_buf)
    sender.recv(send_buf)
    
    # 2. OT扩展阶段
    sender.send(messages)
    results = receiver.recv(choices)
    
    # 验证结果
    for i in range(5):  # 只打印前5个结果
        expected = messages[i][choices[i]].decode()
        print(f"选择{choices[i]}, 获取: {results[i].decode()}, 应得: {expected}")

run_ot_extension()

OT扩展技术通过少量基础OT和对称加密操作，实现了大量OT的高效执行，性能比直接使用基础OT提升几个数量级。

5. 实际应用：隐私保护数据查询

让我们看一个更实际的例子：隐私保护的数据库查询。假设有一个医疗数据库，用户想查询自己的某项检测结果，但不想透露查询的是哪项检测。

from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF

class PrivacyPreservingQuery:
    def __init__(self):
        self.ot = IKNPOTExt()
        
    def setup_database(self):
        # 模拟医疗检测结果数据库
        self.tests = {
            "血糖": "血糖值: 5.2 mmol/L (正常)",
            "胆固醇": "总胆固醇: 4.8 mmol/L (正常)",
            "血压": "血压: 120/80 mmHg (正常)"
        }
        self.test_names = list(self.tests.keys())
        
    def run_query(self, test_index):
        # 数据库方(发送方)
        def sender_process():
            sender = OTSender(self.ot)
            
            # 将每个检测结果转换为OT消息对
            messages = []
            for name in self.test_names:
                # 使用HKDF派生密钥加密结果
                hkdf = HKDF(
                    algorithm=hashes.SHA256(),
                    length=32,
                    salt=None,
                    info=b'medical record',
                )
                key = hkdf.derive(name.encode())
                encrypted = bytes(a ^ b for a, b in zip(
                    self.tests[name].encode(), key))
                
                # 每个检测名称对应一对(加密结果, 空消息)
                messages.append((encrypted, b""))
            
            # 执行OT协议
            send_buf = bytearray(1024)
            sender.recv(send_buf)  # 接收初始消息
            sender.send(messages)  # 发送加密结果
            
            return sender
        
        # 用户方(接收方)
        def receiver_process():
            receiver = OTReceiver(self.ot)
            
            # 发送初始消息
            send_buf = bytearray(1024)
            receiver.send(send_buf)
            
            # 执行OT获取结果
            choices = [False] * len(self.test_names)
            choices[test_index] = True  # 只选择感兴趣的检测
            
            results = receiver.recv(choices)
            
            # 解密结果
            hkdf = HKDF(
                algorithm=hashes.SHA256(),
                length=32,
                salt=None,
                info=b'medical record',
            )
            key = hkdf.derive(self.test_names[test_index].encode())
            decrypted = bytes(a ^ b for a, b in zip(results[test_index], key))
            
            return decrypted.decode()
        
        return sender_process(), receiver_process()

# 使用示例
ppq = PrivacyPreservingQuery()
ppq.setup_database()

# 用户想查询"胆固醇"结果(索引1)
_, result = ppq.run_query(1)
print("查询结果:", result)

这个例子展示了OT协议如何在实际应用中保护双方的隐私：用户获取了需要的检测结果，而数据库方不知道用���查询的是哪项检测。

6. 性能优化与注意事项

在实际使用OT协议时，性能是需要重点考虑的因素。以下是一些优化建议：

1. 批量处理 ：OT扩展技术最适合批量处理大量OT，尽量将多个OT合并执行
2. 网络优化 ：减少通信轮次，使用高效的序列化格式
3. 参数选择 ：根据安全需求选择适当的参数，不必过度安全

libOTe提供了一些内置的优化功能：

from libote import SilentOT

def run_silent_ot():
    # 静默OT进一步优化了通信量
    ot = SilentOT()
    
    # 准备1000个OT
    n = 1000
    messages = [(os.urandom(32), os.urandom(32)) for _ in range(n)]
    choices = np.random.randint(0, 2, n, dtype=bool)
    
    sender = OTSender(ot)
    receiver = OTReceiver(ot)
    
    # 执行协议
    receiver.send(bytearray(1024))
    sender.recv(bytearray(1024))
    
    sender.send(messages)
    results = receiver.recv(choices)
    
    # 验证
    for i in range(n):
        assert results[i] == messages[i][choices[i]]

run_silent_ot()

提示：在生产环境中使用OT协议时，还需要考虑恶意敌手安全、协议可组合性等高级特性。libOTe也支持这些功能，但配置会更复杂。