Kuantum Sonrası Kriptografi için Performans Optimizasyonu: Geliştirici Kılavuzu

içe aktarmak eşzamanlı.vadeli işlemler içe aktarmak tamam itibaren yazarak içe aktarmak Liste, Tuple içe aktarmak zaman sınıf Paralelİmzalayan: """ Toplu işlemler için paralel SPHINCS+ imzalama Birden fazla bağımsız mesajı imzalarken kullanın. """ def __init__(self, max_workers: int = Yok): """ Paralel imzalayan Args'ı başlat: max_workers: Kullanılacak CPU iş parçacıkları (varsayılan: CPU sayısı) """ self.max_workers = max_workers or os.cpu_count() def işaret_batch( self, mesajlar: Liste[bayt], gizli_anahtar: bayt) -> Liste[bayt]: """ Birden fazla mesajı paralel olarak imzala Args: mesajlar: İmzalanacak mesajların listesi secret_key: SPHINCS+ gizli anahtarı Döndürür: İmzaların listesi """ mesajlarıyla aynı sıradadır def Sign_single(mesaj: bayt) -> bayt: sig = oqs.Signature("SPHINCS+-SHAKE-128s-basit", gizli_anahtar) geri dönmek sig.sign(mesaj) ile concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as yürütücü: imzalar = liste(executor.map(sign_single, mesajlar)) geri dönmek imzalar def Sign_with_keys( öz, öğeler: Liste[Grup[bayt, bayt]] # (mesaj, gizli_anahtar) ) -> Liste[bayt]: """Mesajları farklı tuşlarla paralel olarak imzala""" def işaret_item(öğe: Grup[bayt, bayt]) -> bayt: mesaj, sk = öğe sig = oqs.Signature("SPHINCS+-SHAKE-128s-basit", Sk) geri dönmek sig.sign(mesaj) ile concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as icracı: geri dönmek list(executor.map(sign_item, items)) # Karşılaştırma karşılaştırması def Benchmark_parallel_vs_sequential(): # Anahtar oluştur sig = oqs.Signature("SPHINCS+-SHAKE-128s-basit") sig.generate_keypair() sk = sig.export_secret_key() # Test mesajları oluşturun mesajlar = [f"Mesaj {i}".encode() için i in aralık(16)] # Sıralı başlangıç ​​= time.perf_counter() sequential_sigs = [] için mesaj in mesajlar: s = oqs.Signature("SPHINCS+-SHAKE-128s-basit", sk) sequential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - başlangıç # Paralel imzalayan = Paralelİmzalayan() start = time.perf_counter() paralel_sigs =signer.sign_batch(mesajlar, sk) par_time = time.perf_counter() - print(f'yi başlat)"Sıralı: {seq_time:.2f}s ({len(messages)/seq_time:.1f} msg/s)") yazdır(f"Paralel: {par_time:.2f}s ({len(messages)/par_time:.1f} msg/s)") yazdır(f"Hızlanma: {seq_time/par_time:.2f}x")

sınıf Paralel Doğrulayıcı: """Doğrulayıcılar için paralel imza doğrulaması""" def __init__(self, max_workers: int = Yok): self.max_workers = max_workers or os.cpu_count() def doğrulama_batch( öz, öğeler: Liste[Grup[bayt, bayt, bayt]] # (mesaj, imza, pk) ) -> Liste[bool]: """ Birden fazla imzayı paralel olarak doğrula Doğrulama sonuçlarının listesini döndürür """ def doğrulama_single(öğe: Grup[bayt, bayt, bayt]) -> bool: mesaj, imza, public_key = öğe denemek: sig = oqs.Signature("SPHINCS+-SHAKE-128s-basit") geri dönmek sig.verify(mesaj, imza, genel_anahtar) hariç: geri dönmek YANLIŞ ile concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as icracı: geri dönmek list(executor.map(verify_single, items)) def hepsi_geçerli( öz, öğeler: Liste[Grup[bayt, bayt, bayt]] ) -> bool: """Tüm imzaların geçerli olup olmadığını hızlı kontrol edin""" sonuçlar = self.verify_batch(items) geri dönmek hepsi(sonuçlar) # Blokları işleyen doğrulayıcılar için: eşzamansız tanım validate_block_transactions(işlemler: Liste): doğrulayıcı = Paralel Doğrulayıcı(max_workers=8) # Doğrulama öğelerini hazırlayın öğeler = [ (tx.signing_message, tx.signature, tx.public_key) için tx in işlemler ] # Hepsini paralel olarak doğrulayın sonuçlar = verifier.verify_batch(items) # Geçerli işlemleri filtrele geçerli_txs = [tx için tx, geçerli in zip(işlemler, sonuçlar) if geçerli] geri dönmek geçerli_txs

itibaren işlev araçları içe aktarmak lru_cache içe aktarmak hashlib sınıf Anahtar Önbellek: """ Tekrarlanan türetmeyi önlemek için türetilmiş anahtarları önbelleğe alın. Aynı yollara sıklıkla erişilen HD cüzdanlar için kullanışlıdır. """ def __init__(self, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} def get_or_derive( self, master_seed: bayt, yol: str, derive_func) -> Grup[bayt, bayt]: """ Önbelleğe alınmış anahtarı alın veya türetin ve önbelleğe alın Args: master_seed: Cüzdan ana tohum yolu: Türetme yolu derive_func: Önbellek kaçırılırsa çağrılacak işlev Döndürür: (public_key, secret_key) tuple """ # Önbellek anahtarı oluşturun (gerçek tohumu anahtarda saklamayın) önbellek_anahtarı = hashlib.Blake2b(master_seed + path.encode()).hexdigest()[:32] if önbellek_anahtarı in self._cache: geri dönmek self._cache[cache_key] # Anahtarları türet pk, sk = derive_func(master_seed, yol) # Tahliye ile önbellek if len(self._cache) >= self.max_size: # Basit FIFO tahliyesi (üretimde OrderedDict'i kullanın) en eski = sonraki(iter(self._cache)) del self._cache[en eski] self._cache[cache_key] = (pk, sk) geri dönmek pk, sk def temizlemek(kendisi): """Önbelleğe alınmış tüm anahtarları temizle (cüzdan kilidinde çağrı)""" # Güvenli silme için anahtar in list(self._cache.keys()): pk, sk = self._cache[anahtar] # Silmeden önce üzerine yaz self._cache[anahtar] = (b'\x00' * len(pk), b'\x00' *len(sk)) del self._cache[anahtar] # Cüzdanda kullanım sınıf Optimize Edilmiş Cüzdan: def __init__(self, master_seed: bayt): self.master_seed = master_seed self.key_cache = Anahtar Önbellek(max_size=500) def get_address_keys(kendisi, yol: str) -> Grup[bayt, bayt]: geri dönmek self.key_cache.get_or_derive( self.master_seed, yol, self._derive_keys) def _derive_keys(self, tohum: bayt, yol: str): # Gerçek türetme mantığı ...

sınıf İmza Önbelleği: """ Önbellek imza doğrulama sonuçları Doğrulayıcıların görülen işlemleri yeniden doğrulamaktan kaçınmaları için. """ def __init__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} def _signature_id( self, mesaj: bytes, imza: bytes, public_key: bytes) -> str: """İmza doğrulaması için benzersiz kimlik oluşturun""" geri dönmek hashlib.Blake2b( mesaj + imza[:64] + public_key, # İlk 64 baytlık imza yeterli sindirim_boyutu=16 ).hexdigest() def check_or_verify( self, mesaj: bayt, imza: bayt, genel_anahtar: bayt) -> bool: """Önbelleği kontrol edin veya sonucu doğrulayın ve önbelleğe alın""" sig_id = self._signature_id(mesaj, imza, genel_anahtar) if sig_id in self._doğrulandı: geri dönmek self._verified[sig_id] # Doğrulamak sig = oqs.Signature("SPHINCS+-SHAKE-128s-basit") is_valid = sig.verify(mesaj, imza, genel_anahtar) # Önbellek (tahliye ile) if len(self._verified) >= self.max_size: # En eski girişlerin ~%10'unu kaldırın to_remove = liste(self._verified.keys())[:self.max_size // 10] için anahtar in to_remove: del self._verified[anahtar] self._verified[sig_id] = is_valid geri dönmek geçerli

içe aktarmak gc sınıf MemoryEfficientSigner: """ Gömülü/mobil cihazlar için bellek açısından verimli imzalama """ def Sign_and_release( self, mesaj: bayt, gizli_anahtar: bayt) -> bayt: """ Mesajı imzalayın ve anahtar belleğini hemen serbest bırakın Anahtarın kalmaması gereken tek seferlik imzalar için kullanın. """ sig_obj = oqs.Signature("SPHINCS+-SHAKE-128s-basit", gizli_anahtar) imza = sig_obj.sign(mesaj) # OQS nesnesini serbest bırakın del sig_obj # Gizli anahtarın üzerine yaz if örnek(gizli_anahtar, bytearray): için i in aralık(uzunluk(gizli_anahtar)): gizli_anahtar[i] = 0 # Çöp toplamayı zorla gc.collect() geri dönmek imza def akış_işareti( self, message_chunks: Iterator[bytes], secret_key: bytes) -> bytes: """ Tümünü belleğe yüklemeden akış mesajını imzalayın Mesaja parçalar halinde ön karma işlemi uygulayın, ardından karma değerini imzalayın. """ # Parçalar halinde karma mesaj karma = hashlib.Blake2b(digest_size=32) için yığın in message_chunks: hasher.update(chunk) message_hash = hasher.digest() # Karmayı imzala sig = oqs.Signature("SPHINCS+-SHAKE-128s-basit", gizli_anahtar) geri dönmek sig.sign(message_hash)

# Optimum algoritma seçimi için CPU özelliklerini kontrol edin içe aktarmak alt süreç def get_cpu_features() -> ayarla: """Mevcut CPU SIMD özelliklerini tespit et""" denemek: #Linux ile açık("/proc/cpuinfo") as f: cpuinfo = f.read() özellikleri = set() if "avx2" in işlemci bilgisi: özellikler.add("avx2") if "avx512" in işlemci bilgisi: özellikler.add("avx512") if "evet" in işlemci bilgisi: özellikler.add("esni") geri dönmek özellikler hariç: geri dönmek ayarlamak() def select_optimal_variant() -> dizi: """Bu CPU için en iyi SPHINCS+ modelini seçin""" özellikler = get_cpu_features() if "avx512" in özellikler: # AVX-512 ~%20-30 hızlanma sağlar yazdır("AVX-512 optimize edilmiş uygulamasını kullanma") geri dönmek "SPHINCS+-SHAKE-128s-basit" # liboqs otomatik seçer elif "avx2" in özellikler: yazdır("AVX2 optimize edilmiş uygulamasını kullanma") geri dönmek "SPHINCS+-SHAKE-128s-basit" başka: yazdır("Referans uygulamasını kullanma") geri dönmek "SPHINCS+-SHAKE-128s-basit" # Liboq'ları en uygun bayraklarla derleyin # cmake -DOQS_USE_AVX2_INSTRUCTIONS=AÇIK -DOQS_USE_AVX512_INSTRUCTIONS=AÇIK ..

içe aktarmak istatistikler içe aktarmak zaman sınıf KriptoBenchmark: """Kapsamlı PQC kıyaslaması""" def __init__(self, yinelemeler: int = 100): self. yinelemeler = yinelemeler def kıyaslama_işlemi( self, name: str, operasyon, setup=Hiçbiri ) -> dikte: """Tek bir işlemi kıyaslayın""" kez = [] için _ in aralık(kendi. yinelemeler): if kurulum: ctx = setup() başlangıç ​​= time.perf_counter() if kurulum: işlem(ctx) başka: işlem() geçen = (time.perf_counter() - başlangıç) * 1000 # ms times.append(geçen) geri dönmek { "isim": isim, "Anlam": istatistik.mean(kez), "medyan": istatistik.medyan(kez), "stdev": istatistik.stdev(kez), "dakika": dk(kez), "maks": maksimum(kez), "ops_per_sec": 1000 / istatistik.ortalama(kez) } def run_full_benchmark(kendisi) -> dikte: """PQC karşılaştırma paketinin tamamını çalıştırın""" sonuçlar = {} # Kyber kıyaslamaları sonuçlar["kyber_keygen"] = self.benchmark_operation( "Kyber-768 KeyGen", lambda: oqs.KeyEncapsulation("Kyber768").generate_keypair() ) # Encap/decap kurulumu kem = oqs.KeyEncapsulation("Kyber768") pk = kem.generate_keypair() sk = kem.export_secret_key() sonuçlar["kyber_encap"] = self.benchmark_operation( "Kyber-768 Kapama", lambda: kem.encap_secret(pk) ) ct, _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation("Kyber768", sk) sonuçlar["kyber_decap"] = self.benchmark_operation( "Kyber-768 Dekapajı", lambda: kem_dec.decap_secret(ct)) # SPHINCS+ kıyaslamaları sonuçlar["sfinkler_keygen"] = self.benchmark_operation( "SPHINCS+-SHAKE-128s KeyGen", lambda: oqs.Signature("SPHINCS+-SHAKE-128s-basit").generate_keypair() ) sig = oqs.Signature("SPHINCS+-SHAKE-128s-basit") sig.generate_keypair() spx_sk = sig.export_secret_key() msg ​​= b"x" * 256 sonuç["sfink_işareti"] = self.benchmark_operation( "SPHINCS+-SHAKE-128s İşareti", lambda: oqs.Signature("SPHINCS+-SHAKE-128s-basit", spx_sk).sign(msg), yinelemeler=20 # Yavaşlama nedeniyle daha az ) geri dönmek sonuçlar def baskı_sonuçları(kendisi, sonuçlar: dict): """Oldukça baskı kıyaslama sonuçları""" yazdır("\n=== PQC Performans Karşılaştırması ===") yazdır(f"Yinelemeler: {self. yinelemeler}\n") için anahtar, veri in sonuçlar.items(): print(f"{veri['isim']}:") yazdır(f" Anlamı: {veri['mean']:.3f}ms") yazdır(f" Medyan: {veri['medyan']:.3f}ms") yazdır(f" İşlem/sn: {data['ops_per_sec']:.1f}") yazdır() # Karşılaştırmayı çalıştır if __isim__ == "__ana__": bank = KriptoBenchmark(yinelemeler=50) sonuçlar = bench.run_full_benchmark() bench.print_results(results)