Differential D9497 Diff 28414 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.mininode import P2PInterface, mininode_lock		from test_framework.mininode import P2PInterface, mininode_lock
from test_framework.messages import (		from test_framework.messages import (
		AvalancheDelegation,
		AvalancheProof,
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=mininode_lock)		lock=mininode_lock)

with mininode_lock:		with mininode_lock:
return self.avahello		return self.avahello

		def _getLocalSigHash(self, delegation_id: bytes):
		b = delegation_id
		b += struct.pack("<Q", self.local_nonce)
		b += struct.pack("<Q", self.remote_nonce)
		b += struct.pack("<Q", self.local_extra_entropy)
		b += struct.pack("<Q", self.remote_extra_entropy)
		return hash256(b)

		def _send_avahello(self, delegation: AvalancheDelegation, sig: bytes):
		msg = msg_avahello()
		msg.hello.delegation = delegation
		msg.hello.sig = sig
		self.send_message(msg)

		def send_avahello(self, delegation_hex: str, privkey: ECKey):
		delegation = FromHex(AvalancheDelegation(), delegation_hex)
		sig = privkey.sign_schnorr(self._getLocalSigHash(delegation.getid()))
		self._send_avahello(delegation, sig)

		def send_avahello_wrong_signature(self, delegation_hex: str):
		delegation = FromHex(AvalancheDelegation(), delegation_hex)
		sig = bytes(64)
		self._send_avahello(delegation, sig)

		def send_avahello_empty_delegation(self, proofid: bytes, privkey: ECKey):
		delegation = AvalancheDelegation(proofid, levels=[])
		sig = privkey.sign_schnorr(self._getLocalSigHash(delegation.getid()))
		self._send_avahello(delegation, sig)


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 225 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()		interface = get_node()
poll_node = quorum[0]

avahello = poll_node.wait_for_avahello().hello

		avahello = interface.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(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_privkey = ECKey()
		delegated_privkey.generate()
		interface_delegation_hex = node.delegateavalancheproof(
		interface_proof_hex,
		bytes_to_wif(privkey.get_bytes()),
		delegated_privkey.get_pubkey().get_bytes().hex(),
		None
		)

		with self.nodes[0].assert_debug_log(["Successfully checked AVAHELLO signature for peer=0"]):
		interface.send_avahello(
		interface_delegation_hex,
		delegated_privkey)

		self.log.info("Test a wrong avahello signature")
		interface = get_node()
		with self.nodes[0].assert_debug_log(
		["Misbehaving", "peer=1 (0 -> 100) BAN THRESHOLD EXCEEDED: invalid-avahello-signature"]):
		interface.send_avahello_wrong_signature(interface_delegation_hex)

		self.log.info("Test an avahello with an empty delegation")
		interface = get_node()
		proofid = FromHex(AvalancheProof(), interface_proof_hex).proofid
		with self.nodes[0].assert_debug_log(
		["Ignoring AVAHELLO from peer=2 with empty delegation"]):
		interface.send_avahello_empty_delegation(
		proofid, delegated_privkey)


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