Differential D9353 Diff 28043 test/functional/abc_rpc_avalancheproof.py

Changeset View

Standalone View

test/functional/abc_rpc_avalancheproof.py

#!/usr/bin/env python3		#!/usr/bin/env python3
# Copyright (c) 2021 The Bitcoin developers		# Copyright (c) 2021 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying		# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.		# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test building avalanche proofs and using them to add avalanche peers."""		"""Test building avalanche proofs and using them to add avalanche peers."""

from test_framework.avatools import get_stakes		from test_framework.avatools import get_stakes
from test_framework.key import ECKey		from test_framework.key import ECKey, bytes_to_wif
		from test_framework.messages import AvalancheDelegation
from test_framework.mininode import P2PInterface		from test_framework.mininode import P2PInterface
from test_framework.test_framework import BitcoinTestFramework		from test_framework.test_framework import BitcoinTestFramework
from test_framework.test_node import ErrorMatch		from test_framework.test_node import ErrorMatch
from test_framework.util import (		from test_framework.util import (
append_config,		append_config,
wait_until,		wait_until,
		assert_raises_rpc_error,
)		)

AVALANCHE_MAX_PROOF_STAKES = 1000		AVALANCHE_MAX_PROOF_STAKES = 1000


def add_interface_node(test_node) -> str:		def add_interface_node(test_node) -> str:
"""Create a mininode, connect it to test_node, return the nodeid		"""Create a mininode, connect it to test_node, return the nodeid
of the mininode as registered by test_node.		of the mininode as registered by test_node.
Show All 19 Lines	def run_test(self):
blockhashes = node.generatetoaddress(100, addrkey0.address)		blockhashes = node.generatetoaddress(100, addrkey0.address)

self.log.info(		self.log.info(
"Make build a valid proof and restart the node to use it")		"Make build a valid proof and restart the node to use it")
privkey = ECKey()		privkey = ECKey()
privkey.set(bytes.fromhex(		privkey.set(bytes.fromhex(
"12b004fff7f4b69ef8650e767f18f11ede158148b425660723b9f9a66e61f747"), True)		"12b004fff7f4b69ef8650e767f18f11ede158148b425660723b9f9a66e61f747"), True)

proof_master = privkey.get_pubkey().get_bytes().hex()		def get_hex_pubkey(privkey):
		return privkey.get_pubkey().get_bytes().hex()

		proof_master = get_hex_pubkey(privkey)
proof_sequence = 11		proof_sequence = 11
proof_expiration = 12		proof_expiration = 12
proof = node.buildavalancheproof(		proof = node.buildavalancheproof(
proof_sequence, proof_expiration, proof_master,		proof_sequence, proof_expiration, proof_master,
get_stakes(node, [blockhashes[0]], addrkey0.key))		get_stakes(node, [blockhashes[0]], addrkey0.key))

# Restart the node, making sure it is initially in IBD mode		# Restart the node, making sure it is initially in IBD mode
minchainwork = int(node.getblockchaininfo()["chainwork"], 16) + 1		minchainwork = int(node.getblockchaininfo()["chainwork"], 16) + 1
Show All 29 Lines	def run_test(self):

self.log.info("A proof using too many stakes should be rejected...")		self.log.info("A proof using too many stakes should be rejected...")
too_many_utxos = node.buildavalancheproof(		too_many_utxos = node.buildavalancheproof(
proof_sequence, proof_expiration,		proof_sequence, proof_expiration,
proof_master, too_many_stakes)		proof_master, too_many_stakes)
peerid2 = add_interface_node(node)		peerid2 = add_interface_node(node)
assert not node.addavalanchenode(peerid2, proof_master, too_many_utxos)		assert not node.addavalanchenode(peerid2, proof_master, too_many_utxos)

		self.log.info("Generate delegations for the proof")

		# Stack up a few delegation levels
		def gen_privkey():
		pk = ECKey()
		pk.generate()
		return pk

		delegator_privkey = privkey
		delegation = None
		for _ in range(10):
		delegated_privkey = gen_privkey()
		delegation = node.delegateavalancheproof(
		proof,
		bytes_to_wif(delegator_privkey.get_bytes()),
		get_hex_pubkey(delegated_privkey),
		delegation,
		)
		delegator_privkey = delegated_privkey

		random_privkey = gen_privkey()
		random_pubkey = get_hex_pubkey(random_privkey)

		# Invalid proof
		assert_raises_rpc_error(-8, "The proof is invalid",
		node.delegateavalancheproof,
		too_many_utxos,
		bytes_to_wif(privkey.get_bytes()),
		random_pubkey,
		)

		# Invalid privkey
		assert_raises_rpc_error(-5, "The private key is invalid",
		node.delegateavalancheproof,
		proof,
		bytes_to_wif(bytes(32)),
		random_pubkey,
		)

		# Invalid delegation
		bad_dg = AvalancheDelegation()
		assert_raises_rpc_error(-8, "The supplied delegation is not valid",
		node.delegateavalancheproof,
		proof,
		bytes_to_wif(privkey.get_bytes()),
		random_pubkey,
		bad_dg.serialize().hex(),
		)

		# Wrong privkey, does not match the proof
		assert_raises_rpc_error(-8, "The private key does not match the proof or the delegation",
		node.delegateavalancheproof,
		proof,
		bytes_to_wif(random_privkey.get_bytes()),
		random_pubkey,
		)

		# Wrong privkey, match the proof but does not match the delegation
		assert_raises_rpc_error(-8, "The private key does not match the proof or the delegation",
		node.delegateavalancheproof,
		proof,
		bytes_to_wif(privkey.get_bytes()),
		random_pubkey,
		delegation,
		)

# Test invalid proofs		# Test invalid proofs
self.log.info("Bad proof should be rejected at startup")		self.log.info("Bad proof should be rejected at startup")
no_stake = node.buildavalancheproof(		no_stake = node.buildavalancheproof(
proof_sequence, proof_expiration, proof_master, [])		proof_sequence, proof_expiration, proof_master, [])

dust = node.buildavalancheproof(		dust = node.buildavalancheproof(
proof_sequence, proof_expiration, proof_master,		proof_sequence, proof_expiration, proof_master,
get_stakes(node, [blockhashes[0]], addrkey0.key, amount="0"))		get_stakes(node, [blockhashes[0]], addrkey0.key, amount="0"))
▲ Show 20 Lines • Show All 46 Lines • Show Last 20 Lines