Differential D9565 Diff 28657 test/functional/abc_p2p_avalanche.py

Changeset View

Standalone View

test/functional/abc_p2p_avalanche.py

#!/usr/bin/env python3		#!/usr/bin/env python3
# Copyright (c) 2020-2021 The Bitcoin developers		# Copyright (c) 2020-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 the resolution of forks via avalanche."""		"""Test the resolution of forks via avalanche."""
import random		import random
		import struct

from test_framework.avatools import create_coinbase_stakes		from test_framework.avatools import create_coinbase_stakes
from test_framework.key import (		from test_framework.key import (
		bytes_to_wif,
ECKey,		ECKey,
ECPubKey,		ECPubKey,
)		)
from test_framework.p2p import P2PInterface, p2p_lock		from test_framework.p2p import P2PInterface, p2p_lock
from test_framework.messages import (		from test_framework.messages import (
		AvalancheDelegation,
AvalancheResponse,		AvalancheResponse,
AvalancheVote,		AvalancheVote,
CInv,		CInv,
		FromHex,
		hash256,
		msg_avahello,
msg_avapoll,		msg_avapoll,
msg_tcpavaresponse,		msg_tcpavaresponse,
NODE_AVALANCHE,		NODE_AVALANCHE,
NODE_NETWORK,		NODE_NETWORK,
TCPAvalancheResponse,		TCPAvalancheResponse,
)		)
from test_framework.test_framework import BitcoinTestFramework		from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (		from test_framework.util import (
▲ Show 20 Lines • Show All 79 Lines • ▼ Show 20 Lines	def wait_for_avahello(self, timeout=5):
wait_until(		wait_until(
lambda: self.avahello is not None,		lambda: self.avahello is not None,
timeout=timeout,		timeout=timeout,
lock=p2p_lock)		lock=p2p_lock)

with p2p_lock:		with p2p_lock:
return self.avahello		return self.avahello

		def send_avahello(self, delegation_hex: str, delegated_privkey: ECKey):
		delegation = FromHex(AvalancheDelegation(), delegation_hex)
		local_sighash = hash256(
		delegation.getid() +
		struct.pack("<QQQQ", self.local_nonce, self.remote_nonce,
		self.local_extra_entropy, self.remote_extra_entropy))

		msg = msg_avahello()
		msg.hello.delegation = delegation
		msg.hello.sig = delegated_privkey.sign_schnorr(local_sighash)
		self.send_message(msg)


class AvalancheTest(BitcoinTestFramework):		class AvalancheTest(BitcoinTestFramework):
def set_test_params(self):		def set_test_params(self):
self.setup_clean_chain = True		self.setup_clean_chain = True
self.num_nodes = 2		self.num_nodes = 2
self.extra_args = [		self.extra_args = [
['-enableavalanche=1', '-avacooldown=0'],		['-enableavalanche=1', '-avacooldown=0'],
['-enableavalanche=1', '-avacooldown=0', '-noparkdeepreorg', '-maxreorgdepth=-1']]		['-enableavalanche=1', '-avacooldown=0', '-noparkdeepreorg', '-maxreorgdepth=-1']]
▲ Show 20 Lines • Show All 228 Lines • ▼ Show 20 Lines	def run_test(self):
"-avaproof={}".format(proof),		"-avaproof={}".format(proof),
"-avamasterkey=cND2ZvtabDbJ1gucx9GWH6XT9kgTAqfb6cotPt5Q5CyxVDhid2EN",		"-avamasterkey=cND2ZvtabDbJ1gucx9GWH6XT9kgTAqfb6cotPt5Q5CyxVDhid2EN",
])		])

assert_equal(		assert_equal(
int(node.getnetworkinfo()['localservices'], 16) & NODE_AVALANCHE,		int(node.getnetworkinfo()['localservices'], 16) & NODE_AVALANCHE,
NODE_AVALANCHE)		NODE_AVALANCHE)

self.log.info("Test the avahello signature")		self.log.info("Test the avahello signature (node -> P2PInterface)")
quorum = get_quorum()		good_interface = get_node()
poll_node = quorum[0]		avahello = good_interface.wait_for_avahello().hello

avahello = poll_node.wait_for_avahello().hello

avakey.set(bytes.fromhex(node.getavalanchekey()))		avakey.set(bytes.fromhex(node.getavalanchekey()))
assert avakey.verify_schnorr(		assert avakey.verify_schnorr(
avahello.sig, avahello.get_sighash(poll_node))		avahello.sig, avahello.get_sighash(good_interface))

		self.log.info("Test the avahello signature (P2PInterface -> node)")
		stakes = create_coinbase_stakes(node, [blockhashes[1]], addrkey0.key)
		interface_proof_hex = node.buildavalancheproof(
		proof_sequence, proof_expiration, pubkey.get_bytes().hex(),
		stakes)
		# delegate
		delegated_key = ECKey()
		delegated_key.generate()
		interface_delegation_hex = node.delegateavalancheproof(
		interface_proof_hex,
		bytes_to_wif(privkey.get_bytes()),
		delegated_key.get_pubkey().get_bytes().hex(),
		None)
		good_interface.send_avahello(interface_delegation_hex, delegated_key)

		self.log.info("Test that wrong avahello signature causes a ban")
		bad_interface = get_node()
		wrong_key = ECKey()
		wrong_key.generate()
		bad_interface.send_avahello(interface_delegation_hex, wrong_key)
		bad_interface.wait_for_disconnect()
		FabienUnsubmitted Not Done Inline Actions Please do both: check the node is disconnected AND the reason it is disconnected. Otherwise you're not sure what you are testing, here you are looking for invalid-avahello-signature but you if you get an invalid-delegation the test will still pass. Fabien: Please do both: check the node is disconnected AND the reason it is disconnected. Otherwise…

		# Quick check that the good interface is still connected
		# FIXME: when proof relaying is implemented, replace this with a check
		# that the interface is added as a peer.
		good_interface.sync_with_ping()
		FabienUnsubmitted Not Done Inline Actions Move next to the good interface test instead of having it dandling Fabien: Move next to the good interface test instead of having it dandling


if __name__ == '__main__':		if __name__ == '__main__':
AvalancheTest().main()		AvalancheTest().main()