Python으로 ZK 증명, L1↔L2 브릿지, 애플리케이션 특화 ZK 롤업을 구축하는 완전 가이드
- Part 1: Hello World ZK
- Part 2: ZK Note Settlement — ERC20 토큰 비공개 정산
- Part 3: 팁과 제약사항
- Part 4: L1↔L2 General State Bridge
- Part 5: 애플리케이션 특화 ZK 롤업
- Part 6: 프로덕션 배포
"나는 두 비밀 숫자를 알고 있고, 그 곱이 15라는 것을 증명하겠다."
이것이 ZK 증명의 본질입니다. 비밀을 공개하지 않으면서 그 비밀에 대한 사실을 증명하는 것.
ZK 증명의 시작점은 circuit (회로)입니다. circuit은 "무엇을 증명할 것인가"를 수학적으로 정의합니다.
from ethclient.zk import Circuit, groth16
# Circuit 생성
c = Circuit()
# 비밀 입력 (prover만 알고 있음)
x = c.private("x")
y = c.private("y")
# 공개 입력 (verifier도 알고 있음)
z = c.public("z")
# 제약 조건: x * y = z
c.constrain(x * y, z)여기서 핵심 개념:
| 개념 | 설명 | 예시 |
|---|---|---|
| private | prover만 아는 비밀 값 | x=3, y=5 |
| public | 누구나 볼 수 있는 값 | z=15 |
| constrain | 반드시 만족해야 하는 수학적 관계 | x * y = z |
circuit의 구조를 확인해 봅시다:
print(f"Constraints: {c.num_constraints}") # 1
print(f"Public inputs: {c.num_public}") # 1
print(f"Private inputs: {c.num_private}") # 2이 circuit은 R1CS (Rank-1 Constraint System)로 내부 변환됩니다:
A · witness ⊙ B · witness = C · witness
여기서 witness = [1, z, x, y, ...] (상수 1 + 공개 + 비밀 + 중간값).
Groth16은 circuit마다 한 번의 trusted setup이 필요합니다. 이 과정에서 proving key와 verification key가 생성됩니다.
pk, vk = groth16.setup(c)
print(f"Public inputs: {vk.num_public_inputs}") # 1
print(f"IC points: {len(vk.ic)}") # 2 (IC[0] + IC[1])- pk (Proving Key): prover가 proof를 만들 때 사용
- vk (Verification Key): verifier가 proof를 검증할 때 사용
참고: setup 과정에서 생성되는 "toxic waste" (τ, α, β, γ, δ)는 자동으로 폐기됩니다. 이 값이 유출되면 가짜 proof를 만들 수 있으므로, 실무에서는 MPC ceremony를 통해 setup을 수행합니다.
비밀 값(x=3, y=5)을 알고 있는 prover가 "x * y = 15"임을 증명합니다:
proof = groth16.prove(
pk,
private={"x": 3, "y": 5},
public={"z": 15},
circuit=c,
)
print(f"Proof.A = G1({proof.a.x}, {proof.a.y})")
print(f"Proof.B = G2(...)")
print(f"Proof.C = G1({proof.c.x}, {proof.c.y})")proof는 세 개의 타원곡선 점으로 구성됩니다:
- A ∈ G1 (BN128 curve)
- B ∈ G2 (BN128 twist curve)
- C ∈ G1
이 proof 안에는 x=3, y=5라는 정보가 전혀 포함되지 않습니다. 하지만 수학적으로 "x * y = 15를 만족하는 x, y를 알고 있다"는 것을 증명합니다.
verifier는 proof와 공개 입력(z=15)만으로 검증합니다:
valid = groth16.verify(vk, proof, [15])
print(f"Valid: {valid}") # True
assert valid내부적으로 pairing 연산을 수행합니다:
e(A, B) == e(α, β) × e(IC_acc, γ) × e(C, δ)
여기서 IC_acc = IC[0] + 15 × IC[1].
py-ethclient의 진짜 강점: proof를 이더리움 EVM에서 바로 검증할 수 있습니다.
from ethclient.zk.evm_verifier import EVMVerifier
# Verification key로부터 verifier 컨트랙트 바이트코드 자동 생성
verifier = EVMVerifier(vk)
print(f"Bytecode: {len(verifier.bytecode)} bytes")
# 인메모리 EVM에서 실행
result = verifier.verify_on_evm(proof, [15])
print(f"EVM 검증 성공: {result.success}") # True
print(f"Gas 사용량: {result.gas_used}") # ≈ 210,000이것이 내부적으로 하는 일:
- vk의 α, β, γ, δ 포인트를 EVM 바이트코드에 하드코딩
ecMul(0x07) 프리컴파일로 IC 누적 계산ecAdd(0x06) 프리컴파일로 점 덧셈ecPairing(0x08) 프리컴파일로 최종 pairing check- 결과 반환 (1 = valid, 0 = invalid)
Solidity verifier를 직접 작성하고 테스트넷에 배포할 필요 없이, Python 한 줄로 on-chain 검증을 테스트할 수 있습니다.
잘못된 공개 입력으로 검증하면 실패합니다:
# z=16은 틀림 (3 * 5 = 15, not 16)
wrong = verifier.verify_on_evm(proof, [16])
print(f"z=16: success={wrong.success}") # False
# 네이티브 검증도 동일
assert not groth16.verify(vk, proof, [16])on-chain 검증 비용을 프리컴파일별로 분석합니다:
profile = verifier.gas_profile(proof, [15])
print(f"ecAdd: {profile.ecadd_calls} calls, {profile.ecadd_gas:,} gas")
print(f"ecMul: {profile.ecmul_calls} calls, {profile.ecmul_gas:,} gas")
print(f"ecPairing: {profile.ecpairing_calls} calls, {profile.ecpairing_gas:,} gas")
print(f"Total: {profile.total_gas:,} gas")출력 예시:
ecAdd: 1 calls, 150 gas
ecMul: 1 calls, 6,000 gas
ecPairing: 1 calls, 181,000 gas
Total: 187,150 gas
insight: Groth16 검증 gas의 96% 이상이 ecPairing에서 발생합니다. public input 수가 늘어나면 ecMul/ecAdd 호출이 증가하지만, pairing은 항상 1회(4쌍)입니다.
proof 검증이 실패할 때, 어디서 문제가 생겼는지 확인할 수 있습니다:
debug = groth16.debug_verify(vk, proof, [15])
print(f"Valid: {debug.valid}")
# 개별 pairing 값 확인
print(f"e(A, B) = {debug.e_ab}")
print(f"e(α, β) = {debug.e_alpha_beta}")
print(f"e(IC_acc, γ) = {debug.e_ic_gamma}")
print(f"e(C, δ) = {debug.e_c_delta}")
# 유효한 proof라면:
# e(A,B) == e(α,β) * e(IC_acc,γ) * e(C,δ)
assert debug.e_ab == debug.e_alpha_beta * debug.e_ic_gamma * debug.e_c_delta이는 circom/snarkjs에서는 불가능한 디버깅 방법입니다. hex 덤프 대신 실제 pairing 값을 Python 변수로 검사할 수 있습니다.
아래를 hello_zk.py로 저장하고 실행하세요:
"""Hello World ZK — 가장 간단한 Groth16 증명"""
from ethclient.zk import Circuit, groth16
from ethclient.zk.evm_verifier import EVMVerifier
# ── Circuit 정의 ──
c = Circuit()
x = c.private("x")
y = c.private("y")
z = c.public("z")
c.constrain(x * y, z)
print(f"Circuit: {c.num_constraints} constraint, "
f"{c.num_public} public, {c.num_private} private")
# ── Trusted Setup ──
pk, vk = groth16.setup(c)
print(f"Setup 완료: IC points = {len(vk.ic)}")
# ── Proof 생성 ──
proof = groth16.prove(pk, private={"x": 3, "y": 5}, public={"z": 15}, circuit=c)
print(f"Proof 생성 완료")
# ── 네이티브 검증 ──
assert groth16.verify(vk, proof, [15]), "검증 실패!"
print("네이티브 검증: PASS")
# ── EVM 검증 ──
verifier = EVMVerifier(vk)
result = verifier.verify_on_evm(proof, [15])
assert result.success, "EVM 검증 실패!"
print(f"EVM 검증: PASS (gas: {result.gas_used:,})")
# ── 틀린 입력 ──
assert not verifier.verify_on_evm(proof, [16]).success
print("틀린 입력(z=16) 거부: PASS")
print("\nAll checks passed!")python hello_zk.pyZK DEX에서 ZK Note는 토큰 소유권을 나타내는 커밋먼트입니다. Tornado Cash나 Zcash의 노트와 같은 개념입니다.
┌─────────────────────────────────────────────────────┐
│ ZK Note = commitment(secret, amount) │
│ │
│ On-chain에는 commitment만 저장 │
│ secret과 amount는 소유자만 알고 있음 │
│ │
│ 정산(settle)할 때: │
│ → "나는 이 커밋먼트의 preimage를 안다" 를 ZK로 증명 │
│ → nullifier를 공개해서 이중 사용 방지 │
│ → secret은 절대 공개하지 않음 │
└─────────────────────────────────────────────────────┘
전체 흐름:
Deposit (공개) Settle (비공개)
───────────── ────────────────
1. secret, amount 선택 1. ZK proof 생성
2. commitment = secret × amount 2. nullifier = secret² 공개
3. ERC20.transfer(vault, amount) 3. Verifier가 proof 검증
4. commitment를 on-chain 등록 4. nullifier 중복 체크
5. ERC20.transfer(receiver, amount)
왜 이 구조가 프라이버시를 보장하는가?
| 정보 | Deposit 시 | Settle 시 | 관찰자가 아는 것 |
|---|---|---|---|
secret |
소유자만 | 비공개 (ZK proof) | 모름 |
amount |
commitment에 숨김 | 비공개 (ZK proof) | 모름 |
commitment |
공개 등록 | proof의 공개 입력 | 알지만 preimage 모름 |
nullifier |
— | 공개 | 어떤 commitment인지 연결 불가 |
from ethclient.zk import Circuit, groth16
from ethclient.zk.evm_verifier import EVMVerifier
c = Circuit()
# ── 비밀 입력: 노트 소유자만 알고 있음 ──
secret = c.private("secret") # 소유자의 비밀키
amount = c.private("amount") # 노트에 담긴 토큰 양
# ── 공개 입력: 체인에 기록됨 ──
commitment = c.public("commitment") # 노트 커밋먼트 (Deposit 시 등록)
nullifier = c.public("nullifier") # 이중 사용 방지 태그
# ── Constraint 1: 소유권 증명 ──
# "나는 이 commitment의 preimage(secret, amount)를 알고 있다"
c.constrain(secret * amount, commitment)
# ── Constraint 2: Nullifier 결정적 생성 ──
# 같은 secret → 항상 같은 nullifier → 이중 사용 감지
c.constrain(secret * secret, nullifier)
print(f"ZK Note Settlement Circuit:")
print(f" Constraints: {c.num_constraints}") # 2
print(f" Public: {c.num_public}") # 2 (commitment, nullifier)
print(f" Private: {c.num_private}") # 2 (secret, amount)circuit이 보장하는 것:
- prover는
secret × amount == commitment을 만족하는 secret과 amount를 알고 있다 → 소유권 nullifier == secret²이므로, 같은 secret으로 두 번 settle하면 같은 nullifier가 나온다 → 이중 사용 방지- verifier는 secret을 전혀 모른다 → 프라이버시
Private (비밀) Public (공개)
┌───────────┐ ┌──────────────┐
│ secret ────┤──×──────→│ commitment │ Constraint 1: secret × amount == commitment
│ │ │ │
│ amount ────┘ │ │
│ │ │ │
│ secret ────┤──×──────→│ nullifier │ Constraint 2: secret × secret == nullifier
│ secret ────┘ │ │
└───────────┘ └──────────────┘
실제 DApp에서는 Deposit 시 다음이 일어납니다:
# ━━━ Alice가 노트를 생성한다 ━━━
# Alice의 비밀값 (안전하게 보관해야 함!)
alice_secret = 42
alice_amount = 100 # 100 USDC
# Commitment 계산 (off-chain, 누구나 수학 검증 가능)
alice_commitment = alice_secret * alice_amount # = 4200
alice_nullifier = alice_secret * alice_secret # = 1764
print(f"Alice의 노트:")
print(f" secret: {alice_secret} (비밀!)")
print(f" amount: {alice_amount} USDC")
print(f" commitment: {alice_commitment} (on-chain 등록)")
print(f" nullifier: {alice_nullifier} (settle 때까지 비밀)")
# On-chain에서 일어나는 일 (Solidity 의사코드):
# vault.deposit(commitment=4200)
# USDC.transferFrom(alice, vault, 100)
# noteTree.insert(4200)참고: 실제 시스템에서는
commitment = hash(secret, amount, salt)로 해시 커밋먼트를 사용합니다. 이 튜토리얼에서는 R1CS로 표현 가능한 곱셈(secret × amount)을 사용합니다.
Alice가 나중에 토큰을 정산(withdraw/transfer)합니다. 이때 secret을 공개하지 않고 ZK proof로 소유권을 증명합니다.
# ━━━ Trusted Setup (DEX 배포 시 1회) ━━━
pk, vk = groth16.setup(c)
print(f"\nSetup 완료: {vk.num_public_inputs} public inputs")
# ━━━ Alice가 정산 proof를 생성한다 ━━━
proof = groth16.prove(
pk,
private={
"secret": 42, # 비밀! proof에 포함되지 않음
"amount": 100, # 비밀! proof에 포함되지 않음
},
public={
"commitment": 4200, # on-chain에 등록된 값
"nullifier": 1764, # 이중 사용 방지용
},
circuit=c,
)
print(f"Proof 생성 완료")
# ━━━ Verifier (DEX 컨트랙트)가 검증한다 ━━━
# verifier는 commitment=4200, nullifier=1764만 본다
# secret=42, amount=100은 절대 알 수 없다
valid = groth16.verify(vk, proof, [4200, 1764])
print(f"검증 결과: {valid}") # True
assert validverifier(DEX 컨트랙트)의 검증 로직:
1. proof 검증 → True (ZK proof가 유효함)
2. commitment 4200이 noteTree에 있는지 확인 → 있음
3. nullifier 1764가 이미 사용되었는지 확인 → 아직 없음
4. nullifier 1764를 사용 완료 목록에 추가
5. Alice에게 토큰 전송
같은 노트로 두 번 정산하려고 하면? nullifier가 동일하므로 컨트랙트가 거부합니다.
# ━━━ Alice가 같은 노트로 두 번째 정산을 시도한다 ━━━
proof_again = groth16.prove(
pk,
private={"secret": 42, "amount": 100},
public={"commitment": 4200, "nullifier": 1764},
circuit=c,
)
# proof 자체는 유효하다! (수학적으로 맞으니까)
assert groth16.verify(vk, proof_again, [4200, 1764])
print("두 번째 proof도 수학적으로 유효")
# 하지만 on-chain에서:
# → nullifier 1764가 이미 사용됨 → 트랜잭션 revert!
# 이것이 nullifier의 역할
spent_nullifiers = {1764} # on-chain에서 관리하는 사용 완료 목록
if 1764 in spent_nullifiers:
print("이중 사용 감지! nullifier 1764는 이미 사용됨 → 거부")또 다른 사람이 위조된 commitment를 사용하려면?
# ━━━ Bob이 Alice의 commitment에 대한 가짜 proof를 만들려 한다 ━━━
try:
# Bob은 secret=42를 모르므로, 아무 값이나 넣는다
# secret=10, amount=420 → 10 × 420 = 4200 (commitment 일치!)
# 하지만 nullifier = 10² = 100 ≠ 1764
fake_proof = groth16.prove(
pk,
private={"secret": 10, "amount": 420},
public={"commitment": 4200, "nullifier": 1764}, # nullifier가 맞지 않음
circuit=c,
)
print("여기 도달하면 안 됨")
except ValueError as e:
print(f"Bob의 위조 시도 실패: {e}")
print("→ secret=10이면 nullifier=100이어야 하는데, 1764라고 주장 → R1CS 불만족")핵심: commitment만 맞추는 건 쉽지만 (10 × 420 = 4200), 동시에 nullifier도 맞춰야 하므로 (10² = 100 ≠ 1764) 원래 secret을 모르면 유효한 proof를 만들 수 없습니다.
실제 이더리움에서 이 검증을 실행하면 얼마나 gas가 드는지 확인합니다:
# ━━━ EVM Verifier 생성 ━━━
verifier = EVMVerifier(vk)
print(f"Verifier 바이트코드: {len(verifier.bytecode)} bytes")
# ━━━ EVM에서 정산 proof 검증 ━━━
result = verifier.verify_on_evm(proof, [4200, 1764])
print(f"EVM 검증: {'PASS' if result.success else 'FAIL'}")
print(f"Gas 사용량: {result.gas_used:,}")
# 틀린 nullifier로 시도
bad_result = verifier.verify_on_evm(proof, [4200, 9999])
print(f"틀린 nullifier: {'PASS' if bad_result.success else 'FAIL (거부됨)'}")
# ━━━ Gas 프로파일 ━━━
profile = verifier.gas_profile(proof, [4200, 1764])
print(f"\nGas 상세:")
print(f" ecAdd: {profile.ecadd_calls} calls, {profile.ecadd_gas:,} gas")
print(f" ecMul: {profile.ecmul_calls} calls, {profile.ecmul_gas:,} gas")
print(f" ecPairing: {profile.ecpairing_calls} call, {profile.ecpairing_gas:,} gas")
print(f" Total: {profile.total_gas:,} gas")Groth16 on-chain 검증 gas는 public input 수에 약간 비례합니다. 2개 public input (commitment + nullifier)에서 ecMul 2회, ecAdd 2회, ecPairing 1회(4쌍)가 필요합니다.
py-ethclient에서 만든 proof를 snarkjs 포맷으로 내보낼 수 있습니다:
from ethclient.zk.snarkjs_compat import (
export_snarkjs_vkey, export_snarkjs_proof, verify_snarkjs,
)
vk_json = export_snarkjs_vkey(vk)
proof_json = export_snarkjs_proof(proof)
public_json = ["4200", "1764"] # snarkjs는 문자열 사용
assert verify_snarkjs(vk_json, proof_json, public_json)
print(f"snarkjs 포맷 검증: PASS")아래를 zk_note_settle.py로 저장하고 실행하세요:
"""ZK Note Settlement — ERC20 토큰 비공개 정산 데모
Deposit: secret과 amount로 커밋먼트를 만들고, ERC20을 vault에 예치
Settle: ZK proof로 소유권 증명 + nullifier로 이중 사용 방지
증명하는 것:
1. secret × amount == commitment (소유권)
2. secret × secret == nullifier (이중 사용 방지)
비밀로 유지하는 것:
- secret (소유자 비밀키)
- amount (토큰 양)
"""
import time
from ethclient.zk import Circuit, groth16
from ethclient.zk.evm_verifier import EVMVerifier
from ethclient.zk.snarkjs_compat import export_snarkjs_vkey, export_snarkjs_proof, verify_snarkjs
print("=" * 60)
print(" ZK Note Settlement — Private ERC20 Settle")
print("=" * 60)
# ━━━ 1. Circuit 정의 ━━━
c = Circuit()
secret = c.private("secret")
amount = c.private("amount")
commitment = c.public("commitment")
nullifier = c.public("nullifier")
c.constrain(secret * amount, commitment) # 소유권 증명
c.constrain(secret * secret, nullifier) # Nullifier 생성
print(f"\n[1] Circuit")
print(f" {c.num_constraints} constraints, {c.num_public} public, {c.num_private} private")
# ━━━ 2. Trusted Setup ━━━
print(f"\n[2] Trusted setup...")
t0 = time.time()
pk, vk = groth16.setup(c)
print(f" {time.time() - t0:.1f}s")
# ━━━ 3. Deposit (off-chain 계산) ━━━
alice_secret, alice_amount = 42, 100
alice_commitment = alice_secret * alice_amount # 4200
alice_nullifier = alice_secret * alice_secret # 1764
print(f"\n[3] Alice deposits 100 USDC")
print(f" commitment: {alice_commitment}")
print(f" (secret={alice_secret}, amount={alice_amount} — 비밀!)")
# ━━━ 4. Settle (ZK proof 생성) ━━━
print(f"\n[4] Alice settles with ZK proof")
t0 = time.time()
proof = groth16.prove(
pk,
private={"secret": alice_secret, "amount": alice_amount},
public={"commitment": alice_commitment, "nullifier": alice_nullifier},
circuit=c,
)
print(f" Proof 생성: {time.time() - t0:.1f}s")
# ━━━ 5. Verify ━━━
print(f"\n[5] DEX 컨트랙트가 proof 검증")
t0 = time.time()
valid = groth16.verify(vk, proof, [alice_commitment, alice_nullifier])
print(f" 네이티브: {'PASS' if valid else 'FAIL'} ({time.time() - t0:.2f}s)")
assert valid
# ━━━ 6. EVM on-chain 검증 ━━━
verifier = EVMVerifier(vk)
result = verifier.verify_on_evm(proof, [alice_commitment, alice_nullifier])
print(f"\n[6] EVM on-chain 검증")
print(f" result: {'PASS' if result.success else 'FAIL'}")
print(f" gas: {result.gas_used:,}")
# ━━━ 7. 보안 테스트 ━━━
print(f"\n[7] 보안 테스트")
# 7a. 이중 사용
spent = {alice_nullifier}
print(f" 이중 사용: nullifier {alice_nullifier} in spent_set → 거부")
# 7b. 틀린 nullifier
assert not verifier.verify_on_evm(proof, [alice_commitment, 9999]).success
print(f" 틀린 nullifier(9999): EVM 거부")
# 7c. Bob의 위조 시도
try:
groth16.prove(
pk,
private={"secret": 10, "amount": 420},
public={"commitment": 4200, "nullifier": 1764},
circuit=c,
)
print(" Bob 위조: 예상 외 성공 (secret=10에서는 nullifier=100이어야 함)")
except ValueError:
print(f" Bob 위조(secret=10, amount=420): proof 생성 실패 — R1CS 불만족")
# ━━━ 8. 다른 유저 ━━━
print(f"\n[8] Bob deposits 200 USDC (secret=77)")
bob_secret, bob_amount = 77, 200
bob_commitment = bob_secret * bob_amount # 15400
bob_nullifier = bob_secret * bob_secret # 5929
proof_bob = groth16.prove(
pk,
private={"secret": bob_secret, "amount": bob_amount},
public={"commitment": bob_commitment, "nullifier": bob_nullifier},
circuit=c,
)
assert groth16.verify(vk, proof_bob, [bob_commitment, bob_nullifier])
assert verifier.verify_on_evm(proof_bob, [bob_commitment, bob_nullifier]).success
print(f" Bob settle: PASS (commitment={bob_commitment}, nullifier={bob_nullifier})")
# ━━━ 9. snarkjs 호환 ━━━
vk_json = export_snarkjs_vkey(vk)
proof_json = export_snarkjs_proof(proof)
assert verify_snarkjs(vk_json, proof_json, [str(alice_commitment), str(alice_nullifier)])
print(f"\n[9] snarkjs 포맷: PASS")
# ━━━ Gas 프로파일 ━━━
profile = verifier.gas_profile(proof, [alice_commitment, alice_nullifier])
print(f"\n{'=' * 60}")
print(f" All checks passed!")
print(f" Notes: Alice(100 USDC), Bob(200 USDC)")
print(f" EVM gas: {result.gas_used:,}")
print(f" Gas: ecMul={profile.ecmul_gas:,} + ecAdd={profile.ecadd_gas:,}"
f" + ecPairing={profile.ecpairing_gas:,}")
print(f"{'=' * 60}")python zk_note_settle.py| 패턴 | 코드 | R1CS 변환 |
|---|---|---|
| 곱셈 | c.constrain(a * b, result) |
a × b = result |
| 덧셈 | c.constrain(a + b, result) |
(a + b - result) × 1 = 0 |
| 상수 곱 | c.constrain(a * 3, result) |
a × 3 = result |
| 혼합 | c.constrain(a * b, x + y) |
a × b = x + y |
# 지원되는 연산
a + b # Signal + Signal → Signal (선형)
a + 5 # Signal + 상수 → Signal (선형)
a - b # 뺄셈
a * b # Signal × Signal → 중간 변수 생성 + 자동 constraint
a * 3 # Signal × 상수 → Signal (선형, constraint 없음)
-a # 부정
3 + a # 역방향 연산도 지원| Circuit 크기 | Setup | Prove | Verify | EVM Verify |
|---|---|---|---|---|
| 1 constraint | ~2s | ~2s | ~1s | ~0.1s |
| 2 constraints | ~4s | ~4s | ~1s | ~0.1s |
| 5 constraints | ~10s | ~10s | ~1s | ~0.1s |
순수 Python (py_ecc)이므로 10개 이상의 constraint는 시간이 오래 걸릴 수 있습니다. 교육/프로토타이핑 용도에 적합하며, 프로덕션 증명 생성은 snarkjs/rapidsnark를 사용하세요.
-
Range proof 미포함:
a >= b같은 부등식은 R1CS로 직접 표현 불가. 비트 분해가 필요합니다. -
유한체 연산: 모든 연산은 BN128 scalar field (
p ≈ 2^254) 위에서 수행됩니다. 음수는p - |n|으로 변환됩니다. -
Toxic waste:
groth16.setup()은 매 호출마다 새로운 랜덤 값을 생성합니다. 같은 circuit이라도 setup을 다시 하면 이전 proof는 무효화됩니다. -
Circuit 재사용: setup 후에는 같은 pk/vk로 여러 proof를 생성할 수 있습니다. 다른 witness(비밀 값)에 대해 반복 가능.
-
public input 순서:
verify(vk, proof, [a, b])에서 리스트 순서는c.public()호출 순서와 동일해야 합니다.
examples/zk_notebook_demo.py— 전체 워크플로우 데모- snarkjs로 생성한 proof를 py-ethclient에서 검증해 보세요
zk_verifyGroth16RPC 메서드로 원격 검증 서비스 구축- Part 4에서 L1↔L2 브릿지를 배워보세요
"L1에서 보낸 메시지가 L2의 EVM에서 실행되고, 오퍼레이터가 검열해도 사용자가 강제로 포함시킬 수 있다."
py-ethclient의 L2 브릿지는 Optimism의 CrossDomainMessenger 패턴을 따릅니다. 메신저가 유일한 프리미티브이고, 브릿지 컨트랙트(Token, State, ZK 등)는 메신저 위에 올라간 사용자 레벨 코드입니다.
┌─── L1 ────────────────────┐ ┌─── L2 ────────────────────┐
│ Store (MemoryBackend) │ │ Store (MemoryBackend) │
│ │ │ │
│ CrossDomainMessenger │ │ CrossDomainMessenger │
│ send_message() │ │ send_message() │
│ relay_message() │ │ relay_message() │
│ force_include() │ │ │
│ escape_hatch() │ │ │
└───────────┬───────────────┘ └───────────┬───────────────┘
│ │
└────────── BridgeWatcher ─────────────┘
(outbox 스캔 → relay)
핵심 개념:
| 개념 | 설명 |
|---|---|
| CrossDomainMessenger | 임의 메시지를 다른 도메인으로 전송. 릴레이 시 타겟 EVM에서 실행 |
| BridgeWatcher | 양쪽 outbox를 드레인하고 메시지를 릴레이하는 자동 릴레이어 |
| BridgeEnvironment | L1 Store + L2 Store + 2개 Messenger + Watcher를 묶은 편의 래퍼 |
| Force Inclusion | 오퍼레이터가 검열할 때 사용자가 L1에 등록 → 50블록 후 강제 릴레이 |
| Escape Hatch | L2가 완전히 무응답일 때 L1에서 입금 가치를 직접 복구 |
가장 기본적인 사용법: Alice가 L1에서 Bob에게 ETH를 보내고, watcher가 L2로 릴레이합니다.
from ethclient.bridge import BridgeEnvironment
ALICE = b"\x01" * 20
BOB = b"\x02" * 20
# L1 + L2 환경 생성
env = BridgeEnvironment()
# Alice가 L1에서 Bob에게 1000 wei 전송
msg = env.send_l1(sender=ALICE, target=BOB, value=1000)
# 아직 L2에 반영 안 됨
assert env.l2_balance(BOB) == 0
# Watcher가 릴레이
result = env.relay()
assert result.all_success
assert len(result.l1_to_l2) == 1
# L2에 반영됨
assert env.l2_balance(BOB) == 1000
print(f"Bob의 L2 잔액: {env.l2_balance(BOB)} wei")relay()가 내부적으로 하는 일:
- L1 messenger의 outbox에서 메시지를 꺼냄 (drain)
- 각 메시지를 L2 messenger의
relay_message()로 전달 relay_message()는 메시지를 L2의 EVM에서 실행- 코드가 없는 주소로 전송 시 → 단순 value transfer
- 코드가 있는 주소로 전송 시 → calldata를 EVM에서 실행
반대 방향(L2→L1)도 동일합니다:
# Bob이 L2에서 Alice에게 500 wei 출금
env.send_l2(sender=BOB, target=ALICE, value=500)
result = env.relay()
assert result.all_success
assert env.l1_balance(ALICE) == 500value transfer뿐 아니라, 임의 calldata를 L2 컨트랙트에 전달할 수 있습니다. 이것이 "General State Bridge"의 핵심입니다.
from ethclient.bridge import BridgeEnvironment
from ethclient.common.types import Account
from ethclient.common.crypto import keccak256
ALICE = b"\x01" * 20
ORACLE = b"\x0a" * 20
env = BridgeEnvironment()
# L2에 오라클 컨트랙트 배포
# 바이트코드: CALLDATALOAD(0) → SSTORE(slot=0, value)
# PUSH1 0x00 CALLDATALOAD PUSH1 0x00 SSTORE STOP
code = bytes([0x60, 0x00, 0x35, 0x60, 0x00, 0x55, 0x00])
acc = Account()
acc.code_hash = keccak256(code)
env.l2_store.put_account(ORACLE, acc)
env.l2_store.put_code(acc.code_hash, code)
# L1에서 오라클 가격을 L2로 전달 (ETH/USD = 1850)
price = (1850).to_bytes(32, "big")
env.send_l1(sender=ALICE, target=ORACLE, data=price)
result = env.relay()
assert result.all_success
# L2 오라클 스토리지에 가격이 기록됨
assert env.l2_storage(ORACLE, 0) == 1850
print(f"L2 오라클 가격: {env.l2_storage(ORACLE, 0)} USD")메시지가 릴레이될 때, relay_message()는 실제 EVM을 실행합니다:
msg.data가 calldata로 전달됨msg.target주소의 코드가 실행됨- SSTORE, SLOAD 등 모든 opcode 사용 가능
- 실행 성공 시 상태 변경이 커밋됨
L2 오퍼레이터가 정직하다면 모든 메시지가 정상 릴레이됩니다. 하지만 오퍼레이터가 특정 사용자의 메시지를 의도적으로 릴레이하지 않을 수 있습니다 (검열).
Force Inclusion은 이에 대한 해결책입니다:
from ethclient.bridge import BridgeEnvironment, FORCE_INCLUSION_WINDOW
ALICE = b"\x01" * 20
BOB = b"\x02" * 20
env = BridgeEnvironment()
# Alice가 L1→L2 메시지를 보냄
msg = env.send_l1(sender=ALICE, target=BOB, value=1000)
# 오퍼레이터가 검열: outbox에서 꺼내지만 릴레이하지 않음
env.l1_messenger.drain_outbox()
assert env.l2_balance(BOB) == 0 # 릴레이 안 됨
# Alice가 L1에 force inclusion 등록
entry = env.force_include(msg)
print(f"등록된 블록: {entry.registered_block}")
# 아직 윈도우가 안 지남 → 강제 릴레이 실패
result = env.force_relay(msg)
assert not result.success
print(f"너무 이름: {result.error}")
# 50블록 진행
env.advance_l1_block(FORCE_INCLUSION_WINDOW)
# 이제 강제 릴레이 성공!
result = env.force_relay(msg)
assert result.success
assert env.l2_balance(BOB) == 1000
print(f"강제 릴레이 성공! Bob의 L2 잔액: {env.l2_balance(BOB)}")Force Inclusion의 핵심:
- 누구나 등록할 수 있음 (L1 트랜잭션만 보내면 됨)
- 50블록(약 10분) 대기 후 누구나 force relay 가능
- 오퍼레이터의 협조가 전혀 필요 없음
- Watcher도 force queue를 자동 처리:
env.relay()가 eligible한 force inclusion을 자동 릴레이
Force relay도 실패하는 극단적 상황(L2가 완전히 다운)에서는 escape hatch로 L1에서 가치를 복구합니다.
env = BridgeEnvironment()
# Alice가 5000 wei 입금
msg = env.send_l1(sender=ALICE, target=BOB, value=5000)
env.l1_messenger.drain_outbox()
# Force include 등록
env.force_include(msg)
env.advance_l1_block(FORCE_INCLUSION_WINDOW)
# L2가 완전 무응답 → escape hatch로 L1에서 복구
result = env.escape_hatch(msg)
assert result.success
assert env.l1_balance(ALICE) == 5000 # 가치가 Alice에게 반환!
print(f"Escape hatch 성공! Alice의 L1 잔액: {env.l1_balance(ALICE)}")Escape hatch 제약:
- value > 0 인 메시지만 가능 (calldata-only 메시지는 불가)
- 이중 탈출 불가: 같은 메시지로 두 번 escape 불가
- force relay 후 불가: 이미 성공적으로 릴레이된 메시지는 escape 불가
- 가치는 msg.sender에게 반환됨 (msg.target이 아님)
아래를 bridge_tutorial.py로 저장하고 실행하세요:
"""L1↔L2 General State Bridge Tutorial"""
from ethclient.bridge import BridgeEnvironment, FORCE_INCLUSION_WINDOW
from ethclient.common.types import Account
from ethclient.common.crypto import keccak256
ALICE = b"\x01" * 20
BOB = b"\x02" * 20
ORACLE = b"\x0a" * 20
print("=" * 60)
print(" L1↔L2 General State Bridge Tutorial")
print("=" * 60)
# ━━━ 1. ETH 입금 ━━━
print("\n[1] ETH Deposit (L1→L2)")
env = BridgeEnvironment()
env.send_l1(sender=ALICE, target=BOB, value=1000)
result = env.relay()
assert result.all_success
assert env.l2_balance(BOB) == 1000
print(f" Bob L2 balance: {env.l2_balance(BOB)}")
# ━━━ 2. ETH 출금 ━━━
print("\n[2] ETH Withdraw (L2→L1)")
env.send_l2(sender=BOB, target=ALICE, value=500)
result = env.relay()
assert result.all_success
assert env.l1_balance(ALICE) == 500
print(f" Alice L1 balance: {env.l1_balance(ALICE)}")
# ━━━ 3. 상태 릴레이 ━━━
print("\n[3] State Relay (Oracle Price)")
env2 = BridgeEnvironment()
code = bytes([0x60, 0x00, 0x35, 0x60, 0x00, 0x55, 0x00])
acc = Account()
acc.code_hash = keccak256(code)
env2.l2_store.put_account(ORACLE, acc)
env2.l2_store.put_code(acc.code_hash, code)
price = (1850).to_bytes(32, "big")
env2.send_l1(sender=ALICE, target=ORACLE, data=price)
env2.relay()
assert env2.l2_storage(ORACLE, 0) == 1850
print(f" L2 oracle price: {env2.l2_storage(ORACLE, 0)} USD")
# ━━━ 4. Force Inclusion ━━━
print("\n[4] Force Inclusion (Anti-Censorship)")
env3 = BridgeEnvironment()
msg = env3.send_l1(sender=ALICE, target=BOB, value=777)
env3.l1_messenger.drain_outbox() # operator censors
assert env3.l2_balance(BOB) == 0
env3.force_include(msg)
env3.advance_l1_block(FORCE_INCLUSION_WINDOW)
result = env3.force_relay(msg)
assert result.success
assert env3.l2_balance(BOB) == 777
print(f" Force relay success! Bob L2: {env3.l2_balance(BOB)}")
# ━━━ 5. Escape Hatch ━━━
print("\n[5] Escape Hatch (Value Recovery)")
env4 = BridgeEnvironment()
msg = env4.send_l1(sender=ALICE, target=BOB, value=5000)
env4.l1_messenger.drain_outbox()
env4.force_include(msg)
env4.advance_l1_block(FORCE_INCLUSION_WINDOW)
result = env4.escape_hatch(msg)
assert result.success
assert env4.l1_balance(ALICE) == 5000
print(f" Escape success! Alice L1: {env4.l1_balance(ALICE)}")
# ━━━ 6. Replay Protection ━━━
print("\n[6] Replay Protection")
env5 = BridgeEnvironment()
env5.send_l1(sender=ALICE, target=BOB, value=100)
env5.relay()
# Same message can't be relayed twice
env5.send_l1(sender=ALICE, target=BOB, value=100)
env5.relay()
assert env5.l2_balance(BOB) == 200 # two different messages, both relayed
print(f" Two deposits: Bob L2 = {env5.l2_balance(BOB)}")
print(f"\n{'=' * 60}")
print(f" All checks passed!")
print(f"{'=' * 60}")python bridge_tutorial.py기본 브릿지는 L2에서 메시지를 EVM으로 실행합니다. 하지만 플러거블 릴레이 핸들러를 사용하면 L2가 EVM이 아닌 어떤 런타임이든 사용할 수 있습니다.
5가지 릴레이 모드:
| 핸들러 | 신뢰 모델 | EVM 필요 |
|---|---|---|
EVMRelayHandler |
On-chain 실행 (기본) | Yes |
MerkleProofHandler |
L1 상태 루트 Merkle proof | No |
ZKProofHandler |
Groth16 영지식 증명 | No |
TinyDBHandler |
문서 DB 백엔드 | No |
DirectStateHandler |
신뢰 릴레이어 | No |
신뢰 릴레이어가 상태를 직접 적용합니다. EVM 실행이 필요 없습니다.
from ethclient.bridge import BridgeEnvironment, StateUpdate, encode_state_updates
ALICE = b"\x01" * 20
BOB = b"\x02" * 20
# Direct state 모드로 환경 생성
env = BridgeEnvironment.with_direct_state()
# 상태 업데이트를 msg.data에 인코딩
updates = [
StateUpdate(address=ALICE, balance=5_000, nonce=1),
StateUpdate(address=BOB, balance=3_000, nonce=2),
]
data = encode_state_updates(updates)
# L1→L2 전송 + 릴레이
env.send_l1(sender=ALICE, target=BOB, data=data)
result = env.relay()
assert result.all_success
# 상태가 적용됨 (EVM 실행 없이)
assert env.l2_store.get_balance(ALICE) == 5_000
assert env.l2_store.get_balance(BOB) == 3_000L1의 상태 루트에 대한 Merkle proof를 검증한 후 상태를 적용합니다.
from ethclient.bridge import BridgeEnvironment
from ethclient.common import rlp
from ethclient.common.crypto import keccak256
from ethclient.common.trie import Trie
ALICE = b"\x01" * 20
BOB = b"\x02" * 20
EMPTY_ROOT = b"\x56\xe8\x1f\x17\x1b\xcc\x55\xa6\xff\x83\x45\xe6\x92\xc0\xf8\x6e\x5b\x48\xe0\x1b\x99\x6c\xad\xc0\x01\x62\x2f\xb5\xe3\x63\xb4\x21"
EMPTY_CODE = b"\xc5\xd2\x46\x01\x86\xf7\x23\x3c\x92\x7e\x7d\xb2\xdc\xc7\x03\xc0\xe5\x00\xb6\x53\xca\x82\x27\x3b\x7b\xfa\xd8\x04\x5d\x85\xa4\x70"
env = BridgeEnvironment.with_merkle_proof()
# L1 상태 설정
env.l1_store.set_balance(ALICE, 10_000)
env.l1_store.set_nonce(ALICE, 42)
# Merkle trie에서 proof 생성
trie = Trie()
for addr, acc in env.l1_store.iter_accounts():
account_rlp = rlp.encode([acc.nonce, acc.balance, EMPTY_ROOT, EMPTY_CODE])
trie.put_raw(keccak256(addr), account_rlp)
root = trie.root_hash
proof_nodes = trie.prove(keccak256(ALICE))
account_rlp = rlp.encode([42, 10_000, EMPTY_ROOT, EMPTY_CODE])
# 신뢰 루트 등록 + 전송
handler = env.l2_messenger.relay_handler
handler.add_trusted_root(root)
data = rlp.encode([root, ALICE, account_rlp, proof_nodes])
env.send_l1(sender=ALICE, target=BOB, data=data)
result = env.relay()
assert result.all_success
# Merkle proof로 검증된 상태가 L2에 적용됨
assert env.l2_store.get_balance(ALICE) == 10_000
assert env.l2_store.get_nonce(ALICE) == 42Groth16 증명을 검증한 후 상태를 적용합니다. 가장 강력한 신뢰 모델입니다.
from ethclient.bridge import BridgeEnvironment, StateUpdate
from ethclient.common import rlp
from ethclient.zk import Circuit, groth16
ALICE = b"\x01" * 20
BOB = b"\x02" * 20
# 1. Circuit 정의: old_balance + amount = new_balance
c = Circuit()
old_bal = c.public("old_balance")
amount = c.public("amount")
new_bal = c.public("new_balance")
one = c.private("one")
product = (old_bal + amount) * one
c.constrain(product, new_bal)
# 2. Trusted setup
pk, vk = groth16.setup(c)
# 3. Proof 생성
proof = groth16.prove(
pk,
private={"one": 1},
public={"old_balance": 1000, "amount": 500, "new_balance": 1500},
circuit=c,
)
# 4. msg.data 구성
proof_a = proof.a.to_evm_bytes()
proof_b = proof.b.to_evm_bytes()
proof_c = proof.c.to_evm_bytes()
public_inputs = [
(1000).to_bytes(32, "big"),
(500).to_bytes(32, "big"),
(1500).to_bytes(32, "big"),
]
updates = [StateUpdate(address=BOB, balance=1500)]
state_updates_rlp = [rlp.decode(u.encode()) for u in updates]
zk_data = rlp.encode([
proof_a, proof_b, proof_c,
public_inputs, state_updates_rlp,
])
# 5. 릴레이
env = BridgeEnvironment.with_zk_proof(vk)
env.send_l1(sender=ALICE, target=BOB, data=zk_data)
result = env.relay()
assert result.all_success
assert env.l2_store.get_balance(BOB) == 1500Proof 기반 릴레이의 핵심은 L2가 EVM이 아닌 어떤 런타임이든 사용할 수 있다는 것입니다. TinyDB 핸들러는 이를 증명합니다.
from ethclient.bridge import BridgeEnvironment, StateUpdate, TinyDBHandler, encode_state_updates
ALICE = b"\x01" * 20
BOB = b"\x02" * 20
# TinyDB 핸들러로 환경 생성
handler = TinyDBHandler()
env = BridgeEnvironment(l2_handler=handler)
# 상태 업데이트 (스토리지 포함)
updates = [
StateUpdate(address=ALICE, balance=7_777, nonce=10, storage={1: 42}),
StateUpdate(address=BOB, balance=3_333, storage={100: 200}),
]
data = encode_state_updates(updates)
env.send_l1(sender=ALICE, target=BOB, data=data)
result = env.relay()
assert result.all_success
# TinyDB에 JSON 문서로 저장됨 (Ethereum Store가 아닌!)
for doc in handler.db.all():
print(f" {doc['address'][:14]}... → balance={doc.get('balance')}")
# Ethereum Store는 비어 있음
assert env.l2_store.get_balance(ALICE) == 0전체 릴레이 모드 비교 데모:
python examples/bridge_relay_modes.py"상태 전이 함수를 Python 함수로 작성하면, 시퀀싱, 배치, Groth16 증명, L1 검증을 프레임워크가 처리한다."
롤업은 트랜잭션을 L2에서 실행하고, 그 결과를 L1에 증명하는 스케일링 기법입니다. py-ethclient의 L2 롤업 프레임워크는 애플리케이션 특화 접근법을 사용합니다 — EVM 블록을 재실행하는 대신, 개발자가 정의한 **State Transition Function(STF)**을 실행합니다.
┌─────────────────────────────────────────────────────────────┐
│ User Tx → Sequencer → STF 실행 → Batch 봉인 │
│ ↓ │
│ L1 검증 ← Groth16 Proof ← DA 저장 │
└─────────────────────────────────────────────────────────────┘
핵심 개념:
| 개념 | 설명 |
|---|---|
| STF | (state, tx) → output — 롤업의 비즈니스 로직. 일반 Python 함수 |
| Sequencer | 트랜잭션을 수집하고, 논스를 검증하고, STF를 실행하고, 배치를 조립 |
| Batch | 트랜잭션 모음 + old/new 상태 루트 + DA commitment |
| Groth16 Prover | 상태 전이(old_root → new_root)에 대한 ZK 증명 생성 |
| L1 Backend | 증명을 검증하고 새 상태 루트를 기록 |
4개 컴포넌트 모두 플러거블 인터페이스입니다 — 기본 구현이 제공되며, 필요하면 교체할 수 있습니다.
카운터를 증가시키는 가장 단순한 롤업입니다:
from ethclient.l2 import Rollup, L2Tx, L2TxType
# ━━━ 1. State Transition Function 정의 ━━━
# 일반 Python 함수입니다!
def counter_stf(state, tx):
"""카운터를 증가시키는 STF"""
count = state.get("count", 0)
if tx.data.get("action") == "increment":
state["count"] = count + 1
return {"new_count": count + 1}
# ━━━ 2. Rollup 생성 ━━━
rollup = Rollup(stf=counter_stf)
# ━━━ 3. Trusted Setup ━━━
# Groth16 circuit 빌드 + trusted setup + L1 verifier 배포
rollup.setup()
print(f"Setup 완료: {rollup.is_setup}")
# ━━━ 4. 트랜잭션 제출 ━━━
tx = L2Tx(
sender=b"\x01" * 20,
nonce=0,
data={"action": "increment"},
tx_type=L2TxType.CALL,
)
rollup.submit_tx(tx)
print(f"상태: count = {rollup.state.get('count', 0)}")
# → count = 1 (STF가 즉시 실행됨)
# ━━━ 5. 배치 생성 ━━━
batch = rollup.produce_batch()
print(f"Batch #{batch.number}: {len(batch.transactions)} txs")
print(f" old_root: {batch.old_state_root.hex()[:16]}...")
print(f" new_root: {batch.new_state_root.hex()[:16]}...")
# ━━━ 6. 증명 + L1 제출 ━━━
receipt = rollup.prove_and_submit(batch)
print(f"L1 검증: {'PASS' if receipt.verified else 'FAIL'}")
print(f"State root: {receipt.state_root.hex()[:16]}...")
assert receipt.verifiedRollup이 내부적으로 하는 일:
counter_stf함수를PythonRuntime으로 래핑Sequencer가 트랜잭션을 수신, 논스 체크 후 STF 실행produce_batch()가Sequencer.force_seal()호출 → Batch 생성prove_and_submit()가 Groth16 proof 생성 → L1 검증
좀 더 현실적인 예제 — 토큰 발행(mint)과 이체(transfer)를 지원하는 롤업:
from ethclient.l2 import Rollup, L2Tx, L2TxType
def balance_stf(state, tx):
"""잔액 mint + transfer STF"""
action = tx.data.get("action")
if action == "mint":
addr = tx.data["to"]
amount = tx.data["amount"]
state[addr] = state.get(addr, 0) + amount
return {"minted": amount, "to": addr}
elif action == "transfer":
src = tx.data["from"]
dst = tx.data["to"]
amount = tx.data["amount"]
# 잔액 부족 시 예외 → STF가 실패로 처리, 상태 롤백
if state.get(src, 0) < amount:
raise ValueError(f"insufficient balance: {state.get(src, 0)} < {amount}")
state[src] -= amount
state[dst] = state.get(dst, 0) + amount
return {"transferred": amount, "from": src, "to": dst}
# Rollup 생성 + setup
rollup = Rollup(stf=balance_stf)
rollup.setup()
# Alice에게 1000 토큰 발행
mint_tx = L2Tx(
sender=b"\x00" * 20, # minter
nonce=0,
data={"action": "mint", "to": "alice", "amount": 1000},
tx_type=L2TxType.CALL,
)
rollup.submit_tx(mint_tx)
print(f"Alice 잔액: {rollup.state.get('alice', 0)}") # 1000
# Alice → Bob 300 이체
transfer_tx = L2Tx(
sender=b"\x01" * 20,
nonce=0,
data={"action": "transfer", "from": "alice", "to": "bob", "amount": 300},
tx_type=L2TxType.CALL,
)
rollup.submit_tx(transfer_tx)
print(f"Alice: {rollup.state.get('alice', 0)}, Bob: {rollup.state.get('bob', 0)}")
# → Alice: 700, Bob: 300
# 잔액 부족 이체 시도 → STF 실패, 상태 롤백
bad_tx = L2Tx(
sender=b"\x01" * 20,
nonce=1,
data={"action": "transfer", "from": "bob", "to": "alice", "amount": 9999},
tx_type=L2TxType.CALL,
)
rollup.submit_tx(bad_tx)
# Bob 잔액은 여전히 300 (롤백됨)
print(f"실패 후 Bob: {rollup.state.get('bob', 0)}") # 300
# 배치 생성 + 증명 + L1 검증
batch = rollup.produce_batch()
receipt = rollup.prove_and_submit(batch)
assert receipt.verified
print(f"\nL1 검증 성공! Batch #{batch.number}")STF가 예외를 발생시키면 Sequencer가 자동으로 상태를 롤백합니다:
state_store.snapshot()→ STF 실행 → 예외 발생 →state_store.rollback()- 실패한 트랜잭션은 배치에 포함되지만, 상태에는 영향 없음
4개 인터페이스는 모두 교체 가능합니다:
from ethclient.l2 import Rollup, L2Config
from ethclient.l2.interfaces import (
StateTransitionFunction, DAProvider, L1Backend, ProofBackend,
)
# 커스텀 DA Provider 예시
class MyDAProvider(DAProvider):
def store_batch(self, batch_number: int, data: bytes) -> bytes:
# S3, IPFS, EigenDA 등에 저장
...
return commitment
def retrieve_batch(self, batch_number: int):
...
def verify_commitment(self, batch_number: int, commitment: bytes) -> bool:
...
# 커스텀 설정
config = L2Config(
name="my-dex-rollup",
chain_id=42170,
max_txs_per_batch=128, # 배치당 최대 트랜잭션
batch_timeout=30, # 초
rpc_port=9545,
)
# 조립
rollup = Rollup(
stf=my_stf_function,
da=MyDAProvider(),
config=config,
# l1, prover는 기본값 사용 (InMemoryL1Backend, Groth16ProofBackend)
)기본 제공 구현:
| 인터페이스 | 기본 구현 | 설명 |
|---|---|---|
StateTransitionFunction |
PythonRuntime |
callable을 STF로 래핑 |
DAProvider |
LocalDAProvider |
인메모리 dict, keccak256 commitment |
ProofBackend |
Groth16ProofBackend |
BN128 Groth16, 128-bit 필드 절삭 |
L1Backend |
InMemoryL1Backend |
groth16.verify로 직접 검증 |
새 롤업 프로젝트를 CLI로 빠르게 생성할 수 있습니다:
ethclient l2 init --name my-rollup생성되는 파일:
l2.json — 롤업 설정:
{
"name": "my-rollup",
"chain_id": 42170,
"max_txs_per_batch": 64,
"batch_timeout": 10,
"rpc_port": 9545
}stf.py — State Transition Function 템플릿:
def apply(state, tx):
"""State Transition Function — 이 함수를 수정하세요."""
action = tx.data.get("action")
if action == "set":
key = tx.data.get("key")
value = tx.data.get("value")
if key is not None:
state[key] = value
return {"key": key, "value": value}
return None이 파일을 수정해서 원하는 롤업 로직을 구현하면 됩니다.
아래를 rollup_tutorial.py로 저장하고 실행하세요:
"""Application-Specific ZK Rollup Tutorial
Counter STF → 멀티 배치 → 잔액 이체 → 실패 롤백 → L1 검증
"""
import time
from ethclient.l2 import Rollup, L2Tx, L2TxType
print("=" * 60)
print(" Application-Specific ZK Rollup Tutorial")
print("=" * 60)
# ━━━ 1. Counter STF ━━━
print("\n[1] Counter Rollup")
def counter_stf(state, tx):
count = state.get("count", 0)
if tx.data.get("action") == "increment":
state["count"] = count + 1
return {"new_count": count + 1}
rollup = Rollup(stf=counter_stf)
print(" Setting up (trusted setup + verifier deploy)...")
t0 = time.time()
rollup.setup()
print(f" Setup: {time.time() - t0:.1f}s")
# 3번 증가
for i in range(3):
tx = L2Tx(sender=b"\x01" * 20, nonce=i,
data={"action": "increment"}, tx_type=L2TxType.CALL)
rollup.submit_tx(tx)
assert rollup.state["count"] == 3
print(f" count = {rollup.state['count']} (3번 increment)")
# 배치 + 증명
batch = rollup.produce_batch()
print(f" Batch #{batch.number}: {len(batch.transactions)} txs")
t0 = time.time()
receipt = rollup.prove_and_submit(batch)
print(f" Prove + L1 verify: {time.time() - t0:.1f}s")
assert receipt.verified
print(f" L1 verified: YES")
# ━━━ 2. 멀티 배치 ━━━
print("\n[2] Multi-batch (chained state roots)")
for i in range(3, 6):
tx = L2Tx(sender=b"\x01" * 20, nonce=i,
data={"action": "increment"}, tx_type=L2TxType.CALL)
rollup.submit_tx(tx)
batch2 = rollup.produce_batch()
receipt2 = rollup.prove_and_submit(batch2)
assert receipt2.verified
assert rollup.state["count"] == 6
print(f" count = {rollup.state['count']} (batch #0 → #1)")
print(f" Batch #{batch2.number} L1 verified: YES")
# ━━━ 3. 잔액 이체 ━━━
print("\n[3] Balance Transfer Rollup")
def balance_stf(state, tx):
action = tx.data.get("action")
if action == "mint":
addr = tx.data["to"]
amount = tx.data["amount"]
state[addr] = state.get(addr, 0) + amount
return {"minted": amount}
elif action == "transfer":
src, dst = tx.data["from"], tx.data["to"]
amount = tx.data["amount"]
if state.get(src, 0) < amount:
raise ValueError("insufficient balance")
state[src] -= amount
state[dst] = state.get(dst, 0) + amount
return {"transferred": amount}
rollup2 = Rollup(stf=balance_stf)
rollup2.setup()
# Mint 1000 to Alice
rollup2.submit_tx(L2Tx(
sender=b"\x00" * 20, nonce=0,
data={"action": "mint", "to": "alice", "amount": 1000},
tx_type=L2TxType.CALL,
))
print(f" Mint: alice = {rollup2.state.get('alice', 0)}")
# Transfer 300 from Alice to Bob
rollup2.submit_tx(L2Tx(
sender=b"\x01" * 20, nonce=0,
data={"action": "transfer", "from": "alice", "to": "bob", "amount": 300},
tx_type=L2TxType.CALL,
))
print(f" Transfer: alice = {rollup2.state.get('alice', 0)}, bob = {rollup2.state.get('bob', 0)}")
# Insufficient balance → rollback
rollup2.submit_tx(L2Tx(
sender=b"\x01" * 20, nonce=1,
data={"action": "transfer", "from": "bob", "to": "alice", "amount": 9999},
tx_type=L2TxType.CALL,
))
assert rollup2.state.get("bob", 0) == 300 # unchanged!
print(f" Failed transfer: bob still = {rollup2.state.get('bob', 0)} (rollback)")
# Batch + prove
batch3 = rollup2.produce_batch()
receipt3 = rollup2.prove_and_submit(batch3)
assert receipt3.verified
print(f" L1 verified: YES")
# ━━━ 4. Chain Info ━━━
print("\n[4] Chain Info")
info = rollup2.chain_info()
print(f" name: {info['name']}")
print(f" chain_id: {info['chain_id']}")
print(f" batches: {info['sealed_batches']}")
print(f"\n{'=' * 60}")
print(f" All checks passed!")
print(f"{'=' * 60}")python rollup_tutorial.py다음 단계:
StateTransitionFunction을 직접 구현해 보세요 (ABC 상속)DAProvider를 파일 시스템이나 외부 저장소로 교체해 보세요l2_*RPC 메서드를 사용해 HTTP로 롤업과 상호작용해 보세요- Part 6에서 프로덕션 배포를 배워보세요
"개발 모드에서 한 줄도 안 바꾸고
L2Config만 교체하면 프로덕션으로 전환된다."
Part 5에서는 인메모리 백엔드로 롤업의 기본 동작을 배웠습니다. Part 6에서는 실제 프로덕션 환경에서 필요한 모든 컴포넌트를 다룹니다.
개발 모드와 프로덕션 모드의 차이는 L2Config의 4가지 백엔드 설정뿐입니다:
개발 (Part 5) 프로덕션 (Part 6)
───────────── ──────────────────
state_backend = "memory" → state_backend = "lmdb"
da_provider = "local" → da_provider = "blob" | "calldata" | "s3"
prover_backend = "python" → prover_backend = "native"
l1_backend = "memory" → l1_backend = "eth_rpc"
전체 프로덕션 스택:
┌──────────────────────────────────────────────────────────────────────┐
│ Client │
│ curl -H "X-API-Key: ..." http://localhost:9545/ │
└──────────┬───────────────────────────────────────────────────────────┘
│ HTTP
┌──────────▼───────────────────────────────────────────────────────────┐
│ RPC Server (FastAPI + uvicorn) │
│ │
│ ┌─ Middleware ──────────────────────────────────────────────────┐ │
│ │ APIKeyMiddleware → RateLimitMiddleware → RequestSizeLimit │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─ Endpoints ──────────────────────────────────────────────────┐ │
│ │ l2_submitTx │ l2_getState │ l2_getBatch │ /health │ /metrics│ │
│ └──────────────────────────────────────────────────────────────┘ │
└──────────┬───────────────────────────────────────────────────────────┘
│
┌──────────▼───────────────────────────────────────────────────────────┐
│ Rollup Orchestrator │
│ │
│ ┌─ Sequencer ────┐ ┌─ DA Provider ────┐ ┌─ Prover ────────────┐ │
│ │ Mempool │ │ S3 / Calldata / │ │ rapidsnark │ │
│ │ Nonce tracking │ │ Blob (EIP-4844) │ │ (Python fallback) │ │
│ │ Auto-seal │ └──────────────────┘ └─────────────────────┘ │
│ └────────────────┘ │
│ │
│ ┌─ State Store ──┐ ┌─ L1 Backend ─────────────────────────────┐ │
│ │ LMDB │ │ EthL1Backend (EIP-1559 tx) │ │
│ │ Overlay + WAL │ │ EVMVerifier 컨트랙트 배포 + 검증 │ │
│ └────────────────┘ └──────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────────┘
개발 모드의 L2StateStore는 인메모리 dict입니다 — 프로세스가 종료되면 모든 상태가 사라집니다. L2PersistentStateStore는 LMDB를 사용하여 상태를 디스크에 영속화합니다.
from ethclient.l2 import Rollup, L2Config
# ━━━ LMDB 활성화 — config 한 줄 ━━━
config = L2Config(
state_backend="lmdb", # "memory" → "lmdb"
data_dir="./my-rollup-data",
)
def counter_stf(state, tx):
count = state.get("count", 0)
if tx.data.get("action") == "increment":
state["count"] = count + 1
return {"new_count": count + 1}
rollup = Rollup(stf=counter_stf, config=config)
rollup.setup()STF 코드는 한 줄도 바꿀 필요 없습니다. state["count"] 같은 dict 인터페이스가 그대로 동작합니다.
내부 동작:
┌── STF ───────────────────────────────────┐
│ state["count"] = 42 │ ← 일반 dict 문법
└──────────────────────────────────────────┘
│
▼
┌── L2PersistentState (overlay dict) ──────┐
│ overlay = {"count": 42} │ ← 메모리에서 빠르게 처리
│ │
│ state.get("key") │
│ → overlay에 있으면 반환 │
│ → 없으면 LMDB에서 read-through │
└──────────────────────────────────────────┘
│ flush()
▼
┌── LMDB (디스크) ─────────────────────────┐
│ l2_state — 상태 키-값 │
│ l2_batches — 봉인된 배치 │
│ l2_proofs — 증명 데이터 │
│ l2_meta — 카운터, 논스 등 │
│ l2_wal — Write-Ahead Log │
└──────────────────────────────────────────┘
Snapshot/Rollback:
STF가 예외를 발생시키면 Sequencer가 자동으로 상태를 롤백합니다. LMDB 모드에서도 동일하게 동작합니다:
# Sequencer 내부 (자동으로 처리됨)
snapshot_id = state_store.snapshot() # overlay 복사본 저장
try:
stf(state, tx) # STF 실행
state_store.commit() # 성공 시 snapshot 폐기
except Exception:
state_store.rollback(snapshot_id) # 실패 시 overlay 복원배치 데이터를 어디에 저장할 것인가? 3가지 프로덕션 DA Provider가 제공됩니다.
| Provider | 비용 | 가용성 | 적합한 용도 |
|---|---|---|---|
S3DAProvider |
AWS 비용 | 높음 (S3 SLA) | 프라이빗 롤업, 테스트 |
CalldataDAProvider |
L1 gas (16 gas/byte) | 영구 (L1 history) | 소규모 배치, 강한 보증 |
BlobDAProvider |
L1 blob gas (저렴) | ~18일 (pruned) | 대규모 배치, 비용 최적화 |
config = L2Config(
da_provider="s3",
s3_bucket="my-rollup-batches",
s3_prefix="mainnet/batches/",
s3_region="ap-northeast-2",
# s3_endpoint_url="http://localhost:9000", # MinIO 등 호환 스토리지
)배치 데이터가 S3에 mainnet/batches/00000000, mainnet/batches/00000001, ... 형태로 저장됩니다. Commitment은 keccak256(batch_number || data)입니다.
config = L2Config(
da_provider="calldata",
l1_rpc_url="https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY",
l1_private_key="abcdef...", # hex, 0x 없이
l1_chain_id=1,
)배치 데이터가 EIP-1559 (type-2) 트랜잭션의 calldata로 L1에 게시됩니다.
- 장점: L1에 영구 기록, 가장 강한 DA 보증
- 단점: 비용이 높음 (16 gas/byte 비zero + 4 gas/byte zero)
- Gas 자동 추정:
21000 + calldata_gas + 5000 overhead
config = L2Config(
da_provider="blob",
l1_rpc_url="https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY",
l1_private_key="abcdef...",
l1_chain_id=1,
beacon_url="http://localhost:5052", # Beacon API for blob retrieval
)EIP-4844 blob 트랜잭션(type-3)으로 데이터를 게시합니다.
- 장점: calldata 대비 10~100배 저렴
- 단점: blob 데이터는 ~18일 후 pruning됨 (이후 beacon archive 노드 필요)
- 최대 blob 크기: 126,972 bytes (4096 필드 엘리먼트 × 31 usable bytes - 4 byte header)
- KZG commitment + proof가 자동 생성됩니다 (ckzg 라이브러리)
Blob 인코딩 구조:
┌─ Blob (131,072 bytes = 4096 × 32) ──────────────────────────┐
│ Element 0: [0x00] [4-byte length header] [27 bytes data] │
│ Element 1: [0x00] [31 bytes data] │
│ Element 2: [0x00] [31 bytes data] │
│ ... │
│ Element N: [0x00] [31 bytes data + padding] │
│ Element N+1..4095: [0x00] [31 bytes zero padding] │
└──────────────────────────────────────────────────────────────┘
각 32-byte 필드 엘리먼트의 첫 바이트는 0x00 (BLS12-381 필드 안전)
순수 Python Groth16 prover는 교육/프로토타이핑에 적합하지만, 프로덕션에서는 C++/WASM 기반 prover가 필요합니다.
NativeProverBackend는 rapidsnark 또는 snarkjs를 subprocess로 실행하고, 실패 시 자동으로 Python fallback합니다.
config = L2Config(
prover_backend="native",
prover_binary="rapidsnark", # rapidsnark 바이너리 경로
prover_working_dir="./proofs", # R1CS, zkey, witness 파일 디렉토리
)Setup 과정 (자동):
1. Circuit 생성 (max_txs_per_batch 개의 tx signal)
2. R1CS 바이너리 내보내기 → circuit.r1cs
3. snarkjs powersoftau (Phase 1) → pot.ptau
4. snarkjs groth16 setup (Phase 2) → circuit.zkey + vkey.json
5. 실패 시 → Python groth16.setup() fallback
Prove 과정:
1. Witness 생성 → witness.json
2. rapidsnark circuit.zkey witness.json proof.json public.json
(또는 snarkjs groth16 prove ...)
3. proof.json 파싱 → Proof 객체
4. 실패 시 → Python groth16.prove() fallback
Verify는 항상 Python에서 실행됩니다 (충분히 빠르고, 외부 의존성 불필요).
참고: rapidsnark은
brew install rapidsnark또는 소스 빌드로 설치합니다. snarkjs는npm install -g snarkjs로 설치합니다. 둘 다 없어도 Python fallback이 동작합니다.
InMemoryL1Backend는 groth16.verify()를 직접 호출합니다. EthL1Backend는 실제 이더리움 네트워크에 verifier 컨트랙트를 배포하고 on-chain 검증을 수행합니다.
config = L2Config(
l1_backend="eth_rpc",
l1_rpc_url="https://eth-sepolia.g.alchemy.com/v2/YOUR_KEY",
l1_private_key="abcdef...",
l1_chain_id=11155111, # Sepolia
)동작 흐름:
rollup.setup()
│
├─ prover.setup() # Groth16 trusted setup
│
└─ l1.deploy_verifier(vk) # Verifier 컨트랙트 배포
│
├─ EVMVerifier(vk).bytecode # vk 하드코딩된 바이트코드 생성
├─ EIP-1559 tx (to=None, data=bytecode)
├─ gas_limit = 5,000,000
└─ → contract_address (20 bytes)
rollup.prove_and_submit(batch)
│
├─ prover.prove(...) # Groth16 proof 생성
│
└─ l1.submit_batch(...) # On-chain 검증
│
├─ EVMVerifier.encode_calldata(proof, public_inputs)
├─ EIP-1559 tx (to=verifier_contract, data=calldata)
├─ gas_limit = 500,000
└─ receipt.status == 1 → verified!
공개 입력: [old_state_root, new_state_root, tx_commitment] — 3개의 256-bit 정수가 verifier 컨트랙트에 전달됩니다.
ethclient l2 start 명령은 FastAPI 기반 RPC 서버를 시작합니다. 3개 미들웨어가 자동으로 장착됩니다.
API Key 인증:
config = L2Config(
api_keys=["key-alpha-1234", "key-beta-5678"],
)# 인증된 요청
curl -H "X-API-Key: key-alpha-1234" \
-d '{"method": "l2_submitTx", ...}' \
http://localhost:9545/
# 쿼리 파라미터도 가능
curl http://localhost:9545/?api_key=key-alpha-1234
# 인증 실패 → 401
curl http://localhost:9545/
# {"error": "Invalid or missing API key"}/health, /ready, /metrics 엔드포인트는 인증 없이 접근 가능합니다.
Rate Limiting (토큰 버킷):
config = L2Config(
rate_limit_rps=10.0, # 초당 10 요청
rate_limit_burst=50, # 버스트 허용량 50
)IP별로 토큰 버킷이 생성됩니다. 초당 rps개의 토큰이 채워지고, 최대 burst개까지 누적됩니다. 토큰이 부족하면 429 응답:
{"error": "Rate limit exceeded"}Request Size Limit:
config = L2Config(
max_request_size=1_048_576, # 1 MB (기본값)
)Content-Length가 초과하면 413 응답:
{"error": "Request too large (max 1048576 bytes)"}Health/Ready/Metrics 엔드포인트:
# Health check (항상 200)
curl http://localhost:9545/health
# {"status": "ok"}
# Readiness (setup 완료 시 200, 아니면 503)
curl http://localhost:9545/ready
# {"status": "ready", "chain_id": 42170, "state_root": "0x...",
# "pending_txs": 5, "sealed_batches": 3}
# Metrics (JSON)
curl http://localhost:9545/metrics
# {"l2_mempool_size": 5, "l2_sealed_batches_total": 3,
# "l2_proven_batches_total": 2, "l2_submitted_batches_total": 1, ...}CLI 4개 명령으로 전체 라이프사이클을 관리합니다:
# ━━━ 1. 프로젝트 생성 ━━━
ethclient l2 init --name my-dex-rollup
# 생성되는 파일:
# l2.json — 롤업 설정 (L2Config 필드)
# stf.py — State Transition Function 템플릿
# ━━━ 2. 설정 수정 ━━━
# l2.json을 열어 프로덕션 설정으로 변경:{
"name": "my-dex-rollup",
"chain_id": 42170,
"max_txs_per_batch": 128,
"batch_timeout": 30,
"rpc_port": 9545,
"state_backend": "lmdb",
"data_dir": "./data",
"da_provider": "blob",
"l1_rpc_url": "https://eth-sepolia.g.alchemy.com/v2/KEY",
"l1_private_key": "YOUR_PRIVATE_KEY_HEX",
"l1_chain_id": 11155111,
"beacon_url": "http://localhost:5052",
"prover_backend": "native",
"prover_binary": "rapidsnark",
"l1_backend": "eth_rpc",
"api_keys": ["prod-key-1234"],
"rate_limit_rps": 50.0,
"rate_limit_burst": 200,
"enable_metrics": true
}# ━━━ 3. stf.py 수정 ━━━
# 비즈니스 로직 구현 (Part 5의 balance_stf 등)
# ━━━ 4. 시퀀서 + RPC 서버 시작 ━━━
ethclient l2 start --config l2.json
# → Trusted setup 수행
# → L1 verifier 컨트랙트 배포
# → Middleware 장착
# → uvicorn 0.0.0.0:9545 시작
# ━━━ 5. 증명 생성 (별도 프로세스) ━━━
ethclient l2 prove --config l2.json
# → 봉인된 배치를 순회하며 Groth16 proof 생성
# ━━━ 6. L1 제출 (별도 프로세스) ━━━
ethclient l2 submit --config l2.json
# → 증명 완료된 배치를 L1에 제출
# → On-chain 검증 결과 출력운영 패턴: start는 상시 실행, prove와 submit은 cron이나 별도 worker로 주기적 실행합니다.
LMDB 모드에서는 **Write-Ahead Log (WAL)**를 통해 크래시 복구가 가능합니다.
rollup = Rollup(stf=my_stf, config=config)
# 프로세스 재시작 후 복구
rollup.recover()WAL에 기록되는 이벤트:
| entry_type | 시점 | 데이터 |
|---|---|---|
tx_applied |
STF 실행 성공 후 | 트랜잭션 정보 |
batch_sealed |
배치 봉인 후 | 배치 번호 + 상태 루트 |
batch_proven |
증명 생성 후 | 배치 번호 + proof |
batch_submitted |
L1 제출 후 | 배치 번호 + tx_hash |
recover()는 WAL을 순서대로 재생하여 마지막 일관된 상태로 복원합니다. 복구 완료 후 WAL은 truncate됩니다.
개발과 프로덕션의 전체 비교:
from ethclient.l2 import Rollup, L2Config
# ━━━ STF는 동일 — 환경에 무관 ━━━
def balance_stf(state, tx):
action = tx.data.get("action")
if action == "mint":
addr = tx.data["to"]
state[addr] = state.get(addr, 0) + tx.data["amount"]
return {"minted": tx.data["amount"]}
elif action == "transfer":
src, dst = tx.data["from"], tx.data["to"]
amount = tx.data["amount"]
if state.get(src, 0) < amount:
raise ValueError("insufficient balance")
state[src] -= amount
state[dst] = state.get(dst, 0) + amount
return {"transferred": amount}
# ━━━ 개발 모드 (Part 5와 동일) ━━━
dev_rollup = Rollup(stf=balance_stf)
# ━━━ 프로덕션 모드 — config만 교체 ━━━
prod_config = L2Config(
name="my-dex-rollup",
chain_id=42170,
max_txs_per_batch=128,
batch_timeout=30,
rpc_port=9545,
# 영속 상태
state_backend="lmdb",
data_dir="./data",
# Blob DA (EIP-4844)
da_provider="blob",
l1_rpc_url="https://eth-sepolia.g.alchemy.com/v2/KEY",
l1_private_key="YOUR_HEX_KEY",
l1_chain_id=11155111,
beacon_url="http://localhost:5052",
# Native prover
prover_backend="native",
prover_binary="rapidsnark",
prover_working_dir="./proofs",
# 실제 L1
l1_backend="eth_rpc",
# 미들웨어
api_keys=["prod-key-1234", "prod-key-5678"],
rate_limit_rps=50.0,
rate_limit_burst=200,
max_request_size=2_097_152, # 2 MB
enable_metrics=True,
)
prod_rollup = Rollup(stf=balance_stf, config=prod_config)백엔드 자동 선택 요약:
| Config 필드 | 개발 (기본값) | 프로덕션 옵션 |
|---|---|---|
state_backend |
"memory" → L2StateStore |
"lmdb" → L2PersistentStateStore |
da_provider |
"local" → LocalDAProvider |
"s3" / "calldata" / "blob" |
prover_backend |
"python" → Groth16ProofBackend |
"native" → NativeProverBackend |
l1_backend |
"memory" → InMemoryL1Backend |
"eth_rpc" → EthL1Backend |
핵심: STF 코드는 환경에 독립적입니다. 같은
balance_stf함수가 인메모리 테스트에서도, Sepolia에서도, 메인넷에서도 동일하게 동작합니다. 이것이 py-ethclient L2 프레임워크의 플러거블 아키텍처가 제공하는 가치입니다.