تحسين الأداء لتشفير ما بعد الكم: دليل المطور

يستورد المتزامنة. العقود الآجلة يستورد oqs من الكتابة يستورد القائمة، توبل يستورد وقت فصل ParallelSigner: """ التوقيع المتوازي SPHINCS+ للعمليات المجمعة يُستخدم عند توقيع رسائل مستقلة متعددة. """ مواطنه __الحرف الأولي__(الذات، max_workers: int = لا شيء): """ تهيئة وسيطات المُوقع المتوازي: max_workers: سلاسل CPU المراد استخدامها (الافتراضي: عدد CPU) """ self.max_workers = max_workers or os.cpu_count() مواطنه Sign_batch(الذات، الرسائل: قائمة [بايت]، مفتاح_سري: بايت) -> قائمة [بايت]: """ تسجيل رسائل متعددة بالتوازي الوسائط: الرسائل: قائمة الرسائل المراد التوقيع عليها Secret_key: المفتاح السري SPHINCS+ الإرجاعات: قائمة التوقيعات بنفس ترتيب الرسائل """ مواطنه Sign_single(الرسالة: بايت) -> بايت: sig = oqs.Signature("SPHINCS+-SHAKE-128s-بسيط"، سر_المفتاح) يعود علامة (رسالة) مع concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as المنفذ: التوقيعات = قائمة (executor.map (sign_single، messages)) يعود التوقيعات مواطنه Sign_with_keys(الذات، العناصر: قائمة[مترابطة بيانية[بايت، بايت]] # (رسالة، مفتاح_سري) ) -> قائمة [بايت]: """توقيع الرسائل بمفاتيح مختلفة بالتوازي""" مواطنه Sign_item(غرض: مترابطة بيانية[بايت، بايت]) -> بايت: رسالة، sk = العنصر sig = oqs.Signature("SPHINCS+-SHAKE-128s-بسيط"، كورونا) يعود علامة (رسالة) مع concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as المنفذ: يعود قائمة (executor.map (sign_item، items)) # المقارنة المعيارية مواطنه Benchmark_parallel_vs_sequential(): # توليد المفتاح سيج = oqs. التوقيع ("SPHINCS+-SHAKE-128s-بسيط") sig.generate_keypair() sk = sig.export_secret_key() # إنشاء رسائل اختبارية الرسائل = [و"الرسالة {i}".تشفير() ل i in النطاق (16)] # متسلسل البدء = time.perf_counter() sequential_sigs = [] ل رسالة in الرسائل: s = oqs.Signature("SPHINCS+-SHAKE-128s-بسيط"، sk) sequential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - ابدأ # موازي الموقع = ParallelSigner() start = time.perf_counter() موازي_sigs =signer.sign_batch(messages, sk) par_time = time.perf_counter() - بدء الطباعة (f)"تسلسلي: {seq_time:.2f}s ({len(messages)/seq_time:.1f} msg/s)") طباعة (ف"بالتوازي: {par_time:.2f}s ({len(messages)/par_time:.1f} msg/s)") طباعة (ف"التسريع: {seq_time/par_time:.2f}x")

فصل ParallelVerifier: """التحقق من التوقيع المتوازي للمدققين""" مواطنه __الحرف الأولي__(self, max_workers: int = لا شيء): self.max_workers = max_workers or os.cpu_count() مواطنه Verified_batch(الذات، العناصر: قائمة[مترابطة بيانية[بايت، بايت، بايت]] # (رسالة، سيج، بك) ) -> القائمة[bool]: """ التحقق من التوقيعات المتعددة بالتوازي إرجاع قائمة نتائج التحقق """ مواطنه Verified_single(غرض: مترابطة بيانية[بايت، بايت، بايت]) -> منطقي: رسالة، توقيع، public_key = العنصر يحاول: سيج = oqs.Signature("SPHINCS+-SHAKE-128s-بسيط") يعود sig.verify (رسالة، توقيع، مفتاح_عام) يستثني: يعود خطأ شنيع مع concurrent.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as المنفذ: يعود قائمة (executor.map (verify_single، items)) مواطنه all_valid(الذات، العناصر: قائمة[مترابطة بيانية[بايت، بايت، بايت]] ) -> منطقي: """التحقق السريع من صحة كافة التوقيعات""" النتائج = self.verify_batch (العناصر) يعود الكل (النتائج) # بالنسبة لكتل ​​معالجة المدققين: تعريف غير متزامن validate_block_transactions(المعاملات: قائمة): المدقق = ParallelVerifier(max_workers=8) # تحضير عناصر التحقق العناصر = [(tx.signing_message، tx.signature، tx.public_key) ل tx in المعاملات ] # التحقق من كل شيء بالتوازي النتائج = verifier.verify_batch(العناصر) # تصفية المعاملات الصحيحة valid_txs = [tx ل تكساس، صالح in الرمز البريدي (المعاملات والنتائج) if صالح] يعود valid_txs

من functools يستورد lru_cache يستورد هاشليب فصل KeyCache: """ المفاتيح المشتقة من ذاكرة التخزين المؤقت لتجنب الاشتقاق المتكرر مفيدة لمحافظ HD حيث يتم الوصول إلى نفس المسارات بشكل متكرر. """ مواطنه __الحرف الأولي__(الذات، max_size: int = 1000): self.max_size = max_size self._cache: dict = {} مواطنه get_or_derive(الذات، master_seed: بايت، المسار: str، derive_func) -> مترابطة بيانية[بايت، بايت]: """ الحصول على مفتاح مخبأ أو اشتقاق وذاكرة تخزين مؤقت Args: master_seed: المسار الأساسي للمحفظة الرئيسية: مسار الاشتقاق derive_func: وظيفة للاتصال في حالة تفويت ذاكرة التخزين المؤقت الإرجاعات: (مفتاح_عام، مفتاح_سري) صف """ # إنشاء مفتاح ذاكرة التخزين المؤقت (لا تقم بتخزين البذور الفعلية في المفتاح) Cache_key = hashlib.Blake2b(master_seed + path.encode()).hexdigest()[:32] if Cache_key in self._cache: يعود self._cache[cache_key] #اشتقاق المفاتيح pk، sk = derive_func(master_seed، المسار) # مخبأ مع الإخلاء if لين (self._cache) >= self.max_size: # إخلاء بسيط من FIFO (استخدم OrderedDict في الإنتاج) الأقدم = التالي(iter(self._cache)) ديل self._cache[الأقدم] self._cache[cache_key] = (pk, sk) يعود بك، كورونا مواطنه واضح(النفس): """مسح كافة المفاتيح المخزنة مؤقتًا (استدعاء قفل المحفظة)""" # المحو الآمن ل مفتاح in القائمة (self._cache.keys()): pk، sk = self._cache[key] # الكتابة قبل الحذف self._cache[مفتاح] = (ب'\x00' * لين (بك)، ب'\x00' * لين (سك)) ديل self._cache[مفتاح] # الاستخدام في المحفظة فصل المحفظة المحسنة: مواطنه __الحرف الأولي__(الذات، master_seed: بايت): self.master_seed = master_seed self.key_cache = KeyCache(الحجم الأقصى=500) مواطنه get_address_keys(الذات، المسار: شارع) -> مترابطة بيانية[بايت، بايت]: يعود self.key_cache.get_or_derive (self.master_seed، المسار، self._derive_keys) مواطنه _derive_keys(الذات، البذور: بايت، المسار: شارع): # منطق الاشتقاق الفعلي ...

فصل SignatureCache: """ تخزين نتائج التحقق من التوقيع مؤقتًا للمدققين لتجنب إعادة التحقق من المعاملات التي تمت مشاهدتها. """ مواطنه __الحرف الأولي__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} مواطنه _signature_id(ذاتي، الرسالة: بايت، التوقيع: بايت، مفتاح_عام: بايت) -> شارع: """إنشاء معرف فريد للتحقق من صحة التوقيع""" يعود hashlib.Blake2b( الرسالة + التوقيع[:64] + public_key، # أول 64 بايت من sig كافية Digest_size=16).hexdigest() مواطنه check_or_verify(ذاتي، الرسالة: بايت، التوقيع: بايت، مفتاح_عام: بايت) -> منطقي: """التحقق من ذاكرة التخزين المؤقت أو التحقق من نتيجة ذاكرة التخزين المؤقت""" sig_id = self._signature_id (رسالة، توقيع، مفتاح_عام) if sig_id in ذاتي._تم التحقق: يعود self._verified[sig_id] # يؤكد سيج = oqs. التوقيع ("SPHINCS+-SHAKE-128s-بسيط") is_valid = sig.verify(رسالة، توقيع، مفتاح_عام) # ذاكرة التخزين المؤقت (مع الإخلاء) if لين (self._verified) >= self.max_size: # إزالة ~ 10% من الإدخالات الأقدم to_remove = list(self._verified.keys())[:self.max_size // 10] ل مفتاح in to_remove: ديل self._verified[key] self._verified[sig_id] = is_valid يعود is_valid

يستورد gc فصل MemoryEfficientSigner: """ التوقيع الموفر للذاكرة للأجهزة المدمجة/المحمولة """ مواطنه Sign_and_release(الذات، الرسالة: بايت، مفتاح_المفتاح السري: بايت) -> بايت: """ قم بتسجيل الرسالة وتحرير ذاكرة المفتاح على الفور استخدم التوقيعات لمرة واحدة حيث لا ينبغي أن يستمر المفتاح. """ sig_obj = oqs.Signature("SPHINCS+-SHAKE-128s-بسيط"، Secret_key) التوقيع = sig_obj.sign(message) # الافراج عن كائن OQS ديل sig_obj # الكتابة فوق المفتاح السري if isinstance(secret_key, bytearray): ل i in النطاق (لين (مفتاح_سري)): مفتاح_سري[i] = 0 # قوة جمع القمامة جي سي. جمع () يعود إمضاء مواطنه Streaming_sign( self، message_chunks: Iterator[bytes]، Secret_key: bytes ) -> bytes: """ تسجيل بث الرسالة دون تحميل الكل في الذاكرة قم بتجزئة الرسالة مسبقًا إلى أجزاء، ثم قم بالتوقيع على التجزئة. """ # رسالة التجزئة في أجزاء تجزئة = hashlib.Blake2b(digest_size=32) ل قطعة in message_chunks: hasher.update(chunk) message_hash = hasher.digest() # التوقيع على التجزئة سيج = oqs. التوقيع ("SPHINCS+-SHAKE-128s-بسيط"، سر_المفتاح) يعود sig.sign(message_hash)

# تحقق من ميزات CPU لاختيار الخوارزمية الأمثل يستورد عملية فرعية مواطنه get_cpu_features() -> مجموعة: """الكشف عن ميزات CPU SIMD المتوفرة""" يحاول: # لينكس مع يفتح("/بروك/كبوينفو") as f: cpuinfo = ميزات f.read() = set() if "افكس2" in cpuinfo:ميزات.إضافة("افكس2") if "افكس512" in cpuinfo:ميزات.إضافة("افكس512") if "ايه اس" in cpuinfo:ميزات.إضافة("ايسني") يعود سمات يستثني: يعود تعيين() مواطنه Select_optimal_variant() -> شارع: """تحديد أفضل متغير SPHINCS+ لهذا CPU""" الميزات = get_cpu_features () if "افكس512" in سمات: # يوفر AVX-512 سرعة تصل إلى 20-30% تقريبًا مطبعة("استخدام التنفيذ الأمثل لـ AVX-512") يعود "SPHINCS+-SHAKE-128s-بسيط" # liboqs يتم تحديده تلقائيًا إليف "افكس2" in الميزات: طباعة("استخدام التنفيذ الأمثل لـ AVX2") يعود "SPHINCS+-SHAKE-128s-بسيط" آخر: مطبعة("باستخدام التنفيذ المرجعي") يعود "SPHINCS+-SHAKE-128s-بسيط" # تجميع liboqs بالأعلام المثالية # cmake -DOQS_USE_AVX2_INSTRUCTIONS=ON -DOQS_USE_AVX512_INSTRUCTIONS=ON ..

يستورد إحصائيات يستورد وقت فصل CryptoBenchmark: """قياس الأداء الشامل لجودة الجودة""" مواطنه __الحرف الأولي__(الذات، التكرارات: int = 100): التكرارات الذاتية = التكرارات مواطنه Benchmark_Operation(الذات، الاسم: str، العملية، الإعداد =لا أحد ) -> الإملاء: """قياس أداء عملية واحدة""" مرات = [] ل _ in النطاق (التكرارات الذاتية): if الإعداد: ctx = setup() start = time.perf_counter() if الإعداد: العملية (ctx) آخر: العملية () المنقضية = (time.perf_counter () - البداية) * 1000 # آنسة مرات.إلحاق (المنقضية) يعود { "اسم": اسم، "يقصد": الإحصائيات.يعني (مرات)، "متوسط": الإحصاءات.الوسيط (مرات)، "ستديف": الإحصائيات.stdev(مرات)، "دقيقة": دقيقة (مرات)، "الأعلى": الحد الأقصى (مرات)، "ops_per_sec": 1000 / متوسط ​​الإحصائيات (مرات) } مواطنه run_full_benchmark(الذات) -> الإملاء: """تشغيل مجموعة معايير PQC الكاملة""" النتائج = {} # معايير Kyber نتائج["kyber_keygen"] = self.benchmark_operation( "Kyber-768 كيجن", لامدا: oqs.KeyEncapsulation("كايبر 768").generate_keypair() ) # إعداد encap/decap كيم = oqs.KeyEncapsulation("كايبر 768") pk = kem.generate_keypair() sk = kem.export_secret_key() النتائج["kyber_encap"] = self.benchmark_operation( "Kyber-768 إنكاب", لامدا: kem.encap_secret(pk)) ط م، _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation("كايبر 768"، كورونا) النتائج["kyber_decap"] = self.benchmark_operation( "Kyber-768 ديكاب", لامدا: kem_dec.decap_secret(ct)) # معايير SPHINCS+ نتائج["sphincs_keygen"] = self.benchmark_operation( "SPHINCS+-SHAKE-128s KeyGen", لامدا: oqs. التوقيع ("SPHINCS+-SHAKE-128s-بسيط".generate_keypair() ) sig = oqs.Signature("SPHINCS+-SHAKE-128s-بسيط") sig.generate_keypair() spx_sk = sig.export_secret_key() msg ​​= b"x" * 256 نتيجة["shincs_sign"] = self.benchmark_operation( "علامة SPHINCS+-SHAKE-128s", لامدا: oqs. التوقيع ("SPHINCS+-SHAKE-128s-بسيط"، spx_sk).sign(msg)، التكرارات=20 # أقل بسبب البطء ) يعود نتائج مواطنه print_results(الذات، النتائج: إملاء): """نتائج قياس الأداء رائعة""" مطبعة("\n=== قياس أداء PQC ===") طباعة (ف"التكرارات: {self.iterations}\n") ل المفتاح، البيانات in results.items(): طباعة(f"{بيانات['اسم']}:") طباعة (ف" يعني: {data['mean']:.3f}ms") طباعة (ف"الوسيط: {data['median']:.3f}ms") طباعة (ف" العمليات/الثانية: {data['ops_per_sec']:.1f}") مطبعة() # تشغيل المعيار if __الاسم__ == "__رئيسي__": مقاعد البدلاء = CryptoBenchmark(التكرارات = 50) النتائج = bench.run_full_benchmark() bench.print_results(results)