Оптимізація продуктивності для постквантової криптографії: посібник розробника

імпорт concurrent.futures імпорт oqs від введення тексту імпорт Список, Кортеж імпорт час клас ParallelSigner: """ Паралельний підпис SPHINCS+ для пакетних операцій. Використовуйте під час підписання кількох незалежних повідомлень. """ зах __init__(self, max_workers: int = None): """ Ініціалізувати аргументи паралельного підписувача: max_workers: потоки CPU для використання (за замовчуванням: кількість CPU) """ self.max_workers = max_workers or os.cpu_count() зах sign_batch( self, messages: List[bytes], secret_key: bytes ) -> List[bytes]: """ Підпишіть кілька повідомлень паралельно Args: messages: Список повідомлень для підпису secret_key: SPHINCS+ секретний ключ Повернення: Список підписів у тому самому порядку, що й повідомлення """ зах sign_single(повідомлення: байти) -> байти: sig = oqs.Signature("SPHINCS+-SHAKE-128s-простий", секретний_ключ) повернення sig.sign(повідомлення) з concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as виконавець: підписи = список(executor.map(sign_single, messages)) повернення підписи зах знак_з_ключами( self, items: List[Кортеж[байти, байти]] # (повідомлення, секретний_ключ) ) -> Список[байти]: """Підписуйте повідомлення різними ключами паралельно""" зах sign_item(пункт: Кортеж[байти, байти]) -> байти: повідомлення, sk = елемент sig = oqs.Signature("SPHINCS+-SHAKE-128s-простий", ск) повернення sig.sign(повідомлення) з concurrent.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() # Створення тестових повідомлень повідомлення = [ф"Повідомлення {i}".encode() для i in діапазон(16)] # Послідовний start = time.perf_counter() sequential_sigs = [] для msg in повідомлення: s = oqs.Signature("SPHINCS+-SHAKE-128s-простий", sk) sequential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - початок # Паралельний підписувач = ParallelSigner() 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)") print(f"Паралельно: {par_time:.2f}s ({len(messages)/par_time:.1f} msg/s)") print(f"Прискорення: {seq_time/par_time:.2f}x")

клас ParallelVerifier: """Паралельна перевірка підпису для валідаторів""" зах __init__(self, max_workers: int = None): self.max_workers = max_workers or os.cpu_count() зах verify_batch( self, items: List[Кортеж[байти, байти, байти]] # (msg, sig, pk) ) -> Список [bool]: """ Перевірити кілька підписів паралельно Повертає список результатів перевірки """ зах verify_single(пункт: Кортеж[байти, байти, байти]) -> bool: повідомлення, підпис, відкритий_ключ = елемент try: sig = oqs.Signature("SPHINCS+-SHAKE-128s-простий") повернення sig.verify(повідомлення, підпис, відкритий_ключ) крім: повернення Неправда з concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as виконавець: повернення список(executor.map(verify_single, items)) зах all_valid( self, items: List[Кортеж[байти, байти, байти]] ) -> bool: """Швидка перевірка, чи всі підписи дійсні""" результати = self.verify_batch(items) повернення всі (результати) # Для блоків обробки валідаторів: асинхронне визначення validate_block_transactions(транзакції: Список): verifier = ParallelVerifier(max_workers=8) # Підготуйте елементи перевірки елементи = [ (tx.signing_message, tx.signature, tx.public_key) для tx in транзакції] # Перевірити все паралельно результати = verifier.verify_batch(items) # Фільтрувати дійсні транзакції valid_txs = [tx для tx, дійсний in zip(транзакції, результати) if дійсний] повернення valid_txs

від functools імпорт lru_cache імпорт хешліб клас KeyCache: """ Кешувати отримані ключі, щоб уникнути повторного виведення. Корисно для HD-гаманців, де часто використовуються ті самі шляхи. """ зах __init__(self, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} зах get_or_derive(self, master_seed: байти, шлях: str, derive_func) -> Кортеж[байти, байти]: """ Отримати кешований ключ або отримати та кешувати аргументи: master_seed: Шлях основного початкового значення гаманця: Шлях похідного derive_func: Функція для виклику, якщо кеш пропадає. Повертає: (public_key, secret_key) tuple """ # Створіть ключ кешу (не зберігайте фактичне початкове значення в ключі) cache_key = hashlib.Blake2b(master_seed + path.encode()).hexdigest()[:32] if ключ_кешу in self._cache: повернення self._cache[кеш_ключ] # Вивести ключі pk, sk = derive_func(master_seed, шлях) # Кеш з виселенням if len(self._cache) >= self.max_size: # Просте вилучення FIFO (використовуйте OrderedDict у виробництві) oldest = next(iter(self._cache)) дел self._cache[старіший] self._cache[кеш_ключ] = (pk, sk) повернення пк, ск зах ясно(само): """Очистити всі кешовані ключі (виклик блокування гаманця)""" # Безпечне видалення для ключ in list(self._cache.keys()): pk, sk = self._cache[ключ] # Перезаписати перед видаленням self._cache[ключ] = (b'\x00' * len(pk), b'\x00' * len(sk)) дел self._cache[ключ] # Використання в гаманці клас Оптимізований гаманець: зах __init__(self, master_seed: байти): self.master_seed = master_seed self.key_cache = KeyCache(max_size=500) зах get_address_keys(self, шлях: str) -> Кортеж[байти, байти]: повернення self.key_cache.get_or_derive( self.master_seed, шлях, self._derive_keys ) зах _derive_keys(self, seed: bytes, path: str): # Справжня логіка виведення ...

клас SignatureCache: """ Кешувати результати перевірки підпису Для валідаторів, щоб уникнути повторної перевірки побачених транзакцій. """ зах __init__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} зах _signature_id(я, повідомлення: байти, підпис: байти, відкритий_ключ: байти) -> str: """Створити унікальний ідентифікатор для перевірки підпису""" повернення hashlib.Blake2b( повідомлення + підпис[:64] + відкритий_ключ, # Достатньо перших 64 байтів sig digest_size=16 ).hexdigest() зах перевірити_або_перевірити(я, повідомлення: байти, підпис: байти, відкритий_ключ: байти) -> логічний: """Перевірити кеш-пам'ять або перевірити та кешувати результат""" 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 to_remove: дел self._verified[key] self._verified[sig_id] = is_valid повернення is_valid

імпорт gc клас MemoryEfficientSigner: """ Підписання з ефективним використанням пам'яті для вбудованих/мобільних пристроїв """ зах sign_and_release(я, повідомлення: байти, секретний_ключ: байти) -> байти: """ Підпишіть повідомлення та негайно звільніть пам'ять ключів. Використовуйте для одноразових підписів, де ключ не повинен зберігатися. """ sig_obj = oqs.Signature("SPHINCS+-SHAKE-128s-простий", секретний_ключ) підпис = sig_obj.sign(повідомлення) # Вивільнення об'єкта OQS дел sig_obj # Перезаписати секретний ключ if isinstance(secret_key, bytearray): для i in діапазон (len(secret_key)): secret_key[i] = 0 # Примусове збирання сміття gc.collect() повернення підпис зах streaming_sign( self, message_chunks: Ітератор[байти], секретний_ключ: байти) -> байти: """ Підпишіть потокове повідомлення, не завантажуючи все в пам'ять. Попередньо хешуйте повідомлення частинами, а потім підпишіть хеш. """ # Хеш-повідомлення у шматках hasher = 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 для вибору оптимального алгоритму імпорт підпроцес зах get_cpu_features() -> встановити: """Виявлення доступних функцій CPU SIMD""" try: # Linux з ВІДЧИНЕНО("/proc/cpuinfo") as f: cpuinfo = f.read() features = set() if "avx2" in cpuinfo: features.add("avx2") if "avx512" in cpuinfo: features.add("avx512") if "aes" in cpuinfo: features.add("aesni") повернення особливості крім: повернення набір() зах select_optimal_variant() -> str: """Виберіть найкращий варіант SPHINCS+ для цього CPU""" особливості = 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 ..

імпорт статистика імпорт час клас CryptoBenchmark: """Комплексний порівняльний аналіз PQC""" зах __init__(self, ітерації: int = 100): self.iterations = ітерації зах benchmark_operation( self, name: str, operation, setup=Жодного ) -> dict: """Порівняльний тест однієї операції""" раз = [] для _ in діапазон (самоітерації): if налаштування: ctx = setup() start = time.perf_counter() if налаштування: операція (ctx) інше: операція() минула = (time.perf_counter() - початок) * 1000 # РС times.append(минуло) повернення { "ім'я": ім'я, "означати":statistic.mean(times), "медіана":statistic.median(times), "stdev":statistic.stdev(рази), "хв": хв (раз), "макс": макс.(разів), "ops_per_sec": 1000 / statistics.mean(times) } зах run_full_benchmark(само) -> dict: """Запустити повний набір тестів PQC""" результати = {} # Тести Kyber результати["kyber_keygen"] = self.benchmark_operation( "Kyber-768 KeyGen", лямбда: oqs.KeyEncapsulation("Kyber768").generate_keypair() ) # Налаштування для encap/decap kem = oqs.KeyEncapsulation("Kyber768") pk = kem.generate_keypair() sk = kem.export_secret_key() результати["kyber_encap"] = self.benchmark_operation( "Kyber-768 Encap", лямбда: kem.encap_secret(pk) ) ct, _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation("Kyber768", sk) результати["kyber_decap"] = self.benchmark_operation( "Kyber-768 Decap", лямбда: kem_dec.decap_secret(ct) ) # Тести SPHINCS+ результати["sphincs_keygen"] = self.benchmark_operation( "SPHINCS+-SHAKE-128s KeyGen", лямбда: 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 # Менше через повільність ) повернення результати зах print_results(само, результати: dict): """Гарні результати порівняльного тестування друку""" друкувати("\n=== Тест продуктивності PQC ===") print(f"Ітерації: {self.iterations}\n") для ключ, дані in results.items(): print(f"{data['name']}:") print(f" Середнє: {дані['mean']:.3f} мс") print(f" Медіана: {дані['median']:.3f} мс") print(f"Ops/sec: {data['ops_per_sec']:.1f}") print() # Запустіть контрольний тест if __ім'я__ == "__головний__": лавка = CryptoBenchmark(ітерації=50) результати = bench.run_full_benchmark() bench.print_results(результати)