Differential D2483 Diff 7124 test/functional/abc-schnorr-checkdatasig-activation.py

Changeset View

Standalone View

test/functional/abc-schnorr-checkdatasig-activation.py

This file was added.

Property	Old Value	New Value
File Mode	null	100755

				#!/usr/bin/env python3
				# Copyright (c) 2018-2019 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 Schnorr using OP_CHECKDATASIG
				(just OP_CHECKDATASIG so we don't need to sign actual transactions)

				This is based on abc-checkdatasig-activation.py from the previous upgrade.
				"""

				import hashlib
				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
				GREAT_WALL_START_TIME = 2000000000

				# Error due to invalid signature when STRICTENC is on
				BAD_SIG_ERROR = b'mandatory-script-verify-flag-failed (Non-canonical DER signature)'
				RPC_BAD_SIG_ERROR = "16: " + \
				BAD_SIG_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',
				"-greatwallactivationtime=%d" % GREAT_WALL_START_TIME,
				"-replayprotectionactivationtime=%d" % (2 * GREAT_WALL_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"]) * COIN) // count
				tx.vin = [CTxIn(COutPoint(int(utxo["txid"], 16), utxo["vout"]))]
				tx.vout = []
				# dataset from src/test/key_tests.cpp
				signature = bytearray.fromhex(
				'2c56731ac2f7a7e7f11518fc7722a166b02438924ca9d8b4d111347b81d0717571846de67ad3d913a8fdf9d8f3f73161a4c48ae81cb183b214765feb86e255ce')
				# In key_tests.cpp the message was hashed twice; so here we need to
				# hash once and the second hash happens inside OP_CHECKDATASIG.
				message = hashlib.sha256(b"Very deterministic message").digest()
				pubkey = bytearray.fromhex(
				'030b4c866585dd868a9d62348a9cd008d6a312937048fff31670e7e920cfc7a744')
				for _ in range(count):
				tx.vout.append(CTxOut(value, CScript(
				[signature, message, pubkey, OP_CHECKDATASIG])))
				tx.vout[0].nValue -= node.calculate_fee(tx)
				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]

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

				# Create various outputs using the OP_CHECKDATASIG
				# to check for activation.
				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_checkdatasigs = [PreviousSpendableOutput(tx, i)
				for i in range(len(tx.vout))]

				def spend_checkdatasig():
				outpoint = spendable_checkdatasigs.pop()
				out = outpoint.tx.vout[outpoint.n]
				tx = CTransaction()
				tx.vin = [CTxIn(COutPoint(outpoint.tx.sha256, outpoint.n))]
				tx.vout = [CTxOut(out.nValue, CScript([])),
				CTxOut(0, CScript([random.getrandbits(800), OP_RETURN]))]
				tx.vout[0].nValue -= node.calculate_fee(tx)
				tx.rehash()
				return tx

				# Check that transactions using Schnorr are not accepted yet.
				# Because STRICTENC is on, these will fail outright.
				self.log.info("Try to use the Schnorr signature before activation")

				tx0 = spend_checkdatasig()
				tx0_hex = ToHex(tx0)
				assert_raises_rpc_error(-26, RPC_BAD_SIG_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(GREAT_WALL_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):
				b = next_block(GREAT_WALL_START_TIME + i - 1)
				yield accepted(b)

				# Check again just before the activation time
				assert_equal(node.getblockheader(node.getbestblockhash())['mediantime'],
				GREAT_WALL_START_TIME - 1)
				assert_raises_rpc_error(-26, RPC_BAD_SIG_ERROR,
				node.sendrawtransaction, tx0_hex)

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

				# Now we make a block including Schnorr signature just prior to
				# activation. Must still be rejected.
				b = next_block(GREAT_WALL_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(GREAT_WALL_START_TIME + 6)
				yield accepted(fork_block)

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

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

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


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