Page MenuHomePhabricator
Differential D9350 Diff 28021 test/functional/abc_rpc_avalancheproof.py

Changeset View

Standalone View

View Options

test/functional/abc_rpc_avalancheproof.py

  • This file was added.
#!/usr/bin/env python3
# Copyright (c) 2021 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test building avalanche proofs and using them to add avalanche peers."""
from typing import List, Dict
from test_framework.key import ECKey
from test_framework.mininode import P2PInterface
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
assert_equal,
assert_raises_rpc_error,
)
AVALANCHE_MAX_PROOF_STAKES = 1000
def get_stakes(coinbases: List[Dict],
priv_key: str) -> List[Dict]:
return [{
'txid': coinbase['txid'],
'vout': coinbase['n'],
'amount': coinbase['value'],
'height': coinbase['height'],
'iscoinbase': True,
'privatekey': priv_key,
} for coinbase in coinbases]
class AvalancheProofTest(BitcoinTestFramework):
def set_test_params(self):
self.setup_clean_chain = True
self.num_nodes = 1
self.extra_args = [['-enableavalanche=1', '-avacooldown=0'], ]
def run_test(self):
node = self.nodes[0]
def get_interface_node():
"""Return a P2P interface node signaling the avalanche flag."""
n = P2PInterface()
node.add_p2p_connection(n)
n.wait_for_verack()
n.nodeid = node.getpeerinfo()[-1]['id']
return n
def get_coinbase(h):
b = node.getblock(h, 2)
return {
'height': b['height'],
'txid': b['tx'][0]['txid'],
'n': 0,
'value': b['tx'][0]['vout'][0]['value'],
}
addrkey0 = node.get_deterministic_priv_key()
blocks = node.generatetoaddress(100, addrkey0.address)
coinbases = [get_coinbase(h) for h in blocks]
self.log.info(
"Make build a valid proof and restart the node to use it")
privkey = ECKey()
privkey.set(bytes.fromhex(
"12b004fff7f4b69ef8650e767f18f11ede158148b425660723b9f9a66e61f747"), True)
proof_master = privkey.get_pubkey().get_bytes().hex()
proof_sequence = 11
proof_expiration = 12
proof = node.buildavalancheproof(
proof_sequence, proof_expiration, proof_master,
[{
'txid': coinbases[0]['txid'],
'vout': coinbases[0]['n'],
'amount': coinbases[0]['value'],
'height': coinbases[0]['height'],
'iscoinbase': True,
'privatekey': addrkey0.key,
}])
# Restart the node, making sure it is initially in IBD mode
minchainwork = int(node.getblockchaininfo()["chainwork"], 16) + 1
self.restart_node(0, self.extra_args[0] + [
"-avaproof={}".format(proof),
"-avamasterkey=cND2ZvtabDbJ1gucx9GWH6XT9kgTAqfb6cotPt5Q5CyxVDhid2EN",
"-minimumchainwork=0x{:x}".format(minchainwork),
])
self.log.info(
"The proof verification should be delayed until IBD is complete")
assert node.getblockchaininfo()["initialblockdownload"] is True
# Our proof cannot be verified during IBD, so we should have no peer
assert not node.getavalanchepeerinfo()
# Mining a few more blocks should cause us to leave IBD
node.generate(2)
# Our proof is now verified and our node is added as a peer
assert node.getblockchaininfo()["initialblockdownload"] is False
assert_equal(len(node.getavalanchepeerinfo()), 1)
self.log.info(
"A proof using the maximum number of stakes is accepted...")
blocks = node.generatetoaddress(AVALANCHE_MAX_PROOF_STAKES + 1,
addrkey0.address)
too_many_coinbases = [get_coinbase(h) for h in blocks]
too_many_stakes = get_stakes(too_many_coinbases, addrkey0.key)
maximum_stakes = get_stakes(too_many_coinbases[:-1],
addrkey0.key)
good_proof = node.buildavalancheproof(
proof_sequence, proof_expiration,
proof_master, maximum_stakes)
node.addavalanchenode(
get_interface_node().nodeid, proof_master, good_proof)
self.log.info("A proof using too many stakes should be rejected...")
bad_proof = node.buildavalancheproof(
proof_sequence, proof_expiration,
proof_master, too_many_stakes)
assert_raises_rpc_error(-32602, "Avalanche proof has too many UTXOs",
node.addavalanchenode,
get_interface_node().nodeid, proof_master,
bad_proof)
# Test invalid proofs
self.log.info("Bad proof should be rejected at startup")
no_stake = node.buildavalancheproof(
proof_sequence, proof_expiration, proof_master, [])
dust = node.buildavalancheproof(
proof_sequence, proof_expiration, proof_master,
[{
'txid': coinbases[0]['txid'],
'vout': coinbases[0]['n'],
'amount': '0',
'height': coinbases[0]['height'],
'iscoinbase': True,
'privatekey': addrkey0.key,
}])
duplicate_stake = node.buildavalancheproof(
proof_sequence, proof_expiration, proof_master,
[{
'txid': coinbases[0]['txid'],
'vout': coinbases[0]['n'],
'amount': coinbases[0]['value'],
'height': coinbases[0]['height'],
'iscoinbase': True,
'privatekey': addrkey0.key,
}] * 2)
bad_sig = ("0b000000000000000c0000000000000021030b4c866585dd868a9d62348"
"a9cd008d6a312937048fff31670e7e920cfc7a7440105c5f72f5d6da3085"
"583e75ee79340eb4eff208c89988e7ed0efb30b87298fa30000000000f20"
"52a0100000003000000210227d85ba011276cf25b51df6a188b75e604b3"
"8770a462b2d0e9fb2fc839ef5d3faf07f001dd38e9b4a43d07d5d449cc0"
"f7d2888d96b82962b3ce516d1083c0e031773487fc3c4f2e38acd1db974"
"1321b91a79b82d1c2cfd47793261e4ba003cf5")
self.stop_node(0)
def check_proof_init_error(proof, message):
node.assert_start_raises_init_error(
self.extra_args[0] + [
"-avaproof={}".format(proof),
"-avamasterkey=cND2ZvtabDbJ1gucx9GWH6XT9kgTAqfb6cotPt5Q5CyxVDhid2EN",
],
expected_msg="Error: " + message,
)
check_proof_init_error(no_stake,
"the avalanche proof has no stake")
check_proof_init_error(dust,
"the avalanche proof stake is too low")
check_proof_init_error(duplicate_stake,
"the avalanche proof has duplicated stake")
check_proof_init_error(bad_sig,
"the avalanche proof has invalid stake signatures")
if __name__ == '__main__':
AvalancheProofTest().main()