포스트퀀텀 암호화를 위한 성능 최적화: 개발자 가이드

수입 동시 선물 수입 오크스 ~에서 타자 수입 리스트, 튜플 수입 시간 수업 병렬 서명자: """ 일괄 작업을 위한 병렬 SPHINCS+ 서명 여러 개의 독립적인 메시지에 서명할 때 사용합니다. """ 데프 __초기화__(self, max_workers: int = 없음): """ 병렬 서명자 초기화 Args: 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-간단", 비밀_키) 반품 sig.sign(메시지) ~와 함께 동시.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as 실행자: 서명 = 목록(executor.map(sign_single, message)) 반품 서명 데프 sign_with_keys( 자기, 항목: 목록[튜플[바이트, 바이트]] # (메시지, secret_key) ) -> 목록[바이트]: """병렬로 서로 다른 키를 사용하여 메시지에 서명""" 데프 sign_item(목: 튜플[바이트, 바이트]) -> 바이트: 메시지, sk = 항목 sig = oqs.Signature("SPHINCS+-SHAKE-128s-간단", SK) 반품 sig.sign(메시지) ~와 함께 동시.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as 집행자: 반품 목록(executor.map(sign_item, 항목)) # 벤치마크 비교 데프 벤치마크_병렬_vs_순차(): # 키 생성 서명 = oqs.서명("SPHINCS+-SHAKE-128s-간단") sig.generate_keypair() sk = sig.export_secret_key() # 테스트 메시지 생성 메시지 = [f"메시지 {i}".인코드() ~을 위한 i in 범위(16)] # 순차적 시작 = time.perf_counter() ential_sigs = [] ~을 위한 메시지 in 메시지: s = oqs.Signature("SPHINCS+-SHAKE-128s-간단", sk) ential_sigs.append(s.sign(msg)) seq_time = time.perf_counter() - 시작 # 평행한 서명자 = 병렬 서명자() 시작 = time.perf_counter() parallel_sigs = signer.sign_batch(messages, sk) par_time = time.perf_counter() - 인쇄 시작(f"순차: {seq_time:.2f}s ({len(messages)/seq_time:.1f} msg/s)") 인쇄(f"병렬: {par_time:.2f}s ({len(messages)/par_time:.1f} msg/s)") 인쇄(f"속도 향상: {seq_time/par_time:.2f}x")

수업 병렬 검증기: """검증인을 위한 병렬 서명 확인""" 데프 __초기화__(self, max_workers: int = 없음): self.max_workers = max_workers or os.cpu_count() 데프 verify_batch( 자기, 항목: 목록[튜플[바이트, 바이트, 바이트]] # (메시지, 서명, pk) ) -> 목록[bool]: """ 여러 서명을 동시에 검증합니다. 검증 결과 목록을 반환합니다. """ 데프 verify_single(목: 튜플[바이트, 바이트, 바이트]) -> bool: 메시지, 서명, public_key = 항목 노력하다: 시그 = oqs.서명("SPHINCS+-SHAKE-128s-간단") 반품 sig.verify(메시지, 서명, 공개_키) 제외하고: 반품 거짓 ~와 함께 동시.futures.ThreadPoolExecutor( max_workers=self.max_workers ) as 집행자: 반품 목록(executor.map(verify_single, 항목)) 데프 모두_유효하다( 자기, 항목: 목록[튜플[바이트, 바이트, 바이트]] ) -> 부울: """모든 서명이 유효한지 빠르게 확인하세요""" 결과 = self.verify_batch(항목) 반품 모두(결과) # 블록을 처리하는 검증자의 경우: 비동기 정의 verify_block_transactions(거래: 목록): 검증자 = 병렬 검증기(최대_작업자=8) # 검증항목 준비 항목 = [(tx.signing_message, tx.signature, tx.public_key) ~을 위한 tx in 거래 ] # 모두 병렬로 검증 결과 = verifier.verify_batch(항목) # 유효한 거래 필터링 valid_txs = [tx ~을 위한 텍사스, 유효 in zip(거래, 결과) if 유효한] 반품 valid_txs

~에서 기능 도구 수입 lru_cache 수입 해시립 수업 키캐시: """ 파생 키를 캐시하여 반복 파생을 방지합니다. 동일한 경로에 자주 액세스하는 HD 지갑에 유용합니다. """ 데프 __초기화__(self, max_size: int = 1000): self.max_size = max_size self._cache: dict = {} 데프 get_or_derive( self, master_seed: 바이트, 경로: str, 파생_func ) -> 튜플[바이트, 바이트]: """ 캐시된 키 가져오기 또는 파생 및 캐시 인수: master_seed: 지갑 마스터 시드 경로: 파생 경로 파생_func: 캐시 누락 시 호출할 함수 반환: (public_key, secret_key) 튜플 """ # 캐시 키 생성(키에 실제 시드를 저장하지 않음) 캐시_키 = hashlib.Blake2b(master_seed + path.encode()).hexdigest()[:32] if 캐시_키 in self._cache: 반품 self._cache[캐시_키] # 키 파생 pk, sk = 파생_기능(마스터_시드, 경로) # 퇴거와 함께 캐시 if len(self._cache) >= self.max_size: # 단순 FIFO 제거(프로덕션에서는 OrderedDict 사용) 가장 오래된 = 다음(iter(self._cache)) 델 self._cache[오래된 것] self._cache[cache_key] = (pk, sk) 반품 PK, SK 데프 분명한(본인): """캐시된 키 모두 지우기(지갑 잠금 시 호출)""" # 안전한 삭제 ~을 위한 열쇠 in list(self._cache.keys()): pk, sk = self._cache[key] # 삭제하기 전에 덮어쓰기 self._cache[키] = (b'\x00' * len(pk),b'\x00' * len(sk)) 델 self._cache[키] # 지갑에서의 사용 수업 최적화된 지갑: 데프 __초기화__(self, master_seed:바이트): self.master_seed = master_seed self.key_cache = 키캐시(최대_크기=500) 데프 get_address_keys(자신, 경로: str) -> 튜플[바이트, 바이트]: 반품 self.key_cache.get_or_derive( self.master_seed, 경로, self._derive_keys ) 데프 _derive_keys(자체, 시드: 바이트, 경로: str): # 실제 파생 논리 ...

수업 서명캐시: """ 캐시 서명 확인 결과 유효성 검사기가 본 트랜잭션을 다시 확인하는 것을 방지합니다. """ 데프 __초기화__(self, max_size: int = 10000): self.max_size = max_size self._verified: dict[str, bool] = {} 데프 _서명_ID( self, 메시지: 바이트, 서명: 바이트, public_key: 바이트 ) -> str: """서명 확인을 위한 고유 ID 생성""" 반품 hashlib.Blake2b( 메시지 + 서명[:64] + public_key, # sig의 처음 64바이트이면 충분합니다. 다이제스트_크기=16).헥스다이제스트() 데프 check_or_verify( self, 메시지: 바이트, 서명: 바이트, public_key: 바이트 ) -> bool: """캐시를 확인하거나 결과를 확인하고 캐시하세요""" sig_id = self._signature_id(메시지, 서명, 공개_키) if 서명_ID in self._검증됨: 반품 self._verified[서명_ID] # 확인하다 서명 = oqs.서명("SPHINCS+-SHAKE-128s-간단") is_valid = sig.verify(메시지, 서명, 공개_키) # 캐시(제거 포함) if len(self._verified) >= self.max_size: # ~10% 가장 오래된 항목 제거 to_remove = list(self._verified.keys())[:self.max_size // 10] ~을 위한 열쇠 in to_remove: 델 self._verified[키] self._verified[sig_id] = is_valid 반품 is_valid

수입 gc 수업 메모리 효율적인 서명자: """ 임베디드/모바일 장치에 대한 메모리 효율적인 서명 """ 데프 서명 및 해제( self, 메시지: 바이트, secret_key: 바이트 ) -> 바이트: """ 메시지에 서명하고 키 메모리를 즉시 해제합니다. 키가 지속되어서는 안 되는 일회성 서명에 사용합니다. """ sig_obj = oqs.서명("SPHINCS+-SHAKE-128s-간단", secret_key) 서명 = sig_obj.sign(message) # OQS 객체 해제 델 sig_obj # 비밀 키 덮어쓰기 if isinstance(secret_key, bytearray): ~을 위한 i in 범위(len(secret_key)): secret_key[i] = 0 # 강제 가비지 수집 gc.수집() 반품 서명 데프 스트리밍_서명( self, message_chunks: Iterator[bytes], secret_key: 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 기능 감지""" 노력하다: # 리눅스 ~와 함께 열려 있는("/proc/cpu정보") as f: cpuinfo = f.read() 기능 = set() if "avx2" in CPU정보: 기능.추가("avx2") if "avx512" in CPU정보: 기능.추가("avx512") if "에이스" in CPU정보: 기능.추가("에스니") 반품 특징 제외하고: 반품 세트() 데프 select_optimal_variant() -> 문자열: """이 CPU에 가장 적합한 SPHINCS+ 변형을 선택하세요""" 기능 = get_cpu_features() if "avx512" in 특징: # AVX-512는 ~20-30% 속도 향상을 제공합니다. 인쇄("AVX-512 최적화 구현 사용") 반품 "SPHINCS+-SHAKE-128s-간단" # liboqs 자동 선택 엘리프 "avx2" in 특징: 인쇄("AVX2 최적화 구현 사용") 반품 "SPHINCS+-SHAKE-128s-간단" 또 다른: 인쇄("참조 구현 사용") 반품 "SPHINCS+-SHAKE-128s-간단" # 최적의 플래그로 liboq를 컴파일합니다. # cmake -DOQS_USE_AVX2_INSTRUCTIONS=ON -DOQS_USE_AVX512_INSTRUCTIONS=ON ..

수입 통계 수입 시간 수업 암호화폐벤치마크: """종합적인 PQC 벤치마킹""" 데프 __초기화__(self, 반복: int = 100): self.iterations = 반복 데프 벤치마크 작업( self, 이름: str, 작업, 설정=없음 ) -> 받아쓰기: """단일 작업 벤치마크""" 횟수 = [] ~을 위한 _ in 범위(self.iterations): if 설정: ctx = setup() 시작 = time.perf_counter() if 설정: 작업(ctx) 또 다른: 작업() 경과 = (time.perf_counter() - 시작) * 1000 #ms 회.추가(경과) 반품 { "이름": 이름, "평균": 통계.평균(회), "중앙값": 통계.중앙값(회), "표준편차": 통계.stdev(회), "분": 분(회), "최대": 최대(회), "ops_per_sec": 1000 / 통계.평균(회) } 데프 run_full_benchmark(자신) -> 받아쓰기: """전체 PQC 벤치마크 제품군 실행""" 결과 = {} # Kyber 벤치마크 결과["kyber_keygen"] = self.benchmark_Operation( "Kyber-768 키젠", 람다: oqs.KeyEncapsulation("카이버768").generate_keypair() ) # 캡슐화/디캡 설정 kem = oqs.KeyEncapsulation("카이버768") pk = kem.generate_keypair() sk = kem.export_secret_key() 결과["kyber_encap"] = self.benchmark_Operation( "Kyber-768 엔캡", 람다: kem.encap_secret(pk) ) ct, _ = kem.encap_secret(pk) kem_dec = oqs.KeyEncapsulation("카이버768", sk) 결과["kyber_decap"] = self.benchmark_Operation( "Kyber-768 디캡", 람다: kem_dec.decap_secret(ct) ) # SPHINCS+ 벤치마크 결과["스핑크스_키겐"] = 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개["스핑크스_사인"] = self.benchmark_Operation( "SPHINCS+-SHAKE-128s 사인", 람다: oqs.서명("SPHINCS+-SHAKE-128s-간단", spx_sk).sign(msg), 반복=20 # 속도가 느려서 적음 ) 반품 결과 데프 print_results(self, 결과: dict): """예쁜 인쇄 벤치마크 결과""" 인쇄("\n=== PQC 성능 벤치마크 ===") 인쇄(f"반복: {self.iterations}\n") ~을 위한 키, 데이터 in 결과.항목(): 인쇄(f"{데이터['이름']}:") 인쇄(f" 평균: {data['mean']:.3f}ms") 인쇄(f" 중앙값: {data['median']:.3f}ms") 인쇄(f" 작업/초: {data['ops_per_sec']:.1f}") 인쇄() # 벤치마크 실행 if __이름__ == "__기본__": 벤치 = 암호화폐벤치마크(반복=50) 결과 = bench.run_full_benchmark() bench.print_results(결과)