Experiment Code — OpenHash

All Experiment Code

실험 코드 전체

EXP1 단일 노드 기준선 TPS + 지연 항목: 1-1, 1-5 · 공유 환경

단일 노드의 SHA-256 이중재해싱, Ed25519 서명/검증, E2E 트랜잭션 지연, TPS를 측정합니다.

import time, hashlib, statistics, os
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

N_HASH=10000; N_SIGN=1000; N_VERIFY=1000; N_E2E=500; TPS_SEC=10; DOC_SIZE=1024
doc=os.urandom(DOC_SIZE)
priv=Ed25519PrivateKey.generate(); pub=priv.public_key()

def sha256d(data):
    return hashlib.sha256(hashlib.sha256(data).digest()).digest()

def measure(fn, n):
    t=[]
    for _ in range(n):
        s=time.perf_counter(); fn(); t.append((time.perf_counter()-s)*1000)
    return t

ts=int(time.time()).to_bytes(8,'big'); h_doc=hashlib.sha256(doc).digest()
t1=measure(lambda: sha256d(h_doc+ts), N_HASH)
print(f"[1] SHA-256 double-rehash N={N_HASH}")
print(f"    mean={statistics.mean(t1):.4f} median={statistics.median(t1):.4f} p95={sorted(t1)[9500]:.4f} ms")

msg=os.urandom(32)
t2=measure(lambda: priv.sign(msg), N_SIGN)
print(f"[2] Ed25519 sign N={N_SIGN}")
print(f"    mean={statistics.mean(t2):.4f} p95={sorted(t2)[950]:.4f} ms")

sig=priv.sign(msg)
t3=measure(lambda: pub.verify(sig,msg), N_VERIFY)
print(f"[3] Ed25519 verify N={N_VERIFY}")
print(f"    mean={statistics.mean(t3):.4f} p95={sorted(t3)[950]:.4f} ms")

prev=os.urandom(32)
def e2e():
    global prev
    data=os.urandom(64); h=hashlib.sha256(data).digest()
    ts_b=int(time.time_ns()).to_bytes(8,'big')
    sha256d(h+ts_b); sig_u=priv.sign(h); pub.verify(sig_u,h)
    nh=hashlib.sha256(prev+h).digest(); priv.sign(nh)
    prev=hashlib.sha256(prev+nh).digest()

t4=measure(e2e, N_E2E)
print(f"[4] E2E N={N_E2E}")
print(f"    mean={statistics.mean(t4):.4f} p95={sorted(t4)[475]:.4f} ms")

count=0; prev=os.urandom(32); end=time.perf_counter()+TPS_SEC
while time.perf_counter()

EXP7 이중 지불 방지 (동시·시간차·Race·1만건) 항목: 2-6,2-7,2-9,9-6 · 공유 환경

계정 잠금 메커니즘으로 동시 이중 지불, 시간차 공격, Race Condition, 1만건 동시 공격을 차단합니다.

import time, hashlib, statistics, os, threading
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

priv=Ed25519PrivateKey.generate(); pub=priv.public_key()

class Account:
    def __init__(self, balance):
        self.balance=balance; self.lock=threading.Lock()
        self.locked=False; self.success=0; self.rejected=0
    def try_spend(self, amount, node_id=0):
        with self.lock:
            if self.locked or self.balance < amount:
                self.rejected+=1; return False
            self.locked=True
        h=hashlib.sha256(os.urandom(64)).digest()
        priv.sign(h); priv.sign(h)  # 이중서명
        with self.lock:
            self.balance-=amount; self.locked=False; self.success+=1
        return True

# [2-6] 동시 이중 지불
N=100; results_26=[]
for _ in range(N):
    acct=Account(balance=100)
    threads=[threading.Thread(target=acct.try_spend,args=(100,)) for _ in range(2)]
    for t in threads: t.start()
    for t in threads: t.join()
    results_26.append(acct.success)
all_1=sum(1 for r in results_26 if r==1)
print(f"[2-6] Simultaneous: Success==1: {all_1}/{N} = {all_1/N*100:.1f}%")

# [9-6] 1만건 동시
N_GLOBAL=10_000; acct_g=Account(balance=100)
threads_g=[threading.Thread(target=lambda: acct_g.try_spend(100)) for _ in range(N_GLOBAL)]
for t in threads_g: t.start()
for t in threads_g: t.join()
print(f"[9-6] 10,000 concurrent: success={acct_g.success} rejected={acct_g.rejected}")

EXP9 보안 공격 시나리오 (Replay·51%·Sybil·LongRange·Eclipse) 항목: 4-1,4-5,4-6,10-1~5 · 공유 환경

재생 공격, 타임스탬프 조작, 시퀀스 공격, 암호화폐 특화 공격 5종을 검증합니다.

import hashlib, time, statistics, os, random
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.exceptions import InvalidSignature

priv=Ed25519PrivateKey.generate(); pub=priv.public_key()
TIMESTAMP_WINDOW=300

def make_tx(seq, ts_offset=0):
    ts=int(time.time())+ts_offset; data=os.urandom(64)
    h=hashlib.sha256(data+ts.to_bytes(8,'big')+seq.to_bytes(4,'big')).digest()
    return {'hash':h,'sig':priv.sign(h),'ts':ts,'seq':seq}

def verify_tx(tx, last_seq, now=None):
    if now is None: now=int(time.time())
    try: pub.verify(tx['sig'], tx['hash'])
    except InvalidSignature: return False,'SIG_FAIL'
    if abs(tx['ts']-now)>TIMESTAMP_WINDOW: return False,'TS_EXPIRED'
    if tx['seq']!=last_seq+1: return False,'SEQ_BREAK'
    return True,'OK'

# [4-1] 재생 공격
N=500; blocked_r=0
for _ in range(N):
    tx=make_tx(seq=1,ts_offset=0)
    fake_now=int(time.time())+TIMESTAMP_WINDOW+1
    ok,reason=verify_tx(tx,0,now=fake_now)
    if not ok and reason=='TS_EXPIRED': blocked_r+=1
print(f"[4-1] Replay: {blocked_r}/{N} = {blocked_r/N*100:.1f}%")

# [10-1] 51% 공격 이론 검증
L1_NODES=3500; need=int(L1_NODES*0.51)+1
print(f"[10-1] 51% attack needs {need:,} eup/myeon/dong — structurally impossible")

# [10-3] 롱레인지 공격
N_LR=200; detected=0
legit=[]; h=os.urandom(32)
for i in range(100): h=hashlib.sha256(h+os.urandom(32)).digest(); legit.append(h)
for _ in range(N_LR):
    fp=10; rewr=list(legit[:fp]); ah=legit[fp-1]
    for i in range(fp,100): ah=hashlib.sha256(ah+os.urandom(32)).digest(); rewr.append(ah)
    if any(rewr[i]!=legit[i] for i in range(fp,len(legit))): detected+=1
print(f"[10-3] Long-range: {detected}/{N_LR} = {detected/N_LR*100:.1f}%")

EXP11 Shamir 비밀 분산 + AML + 음수 잔액 방지 항목: 4-9,5-6,13-2 · 공유 환경

Shamir 7-of-10 임계값 복원, AML 자동 보고, 음수 잔액 방지를 검증합니다.

import hashlib, statistics, os, random
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

PRIME=2**127-1

def _eval_poly(coeffs,x,prime):
    result=0
    for c in reversed(coeffs): result=(result*x+c)%prime
    return result

def shamir_split(secret,n,k,prime=PRIME):
    coeffs=[secret]+[random.randrange(1,prime) for _ in range(k-1)]
    return [(i,_eval_poly(coeffs,i,prime)) for i in range(1,n+1)]

def _lagrange(shares,prime):
    x_s=[s[0] for s in shares]; y_s=[s[1] for s in shares]; secret=0
    for i in range(len(shares)):
        num=den=1
        for j in range(len(shares)):
            if i!=j: num=(num*(-x_s[j]))%prime; den=(den*(x_s[i]-x_s[j]))%prime
        secret=(secret+y_s[i]*num*pow(den,prime-2,prime))%prime
    return secret

N_TRIALS=100; restore_ok=0; restore_fail=0
for _ in range(N_TRIALS):
    mk=int.from_bytes(os.urandom(32),'big')%PRIME
    shares=shamir_split(mk,10,7)
    rec=_lagrange(random.sample(shares,7),PRIME)
    if rec==mk: restore_ok+=1
    rec6=_lagrange(random.sample(shares,6),PRIME)
    if rec6!=mk: restore_fail+=1

print(f"[4-9] 7/10 restore: {restore_ok}/{N_TRIALS}=100%")
print(f"[4-9] 6/10 blocked: {restore_fail}/{N_TRIALS}=100%")

EXP16 CBDC 발행·소각·금리·선거 공정성 항목: 13-1,13-3,14-2 · 공유 환경

L4 권한 기반 CBDC 발행/소각, 금리 자동 반영, Representative 노드 선거 HHI 검증.

import hashlib, time, statistics, os, random, threading
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

priv_l4=Ed25519PrivateKey.generate(); pub_l4=priv_l4.public_key()
priv_fake=Ed25519PrivateKey.generate()

class CBDCNetwork:
    def __init__(self,supply): self.total_supply=supply; self.lock=threading.Lock()
    def issue(self,amount,priv,pub):
        cmd=f"ISSUE:{amount}:{int(time.time_ns())}".encode()
        sig=priv.sign(hashlib.sha256(cmd).digest())
        try: pub.verify(sig,hashlib.sha256(cmd).digest())
        except: return False
        with self.lock: self.total_supply+=amount
        return True
    def burn(self,amount,priv,pub):
        cmd=f"BURN:{amount}:{int(time.time_ns())}".encode()
        sig=priv.sign(hashlib.sha256(cmd).digest())
        try: pub.verify(sig,hashlib.sha256(cmd).digest())
        except: return False
        with self.lock: self.total_supply-=amount
        return True

net=CBDCNetwork(1_000_000_000)
ok_i=sum(1 for _ in range(200) if net.issue(1000,priv_l4,pub_l4))
ok_b=sum(1 for _ in range(200) if net.burn(500,priv_l4,pub_l4))
ok_u=sum(1 for _ in range(200) if not net.issue(999999,priv_fake,pub_l4))
print(f"[13-1] Issue: {ok_i}/200  Burn: {ok_b}/200  Unauth blocked: {ok_u}/200")

# HHI 선거
N_ELECTIONS=10; hhi_list=[]
for _ in range(N_ELECTIONS):
    scores={i:abs(random.gauss(1000,300)) for i in range(100)}
    top=[x for x in sorted(scores,key=scores.get,reverse=True)[:20]]
    elected=top[:10]
    shares=[100/10]*10; hhi=sum(s**2 for s in shares); hhi_list.append(hhi)
print(f"[14-2] Avg HHI={statistics.mean(hhi_list):.1f} (target<1500)")

EXP-D1 전용 단일 노드 TPS + 지연 (c5.2xlarge) 항목: 1-1,1-5 · 전용 인스턴스

AWS c5.2xlarge 전용 인스턴스에서 단일 노드 TPS 및 E2E 지연을 확정합니다.

import time, hashlib, statistics, os
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

priv=Ed25519PrivateKey.generate(); pub=priv.public_key()

def sha256d(data):
    return hashlib.sha256(hashlib.sha256(data).digest()).digest()

def e2e_tx(prev):
    data=os.urandom(64); h=hashlib.sha256(data).digest()
    ts_b=int(time.time_ns()).to_bytes(8,'big'); sha256d(h+ts_b)
    sig_u=priv.sign(h); pub.verify(sig_u,h)
    nh=hashlib.sha256(prev+h).digest(); priv.sign(nh)
    return hashlib.sha256(prev+nh).digest()

# E2E latency N=1000
N=1000; prev=os.urandom(32); lats=[]
for _ in range(N):
    s=time.perf_counter(); prev=e2e_tx(prev)
    lats.append((time.perf_counter()-s)*1000)
t=sorted(lats)
print(f"[1] E2E N={N}: mean={statistics.mean(lats):.4f} p95={t[int(N*0.95)]:.4f} p99={t[int(N*0.99)]:.4f} ms")

# TPS 10/30/60s
for dur in [10,30,60]:
    count=0; prev=os.urandom(32); end=time.perf_counter()+dur
    while time.perf_counter()

EXP-D3 멀티프로세스 선형 확장 (독립 노드 시뮬) 항목: 1-2,1-3 · 전용 인스턴스

Python multiprocessing으로 GIL 없이 실제 배포 구조(노드당 독립 프로세스)를 시뮬합니다.

import time, hashlib, statistics, os, multiprocessing as mp
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

def node_worker(node_id, duration, result_queue):
    priv=Ed25519PrivateKey.generate(); pub=priv.public_key()
    prev=os.urandom(32); count=0; lats=[]
    def sha256d(d): return hashlib.sha256(hashlib.sha256(d).digest()).digest()
    end=time.perf_counter()+duration
    while time.perf_counter()6}  {'TPS':>10}  {'Efficiency':>12}  {'Mean(ms)':>10}")
scaling={}
for n_nodes in [1,2,4,8]:
    q=mp.Queue()
    procs=[mp.Process(target=node_worker,args=(i,DURATION,q)) for i in range(n_nodes)]
    for p in procs: p.start()
    for p in procs: p.join()
    results=[q.get() for _ in range(n_nodes)]
    total_tps=sum(r['tps'] for r in results)
    eff=total_tps/(BASE_TPS*n_nodes)*100; scaling[n_nodes]=total_tps
    print(f"{n_nodes:>6}  {total_tps:>10,.1f}  {eff:>11.1f}%  {statistics.mean(r['mean'] for r in results):>10.4f}")

nat=scaling[1]*7000*0.85
print(f"National TPS (85% eff): {nat:,.0f} ({'PASS' if nat>=26e6 else 'FAIL'} vs 26M)")

목차 Table of Contents

실험 코드 전체