Differential D2341 Diff 7356 test/functional/abc-nulldummy-activation.py

Changeset View

Standalone View

test/functional/abc-nulldummy-activation.py

This file was added.

Property	Old Value	New Value
File Mode	null	100755

				#!/usr/bin/env python3
				# Copyright (c) 2016 The Bitcoin Core developers
				# Copyright (c) 2019 The Bitcoin developers
				# Distributed under the MIT software license, see the accompanying
				# file COPYING or http://www.opensource.org/licenses/mit-license.php.

				from test_framework.blocktools import create_coinbase, create_block, make_conform_to_ctor
				from test_framework.test_framework import BitcoinTestFramework
				from test_framework.util import assert_equal, assert_raises_rpc_error
				from test_framework.mininode import network_thread_start, COIN, P2PInterface
				from test_framework.messages import ToHex, FromHex, CTransaction, msg_tx, msg_block
				from test_framework.script import CScript

				NULLDUMMY_ERROR = "64: non-mandatory-script-verify-flag (Dummy CHECKMULTISIG argument must be zero)"

				# far into the future
				GREAT_WALL_START_TIME = 2000000000

				# If we don't do this, autoreplay protection will activate simultaneous with
				# great_wall and all our sigs will mysteriously fail.
				REPLAY_PROTECTION_START_TIME = GREAT_WALL_START_TIME * 2


				def swap_dummy_element(tx):
				assert(len(tx.vin[0].scriptSig))
				assert(tx.vin[0].scriptSig[0] == 0)
				scriptSig = CScript(tx.vin[0].scriptSig)
				newscript = []
				for i in scriptSig:
				if (len(newscript) == 0):
				assert(len(i) == 0)
				newscript.append(b'\x51')
				else:
				newscript.append(i)
				tx.vin[0].scriptSig = CScript(newscript)
				tx.rehash()


				class BlockHeader:
				def __init__(self, header):
				self.header = header

				def __getattr__(self, name):
				return self.header[name]

				def hash_as_int(self):
				return int(self.hash, 16)


				def fetch_best_header(node):
				return BlockHeader(node.getblockheader(node.getbestblockhash()))


				class NullDummyActivation(BitcoinTestFramework):
				def set_test_params(self):
				self.num_nodes = 1
				self.setup_clean_chain = True
				self.extra_args = [[
				# promiscuous allows us to accept non-std transactions in ATM
				"-promiscuousmempoolflags=1",
				"-whitelist=127.0.0.1",
				"-greatwallactivationtime={}".format(
				GREAT_WALL_START_TIME),
				"-replayprotectionactivationtime={}".format(
				REPLAY_PROTECTION_START_TIME)]]
				self.mocktime = GREAT_WALL_START_TIME - 200
				self.lastblock = None

				def run_test(self):
				self.nodes[0].add_p2p_connection(P2PInterface())
				network_thread_start()

				coinbase_utxo = self.create_coinbase_utxo(4)

				self.test_nulldummy_before_activation(coinbase_utxo.pop())
				self.move_mtp(GREAT_WALL_START_TIME - 1)
				self.test_nulldummy_before_activation(coinbase_utxo.pop())

				# Activation time.
				before_activation = self.lastblock.hash
				self.move_mtp(GREAT_WALL_START_TIME)
				self.test_nulldummy_after_activation(coinbase_utxo.pop())

				self.log.info("Rollback to before activation time.")
				self.nodes[0].invalidateblock(before_activation)
				self.lastblock = fetch_best_header(self.nodes[0])
				assert(not self.is_great_wall_activated())
				self.test_nulldummy_before_activation(coinbase_utxo.pop())

				def test_nulldummy_before_activation(self, coinbase_utxo):
				self.log.info(
				"Check that nulldummy is accepted to mempool/block before activation")
				assert(not self.is_great_wall_activated())

				p2pk_address = self.nodes[0].getnewaddress()
				ms_address = self.nodes[0].addmultisigaddress(1, [p2pk_address])

				[to_multisig_tx, ms_utxo] = self.create_transaction(
				coinbase_utxo, ms_address)
				self.nodes[0].sendrawtransaction(ToHex(to_multisig_tx))

				[from_multisig_tx, _] = self.create_transaction(ms_utxo, ms_address)

				# Swap out nulldummy element with non-nulldummy, making the tx non-standard.
				swap_dummy_element(from_multisig_tx)

				# promiscuousmempoolflags allows the transaction to be accepted to
				# mempool even though it's non-standard.
				self.nodes[0].p2p.send_message(msg_tx(from_multisig_tx))
				self.nodes[0].p2p.sync_with_ping()
				assert(from_multisig_tx.hash in set(self.nodes[0].getrawmempool()))

				block = self.create_block([to_multisig_tx, from_multisig_tx])
				self.nodes[0].p2p.send_message(msg_block(block))
				self.nodes[0].p2p.sync_with_ping()
				self.lastblock = fetch_best_header(self.nodes[0])
				assert_equal(self.lastblock.hash, block.hash)

				def test_nulldummy_after_activation(self, coinbase_utxo):
				self.log.info(
				"Check that nulldummy is rejected from mempool/block after activation")
				assert(self.is_great_wall_activated())

				p2pk_address = self.nodes[0].getnewaddress()
				ms_address = self.nodes[0].addmultisigaddress(1, [p2pk_address])

				[to_multisig_tx, ms_utxo] = self.create_transaction(
				coinbase_utxo, ms_address)
				self.nodes[0].sendrawtransaction(ToHex(to_multisig_tx))

				[from_multisig_tx, _] = self.create_transaction(ms_utxo, ms_address)

				# Swap out nulldummy element with non-nulldummy, making the tx non-standard.
				swap_dummy_element(from_multisig_tx)

				self.nodes[0].p2p.send_message(msg_tx(from_multisig_tx))
				self.nodes[0].p2p.sync_with_ping()
				assert(from_multisig_tx.hash not in set(self.nodes[0].getrawmempool()))

				assert_raises_rpc_error(-26, NULLDUMMY_ERROR,
				self.nodes[0].sendrawtransaction, ToHex(from_multisig_tx))

				block = self.create_block([to_multisig_tx, from_multisig_tx])
				self.nodes[0].p2p.send_message(msg_block(block))
				self.nodes[0].p2p.sync_with_ping()
				assert_equal(self.lastblock.hash, self.nodes[0].getbestblockhash())

				self.nodes[0].submitblock(ToHex(block))
				assert_equal(self.lastblock.hash, self.nodes[0].getbestblockhash())

				def create_coinbase_utxo(self, n):
				coinbase_utxo = []
				for block in self.nodes[0].generate(n):
				txid = self.nodes[0].getblock(block)['tx'][0]
				coinbase_utxo.append({'txid': txid, 'vout': 0, 'value': 50 * COIN})

				self.nodes[0].generate(101)
				self.lastblock = fetch_best_header(self.nodes[0])
				return coinbase_utxo

				def move_mtp(self, new_mtp):
				assert new_mtp >= self.lastblock.mediantime

				sign = "+" if new_mtp >= GREAT_WALL_START_TIME else "-"
				self.log.info("Moving MTP to GREAT_WALL_START_TIME {} {}".format(
				sign, abs(new_mtp - GREAT_WALL_START_TIME)))
				while self.lastblock.mediantime != new_mtp:
				self.nodes[0].setmocktime(self.lastblock.time + 1)
				self.nodes[0].generate(1)
				self.lastblock = fetch_best_header(self.nodes[0])

				def is_great_wall_activated(self):
				return self.lastblock.mediantime >= GREAT_WALL_START_TIME

				def create_block(self, txs):
				block = create_block(self.lastblock.hash_as_int(), create_coinbase(
				self.lastblock.height + 1), self.lastblock.time + 1)
				block.nVersion = 4
				block.vtx.extend(txs)
				make_conform_to_ctor(block)
				block.hashMerkleRoot = block.calc_merkle_root()
				block.rehash()
				block.solve()
				return block

				# Returns the created transaction + an utxo to spend it
				def create_transaction(self, spend, to_address):
				node = self.nodes[0]
				TX_FEE = 1000

				inputs = [{
				"txid": spend['txid'],
				"vout": spend['vout']}]
				amount = spend['value'] - TX_FEE
				outputs = {to_address: amount / COIN}

				rawtx = node.createrawtransaction(inputs, outputs)
				signresult = node.signrawtransaction(rawtx)
				tx = FromHex(CTransaction(), signresult['hex'])
				tx.calc_sha256()
				return [tx, {"txid": tx.hash, "vout": 0, "value": amount}]


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