247 lines
7.2 KiB
Python
247 lines
7.2 KiB
Python
"""
|
|
BLS test vectors generator
|
|
"""
|
|
|
|
from typing import Tuple
|
|
|
|
from eth_utils import (
|
|
encode_hex,
|
|
int_to_big_endian,
|
|
to_tuple,
|
|
)
|
|
from gen_base import gen_runner, gen_suite, gen_typing
|
|
|
|
from py_ecc import bls
|
|
|
|
|
|
F2Q_COEFF_LEN = 48
|
|
G2_COMPRESSED_Z_LEN = 48
|
|
|
|
|
|
def int_to_hex(n: int, byte_length: int=None) -> str:
|
|
byte_value = int_to_big_endian(n)
|
|
if byte_length:
|
|
byte_value = byte_value.rjust(byte_length, b'\x00')
|
|
return encode_hex(byte_value)
|
|
|
|
|
|
def hex_to_int(x: str) -> int:
|
|
return int(x, 16)
|
|
|
|
|
|
DOMAINS = [
|
|
b'\x00\x00\x00\x00\x00\x00\x00\x00',
|
|
b'\x00\x00\x00\x00\x00\x00\x00\x01',
|
|
b'\xff\xff\xff\xff\xff\xff\xff\xff'
|
|
]
|
|
|
|
MESSAGES = [
|
|
bytes(b'\x00' * 32),
|
|
bytes(b'\x56' * 32),
|
|
bytes(b'\xab' * 32),
|
|
]
|
|
|
|
PRIVKEYS = [
|
|
# Curve order is 256 so private keys are 32 bytes at most.
|
|
# Also not all integers is a valid private key, so using pre-generated keys
|
|
hex_to_int('0x00000000000000000000000000000000263dbd792f5b1be47ed85f8938c0f29586af0d3ac7b977f21c278fe1462040e3'),
|
|
hex_to_int('0x0000000000000000000000000000000047b8192d77bf871b62e87859d653922725724a5c031afeabc60bcef5ff665138'),
|
|
hex_to_int('0x00000000000000000000000000000000328388aff0d4a5b7dc9205abd374e7e98f3cd9f3418edb4eafda5fb16473d216'),
|
|
]
|
|
|
|
|
|
def hash_message(msg: bytes,
|
|
domain: bytes) ->Tuple[Tuple[str, str], Tuple[str, str], Tuple[str, str]]:
|
|
"""
|
|
Hash message
|
|
Input:
|
|
- Message as bytes32
|
|
- domain as bytes8
|
|
Output:
|
|
- Message hash as a G2 point
|
|
"""
|
|
return [
|
|
[
|
|
int_to_hex(fq2.coeffs[0], F2Q_COEFF_LEN),
|
|
int_to_hex(fq2.coeffs[1], F2Q_COEFF_LEN),
|
|
]
|
|
for fq2 in bls.utils.hash_to_G2(msg, domain)
|
|
]
|
|
|
|
|
|
def hash_message_compressed(msg: bytes, domain: bytes) -> Tuple[str, str]:
|
|
"""
|
|
Hash message
|
|
Input:
|
|
- Message as bytes32
|
|
- domain as bytes8
|
|
Output:
|
|
- Message hash as a compressed G2 point
|
|
"""
|
|
z1, z2 = bls.utils.compress_G2(bls.utils.hash_to_G2(msg, domain))
|
|
return [int_to_hex(z1, G2_COMPRESSED_Z_LEN), int_to_hex(z2, G2_COMPRESSED_Z_LEN)]
|
|
|
|
|
|
@to_tuple
|
|
def case01_message_hash_G2_uncompressed():
|
|
for msg in MESSAGES:
|
|
for domain in DOMAINS:
|
|
yield {
|
|
'input': {
|
|
'message': encode_hex(msg),
|
|
'domain': encode_hex(domain),
|
|
},
|
|
'output': hash_message(msg, domain)
|
|
}
|
|
|
|
@to_tuple
|
|
def case02_message_hash_G2_compressed():
|
|
for msg in MESSAGES:
|
|
for domain in DOMAINS:
|
|
yield {
|
|
'input': {
|
|
'message': encode_hex(msg),
|
|
'domain': encode_hex(domain),
|
|
},
|
|
'output': hash_message_compressed(msg, domain)
|
|
}
|
|
|
|
@to_tuple
|
|
def case03_private_to_public_key():
|
|
pubkeys = [bls.privtopub(privkey) for privkey in PRIVKEYS]
|
|
pubkeys_serial = ['0x' + pubkey.hex() for pubkey in pubkeys]
|
|
for privkey, pubkey_serial in zip(PRIVKEYS, pubkeys_serial):
|
|
yield {
|
|
'input': int_to_hex(privkey),
|
|
'output': pubkey_serial,
|
|
}
|
|
|
|
@to_tuple
|
|
def case04_sign_messages():
|
|
for privkey in PRIVKEYS:
|
|
for message in MESSAGES:
|
|
for domain in DOMAINS:
|
|
sig = bls.sign(message, privkey, domain)
|
|
yield {
|
|
'input': {
|
|
'privkey': int_to_hex(privkey),
|
|
'message': encode_hex(message),
|
|
'domain': encode_hex(domain),
|
|
},
|
|
'output': encode_hex(sig)
|
|
}
|
|
|
|
# TODO: case05_verify_messages: Verify messages signed in case04
|
|
# It takes too long, empty for now
|
|
|
|
|
|
@to_tuple
|
|
def case06_aggregate_sigs():
|
|
for domain in DOMAINS:
|
|
for message in MESSAGES:
|
|
sigs = [bls.sign(message, privkey, domain) for privkey in PRIVKEYS]
|
|
yield {
|
|
'input': [encode_hex(sig) for sig in sigs],
|
|
'output': encode_hex(bls.aggregate_signatures(sigs)),
|
|
}
|
|
|
|
@to_tuple
|
|
def case07_aggregate_pubkeys():
|
|
pubkeys = [bls.privtopub(privkey) for privkey in PRIVKEYS]
|
|
pubkeys_serial = [encode_hex(pubkey) for pubkey in pubkeys]
|
|
yield {
|
|
'input': pubkeys_serial,
|
|
'output': encode_hex(bls.aggregate_pubkeys(pubkeys)),
|
|
}
|
|
|
|
|
|
# TODO
|
|
# Aggregate verify
|
|
|
|
# TODO
|
|
# Proof-of-possession
|
|
|
|
|
|
def bls_msg_hash_uncompressed_suite(configs_path: str) -> gen_typing.TestSuiteOutput:
|
|
return ("g2_uncompressed", "msg_hash_g2_uncompressed", gen_suite.render_suite(
|
|
title="BLS G2 Uncompressed msg hash",
|
|
summary="BLS G2 Uncompressed msg hash",
|
|
forks_timeline="mainnet",
|
|
forks=["phase0"],
|
|
config="mainnet",
|
|
runner="bls",
|
|
handler="msg_hash_uncompressed",
|
|
test_cases=case01_message_hash_G2_uncompressed()))
|
|
|
|
|
|
def bls_msg_hash_compressed_suite(configs_path: str) -> gen_typing.TestSuiteOutput:
|
|
return ("g2_compressed", "msg_hash_g2_compressed", gen_suite.render_suite(
|
|
title="BLS G2 Compressed msg hash",
|
|
summary="BLS G2 Compressed msg hash",
|
|
forks_timeline="mainnet",
|
|
forks=["phase0"],
|
|
config="mainnet",
|
|
runner="bls",
|
|
handler="msg_hash_compressed",
|
|
test_cases=case02_message_hash_G2_compressed()))
|
|
|
|
|
|
|
|
def bls_priv_to_pub_suite(configs_path: str) -> gen_typing.TestSuiteOutput:
|
|
return ("priv_to_pub", "priv_to_pub", gen_suite.render_suite(
|
|
title="BLS private key to pubkey",
|
|
summary="BLS Convert private key to public key",
|
|
forks_timeline="mainnet",
|
|
forks=["phase0"],
|
|
config="mainnet",
|
|
runner="bls",
|
|
handler="priv_to_pub",
|
|
test_cases=case03_private_to_public_key()))
|
|
|
|
|
|
def bls_sign_msg_suite(configs_path: str) -> gen_typing.TestSuiteOutput:
|
|
return ("sign_msg", "sign_msg", gen_suite.render_suite(
|
|
title="BLS sign msg",
|
|
summary="BLS Sign a message",
|
|
forks_timeline="mainnet",
|
|
forks=["phase0"],
|
|
config="mainnet",
|
|
runner="bls",
|
|
handler="sign_msg",
|
|
test_cases=case04_sign_messages()))
|
|
|
|
|
|
def bls_aggregate_sigs_suite(configs_path: str) -> gen_typing.TestSuiteOutput:
|
|
return ("aggregate_sigs", "aggregate_sigs", gen_suite.render_suite(
|
|
title="BLS aggregate sigs",
|
|
summary="BLS Aggregate signatures",
|
|
forks_timeline="mainnet",
|
|
forks=["phase0"],
|
|
config="mainnet",
|
|
runner="bls",
|
|
handler="aggregate_sigs",
|
|
test_cases=case06_aggregate_sigs()))
|
|
|
|
|
|
def bls_aggregate_pubkeys_suite(configs_path: str) -> gen_typing.TestSuiteOutput:
|
|
return ("aggregate_pubkeys", "aggregate_pubkeys", gen_suite.render_suite(
|
|
title="BLS aggregate pubkeys",
|
|
summary="BLS Aggregate public keys",
|
|
forks_timeline="mainnet",
|
|
forks=["phase0"],
|
|
config="mainnet",
|
|
runner="bls",
|
|
handler="aggregate_pubkeys",
|
|
test_cases=case07_aggregate_pubkeys()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
gen_runner.run_generator("bls", [
|
|
bls_msg_hash_compressed_suite,
|
|
bls_msg_hash_uncompressed_suite,
|
|
bls_priv_to_pub_suite,
|
|
bls_sign_msg_suite,
|
|
bls_aggregate_sigs_suite,
|
|
bls_aggregate_pubkeys_suite
|
|
])
|