ポスト量子暗号のパフォーマンスの最適化: 開発者ガイド

輸入 同時先物 輸入 オークス から タイピング 輸入 リスト、タプル 輸入 時間 クラス パラレル署名者: """ バッチ操作用の並列 SPHINCS+ 署名。複数の独立したメッセージに署名する場合に使用します。 """ 確かに __初期化__(self、max_workers: int = なし): """ 並列署名者の初期化 引数: max_workers: 使用する CPU スレッド (デフォルト: CPU カウント) """ self.max_workers = max_workers or os.cpu_count() 確かに 署名バッチ( self、メッセージ: リスト[バイト]、秘密キー: バイト) -> リスト[バイト]: """ 複数のメッセージに並行して署名します。 引数:messages:署名するメッセージのリスト Secret_key:SPHINCS+ 秘密鍵 戻り値:メッセージと同じ順序の署名のリスト """ 確かに サイン_シングル(メッセージ: バイト) -> バイト: sig = oqs.Signature(「SPHINCS+-SHAKE-128s-シンプル」、秘密キー) 戻る sig.sign(メッセージ) と concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as executor: 署名 = list(executor.map(sign_single,messages)) 戻る 署名 確かに キー付きの署名( 自分自身、アイテム: リスト[タプル[バイト、バイト]] # (メッセージ、秘密キー) ) -> リスト[バイト]: """異なるキーを使用してメッセージに並行して署名します""" 確かに サインアイテム（アイテム： タプル[バイト, バイト]) -> バイト: メッセージ, sk = アイテム sig = oqs.Signature(「SPHINCS+-SHAKE-128s-シンプル」、スク） 戻る sig.sign(メッセージ) と concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as 執行者: 戻る list(executor.map(sign_item, items)) # ベンチマーク比較 確かに ベンチマーク_パラレル_対_シーケンシャル(): # キーを生成する sig = oqs.Signature(「SPHINCS+-SHAKE-128s-シンプル」) sig.generate_keypair() sk = sig.export_secret_key() # テストメッセージを作成する メッセージ = [f「メッセージ{i}」。エンコード（） のために i in 範囲(16)] ＃ 一連 start = time.perf_counter() sequence_sigs = [] のために メッセージ in メッセージ: s = oqs.Signature(「SPHINCS+-SHAKE-128s-シンプル」、sk) sequence_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - start ＃ 平行 署名者 = パラレル署名者() start = time.perf_counter()Parallel_sigs = Signer.sign_batch(messages, sk) par_time = time.perf_counter() - start print(f「シーケンシャル: {seq_time:.2f}s ({len(messages)/seq_time:.1f} msg/s)」) print(f「並列: {par_time:.2f}s ({len(messages)/par_time:.1f} msg/s)」) print(f「スピードアップ: {seq_time/par_time:.2f}x」)

クラス ParallelVerifier: """バリデーターの並行署名検証""" 確かに __初期化__(self, max_workers: int = None): self.max_workers = max_workers or os.cpu_count() 確かに 検証バッチ( 自分自身、アイテム: リスト[タプル[バイト、バイト、バイト]] # (msg、sig、pk) ) -> リスト[ブール値]: """ 複数の署名を並行して検証し、検証結果のリストを返します """ 確かに verify_single（アイテム： タプル[バイト、バイト、バイト]) -> bool: メッセージ、署名、公開キー = 項目 試す: sig = oqs.Signature(「SPHINCS+-SHAKE-128s-シンプル」) 戻る sig.verify(メッセージ、署名、公開鍵) を除外する: 戻る 間違い と concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as 執行者: 戻る list(executor.map(verify_single, items)) 確かに すべて有効( 自分自身、アイテム: リスト[タプル[バイト、バイト、バイト]] ) -> ブール: """すべての署名が有効かどうかを簡単に確認します""" 結果 = self.verify_batch(項目) 戻る すべて(結果) # ブロックを処理するバリデーターの場合: 非同期定義 validate_block_transactions(トランザクション: リスト): 検証者 = ParallelVerifier(max_workers=8) # 検証項目の準備 items = [ (tx.signing_message、tx.signature、tx.public_key) のために tx in お取引】 # すべてを並行して検証する 結果 = verifier.verify_batch(items) # 有効なトランザクションをフィルタリングする valid_txs = [tx のために TX、有効 in zip(トランザクション、結果) if 有効] 戻る 有効な_txs

から 関数ツール 輸入 lru_キャッシュ 輸入 ハッシュリブ クラス キーキャッシュ: """ 導出キーをキャッシュして、導出の繰り返しを回避します。同じパスが頻繁にアクセスされる HD ウォレットに役立ちます。 """ 確かに __初期化__(self, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} 確かに 取得または派生( self、master_seed: バイト、パス: str、derive_func ) -> タプル[バイト、バイト]: """ キャッシュされたキーの取得、または導出してキャッシュ # キャッシュキーを作成します (実際のシードをキーに保存しません) キャッシュキー = hashlib.Blake2b(マスターシード + path.encode()).hexdigest()[:32] if キャッシュキー in self._cache: 戻る self._cache[キャッシュキー] # キーの派生 pk, sk = 派生_func(マスターシード, パス) # エビクション付きキャッシュ if len(self._cache) >= self.max_size: # 単純な FIFO エビクション (運用環境では OrderedDict を使用) 最も古い = 次(iter(self._cache)) デル self._cache[最も古い] self._cache[キャッシュキー] = (pk, sk) 戻る PK、SK 確かに クリア（自己）： """キャッシュされたキーをすべてクリアします (ウォレット ロックの呼び出し)""" # 安全な消去 のために 鍵 in list(self._cache.keys()): pk, sk = self._cache[key] # 削除前に上書き self._cache[キー] = (b'\x00' * len(pk)、b'\x00' *レン(スク)) デル self._cache[キー] # ウォレット内での使用方法 クラス 最適化されたウォレット: 確かに __初期化__(self, master_seed: バイト): self.master_seed = master_seed self.key_cache = キーキャッシュ(最大サイズ=500) 確かに アドレスキーの取得(自分自身、パス: str) -> タプル[バイト、バイト]: 戻る self.key_cache.get_or_derive( self.master_seed, path, self._derive_keys ) 確かに _derive_keys(self、シード: バイト、パス: str): # 実際の導出ロジック ...

クラス 署名キャッシュ: """ 検証者が確認したトランザクションの再検証を避けるために、署名検証結果をキャッシュします。 """ 確かに __初期化__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} 確かに _signature_id( self、メッセージ: バイト、署名: バイト、公開キー: バイト) -> str: """署名検証用に一意の ID を作成します""" 戻る hashlib.Blake2b(メッセージ + 署名[:64] + public_key, # 署名の最初の 64 バイトで十分 ダイジェストサイズ=16 ).hexdigest() 確かに チェックまたは検証( self、メッセージ: バイト、署名: バイト、公開キー: バイト) -> bool: """キャッシュをチェックするか、結果を検証してキャッシュします""" sig_id = self._signature_id(メッセージ、署名、公開鍵) if sig_id in self._verified: 戻る 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 クラス メモリ効率の高い署名者: """ 組み込み/モバイル デバイス向けのメモリ効率の高い署名 """ 確かに 署名と解放( self、メッセージ: バイト、秘密キー: バイト) -> バイト: """ メッセージに署名し、キー メモリをすぐに解放します。キーを保持すべきではない 1 回限りの署名に使用します。 """ sig_obj = oqs.Signature(「SPHINCS+-SHAKE-128s-シンプル」、secret_key) 署名 = sig_obj.sign(メッセージ) # OQS オブジェクトを解放します デル sig_obj # 秘密鍵を上書きする if isinstance(secret_key, bytearray): のために i in range(len(secret_key)): Secret_key[i] = 0 # ガベージコレクションを強制する gc.collect() 戻る サイン 確かに ストリーミングサイン( self, message_chunks: Iterator[bytes], Secret_key: bytes ) -> バイト: """ すべてをメモリにロードせずにストリーミング メッセージに署名します。メッセージをチャンクで事前にハッシュし、ハッシュに署名します。 """ # チャンク内のハッシュメッセージ ハッシュ = hashlib.Blake2b(digest_size=32) のために かたまり in message_chunks: hasher.update(chunk) message_hash = hasher.digest() # ハッシュに署名する sig = oqs.Signature(「SPHINCS+-SHAKE-128s-シンプル」、秘密キー) 戻る sig.sign(メッセージ_ハッシュ)

# 最適なアルゴリズムを選択するには、CPU の機能を確認してください 輸入 サブプロセス 確かに get_cpu_features() -> 設定: """利用可能な CPU SIMD 機能を検出します""" 試す: # Linux と 開ける（「/proc/cpuinfo」) as f: cpuinfo = f.read() 機能 = set() if 「avx2」 in cpuinfo: features.add(「avx2」) if 「avx512」 in cpuinfo: features.add(「avx512」) if 「エース」 in cpuinfo: features.add(「エースニ」) 戻る 特徴 を除外する: 戻る セット（） 確かに select_optimal_variant() -> 文字列: """この 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-シンプル」 # 最適なフラグを使用して liboq をコンパイルする # cmake -DOQS_USE_AVX2_INSTRUCTIONS=ON -DOQS_USE_AVX512_INSTRUCTIONS=ON ..

輸入 統計 輸入 時間 クラス 暗号ベンチマーク: """包括的な PQC ベンチマーク""" 確かに __初期化__(self、反復: int = 100): self.iterations = 反復 確かに ベンチマーク操作( self、名前: str、操作、setup=なし ) -> 辞書: """単一操作のベンチマークを行う""" 回 = [] のために _ in 範囲(self.iterations): if セットアップ: ctx = setup() start = time.perf_counter() if セットアップ: 操作(ctx) それ以外: 経過した操作() = (time.perf_counter() - 開始) * 1000 ＃ MS 回.追加(経過) 戻る { "名前"： 名前、 "平均": 統計.平均(回)、 「中央値」: 統計.中央値(回)、 「標準偏差」: 統計.stdev(回)、 「分」: 分(回)、 「マックス」: 最大(回)、 「ops_per_sec」: 1000 / 統計.平均(回) } 確かに run_full_benchmark(自分自身) -> 辞書: """完全な PQC ベンチマーク スイートを実行する""" 結果 = {} # Kyber ベンチマーク 結果[「kyber_keygen」] = self.benchmark_operation( 「Kyber-768キージェネ」, ラムダ: oqs.KeyEncapsulation(「カイバー768」).generate_keypair() ) # encap/decap の設定 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 KeyGen」, ラムダ: oqs.署名(「SPHINCS+-SHAKE-128s-シンプル」).generate_keypair() ) sig = oqs.Signature(「SPHINCS+-SHAKE-128s-シンプル」) sig.generate_keypair() spx_sk = sig.export_secret_key() msg​​ = b"x" * 256 件の結果[「スフィンクスサイン」] = self.benchmark_operation( 「SPHINCS+-SHAKE-128sサイン」, ラムダ: oqs.署名(「SPHINCS+-SHAKE-128s-シンプル」、spx_sk).sign(msg)、反復=20 # 遅いため少ない ) 戻る 結果 確かに 印刷結果(自己、結果: dict): """ベンチマーク結果を美しく印刷""" 印刷("\n=== PQC パフォーマンス ベンチマーク ===) print(f"反復: {self.iterations}\n") のために キー、データ in results.items(): print(f"{データ['名前']}:") print(f" 平均値: {data['mean']:.3f}ms") print(f" 中央値: {data['median']:.3f}ms") print(f「オペレーション/秒: {data['ops_per_sec']:.1f}」) プリント() # ベンチマークを実行する if __名前__ == "__主要__": ベンチ = 暗号ベンチマーク(反復=50) 結果 = bench.run_full_benchmark() bench.print_results(results)