Optymalizacja wydajności dla kryptografii postkwantowej: przewodnik dla programistów

import współbieżne.przyszłości import OK z pisanie na maszynie import Lista, Tuple import czas klasa Równoległy podpis: """ Równoległe podpisywanie SPHINCS+ dla operacji wsadowych Użyj podczas podpisywania wielu niezależnych wiadomości. """ def __inicj__(self, max_workers: int = Brak): """ Zainicjuj równoległego podpisującego Args: max_workers: wątki CPU do użycia (domyślnie: liczba CPU) """ self.max_workers = max_workers or liczba_os.cpu() def partia_znaków( self, wiadomości: Lista [bajty], klucz tajny: bajty ) -> Lista [bajty]: """ Podpisz wiele wiadomości równolegle Args: wiadomości: Lista wiadomości do podpisania tajny klucz: tajny klucz SPHINCS+ Zwraca: Lista podpisów w tej samej kolejności co wiadomości """ def znak_pojedynczy(wiadomość: bajty) -> bajty: sig = oqs.Signature(„SPHINCS+-SHAKE-128s-prosty”, sekret_klucz) powrót sig.sign(wiadomość) z concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as executor: podpisy = lista(executor.map(sign_single, Messages)) powrót podpisy def znak_z_kluczami( ja, elementy: Lista [Krotka[bajty, bajty]] # (wiadomość, klucz_tajny) ) -> Lista[bajty]: """Podpisz wiadomości równolegle różnymi kluczami""" def element_znaku(przedmiot: Krotka[bajty, bajty]) -> bajty: wiadomość, sk = pozycja sig = oqs.Signature(„SPHINCS+-SHAKE-128s-prosty”, sk) powrót sig.sign(wiadomość) z concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as wykonawca: powrót list(executor.map(sign_item, items)) # Porównanie benchmarków def benchmark_parallel_vs_sequential(): # Wygeneruj klucz sig = oqs.Podpis(„SPHINCS+-SHAKE-128s-prosty”) sig.generate_keypair() sk = sig.export_secret_key() # Twórz wiadomości testowe wiadomości = [f„Wiadomość {i}”.kodować() Do i in zakres(16)] # Sekwencja start = time.perf_counter() sequential_sigs = [] Do wiadomość in wiadomości: s = oqs.Signature(„SPHINCS+-SHAKE-128s-prosty”, sk) sequential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - start # Równoległy podpisujący = Równoległy podpis() start = time.perf_counter() równoległy_sigs =signer.sign_batch(wiadomości, sk) par_time = time.perf_counter() - rozpocznij drukowanie(f„Sekwencyjny: {seq_time:.2f} s ({len(wiadomości)/seq_time:.1f} msg/s)”) wydrukować (f„Równolegle: {par_time:.2f} s ({len(messages)/par_time:.1f} msg/s)”) wydrukować (f„Przyspieszenie: {seq_time/par_time:.2f}x”)

klasa Weryfikator równoległy: """Równoległa weryfikacja podpisu dla walidatorów""" def __inicj__(self, max_workers: int = Brak): self.max_workers = max_workers or liczba_os.cpu() def zweryfikować_partię( ja, elementy: Lista [Krotka[bajty, bajty, bajty]] # (wiadomość, sygnał, pk) ) -> Lista[bool]: """ Zweryfikuj wiele podpisów równolegle. Zwraca listę wyników weryfikacji """ def zweryfikować_single(przedmiot: Krotka[bajty, bajty, bajty]) -> bool: wiadomość, podpis, klucz_publiczny = pozycja próbować: sig = oqs.Podpis(„SPHINCS+-SHAKE-128s-prosty”) powrót sig.verify(wiadomość, podpis, klucz_publiczny) z wyjątkiem: powrót FAŁSZ z concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as wykonawca: powrót list(executor.map(verify_single, elementy)) def wszystko_ważne( ja, elementy: Lista [Krotka[bajty, bajty, bajty]] ) -> bool: """Szybkie sprawdzenie, czy wszystkie podpisy są ważne""" wyniki = self.verify_batch(elementy) powrót wszystkie(wyniki) # Dla walidatorów przetwarzających bloki: asynchroniczna def valid_block_transactions(transakcje: Lista): weryfikator = Weryfikator równoległy(maks._pracowników=8) # Przygotuj elementy weryfikacyjne elementy = [ (tx.signing_message, tx.signature, tx.public_key) Do tx in transakcje] # Sprawdź wszystko równolegle wyniki = weryfikator.verify_batch(elementy) # Filtruj ważne transakcje valid_txs = [tx Do tx, ważne in zip(transakcje, wyniki) if ważny] powrót ważny_txs

z narzędzia funkcyjne import lru_cache import hashlib klasa Pamięć podręczna kluczy: """ Klucze wyprowadzane z pamięci podręcznej, aby uniknąć wielokrotnego wyprowadzania. Przydatne w przypadku portfeli HD, w których często uzyskiwany jest dostęp do tych samych ścieżek. """ def __inicj__(self, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} def get_or_derive( self, master_seed: bajty, ścieżka: str, pochodna_func ) -> Krotka[bajty, bajty]: """ Pobierz klucz z pamięci podręcznej lub wyprowadź i buforuj Argumenty: master_seed: Ścieżka głównego źródła portfela: Ścieżka wyprowadzenia funkcja_pochodna: Funkcja do wywołania w przypadku braku pamięci podręcznej. Zwraca: (klucz_publiczny, klucz_tajny) krotka """ # Utwórz klucz pamięci podręcznej (nie przechowuj rzeczywistego materiału siewnego w kluczu) cache_key = hashlib.Blake2b(master_seed + path.encode()).hexdigest()[:32] if klucz_pamięci podręcznej in self._cache: powrót self._cache[klucz_cache] # Wyprowadź klucze pk, sk = pochodna_func(główne ziarno, ścieżka) # Pamięć podręczna z eksmisją if len(self._cache) >= self.max_size: # Prosta eksmisja FIFO (użyj OrderedDict w produkcji) najstarszy = następny(iter(self._cache)) del self._cache[najstarszy] self._cache[key_cache] = (pk, sk) powrót pk, sk def jasne(ja): """Wyczyść wszystkie klucze z pamięci podręcznej (zablokuj portfel)""" # Bezpieczne usuwanie Do klawisz in list(self._cache.keys()): pk, sk = self._cache[klucz] # Nadpisz przed usunięciem self._cache[key] = (ur'\x00' *len(pk), ur'\x00' * len(sk)) del self._cache[klucz] # Użycie w portfelu klasa Zoptymalizowany portfel: def __inicj__(self, master_seed: bajty): self.master_seed = master_seed self.key_cache = Pamięć podręczna kluczy(maks._rozmiar=500) def get_address_keys(self, ścieżka: str) -> Krotka[bajty, bajty]: powrót self.key_cache.get_or_derive( self.master_seed, ścieżka, self._derive_keys ) def _derive_keys(self, ziarno: bajty, ścieżka: str): # Rzeczywista logika wyprowadzania ...

klasa Pamięć podręczna podpisów: """ Wyniki weryfikacji podpisu w pamięci podręcznej Dla walidatorów, aby uniknąć ponownej weryfikacji widzianych transakcji. """ def __inicj__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} def _identyfikator_podpisu( self, wiadomość: bajty, podpis: bajty, klucz_publiczny: bajty ) -> str: """Utwórz unikalny identyfikator do weryfikacji podpisu""" powrót hashlib.Blake2b(wiadomość + podpis[:64] +klucz_publiczny, # Wystarczą pierwsze 64 bajty sig Digest_size=16).hexdigest() def sprawdź_lub_weryfikuj( self, wiadomość: bajty, podpis: bajty, klucz_publiczny: bajty ) -> bool: """Sprawdź pamięć podręczną lub zweryfikuj i buforuj wyniki""" sig_id = self._signature_id(wiadomość, podpis, klucz_publiczny) if sig_id in self._verified: powrót self._verified[sig_id] # Sprawdź sig = oqs.Podpis(„SPHINCS+-SHAKE-128s-prosty”) is_valid = sig.verify(wiadomość, podpis, klucz_publiczny) # Pamięć podręczna (z eksmisją) if len(self._verified) >= self.max_size: # Usuń ~10% najstarszych wpisów to_remove = list(self._verified.keys())[:self.max_size // 10] Do klawisz in do_usunięcia: del self._verified[key] self._verified[sig_id] = is_valid powrót jest_ważny

import gc klasa MemoryEfficientSigner: „”” Oszczędzające pamięć podpisywanie dla urządzeń wbudowanych/mobilnych „”” def znak_i_zwolnienie( self, wiadomość: bajty, klucz_tajny: bajty ) -> bajty: """ Podpisz wiadomość i natychmiast zwolnij pamięć klucza. Używaj do podpisów jednorazowych, gdzie klucz nie powinien zostać zachowany. """ sig_obj = oqs.Podpis(„SPHINCS+-SHAKE-128s-prosty”, sekret_klucz) podpis = sig_obj.sign(wiadomość) # Zwolnij obiekt OQS del sig_obj # Zastąp tajny klucz if isinstance(sekret_klucz, bajtarray): Do i in zakres(dł(sekret_klucz)): sekret_klucz[i] = 0 # Wymuś zbieranie śmieci gc.collect() powrót podpis def streaming_sign( self, Message_chunks: Iterator[bajty], secret_key: bytes ) -> bajty: """ Podpisz wiadomość przesyłaną strumieniowo bez ładowania całości do pamięci. Podziel wiadomość wstępnie na fragmenty, a następnie podpisz skrót. """ # Hashuj wiadomość w kawałkach hasher = hashlib.Blake2b(digest_size=32) Do kawałek in Message_chunks: hasher.update(fragment) Message_hash = hasher.digest() # Podpisz skrót sig = oqs.Podpis(„SPHINCS+-SHAKE-128s-prosty”, sekret_klucz) powrót sig.sign(hash_wiadomości)

# Sprawdź funkcje CPU, aby wybrać optymalny algorytm import podproces def get_cpu_features() -> zestaw: """Wykryj dostępne funkcje SIMD CPU""" próbować: #Linux z Otwarte(„/proc/informacje o procesorze”) as f: cpuinfo = f.read() Features = set() if „avx2” in informacje o procesorze: funkcje.add(„avx2”) if „avx512” in informacje o procesorze: funkcje.add(„avx512”) if „aes” in informacje o procesorze: funkcje.add(„aesni”) powrót cechy z wyjątkiem: powrót ustawić() def wybierz_optymalny_wariant() -> str: """Wybierz najlepszy wariant SPHINCS+ dla tego CPU""" funkcje = get_cpu_features() if „avx512” in cechy: # AVX-512 zapewnia przyspieszenie o ~20-30%. wydrukować(„Korzystanie ze zoptymalizowanej implementacji AVX-512”) powrót „SPHINCS+-SHAKE-128s-prosty” # liboqs wybiera automatycznie Elif „avx2” in funkcje: drukuj(„Korzystanie z implementacji zoptymalizowanej pod AVX2”) powrót „SPHINCS+-SHAKE-128s-prosty” else: drukuj(„Korzystanie z implementacji referencyjnej”) powrót „SPHINCS+-SHAKE-128s-prosty” # Skompiluj liboqs z optymalnymi flagami # cmake -DOQS_USE_AVX2_INSTRUCTIONS=ON -DOQS_USE_AVX512_INSTRUCTIONS=ON ..

import statystyka import czas klasa KryptoBenchmark: """Kompleksowy benchmarking PQC""" def __inicj__(self, iteracje: int = 100): self.iteracje = iteracje def benchmark_operacja( self, nazwa: str, operacja, setup=Nic ) -> dykt: """Porównaj pojedynczą operację""" razy = [] Do _ in zakres(samoistne iteracje): if konfiguracja: ctx = setup() start = time.perf_counter() if konfiguracja: operacja (ctx) else: upłynęła operacja() = (time.perf_counter() - start) * 1000 # SM czasy.dołącz(upłynęły) powrót { "nazwa": nazwa, "mieć na myśli": statystyki.średnia(razy), "mediana": statystyki.mediana(razy), „stdev”: statystyki.stdev(razy), „min”: min(razy), „maks”: maks. (razy), „ops_per_sec”: 1000 / statystyka.średnia(razy) } def run_full_benchmark(samo) -> dyktuj: """Uruchom kompletny zestaw testów porównawczych PQC""" wyniki = {} # Testy porównawcze Kyber wyniki[„kyber_keygen”] = self.benchmark_operacja( „Generowanie klucza Kyber-768”, lambda: oqs.KeyEncapsulation(„Kyber768”).generate_keypair() ) # Konfiguracja encap/decap kem = oqs.KeyEncapsulation(„Kyber768”) pk = kem.generate_keypair() sk = kem.export_secret_key() wyniki[„kyber_encap”] = self.benchmark_operacja( „Obudowa Kyber-768”, lambda: kem.encap_secret(pk) ) ct, _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation(„Kyber768”, sk) wyniki[„kyber_decap”] = self.benchmark_operacja( „Kyber-768 Decap”, lambda: kem_dec.decap_secret(ct) ) # Testy porównawcze SPHINCS+ wyniki[„generator_klucza_zwieracza”] = self.benchmark_operacja( „SPHINCS+-SHAKE-128s Generowanie klucza”, lambda: oqs.Podpis(„SPHINCS+-SHAKE-128s-prosty”).generate_keypair() ) sig = oqs.Signature(„SPHINCS+-SHAKE-128s-prosty”) sig.generate_keypair() spx_sk = sig.export_secret_key() msg ​​= b"x" * 256 wyników["znak_zwieracza"] = self.benchmark_operacja( „Znak SPHINCS+-SHAKE-128s”, lambda: oqs.Podpis(„SPHINCS+-SHAKE-128s-prosty”, spx_sk).sign(msg), iteracje=20 # Mniej ze względu na powolność ) powrót wyniki def wydruk_wyniki(samodzielnie, wyniki: dykt): """Wyniki testów porównawczych ładnego druku""" wydrukować("\n=== Test wydajności PQC ===") wydrukować (f"Iteracje: {self.iterations}\n") Do klucz, dane in wyniki.items(): print(f„{dane['nazwa']}:”) wydrukować (f"Średnia: {dane['średnia']:.3f}ms") wydrukować (f" Mediana: {dane['mediana']:.3f}ms") wydrukować (f„ Ops/s: {data['ops_per_sec']:.1f}”) drukuj() # Uruchom test porównawczy if __imię__ == "__główny__": ławka = KryptoBenchmark(iteracji=50) wyniki =bench.run_full_benchmark()bench.print_results(results)