Differential D1625 Diff 4507 test/functional/abc-checkdatasig-activation.py

Changeset View

Standalone View

test/functional/abc-checkdatasig-activation.py

This file was added.

Property	Old Value	New Value
File Mode	null	100755

				#!/usr/bin/env python3
				# Copyright (c) 2018 The Bitcoin developers
				# Distributed under the MIT software license, see the accompanying
				# file COPYING or http://www.opensource.org/licenses/mit-license.php.
				"""
				This test checks activation of OP_CHECKDATASIG
				"""

				from test_framework.test_framework import ComparisonTestFramework
				from test_framework.util import satoshi_round, assert_equal, assert_raises_rpc_error
				from test_framework.comptool import TestManager, TestInstance, RejectResult
				from test_framework.blocktools import *
				from test_framework.script import *

				# far into the future
				MAGNETIC_ANOMALY_START_TIME = 2000000000

				# Error due to invalid opcodes
				BAD_OPCODE_ERROR = b'mandatory-script-verify-flag-failed (Opcode missing or not understood)'
				RPC_BAD_OPCODE_ERROR = "16: " + \
				BAD_OPCODE_ERROR.decode("utf-8")


				class PreviousSpendableOutput():

				def __init__(self, tx=CTransaction(), n=-1):
				self.tx = tx
				self.n = n # the output we're spending


				class CheckDataSigActivationTest(ComparisonTestFramework):

				def set_test_params(self):
				self.num_nodes = 1
				self.setup_clean_chain = True
				self.extra_args = [['-whitelist=127.0.0.1',
				"-magneticanomalyactivationtime=%d" % MAGNETIC_ANOMALY_START_TIME,
				"-replayprotectionactivationtime=%d" % (2 * MAGNETIC_ANOMALY_START_TIME)]]

				def create_checkdatasig_tx(self, count):
				node = self.nodes[0]
				utxos = node.listunspent()
				assert(len(utxos) > 0)
				utxo = utxos[0]
				tx = CTransaction()
				value = int(satoshi_round(
				utxo["amount"] - self.relayfee) * COIN) // count
				tx.vin = [CTxIn(COutPoint(int(utxo["txid"], 16), utxo["vout"]))]
				tx.vout = []
				signature = bytearray.fromhex(
				'30440220052a549456efe256a8c190650648e7ebdd46161f330830733fc7c674aeb7d3da0220519999b4f25b37de557a9c909cb5b155d1f0b11293b0890236696bfa0bd310e1')
				message = bytearray.fromhex(
				'0100000000000000000000000000000000000000000000000000000000000000')
				pubkey = bytearray.fromhex(
				'038282263212c609d9ea2a6e3e172de238d8c39cabd5ac1ca10646e23fd5f51508')
				for _ in range(count):
				tx.vout.append(CTxOut(value, CScript(
				[signature, message, pubkey, OP_CHECKDATASIG])))
				tx_signed = node.signrawtransaction(ToHex(tx))["hex"]
				return tx_signed

				def run_test(self):
				self.test = TestManager(self, self.options.tmpdir)
				self.test.add_all_connections(self.nodes)
				# Start up network handling in another thread
				NetworkThread().start()
				self.test.run()

				def get_tests(self):
				node = self.nodes[0]
				self.relayfee = self.nodes[0].getnetworkinfo()["relayfee"]

				# First, we generate some coins to spend.
				node.generate(125)

				# Create various outputs using the OP_AND to check for activation.
				jasonbcoxUnsubmitted Not Done Inline Actions OP_AND should be OP_CHECKDATASIG. very confusing otherwise. jasonbcox: OP_AND should be OP_CHECKDATASIG. very confusing otherwise.
				tx_hex = self.create_checkdatasig_tx(25)
				txid = node.sendrawtransaction(tx_hex)
				assert(txid in set(node.getrawmempool()))

				node.generate(1)
				assert(txid not in set(node.getrawmempool()))

				# register the spendable outputs.
				tx = FromHex(CTransaction(), tx_hex)
				tx.rehash()
				spendable_ands = [PreviousSpendableOutput(tx, i)
				jasonbcoxUnsubmitted Not Done Inline Actions spendable_ands -> spendable_checkdatasigs jasonbcox: spendable_ands -> spendable_checkdatasigs
				for i in range(len(tx.vout))]

				def spend_checkdatasig():
				outpoint = spendable_ands.pop()
				out = outpoint.tx.vout[outpoint.n]
				value = int(out.nValue - (self.relayfee * COIN))
				tx = CTransaction()
				tx.vin = [CTxIn(COutPoint(outpoint.tx.sha256, outpoint.n))]
				tx.vout = [CTxOut(value, CScript([])),
				CTxOut(0, CScript([random.getrandbits(800), OP_RETURN]))]
				tx.rehash()
				return tx

				# Check that large opreturn are not accepted yet.
				jasonbcoxUnsubmitted Not Done Inline Actions opreturn -> something about checkdatasig jasonbcox: opreturn -> something about checkdatasig
				self.log.info("Try to use the checkdatasig opcodes before activation")

				tx0 = spend_checkdatasig()
				tx0_hex = ToHex(tx0)
				assert_raises_rpc_error(-26, RPC_BAD_OPCODE_ERROR,
				node.sendrawtransaction, tx0_hex)

				# Push MTP forward just before activation.
				self.log.info("Pushing MTP just before the activation and check again")
				node.setmocktime(MAGNETIC_ANOMALY_START_TIME)

				# returns a test case that asserts that the current tip was accepted
				def accepted(tip):
				return TestInstance([[tip, True]])

				# returns a test case that asserts that the current tip was rejected
				def rejected(tip, reject=None):
				if reject is None:
				return TestInstance([[tip, False]])
				else:
				return TestInstance([[tip, reject]])

				def next_block(block_time):
				# get block height
				blockchaininfo = node.getblockchaininfo()
				height = int(blockchaininfo['blocks'])

				# create the block
				coinbase = create_coinbase(height)
				coinbase.rehash()
				block = create_block(
				int(node.getbestblockhash(), 16), coinbase, block_time)

				# Do PoW, which is cheap on regnet
				block.solve()
				return block

				for i in range(6):
				jasonbcoxUnsubmitted Not Done Inline Actions I'm confused how this magic number 6 was selected. With MTP being the last 11 blocks, does selecting fewer potentially make the test flaky and/or fragile to future changes, as there could be blocks mined before these 6 with unknown timestamps? jasonbcox: I'm confused how this magic number 6 was selected. With MTP being the last 11 blocks, does…
				deadalnixAuthorUnsubmitted Not Done Inline Actions We are good for several decades. deadalnix: We are good for several decades.
				jasonbcoxUnsubmitted Not Done Inline Actions You're referring to the MAGNETIC_ANOMALY_START_TIME value, but I'm talking about MTP calculated using 11 blocks' worth of timestamps, not 6. jasonbcox: You're referring to the MAGNETIC_ANOMALY_START_TIME value, but I'm talking about MTP calculated…
				deadalnixAuthorUnsubmitted Not Done Inline Actions You need to pour in 6 new timestamp for the median to budge. I don't see what's the problem here. deadalnix: You need to pour in 6 new timestamp for the median to budge. I don't see what's the problem…
				b = next_block(MAGNETIC_ANOMALY_START_TIME + i - 1)
				yield accepted(b)

				# Check again just before the activation time
				assert_equal(node.getblockheader(node.getbestblockhash())['mediantime'],
				MAGNETIC_ANOMALY_START_TIME - 1)
				assert_raises_rpc_error(-26, RPC_BAD_OPCODE_ERROR,
				jasonbcoxUnsubmitted Not Done Inline Actions good check. I didn't think about testing the RPC in conjunction with the block acceptance. @schancel we should be on the lookout for stuff like this. in the future, a solid testing framework would be able to make checks like this for us. jasonbcox: good check. I didn't think about testing the RPC in conjunction with the block acceptance.
				deadalnixAuthorUnsubmitted Not Done Inline Actions Yes please. deadalnix: Yes please.
				node.sendrawtransaction, tx0_hex)

				def add_tx(block, tx):
				block.vtx.append(tx)
				block.hashMerkleRoot = block.calc_merkle_root()
				block.solve()

				b = next_block(MAGNETIC_ANOMALY_START_TIME + 6)
				add_tx(b, tx0)
				yield rejected(b, RejectResult(16, b'blk-bad-inputs'))

				self.log.info("Activates checkdatasig")
				fork_block = next_block(MAGNETIC_ANOMALY_START_TIME + 6)
				yield accepted(fork_block)

				assert_equal(node.getblockheader(node.getbestblockhash())['mediantime'],
				MAGNETIC_ANOMALY_START_TIME)

				tx0id = node.sendrawtransaction(tx0_hex)
				assert(tx0id in set(node.getrawmempool()))

				# Transactions can also be included in blocks.
				magneticanomalyblock = next_block(MAGNETIC_ANOMALY_START_TIME + 7)
				add_tx(magneticanomalyblock, tx0)
				yield accepted(magneticanomalyblock)

				self.log.info("Cause a reorg that deactivate the checkdatasig opcodes")

				# Invalidate the checkdatasig block, ensure tx0 gets back to the mempool.
				assert(tx0id not in set(node.getrawmempool()))

				node.invalidateblock(format(magneticanomalyblock.sha256, 'x'))
				assert(tx0id in set(node.getrawmempool()))

				node.invalidateblock(format(fork_block.sha256, 'x'))
				assert(tx0id not in set(node.getrawmempool()))


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