Оптимизация производительности пост-квантовой криптографии: руководство для разработчиков

импортировать concurrent.futures импортировать ОК от печатание импортировать Список, Кортеж импортировать время сорт Параллельныйподписчик: """ Параллельная подпись SPHINCS+ для пакетных операций. Используется при подписании нескольких независимых сообщений. """ защита __init__(я, max_workers: int = Нет): """ Инициализировать параллельную подписывающую сторону. Args: max_workers: используемые потоки CPU (по умолчанию: количество CPU) """ self.max_workers = max_workers or os.cpu_count() защита знак_пакета(self, messages: List[bytes], secret_key: bytes) -> List[bytes]: """ Параллельная подпись нескольких сообщений. Аргументы: сообщения: список сообщений для подписи secret_key: секретный ключ SPHINCS+. Возврат: список подписей в том же порядке, что и сообщения """ защита знак_одиночный(сообщение: байты) -> байты: sig = oqs.Signature("SPHINCS+-SHAKE-128s-просто", секретный_ключ) возвращаться sig.sign(сообщение) с concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as исполнитель: подписи = список(executor.map(sign_single, messages)) возвращаться подписи защита Sign_with_keys(я, элементы: Список[Кортеж[байты, байты]] # (сообщение, секретный_ключ) ) -> Список[байты]: """Подписывать сообщения разными ключами параллельно""" защита знак_элемент(элемент: Кортеж[байты, байты]) -> байты: сообщение, sk = item sig = oqs.Signature("SPHINCS+-SHAKE-128s-просто", ск) возвращаться sig.sign(сообщение) с concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as исполнитель: возвращаться список(executor.map(sign_item, элементы)) # Сравнение эталонных показателей защита эталонный_параллельный_vs_sequential(): # Генерируем ключ sig = oqs.Signature("SPHINCS+-SHAKE-128s-просто") sig.generate_keypair() sk = sig.export_secret_key() # Создаем тестовые сообщения сообщения = [ф«Сообщение {i}».кодировать() для i in диапазон(16)] # Последовательный start = time.perf_counter() sequential_sigs = [] для сообщение in сообщения: s = oqs.Signature("SPHINCS+-SHAKE-128s-просто", sk) sequential_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() - start print(f«Последовательно: {seq_time:.2f} с ({len(messages)/seq_time:.1f} мсг/с)») print(f«Параллельно: {par_time:.2f} с ({len(messages)/par_time:.1f} мсг/с)») print(f«Ускорение: {seq_time/par_time:.2f}x»)

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

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

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

импортировать gc сорт ПамятьЭффективныйподписчик: """ Подпись с эффективным использованием памяти для встроенных и мобильных устройств """ защита Sign_and_release(self, message: байты, 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) -> байты: """ Подпишите потоковое сообщение, не загружая все в память. Предварительно хешируйте сообщение по частям, а затем подпишите хеш. """ # Хэширование сообщения по частям хэшер = 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() -> установить: """Обнаружение доступных функций SIMD CPU""" пытаться: # Линукс с открыть("/прок/информация о процессоре") 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() -> ул: """Выберите лучший вариант 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 ..

импортировать статистика импортировать время сорт КриптоБенчмарк: """Комплексный сравнительный анализ PQC""" защита __init__(self, итерации: int = 100): self.iterations = итерации защита эталонная_операция(self, name: str, Operation, setup=Никто ) -> дикт: """Протестируйте одну операцию""" раз = [] для _ in диапазон(само.итерации): if настройка: ctx = setup() start = time.perf_counter() if настройка: операция (ctx) еще: прошедшее время операции() = (time.perf_counter() - начало) * 1000 # РС times.append(прошедшее) возвращаться { "имя": имя, "иметь в виду": статистика.среднее(раз), "медиана": статистика.медиана(раз), "stdev": статистика.stdev(раз), "мин": мин(раз), "макс": макс(раз), "ops_per_sec": 1000 / статистика.среднее(раз) } защита run_full_benchmark(сам) -> дикт: """Запустите полный пакет тестов PQC""" результаты = {} # тесты Kyber результаты["kyber_keygen"] = self.benchmark_operation( "Kyber-768 Генератор ключей", лямбда: oqs.KeyEncapsulation("Кибер768").generate_keypair() ) # Настройка инкап/декап кем = 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", ск) результаты["kyber_decap"] = self.benchmark_operation( "Kyber-768 Декап", лямбда: kem_dec.decap_secret(ct) ) # тесты SPHINCS+ результаты["сфинкс_кейген"] = 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 результатов["сфинкс_знак"] = 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 результаты.items(): print(f"{данные['имя']}:") print(f" Среднее: {данные['среднее']:.3f}мс") print(f«Медиана: {данные['медиана']:.3f}мс») print(f" Операций в секунду: {data['ops_per_sec']:.1f}") печать() # Запускаем тест if __имя__ == "__основной__": скамейка = КриптоБенчмарк(итерации = 50) результаты =bench.run_full_benchmark()bench.print_results(результаты)