Prestatieoptimalisatie voor post-kwantumcryptografie: handleiding voor ontwikkelaars

importeren gelijktijdige toekomsten importeren oké van typen importeren Lijst, Tupel importeren tijd klas ParallelSigner: """ Parallelle SPHINCS+-ondertekening voor batchbewerkingen Te gebruiken bij het ondertekenen van meerdere onafhankelijke berichten. """ zeker __init__(self, max_workers: int = Geen): """ Initialiseer parallelle ondertekenaar Args: max_workers: CPU-threads om te gebruiken (standaard: CPU-aantal) """ self.max_workers = max_werkers or os.cpu_count() zeker teken_batch( zelf, berichten: Lijst[bytes], geheime_sleutel: bytes ) -> Lijst[bytes]: """ Meerdere berichten parallel ondertekenen Args: berichten: Lijst met te ondertekenen berichten geheime_sleutel: SPHINCS+ geheime sleutel Resultaat: Lijst met handtekeningen in dezelfde volgorde als berichten """ zeker teken_single(bericht: bytes) -> bytes: sig = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig", geheime_sleutel) opbrengst sig.teken(bericht) met concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as uitvoerder: handtekeningen = lijst(uitvoerder.map(sign_single, berichten)) opbrengst handtekeningen zeker sign_with_keys( zelf, items: Lijst[Tupel[bytes, bytes]] # (bericht, geheime_sleutel) ) -> Lijst[bytes]: """Berichten ondertekenen met verschillende sleutels parallel""" zeker teken_item(item: Tupel[bytes, bytes]) -> bytes: bericht, sk = item sig = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig", sk) opbrengst sig.teken(bericht) met concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as uitvoerder: opbrengst list(uitvoerder.map(sign_item, items)) # Benchmarkvergelijking zeker benchmark_parallel_vs_sequentieel(): # Genereer sleutel sig = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig") sig.generate_keypair() sk = sig.export_secret_key() # Maak testberichten berichten = [v"Bericht {i}".coderen() voor i in bereik(16)] # Sequentieel start = time.perf_counter() sequential_sigs = [] voor bericht in berichten: s = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig", sk) sequential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - start # Parallel ondertekenaar = ParallelSigner() start = time.perf_counter() parallel_sigs = signer.sign_batch(messages, sk) par_time = time.perf_counter() - start print(f"Opeenvolgend: {seq_time:.2f}s ({len(berichten)/seq_time:.1f} msg/s)") afdrukken(v"Parallel: {par_time:.2f}s ({len(berichten)/par_time:.1f} bericht/s)") afdrukken(v"Versnelling: {seq_time/par_time:.2f}x")

klas ParallelVerifier: """Parallelle handtekeningverificatie voor validators""" zeker __init__(self, max_workers: int = Geen): self.max_workers = max_workers or os.cpu_count() zeker verifieer_batch( zelf, items: Lijst[Tupel[bytes, bytes, bytes]] # (bericht, sig, pk) ) -> Lijst[bool]: """ Meerdere handtekeningen parallel verifiëren Geeft een lijst met verificatieresultaten weer """ zeker verifieer_single(item: Tupel[bytes, bytes, bytes]) -> bool: bericht, handtekening, publieke_sleutel = item poging: sig = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig") opbrengst sig.verify(bericht, handtekening, publieke_sleutel) behalve: opbrengst Vals met concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as uitvoerder: opbrengst list(uitvoerder.map(verify_single, items)) zeker alles_geldig( zelf, items: Lijst[Tupel[bytes, bytes, bytes]] ) -> bool: """Snelle controle of alle handtekeningen geldig zijn""" resultaten = self.verify_batch(items) opbrengst alle(resultaten) # Voor validators die blokken verwerken: asynchroon def valideer_blok_transacties(transacties: Lijst): verificateur = ParallelVerifier(max_workers=8) # Verificatie-items voorbereiden items = [ (tx.signing_message, tx.signature, tx.public_key) voor tx in transacties ] # Controleer alles parallel resultaten = verifier.verify_batch(items) # Filter geldige transacties valid_txs = [tx voor tx, geldig in zip(transacties, resultaten) if geldig] opbrengst geldige_txs

van functionele hulpmiddelen importeren lru_cache importeren hasjlib klas SleutelCache: """ Afgeleide sleutels in cache opslaan om herhaalde afleiding te voorkomen. Handig voor HD-portemonnees waar vaak dezelfde paden worden gebruikt. """ zeker __init__(self, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} zeker haal_of_afleiden(self, master_seed: bytes, pad: str, derivaat_func) -> Tupel[bytes, bytes]: """ Sleutel in cache ophalen of afleiden en cachen Args: master_seed: Portefeuille master-seedpad: Afleidingspad derivaat_func: Functie die moet worden aangeroepen als cache ontbreekt. Retourneert: (public_key, secret_key) tuple """ # Maak een cachesleutel (bewaar het werkelijke zaad niet in de sleutel) cache_key = hashlib.Blake2b(master_seed + pad.encode()).hexdigest()[:32] if cache_sleutel in zelf._cache: opbrengst self._cache[cache_sleutel] # Sleutels afleiden pk, sk = deriveer_func(master_seed, pad) # Cache met uitzetting if len(self._cache) >= self.max_size: # Eenvoudige FIFO-uitzetting (gebruik OrderedDict in productie) oudste = volgende(iter(self._cache)) del self._cache[oudste] self._cache[cache_key] = (pk, sk) opbrengst pk, sk zeker duidelijk(zelf): """Wis alle sleutels in de cache (roep portemonneevergrendeling aan)""" # Veilig wissen voor sleutel in lijst(self._cache.keys()): pk, sk = self._cache[sleutel] # Overschrijven vóór verwijdering self._cache[sleutel] = (b'\x00' * len(pk), b'\x00' * len(sk)) del self._cache[sleutel] # Gebruik in portemonnee klas Geoptimaliseerde portemonnee: zeker __init__(zelf, master_seed: bytes): self.master_seed = master_seed self.key_cache = SleutelCache(max_grootte=500) zeker get_address_keys(zelf, pad: str) -> Tupel[bytes, bytes]: opbrengst self.key_cache.get_or_derive( self.master_seed, pad, self._derive_keys ) zeker _afleiden_sleutels(zelf, zaad: bytes, pad: str): # Werkelijke afleidingslogica ...

klas Handtekeningcache: """ Resultaten van handtekeningverificatie in cache opslaan Voor validators om te voorkomen dat geziene transacties opnieuw worden geverifieerd. """ zeker __init__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} zeker _handtekening_id(self, bericht: bytes, handtekening: bytes, public_key: bytes ) -> str: """Maak een unieke ID voor handtekeningverificatie""" opbrengst hashlib.Blake2b( bericht + handtekening[:64] + publieke_sleutel, # Eerste 64 bytes aan sig genoeg digest_size=16).hexdigest() zeker check_of_verify(zelf, bericht: bytes, handtekening: bytes, public_key: bytes) -> bool: """Cache controleren of verifiëren en resultaat cachen""" sig_id = self._signature_id(bericht, handtekening, publieke_sleutel) if sig_id in zelf._geverifieerd: opbrengst zelf._verifieerd[sig_id] # Verifieer sig = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig") is_valid = sig.verify(bericht, handtekening, publieke_sleutel) # Cache (met uitzetting) if len(self._verified) >= self.max_size: # Verwijder ~10% oudste vermeldingen to_remove = lijst(self._verified.keys())[:self.max_size // 10] voor sleutel in to_remove: del self._verified[sleutel] self._verified[sig_id] = is_geldig opbrengst is_geldig

importeren gc klas GeheugenEfficientSigner: """ Geheugenefficiënt ondertekenen voor ingebouwde/mobiele apparaten """ zeker sign_and_release(zelf, bericht: bytes, geheime_sleutel: bytes) -> bytes: """ Bericht ondertekenen en het sleutelgeheugen onmiddellijk vrijgeven. Gebruik voor eenmalige handtekeningen waarbij de sleutel niet mag blijven bestaan. """ sig_obj = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig", geheime_sleutel) handtekening = sig_obj.sign(bericht) # OQS-object vrijgeven del sig_obj # Overschrijf geheime sleutel if isinstance(geheime_sleutel, bytearray): voor i in bereik(len(geheime_sleutel)): geheime_sleutel[i] = 0 # Forceer afvalinzameling gc.collect() opbrengst handtekening zeker streaming_teken(self, message_chunks: Iterator[bytes], geheime_sleutel: bytes ) -> bytes: """ Stream het bericht ondertekenen zonder alles in het geheugen te laden. Has het bericht vooraf in stukjes en onderteken vervolgens de hash. """ # Hash-bericht in stukjes hasher = hashlib.Blake2b(digest_size=32) voor brok in message_chunks: hasher.update(chunk) message_hash = hasher.digest() # Onderteken de hash sig = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig", geheime_sleutel) opbrengst sig.sign(bericht_hash)

# Controleer de CPU-functies voor optimale algoritmeselectie importeren subproces zeker get_cpu_features() -> instellen: """Beschikbare CPU SIMD-functies detecteren""" poging: #Linux met open("/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") opbrengst functies behalve: opbrengst set() zeker selecteer_optimale_variant() -> str: """Selecteer de beste SPHINCS+-variant voor deze CPU""" features = get_cpu_features() if "avx512" in functies: # AVX-512 biedt ~20-30% versnelling afdrukken("Gebruik van AVX-512 geoptimaliseerde implementatie") opbrengst "SPHINCS+-SHAKE-128s-eenvoudig" # liboqs selecteert automatisch elif "avx2" in Kenmerken: afdrukken("Gebruik van AVX2 geoptimaliseerde implementatie") opbrengst "SPHINCS+-SHAKE-128s-eenvoudig" anders: afdrukken("Referentie-implementatie gebruiken") opbrengst "SPHINCS+-SHAKE-128s-eenvoudig" # Compileer liboqs met optimale vlaggen # cmake -DOQS_USE_AVX2_INSTRUCTIONS=AAN -DOQS_USE_AVX512_INSTRUCTIONS=AAN ..

importeren statistieken importeren tijd klas CryptoBenchmark: """Uitgebreide PQC-benchmarking""" zeker __init__(zelf, iteraties: int = 100): self.iteraties = iteraties zeker benchmark_operatie(self, naam: str, bediening, setup=Geen ) -> dicteer: """Benchmark een enkele operatie""" keer = [] voor _ in bereik(zelf.iteraties): if opstelling: ctx = opstelling() start = time.perf_counter() if opstelling: bediening(ctx) anders: operatie() verstreken = (time.perf_counter() - start) * 1000 # ms tijden.append(verstreken) opbrengst { "naam": naam, "gemeen": statistieken.gemiddelde(tijden), "mediaan": statistieken.mediaan(tijden), "stdev": statistieken.stdev(tijden), "min": min(tijden), "max": maximaal(tijden), "ops_per_sec": 1000 / statistieken.gemiddelde(tijden) } zeker run_full_benchmark(zelf) -> dicteer: """Voer de volledige PQC-benchmarksuite uit""" resultaten = {} # Kyber-benchmarks resultaten["kyber_keygen"] = zelf.benchmark_operation( "Kyber-768 KeyGen", lambda: oqs.KeyEncapsulation("Kyber768").generate_keypair() ) # Instelling voor encap/decap kem = oqs.KeyEncapsulation("Kyber768") pk = kem.generate_keypair() sk = kem.export_secret_key() resultaten["kyber_encap"] = zelf.benchmark_operation( "Kyber-768 Encap", lambda: kem.encap_secret(pk) ) ct, _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation("Kyber768", sk) resultaten["kyber_decap"] = zelf.benchmark_operation( "Kyber-768 Decap", lambda: kem_dec.decap_secret(ct) ) # SPHINCS+-benchmarks resultaten["sfincs_keygen"] = zelf.benchmark_operation( "SPHINCS+-SHAKE-128s KeyGen", lambda: oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig").generate_keypair() ) sig = oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig") sig.generate_keypair() spx_sk = sig.export_secret_key() msg ​​= b"x" * 256 resultaten["sfincs_teken"] = zelf.benchmark_operation( "SPHINCS+-SHAKE-128s-bord", lambda: oqs.Handtekening("SPHINCS+-SHAKE-128s-eenvoudig", spx_sk).sign(msg), iteraties=20 # Minder vanwege traagheid ) opbrengst resultaten zeker afdrukresultaten(zelf, resultaten: dictaat): """Mooie afdrukbenchmarkresultaten""" afdrukken("\n=== PQC-prestatiebenchmark ===) afdrukken(v"Iteraties: {self.iteraties}\n") voor sleutel, gegevens in resultaten.items(): print(f"{gegevens['naam']}:") afdrukken(v" Gemiddelde: {data['gemiddelde']:.3f}ms") afdrukken(v" Mediaan: {data['mediaan']:.3f}ms") afdrukken(v"Ops/sec: {data['ops_per_sec']:.1f}") afdrukken() # Voer een benchmark uit if __naam__ == "__voornaamst__": bank = CryptoBenchmark(iteraties=50) resultaten = bench.run_full_benchmark() bench.print_results(resultaten)