Differential D9351 Diff 28024 test/functional/abc_rpc_avalancheproof.py

Changeset View

Standalone View

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 test_framework.avatools import get_stakes
				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 add_interface_node(test_node) -> str:
				"""Create a mininode, connect it to test_node, return the nodeid
				of the mininode as registered by test_node.
				"""
				n = P2PInterface()
				test_node.add_p2p_connection(n)
				n.wait_for_verack()
				return test_node.getpeerinfo()[-1]['id']


				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]

				addrkey0 = node.get_deterministic_priv_key()
				blockhashes = node.generatetoaddress(100, addrkey0.address)

				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,
				get_stakes(node, [blockhashes[0]], 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...")
				blockhashes = node.generatetoaddress(AVALANCHE_MAX_PROOF_STAKES + 1,
				addrkey0.address)

				too_many_stakes = get_stakes(node, blockhashes, addrkey0.key)
				maximum_stakes = get_stakes(node, blockhashes[:-1], addrkey0.key)
				good_proof = node.buildavalancheproof(
				proof_sequence, proof_expiration,
				proof_master, maximum_stakes)
				peerid1 = add_interface_node(node)
				node.addavalanchenode(
				peerid1, 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)

				peerid2 = add_interface_node(node)
				assert_raises_rpc_error(-32602, "Avalanche proof has too many UTXOs",
				node.addavalanchenode,
				peerid2, 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,
				get_stakes(node, [blockhashes[0]], addrkey0.key, amount="0"))

				duplicate_stake = node.buildavalancheproof(
				proof_sequence, proof_expiration, proof_master,
				get_stakes(node, [blockhashes[0]] * 2, addrkey0.key))

				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()