后量子密码学的性能优化：开发人员指南

进口 并发期货 进口 奥克斯 从 打字 进口 列表、元组 进口 时间 班级 并行签名者: """ 用于批量操作的并行 SPHINCS+ 签名 在签署多个独立消息时使用。 """ 定义 __初始化__（自我，max_workers：int =无）： """ 初始化并行签名者参数：max_workers：要使用的 CPU 线程（默认值：CPU 计数） """ self.max_workers = max_workers or os.cpu_count() 定义 签名批次( self, messages: List[bytes], Secret_key: bytes ) -> List[bytes]: """ 并行签署多条消息 参数： messages：要签名的消息列表 Secret_key：SPHINCS+ 密钥 返回：与消息顺序相同的签名列表 """ 定义 标志_单（消息：字节）->字节：sig = oqs.Signature（“SPHINCS+-SHAKE-128s-简单”，秘密密钥） 返回 sig.sign(消息) 和 并发.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as 执行者：签名=列表（executor.map（sign_single，消息）） 返回 签名 定义 带钥匙签名（自我，项目：列表[元组[字节，字节]] #（消息，秘密密钥） ) -> 列表[字节]: """使用不同的密钥并行签署消息""" 定义 标志项（物品： 元组[字节，字节]) -> 字节：消息，sk = item sig = oqs.Signature(“SPHINCS+-SHAKE-128s-简单”, sk） 返回 sig.sign(消息) 和 并发.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as 执行人： 返回 列表(executor.map(sign_item, items)) # 基准比较 定义 benchmark_parallel_vs_sequential(): # 生成密钥 sig = oqs.Signature(“SPHINCS+-SHAKE-128s-简单”) sig.generate_keypair() sk = sig.export_secret_key() # 创建测试消息 消息 = [f“消息{i}”.encode() 为了 i in 范围(16)] # 顺序 开始= time.perf_counter（）sequential_sigs = [] 为了 味精 in 消息： s = oqs.Signature(“SPHINCS+-SHAKE-128s-简单”, sk)equential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - 开始 ＃ 平行线 签名者= 并行签名者() start = time.perf_counter() parallel_sigs = signer.sign_batch(messages, sk) par_time = time.perf_counter() - 开始 print(f“顺序：{seq_time:.2f}s ({len(messages)/seq_time:.1f} msg/s)”) 打印(f“并行：{par_time:.2f}s ({len(messages)/par_time:.1f} msg/s)”) 打印(f“加速比：{seq_time/par_time:.2f}x”)

班级 并行验证器: """验证器的并行签名验证""" 定义 __初始化__(self, max_workers: int = None): self.max_workers = max_workers or os.cpu_count() 定义 验证批次（自我，项目：列表[元组[字节，字节，字节]] #（消息、信号、PK） ) -> 列表[布尔]: """并行验证多个签名返回验证结果列表""" 定义 验证单（物品： 元组[字节，字节，字节]) -> bool: 消息，签名，public_key = item 尝试: sig = oqs.Signature(“SPHINCS+-SHAKE-128s-简单”) 返回 sig.verify(消息、签名、公钥) 除了: 返回 错误的 和 并发.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as 执行人： 返回 列表（executor.map（verify_single，项目）） 定义 全部有效（自我，项目：列表[元组[字节，字节，字节]] ) -> 布尔值： """快速检查所有签名是否有效""" 结果= self.verify_batch（项目） 返回 全部（结果） # 对于处理块的验证器： 异步定义 验证区块交易（交易：列表）：验证者 = 并行验证器（最大工人数=8） # 准备验证项 items = [ (tx.signing_message, tx.signature, tx.public_key) 为了 tx in 交易] # 并行验证所有内容 结果 = verifier.verify_batch(items) # 过滤有效交易 valid_txs = [tx 为了 发送，有效 in zip（交易、结果） if 有效的] 返回 有效交易

从 功能工具 进口 lru_缓存 进口 哈希库 班级 密钥缓存: """ 缓存派生密钥以避免重复派生 对于频繁访问相同路径的 HD 钱包很有用。""" 定义 __初始化__(self, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} 定义 获取或派生（自我，master_seed：字节，路径：str，derive_func）-> 元组[字节，字节]： """ 获取缓存密钥或派生并缓存参数： master_seed：钱包主种子路径：派生路径derive_func：缓存未命中时调用的函数返回：(public_key，secret_key) 元组 """ # 创建缓存键（不要在键中存储实际种子） cache_key = hashlib.Blake2b(master_seed + path.encode()).hexdigest()[:32] if 缓存键 in self._cache： 返回 self._cache[cache_key] # 派生密钥 pk, sk =derive_func(master_seed, 路径) # 缓存与驱逐 if len(self._cache) >= self.max_size: # 简单的 FIFO 驱逐（在生产中使用 OrderedDict） 最旧的=下一个（iter（self._cache）） 德尔 self._cache[最旧] self._cache[cache_key] = (pk, sk) 返回 PK、SK 定义 清除（自己）： """清除所有缓存的密钥（调用钱包锁）""" # 安全擦除 为了 钥匙 in 列表(self._cache.keys()): pk, sk = self._cache[key] # 删除前覆盖 self._cache[key] = (b'\x00' * len(pk), b'\x00' * 长度（sk）） 德尔 self._cache[键] # 钱包中的使用 班级 优化钱包: 定义 __初始化__(self, master_seed: 字节): self.master_seed = master_seed self.key_cache = 密钥缓存（最大大小=500） 定义 获取地址密钥（自我，路径：str）-> 元组[字节，字节]： 返回 self.key_cache.get_or_derive( self.master_seed, 路径, self._derive_keys ) 定义 _derive_keys（自身，种子：字节，路径：str）： # 实际推导逻辑 ...

班级 签名缓存: """缓存签名验证结果供验证者避免重新验证已看到的交易。""" 定义 __初始化__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} 定义 _signature_id( self, 消息: 字节, 签名: 字节, public_key: 字节 ) -> str: """创建唯一 ID 用于签名验证""" 返回 hashlib.Blake2b( 消息 + 签名[:64] + public_key, # sig 的前 64 字节足够了 摘要_大小 = 16 ).hexdigest() 定义 检查或验证( self, 消息: 字节, 签名: 字节, public_key: 字节 ) -> bool: """检查缓存或验证并缓存结果""" sig_id = self._signature_id(消息、签名、公钥) if sig_id in 自我验证： 返回 self._verified[sig_id] ＃ 核实 sig = oqs.Signature(“SPHINCS+-SHAKE-128s-简单”) is_valid = sig.verify(消息, 签名, 公钥) # 缓存（带驱逐） if len(self._verified) >= self.max_size: # 删除 ~10% 最旧的条目 to_remove = list(self._verified.keys())[:self.max_size // 10] 为了 钥匙 in 删除： 德尔 self._verified[key] self._verified[sig_id] = is_valid 返回 有效

进口 gc 班级 内存效率签名者: """嵌入式/移动设备的内存高效签名""" 定义 签署并释放（自我，消息：字节，secret_key：字节）->字节： """ 签署消息并立即释放密钥内存 用于密钥不应持续存在的一次性签名。 """ sig_obj = oqs.Signature(“SPHINCS+-SHAKE-128s-简单”, Secret_key) 签名 = sig_obj.sign(消息) # 释放OQS对象 德尔 sig_obj # 覆盖密钥 if isinstance(secret_key, bytearray): 为了 i in 范围（len（secret_key））：secret_key [i] = 0 # 强制垃圾回收 GC.collect() 返回 签名 定义 流媒体符号( self, message_chunks: Iterator[bytes], Secret_key: bytes ) -> bytes: """ 对流消息进行签名，而不将所有消息加载到内存中 对消息进行分块预哈希，然后对哈希进行签名。 """ # 散列消息块 哈希器 = hashlib.Blake2b(digest_size=32) 为了 块 in message_chunks: hasher.update(chunk) message_hash = hasher.digest() # 对哈希进行签名 sig = oqs.Signature(“SPHINCS+-SHAKE-128s-简单”，秘密密钥） 返回 sig.sign(message_hash)

# 检查 CPU 特征以选择最佳算法 进口 子流程 定义 获取CPU功能() -> 设置： """检测可用的 CPU SIMD 功能""" 尝试: # Linux 和 打开（“/proc/cpuinfo”) as f: cpuinfo = f.read() 功能 = set() if “avx2” in cpuinfo: 功能.add(“avx2”) if “avx512” in cpuinfo: 功能.add(“avx512”) if “aes” in cpuinfo: 功能.add(“埃斯尼”) 返回 特征 除了: 返回 放（） 定义 选择最优变体() -> 字符串： """为此 CPU 选择最佳的 SPHINCS+ 变体""" 功能 = get_cpu_features() if “avx512” in 特征： # AVX-512 提供约 20-30% 的加速 打印（《使用AVX-512优化实现》) 返回 “SPHINCS+-SHAKE-128s-简单” # liboqs 自动选择 埃利夫 “avx2” in 特点：打印（《使用AVX2优化实现》) 返回 “SPHINCS+-SHAKE-128s-简单” 别的： 打印（“使用参考实现”) 返回 “SPHINCS+-SHAKE-128s-简单” # 使用最佳标志编译 liboqs # cmake -DOQS_USE_AVX2_INSTRUCTIONS=ON -DOQS_USE_AVX512_INSTRUCTIONS=ON ..

进口 统计数据 进口 时间 班级 加密基准测试: """全面的 PQC 基准测试""" 定义 __初始化__（自我，迭代：int = 100）：self.iterations =迭代 定义 基准操作（自我，名称：str，操作，设置=没有任何 ) -> 字典： """对单个操作进行基准测试""" 次数 = [] 为了 _ in 范围（自我迭代）： if 设置： ctx = setup() 开始 = time.perf_counter() if 设置：操作（ctx） 别的：操作（）已过去=（time.perf_counter（）-开始）* 1000 ＃ 多发性硬化症 times.append(已过去) 返回 { “姓名”： 姓名， “意思是”: 统计.平均值(次), “中位数”: 统计.中位数(次), “标准开发”：统计数据.stdev（次）， “分钟”：分钟（次）， “最大限度”：最大（次）， “每秒操作数”: 1000 / 统计.平均值(次) } 定义 运行完整基准（自我）-> 字典： """运行完整的 PQC 基准测试套件""" 结果={} # Kyber 基准测试 结果[“kyber_keygen”] = self.benchmark_operation( “Kyber-768 密钥生成器”, 拉姆达: oqs.KeyEncapsulation(“凯伯768”).generate_keypair() ) # 封装/解盖设置 kem = oqs.KeyEncapsulation(“凯伯768”) pk = kem.generate_keypair() sk = kem.export_secret_key() 结果[“kyber_encap”] = self.benchmark_operation( “Kyber-768 封装”, 拉姆达: kem.encap_secret(pk) ) ct, _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation(“凯伯768”, sk) 结果[“kyber_decap”] = self.benchmark_operation( “Kyber-768 开盖”, 拉姆达: kem_dec.decap_secret(ct) ) # SPHINCS+ 基准测试 结果[“sphincs_keygen”] = self.benchmark_operation( “SPHINCS+-SHAKE-128s 密钥生成器”, 拉姆达: oqs.Signature(“SPHINCS+-SHAKE-128s-简单”).generate_keypair() ) sig = oqs.Signature(“SPHINCS+-SHAKE-128s-简单”) sig.generate_keypair() spx_sk = sig.export_secret_key() msg​​ = b"x" * 256 个结果[“sphincs_sign”] = self.benchmark_operation( “SPHINCS+-SHAKE-128s 标志”, 拉姆达: oqs.Signature(“SPHINCS+-SHAKE-128s-简单”, spx_sk).sign(msg), 迭代=20 # 由于速度较慢而减少 ) 返回 结果 定义 打印结果（自我，结果：字典）： """漂亮的打印基准测试结果""" 打印（"\n=== PQC 性能基准 ===) 打印(f“迭代次数：{self.iterations}\n”) 为了 密钥、数据 in 结果.items(): 打印(f“{数据['名称']}：”) 打印(f“平均值：{数据['平均值']：.3f}毫秒”) 打印(f“中位数：{数据['中位数']：.3f}毫秒”) 打印(f“每秒操作次数：{data['ops_per_sec']:.1f}”） 打印（） # 运行基准测试 if __名字__ == “__主要的__”: 板凳 = 加密基准测试(迭代=50) 结果 = bench.run_full_benchmark() bench.print_results(结果)