Prestandaoptimering för postkvantkryptering: Utvecklarguide

importera concurrent.futures importera oqs från skriver importera Lista, Tuple importera tid klass ParallelSigner: """ Parallell SPHINCS+-signering för batchoperationer Använd när du signerar flera oberoende meddelanden. """ def __init__(self, max_workers: int = Ingen): """ Initiera parallellsignerare Args: max_workers: CPU-trådar att använda (standard: CPU-antal) """ self.max_workers = max_workers or os.cpu_count() def sign_batch( själv, meddelanden: Lista[bytes], hemlig_nyckel: bytes ) -> Lista[bytes]: """ Signera flera meddelanden parallellt Args: meddelanden: Lista över meddelanden att signera secret_key: SPHINCS+ hemlig nyckel Returnerar: Lista över signaturer i samma ordning som meddelanden """ def tecken_singel(meddelande: bytes) -> bytes: sig = oqs.Signature("SPHINCS+-SHAKE-128s-enkel", hemlig_nyckel) återvända sig.sign(meddelande) med concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as executor: signaturer = list(executor.map(sign_single, messages)) återvända signaturer def sign_with_keys( själv, objekt: Lista[Tuple[bytes, bytes]] # (meddelande, hemlig_nyckel) ) -> Lista[bytes]: """Signera meddelanden med olika nycklar parallellt""" def sign_item(punkt: Tuple[bytes, bytes]) -> bytes: meddelande, sk = objekt sig = oqs.Signature("SPHINCS+-SHAKE-128s-enkel", sk) återvända sig.sign(meddelande) med concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as testamentsexekutor: återvända list(executor.map(sign_item, items)) # Benchmark jämförelse def benchmark_parallel_vs_sequential(): # Generera nyckel sig = oqs.Signature("SPHINCS+-SHAKE-128s-enkel") sig.generate_keypair() sk = sig.export_secret_key() # Skapa testmeddelanden meddelanden = [f"Meddelande {i}".koda() för i in range(16)] # Sekventiell start = time.perf_counter() sequential_sigs = [] för medd in meddelanden: s = oqs.Signature("SPHINCS+-SHAKE-128s-enkel", sk) sequential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - start # Parallellt undertecknare = ParallelSigner() start = time.perf_counter() parallell_sigs = signer.sign_batch(messages, sk) par_time = time.perf_counter() - start print(f"Sekventiell: {seq_time:.2f}s ({len(meddelanden)/seq_time:.1f} msg/s)") print(f"Parallell: {par_time:.2f}s ({len(meddelanden)/par_time:.1f} msg/s)") print(f"Speedup: {seq_time/par_time:.2f}x")

klass ParallelVerifier: """Parallell signaturverifiering för validerare""" def __init__(self, max_workers: int = None): self.max_workers = max_workers or os.cpu_count() def verifiera_batch( själv, objekt: Lista[Tuple[bytes, bytes, bytes]] # (msg, sig, pk) ) -> Lista[bool]: """ Verifiera flera signaturer parallellt Returnerar en lista med verifieringsresultat """ def verifiera_singel(punkt: Tuple[bytes, bytes, bytes]) -> bool: meddelande, signatur, public_key = objekt försök: sig = oqs.Signature("SPHINCS+-SHAKE-128s-enkel") återvända sig.verify(meddelande, signatur, public_key) utom: återvända Falsk med concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as testamentsexekutor: återvända list(executor.map(verify_single, items)) def alla_giltiga( själv, objekt: Lista[Tuple[bytes, bytes, bytes]] ) -> bool: """Snabbkontroll om alla signaturer är giltiga""" resultat = self.verify_batch(items) återvända alla (resultat) # För validatorer som bearbetar block: asynkron def validate_block_transactions(transaktioner: Lista): verifier = ParallelVerifier(max_workers=8) # Förbered verifieringsobjekt objekt = [ (tx.signing_message, tx.signature, tx.public_key) för tx in transaktioner ] # Verifiera allt parallellt resultat = verifier.verify_batch(items) # Filtrera giltiga transaktioner valid_txs = [tx för tx, giltigt in zip(transaktioner, resultat) if giltig] återvända valid_txs

från funktionsverktyg importera lru_cache importera hashlib klass KeyCache: """ Cache-härledda nycklar för att undvika upprepad härledning Användbar för HD-plånböcker där samma sökvägar används ofta. """ def __init__(self, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} def get_or_derive( self, master_seed: bytes, path: str, derive_func ) -> Tuple[bytes, bytes]: """ Hämta cachelagrad nyckel eller härled och cachelagra Args: master_seed: Plånbokens huvudfrösökväg: Derivationssökväg derive_func: Funktion att anropa om cachen missar Returnerar: (public_key, secret_key) tuple """ # Skapa cache-nyckel (lagra inte faktiska utsäde i nyckel) cache_key = hashlib.Blake2b(master_seed + path.encode()).hexdigest()[:32] if cache_key in self._cache: återvända self._cache[cache_key] # Härleda nycklar pk, sk = derive_func(master_seed, path) # Cache med vräkning if len(self._cache) >= self.max_size: # Enkel FIFO-vräkning (använd OrderedDict i produktionen) äldst = nästa(iter(self._cache)) del self._cache[äldsta] self._cache[cache_key] = (pk, sk) återvända pk, sk def rensa(själv): """Rensa alla cachade nycklar (ringa på plånbokslås)""" # Säker radering för nyckel in list(self._cache.keys()): pk, sk = self._cache[nyckel] # Skriv över före radering self._cache[nyckel] = (b'\x00' * len(pk), f'\x00' * len(sk)) del self._cache[nyckel] # Användning i plånbok klass Optimerad plånbok: def __init__(self, master_seed: bytes): self.master_seed = master_seed self.key_cache = KeyCache(max_size=500) def get_address_keys(själv, väg: str) -> Tuple[bytes, bytes]: återvända self.key_cache.get_or_derive( self.master_seed, path, self._derive_keys ) def _härleda_nycklar(själv, frö: bytes, sökväg: str): # Faktisk härledningslogik ...

klass SignatureCache: """ Cachesignaturverifieringsresultat För validerare att undvika att återverifiera sedda transaktioner. """ def __init__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} def _signatur_id( själv, meddelande: bytes, signatur: bytes, public_key: bytes ) -> str: """Skapa unikt ID för signaturverifiering""" återvända hashlib.Blake2b( meddelande + signatur[:64] + public_key, # Första 64 byte av sig nog digest_size=16 ).hexdigest() def check_or_verify( själv, meddelande: bytes, signatur: bytes, public_key: bytes ) -> bool: """Kontrollera cache eller verifiera och cacheresultat""" sig_id = self._signature_id(meddelande, signatur, public_key) if sig_id in self._verified: återvända self._verified[sig_id] # Verifiera sig = oqs.Signature("SPHINCS+-SHAKE-128s-enkel") is_valid = sig.verify(meddelande, signatur, public_key) # Cache (med vräkning) if len(self._verified) >= self.max_size: # Ta bort ~10% äldsta poster to_remove = list(self._verified.keys())[:self.max_size // 10] för nyckel in att_ta bort: del self._verified[nyckel] self._verified[sig_id] = är_giltig återvända är_giltig

importera gc klass MemoryEfficientSigner: """ Minneseffektiv signering för inbäddade/mobila enheter """ def sign_and_release( själv, meddelande: bytes, hemlig_nyckel: bytes ) -> bytes: """ Signera meddelande och omedelbart släpp nyckelminne Använd för engångssignaturer där nyckeln inte ska finnas kvar. """ sig_obj = oqs.Signature("SPHINCS+-SHAKE-128s-enkel", hemlig_nyckel) signatur = sig_obj.sign(meddelande) # Släpp OQS-objekt del sig_obj # Skriv över hemlig nyckel if isinstance(hemlig_nyckel, bytearray): för i in range(len(hemlig_nyckel)): hemlig_nyckel[i] = 0 # Forcera sophämtning gc.collect() återvända signatur def streaming_sign( self, message_chunks: Iterator[bytes], secret_key: bytes ) -> bytes: """ Signera strömmande meddelande utan att ladda allt i minnet Förhasha meddelandet i bitar och signera sedan hashen. """ # Hash-meddelande i bitar hasher = hashlib.Blake2b(digest_size=32) för bit in message_chunks: hasher.update(chunk) message_hash = hasher.digest() # Signera hash sig = oqs.Signature("SPHINCS+-SHAKE-128s-enkel", hemlig_nyckel) återvända sig.sign(meddelande_hash)

# Kontrollera CPU-funktionerna för optimalt val av algoritm importera delprocess def get_cpu_features() -> ställ in: """Detektera tillgängliga CPU SIMD-funktioner""" försök: # Linux med öppna("/proc/cpuinfo") as f: cpuinfo = f.read() features = set() if "avx2" in cpuinfo: features.add("avx2") if "avx512" in cpuinfo: features.add("avx512") if "äs" in cpuinfo: features.add("aesni") återvända drag utom: återvända uppsättning() def select_optimal_variant() -> str: """Välj bästa SPHINCS+-varianten för denna CPU""" features = get_cpu_features() if "avx512" in drag: # AVX-512 ger ~20-30% snabbhet skriva ut("Använder AVX-512 optimerad implementering") återvända "SPHINCS+-SHAKE-128s-enkel" # liboqs väljer automatiskt elif "avx2" in funktioner: print("Använder AVX2-optimerad implementering") återvända "SPHINCS+-SHAKE-128s-enkel" annan: print("Använder referensimplementering") återvända "SPHINCS+-SHAKE-128s-enkel" # Kompilera liboqs med optimala flaggor # cmake -DOQS_USE_AVX2_INSTRUCTIONS=PÅ -DOQS_USE_AVX512_INSTRUCTIONS=PÅ ..

importera statistik importera tid klass Kryptobenchmark: """Omfattande PQC benchmarking""" def __init__(jag, iterationer: int = 100): self.iterations = iterationer def benchmark_operation( själv, namn: str, operation, setup=Ingen ) -> dikt: """Jämför en enskild operation""" gånger = [] för _ in range(self.iterations): if setup: ctx = setup() start = time.perf_counter() if setup: operation(ctx) annan: operation() elapsed = (time.perf_counter() - start) * 1000 # ms times.append(förfluten) återvända { "namn": namn, "betyda": statistics.mean(times), "median": statistics.median(times), "stdev": statistics.stdev(times), "min": min( gånger), "max": max( gånger), "ops_per_sec": 1000 / statistics.mean(times) } def run_full_benchmark(själv) -> dikt: """Kör komplett PQC benchmark suite""" resultat = {} # Kyber-riktmärken resultat["kyber_keygen"] = self.benchmark_operation( "Kyber-768 KeyGen", lambda: oqs.KeyEncapsulation("Kyber768").generate_keypair() ) # Inställning för encap/decap kem = oqs.KeyEncapsulation("Kyber768") pk = kem.generate_keypair() sk = kem.export_secret_key() resultat["kyber_encap"] = self.benchmark_operation( "Kyber-768 Encap", lambda: kem.encap_secret(pk) ) ct, _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation("Kyber768", sk) resultat["kyber_decap"] = self.benchmark_operation( "Kyber-768 Decap", lambda: kem_dec.decap_secret(ct) ) # SPHINCS+-riktmärken resultat["sfincs_keygen"] = self.benchmark_operation( "SPHINCS+-SHAKE-128s KeyGen", lambda: oqs.Signatur("SPHINCS+-SHAKE-128s-enkel").generate_keypair() ) sig = oqs.Signature("SPHINCS+-SHAKE-128s-enkel") sig.generate_keypair() spx_sk = sig.export_secret_key() msg ​​= b"x" * 256 resultat["sfincs_sign"] = self.benchmark_operation( "SPHINCS+-SHAKE-128s Sign", lambda: oqs.Signatur("SPHINCS+-SHAKE-128s-enkel", spx_sk).sign(msg), iterations=20 # Färre på grund av långsam ) återvända resultat def print_results(själv, resultat: dikt): """Snygga utskriftsresultat för benchmark""" skriva ut("\n=== PQC Performance Benchmark ===") print(f"Iterationer: {self.iterations}\n") för nyckel, data in results.items(): print(f"{data['name']}:") 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 __namn__ == "__huvud__": bänk = Kryptobenchmark(iterations=50) resultat = bench.run_full_benchmark() bench.print_results(results)