Ydeevneoptimering til postkvantekryptering: Udviklervejledning

importere concurrent.futures importere oks fra skrive importere Liste, Tuple importere tid klasse ParallelSigner: """ Parallel SPHINCS+-signering til batch-handlinger Bruges ved signering af flere uafhængige meddelelser. """ def __init__(selv, max_workers: int = Ingen): """ Initialiser parallel underskriver Args: max_workers: CPU-tråde, der skal bruges (standard: CPU-antal) """ self.max_workers = max_workers or os.cpu_count() def sign_batch( selv, beskeder: Liste[bytes], hemmelig_nøgle: bytes ) -> Liste[bytes]: """ Signer flere meddelelser parallelt Args: meddelelser: Liste over meddelelser, der skal signeres hemmelig_nøgle: SPHINCS+ hemmelig nøgle Returnerer: Liste over signaturer i samme rækkefølge som meddelelser """ def tegn_enkelt(meddelelse: bytes) -> bytes: sig = oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt", hemmelig_nøgle) returnere sig.sign(besked) med concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as executor: signaturer = liste(executor.map(sign_single, messages)) returnere underskrifter def underskriv_med_nøgler( selv, elementer: Liste[Tuple[bytes, bytes]] # (meddelelse, hemmelig_nøgle) ) -> Liste[bytes]: """Signér meddelelser med forskellige nøgler parallelt""" def sign_item(punkt: Tuple[bytes, bytes]) -> bytes: besked, sk = item sig = oqs.Signature("SPHINCS+-SHAKE-128s-simpelt", sk) returnere sig.sign(besked) med concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as bobestyrer: returnere list(executor.map(sign_item, items)) # Benchmark sammenligning def benchmark_parallel_vs_sekventiel(): # Generer nøgle sig = oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt") sig.generate_keypair() sk = sig.export_secret_key() # Opret testmeddelelser beskeder = [f"Besked {i}".encode() for i in interval(16)] # Sekventiel start = time.perf_counter() sequential_sigs = [] for besked in beskeder: s = oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt", sk) sequential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - start # Parallel underskriver = ParallelSigner() start = time.perf_counter() parallel_sigs = signer.sign_batch(messages, sk) par_time = time.perf_counter() - start print(f"Sekventiel: {seq_time:.2f}s ({len(messages)/seq_time:.1f} meddelelse/s)") print(f"Parallel: {par_time:.2f}s ({len(messages)/par_time:.1f} meddelelse/s)") print(f"Speedup: {seq_time/par_time:.2f}x")

klasse ParallelVerifier: """Parallel signaturbekræftelse for validatorer""" def __init__(selv, max_arbejdere: int = Ingen): self.max_workers = max_workers or os.cpu_count() def verify_batch( selv, elementer: Liste[Tuple[bytes, bytes, bytes]] # (msg, sig, pk) ) -> Liste[bool]: """ Bekræft flere signaturer parallelt Returnerer liste over verifikationsresultater """ def verify_single(punkt: Tuple[bytes, bytes, bytes]) -> bool: besked, signatur, public_key = element prøve: sig = oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt") returnere sig.verify(meddelelse, signatur, offentlig_nøgle) undtagen: returnere falsk med concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as bobestyrer: returnere list(executor.map(verify_single, items)) def alle_gyldige( selv, elementer: Liste[Tuple[bytes, bytes, bytes]] ) -> bool: """Hurtig tjek om alle signaturer er gyldige""" resultater = self.verify_batch(items) returnere alle (resultater) # For validatorer, der behandler blokke: asynkron def validate_block_transactions(transaktioner: Liste): verifikator = ParallelVerifier(max_workers=8) # Forbered verifikationselementer items = [ (tx.signing_message, tx.signature, tx.public_key) for tx in transaktioner ] # Bekræft alt parallelt resultater = verifier.verify_batch(items) # Filtrer gyldige transaktioner valid_txs = [tx for tx, gyldig in zip (transaktioner, resultater) if gyldig] returnere valid_txs

fra funktionsværktøjer importere lru_cache importere hashlib klasse KeyCache: """ Cache-afledte nøgler for at undgå gentagen udledning Nyttig til HD-tegnebøger, hvor de samme stier ofte tilgås. """ def __init__(selv, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} def get_or_derive(selv, master_seed: bytes, sti: str, derive_func) -> Tuple[bytes, bytes]: """ Hent cachenøgle eller udled og cache Args: master_seed: Wallet master-seed-sti: Derivationssti derive_func: Funktion til at kalde, hvis cache mangler Returnerer: (public_key, secret_key) tuple """ # Opret cache-nøgle (gem ikke faktiske frø i nøgle) cache_key = hashlib.Blake2b(master_seed + path.encode()).hexdigest()[:32] if cache_key in self._cache: returnere self._cache[cache_key] # Udled nøgler pk, sk = derive_func(master_seed, path) # Cache med udsættelse if len(self._cache) >= self.max_size: # Simpel FIFO-udsættelse (brug OrderedDict i produktionen) ældste = næste(iter(selv._cache)) del self._cache[ældste] self._cache[cache_key] = (pk, sk) returnere pk, sk def klar(selv): """Ryd alle cachelagrede nøgler (kald på tegnebogslås)""" # Sikker sletning for nøgle in list(self._cache.keys()): pk, sk = self._cache[key] # Overskriv før sletning self._cache[key] = (b'\x00' * len(pk), f'\x00' * len(sk)) del self._cache[nøgle] # Brug i tegnebogen klasse Optimeret tegnebog: def __init__(selv, master_seed: bytes): self.master_seed = master_seed self.key_cache = KeyCache(max_size=500) def få_adressenøgler(selv, sti: str) -> Tuple[bytes, bytes]: returnere self.key_cache.get_or_derive( self.master_seed, path, self._derive_keys ) def _aflede_nøgler(selv, frø: bytes, sti: str): # Faktisk afledningslogik ...

klasse SignatureCache: """ Cache-signaturbekræftelsesresultater Til validatorer for at undgå genbekræftelse af set transaktioner. """ def __init__(selv, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} def _signatur_id( self, message: bytes, signatur: bytes, public_key: bytes ) -> str: """Opret unikt ID til signaturbekræftelse""" returnere hashlib.Blake2b( besked + signatur[:64] + offentlig_nøgle, # De første 64 bytes af sig nok digest_size=16 ).hexdigest() def check_or_verify( self, message: bytes, signatur: bytes, public_key: bytes ) -> bool: """Tjek cache eller bekræft og cache resultat""" sig_id = self._signature_id(meddelelse, signatur, offentlig_nøgle) if sig_id in self._verified: returnere self._verified[sig_id] # Bekræft sig = oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt") is_valid = sig.verify(meddelelse, signatur, offentlig_nøgle) # Cache (med udsættelse) if len(self._verified) >= self.max_size: # Fjern ~10% ældste poster to_remove = list(self._verified.keys())[:self.max_size // 10] for nøgle in at_fjerne: del self._verified[key] self._verified[sig_id] = er_gyldig returnere er_gyldig

importere gc klasse MemoryEfficientSigner: """ Hukommelseseffektiv signering for indlejrede/mobile enheder """ def sign_and_release( selv, besked: bytes, hemmelig_nøgle: bytes ) -> bytes: """ Signer besked og frigiv øjeblikkeligt nøglehukommelse Bruges til engangssignaturer, hvor nøglen ikke bør bestå. """ sig_obj = oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt", hemmelig_nøgle) signatur = sig_obj.sign(besked) # Slip OQS-objekt del sig_obj # Overskriv hemmelig nøgle if isinstance(hemmelig_nøgle, bytearray): for i in range(len(hemmelig_nøgle)): hemmelig_nøgle[i] = 0 # Tving affaldsindsamling gc.collect() returnere signatur def streaming_sign( self, message_chunks: Iterator[bytes], hemmelig_nøgle: bytes ) -> bytes: """ Signer streaming besked uden at indlæse alt i hukommelsen Forhash beskeden i bidder, og underskriv derefter hashen. """ # Hash-meddelelse i bidder hasher = hashlib.Blake2b(digest_size=32) for luns in message_chunks: hasher.update(chunk) message_hash = hasher.digest() # Signer hashen sig = oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt", hemmelig_nøgle) returnere sig.sign(message_hash)

# Tjek CPU-funktionerne for optimal algoritmevalg importere delproces def get_cpu_features() -> sæt: """Opdag tilgængelige CPU SIMD-funktioner""" prøve: # Linux med åben("/proc/cpuinfo") as f: cpuinfo = f.read() features = sæt() if "avx2" in cpuinfo: features.add("avx2") if "avx512" in cpuinfo: features.add("avx512") if "aes" in cpuinfo: features.add("aesni") returnere funktioner undtagen: returnere sæt() def select_optimal_variant() -> str: """Vælg den bedste SPHINCS+-variant til denne CPU""" features = get_cpu_features() if "avx512" in funktioner: # AVX-512 giver ~20-30% speedup trykke("Bruger AVX-512 optimeret implementering") returnere "SPHINCS+-SHAKE-128s-simpelt" # liboqs vælger automatisk elif "avx2" in funktioner: print("Bruger AVX2 optimeret implementering") returnere "SPHINCS+-SHAKE-128s-simpelt" andet: print("Brug af referenceimplementering") returnere "SPHINCS+-SHAKE-128s-simpelt" # Kompiler liboqs med optimale flag # cmake -DOQS_USE_AVX2_INSTRUCTIONS=TIL -DOQS_USE_AVX512_INSTRUCTIONS=TIL ..

importere statistik importere tid klasse Kryptobenchmark: """Omfattende PQC-benchmarking""" def __init__(selv, iterationer: int = 100): self.iterations = iterationer def benchmark_operation(selv, navn: str, operation, opsætning=Ingen ) -> dikt: """Benchmark en enkelt handling""" gange = [] for _ in range(self.iterations): if opsætning: ctx = setup() start = time.perf_counter() if opsætning: operation(ctx) andet: operation() elapsed = (time.perf_counter() - start) * 1000 # ms times.append(forløbet) returnere { "navn": navn, "betyde": statistics.mean(times), "median": statistics.median(times), "stdev": statistics.stdev(times), "min": min(gange), "max": max (gange), "ops_per_sec": 1000 / statistics.mean(time) } def run_full_benchmark(selv) -> dikt: """Kør komplet PQC benchmark suite""" resultater = {} # Kyber benchmarks resultater["kyber_keygen"] = self.benchmark_operation( "Kyber-768 KeyGen", lambda: oqs.KeyEncapsulation("Kyber768").generate_keypair() ) # Opsætning til encap/decap kem = oqs.KeyEncapsulation("Kyber768") pk = kem.generate_keypair() sk = kem.export_secret_key() resultater["kyber_encap"] = self.benchmark_operation( "Kyber-768 Encap", lambda: kem.encap_secret(pk) ) ct, _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation("Kyber768", sk) resultater["kyber_decap"] = self.benchmark_operation( "Kyber-768 Decap", lambda: kem_dec.decap_secret(ct) ) # SPHINCS+ benchmarks resultater["sphincs_keygen"] = self.benchmark_operation( "SPHINCS+-SHAKE-128s KeyGen", lambda: oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt").generate_keypair() ) sig = oqs.Signature("SPHINCS+-SHAKE-128s-simpelt") sig.generate_keypair() spx_sk = sig.export_secret_key() msg ​​= b"x" * 256 resultater["sphincs_sign"] = self.benchmark_operation( "SPHINCS+-SHAKE-128s Sign", lambda: oqs.Signatur("SPHINCS+-SHAKE-128s-simpelt", spx_sk).sign(msg), iterations=20 # Færre på grund af langsom ) returnere resultater def print_resultater(selv, resultater: dikt): """Smukke print benchmark resultater""" trykke("\n=== PQC Performance Benchmark ===") print(f"Gentagelser: {self.iterations}\n") for nøgle, data in results.items(): print(f"{data['navn']}:") print(f"Mean: {data['mean']:.3f}ms") print(f" Median: {data['median']:.3f}ms") print(f"Ops/sek: {data['ops_per_sec']:.1f}") print() # Kør benchmark if __navn__ == "__main__": bænk = Kryptobenchmark(iterations=50) resultater = bench.run_full_benchmark() bench.print_results(results)