Differential D2483 Diff 7210 test/functional/abc-schnorr-checksig-activation.py

Changeset View

Standalone View

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

This file was added.

Property	Old Value	New Value
File Mode	null	100755

				#!/usr/bin/env python3
				# Copyright (c) 2015-2016 The Bitcoin Core developers
				# Copyright (c) 2017 The Bitcoin developers
				# Distributed under the MIT software license, see the accompanying
				# file COPYING or http://www.opensource.org/licenses/mit-license.php.
				"""
				This tests the activation of Schnorr transaction signatures:
				- rejection prior to upgrade both in mempool and blocks.
				- acceptance after upgrade both in mempool and blocks.
				- advance and rewind mempool drop tests. (note: advance test requires
				a temporary patch to bitcoind; see fakeDER comment below)

				Derived from abc-replay-protection.py
				"""

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

				# far into the future
				GREAT_WALL_START_TIME = 2000000000

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

				# Error due to passing a Schnorr signature before upgrade.
				BAD_SIG_ERROR = b'mandatory-script-verify-flag-failed (Non-canonical DER signature)'
				RPC_BAD_SIG_ERROR = "16: " + \
				BAD_SIG_ERROR.decode("utf-8")
				# Error due to passing Schnorr into multisig after upgrade.
				BAD_MULTISIG_ERROR = b'mandatory-script-verify-flag-failed (Signature cannot be 65 bytes in CHECKMULTISIG)'
				RPC_BAD_MULTISIG_ERROR = "16: " + \
				BAD_MULTISIG_ERROR.decode("utf-8")
				# Error due to invalid signature when STRICTENC is on
				NULLFAIL_ERROR = b'mandatory-script-verify-flag-failed (Signature must be zero for failed CHECK(MULTI)SIG operation)'
				RPC_NULLFAIL_ERROR = "16: " + \
				NULLFAIL_ERROR.decode("utf-8")

				# For normal test running:
				fakeDER = b''

				# To properly test activation, we need to make txes with 64 byte ECDSA sigs.
				# The easiest way to do this is to fake them, and then temporarily modify
				# VerifySignature in src/script/interpreter.cpp to always `return true;`
				# for ECDSA sigs, instead of `return pubkey.VerifyECDSA(sighash, vchSig);`
				# Once that patch is done, you can uncomment the following and tests should
				# pass.
				# fakeDER = bytes.fromhex('303e021d44444444444444444444444444444444444444444'
				# '44444444444444444021d4444444444444444444444444444'
				# '444444444444444444444444444444')

				assert len(fakeDER) in [0, 64]


				class PreviousSpendableOutput(object):

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


				class SchnorrActivationTest(ComparisonTestFramework):

				def set_test_params(self):
				self.num_nodes = 1
				self.setup_clean_chain = True
				self.block_heights = {}
				self.tip = None
				self.blocks = {}
				self.extra_args = [['-whitelist=127.0.0.1',
				"-greatwallactivationtime=%d" % GREAT_WALL_START_TIME,
				"-replayprotectionactivationtime=%d" % REPLAY_PROTECTION_START_TIME]]

				def run_test(self):
				self.test = TestManager(self, self.options.tmpdir)
				self.test.add_all_connections(self.nodes)
				network_thread_start()
				self.nodes[0].setmocktime(GREAT_WALL_START_TIME)
				self.test.run()

				def next_block(self, number):
				if self.tip == None:
				base_block_hash = self.genesis_hash
				block_time = int(time.time()) + 1
				else:
				base_block_hash = self.tip.sha256
				block_time = self.tip.nTime + 1
				# First create the coinbase
				height = self.block_heights[base_block_hash] + 1
				coinbase = create_coinbase(height)
				coinbase.rehash()
				block = create_block(base_block_hash, coinbase, block_time)

				# Do PoW, which is cheap on regnet
				block.solve()
				self.tip = block
				self.block_heights[block.sha256] = height
				assert number not in self.blocks
				self.blocks[number] = block
				return block

				def get_tests(self):
				self.genesis_hash = int(self.nodes[0].getbestblockhash(), 16)
				self.block_heights[self.genesis_hash] = 0
				spendable_outputs = []

				# save the current tip so its coinbase can be spent by a later block
				def save_spendable_output():
				spendable_outputs.append(self.tip)

				# get a coinbase that we previously marked as spendable
				def get_spendable_output():
				return PreviousSpendableOutput(spendable_outputs.pop(0).vtx[0], 0)

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

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

				# move the tip back to a previous block
				def tip(number):
				self.tip = self.blocks[number]

				# adds transactions to the block and updates state
				def update_block(block_number, new_transactions):
				[tx.rehash() for tx in new_transactions]
				block = self.blocks[block_number]
				block.vtx.extend(new_transactions)
				old_sha256 = block.sha256
				make_conform_to_ctor(block)
				block.hashMerkleRoot = block.calc_merkle_root()
				block.solve()
				# Update the internal state just like in next_block
				self.tip = block
				if block.sha256 != old_sha256:
				self.block_heights[
				block.sha256] = self.block_heights[old_sha256]
				del self.block_heights[old_sha256]
				self.blocks[block_number] = block
				return block

				# shorthand for functions
				block = self.next_block
				node = self.nodes[0]

				# Create a new block
				block(0)
				save_spendable_output()
				yield accepted()

				# Now we need that block to mature so we can spend the coinbase.
				test = TestInstance(sync_every_block=False)
				for i in range(199):
				block(5000 + i)
				test.blocks_and_transactions.append([self.tip, True])
				save_spendable_output()
				yield test

				# collect spendable outputs now to avoid cluttering the code later on
				out = []
				for i in range(100):
				out.append(get_spendable_output())

				# Generate a key pair to test P2SH sigops count
				private_key = CECKey()
				private_key.set_secretbytes(b"Schnorr!"*4)
				public_key = private_key.get_pubkey() # uncompressed

				def create_fund_and_spend_tx(spend, multi=False, sig='schnorr'):
				if multi:
				script = CScript([OP_1, public_key, OP_1, OP_CHECKMULTISIG])
				else:
				script = CScript([public_key, OP_CHECKSIG])

				# Fund transaction
				txfund = create_transaction(
				spend.tx, spend.n, b'', 50 * COIN, script)
				txfund.rehash()

				# Spend transaction
				txspend = CTransaction()
				txspend.vout.append(CTxOut(50 * COIN - 1000, CScript([OP_TRUE])))
				txspend.vin.append(CTxIn(COutPoint(txfund.sha256, 0), b''))

				# Sign the transaction
				sighashtype = SIGHASH_ALL \| SIGHASH_FORKID
				hashbyte = bytes([sighashtype & 0xff])
				sighash = SignatureHashForkId(
				script, txspend, 0, sighashtype, 50 * COIN)
				if sig == 'schnorr':
				txsig = private_key.sign_schnorr(sighash) + hashbyte
				elif sig == 'ecdsa':
				txsig = private_key.sign(sighash) + hashbyte
				elif isinstance(sig, bytes):
				txsig = sig + hashbyte
				if multi:
				txspend.vin[0].scriptSig = CScript([b'', txsig])
				else:
				txspend.vin[0].scriptSig = CScript([txsig])
				txspend.rehash()

				return txfund, txspend

				def send_transaction_to_mempool(tx):
				tx_id = node.sendrawtransaction(ToHex(tx))
				assert(tx_id in set(node.getrawmempool()))
				return tx_id

				# Setup fundings
				txfund, txspend = create_fund_and_spend_tx(out[0])
				multisigtxfund, multisigtxspend = create_fund_and_spend_tx(
				out[1], multi=True)
				ecdsatxfund, ecdsatxspend = create_fund_and_spend_tx(
				out[2], sig='ecdsa')
				send_transaction_to_mempool(txfund)
				send_transaction_to_mempool(multisigtxfund)
				send_transaction_to_mempool(ecdsatxfund)

				block(1)
				update_block(1, [txfund, multisigtxfund, ecdsatxfund])
				yield accepted()

				# We are before the upgrade, no Schnorrs get in the mempool.
				assert_raises_rpc_error(-26, RPC_BAD_SIG_ERROR,
				node.sendrawtransaction, ToHex(txspend))
				assert_raises_rpc_error(-26, RPC_BAD_SIG_ERROR,
				node.sendrawtransaction, ToHex(multisigtxspend))

				# And block containing them are rejected as well.
				block(2)
				update_block(2, [txspend])
				yield rejected(RejectResult(16, b'blk-bad-inputs'))
				# Rewind bad block
				tip(1)

				block(3)
				update_block(3, [multisigtxspend])
				yield rejected(RejectResult(16, b'blk-bad-inputs'))
				# Rewind bad block
				tip(1)

				if fakeDER:
				fakedtxfund, fakedtxspend = create_fund_and_spend_tx(
				out[5], sig=fakeDER)
				send_transaction_to_mempool(fakedtxfund)
				block(4)
				update_block(4, [fakedtxfund])
				yield accepted()

				# Create a block that would activate the schnorr
				bfork = block(5555)
				bfork.nTime = GREAT_WALL_START_TIME - 1
				update_block(5555, [])
				yield accepted()

				for i in range(5):
				block(5200 + i)
				test.blocks_and_transactions.append([self.tip, True])
				yield test

				# Check we are just before the activation time
				assert_equal(node.getblockheader(node.getbestblockhash())['mediantime'],
				GREAT_WALL_START_TIME - 1)

				# We are just before the upgrade, still no Schnorrs get in the mempool,
				assert_raises_rpc_error(-26, RPC_BAD_SIG_ERROR,
				node.sendrawtransaction, ToHex(txspend))
				assert_raises_rpc_error(-26, RPC_BAD_SIG_ERROR,
				node.sendrawtransaction, ToHex(multisigtxspend))
				# ... nor in blocks.
				block(10)
				update_block(10, [txspend])
				yield rejected(RejectResult(16, b'blk-bad-inputs'))
				# Rewind bad block
				tip(5204)
				block(11)
				update_block(11, [multisigtxspend])
				yield rejected(RejectResult(16, b'blk-bad-inputs'))
				# Rewind bad block
				tip(5204)

				if fakeDER:
				# Throw a "valid" 65-byte ECDSA signature into the mempool just
				# prior to the activation.
				faked_tx_id = send_transaction_to_mempool(fakedtxspend)

				# Put a proper ECDSA transaction into the mempool but it won't
				# be mined...
				ecdsa_tx_id = send_transaction_to_mempool(ecdsatxspend)

				# Activate the Schnorr!
				block(5556)
				yield accepted()

				# Make sure ECDSA is still in -- we don't want to lose uninvolved txns
				# when the upgrade happens.
				assert ecdsa_tx_id in set(node.getrawmempool())

				if fakeDER:
				# The 65-byte ECDSA sig must be ejected.
				assert(faked_tx_id not in set(node.getrawmempool()))
				# If we try to re-add it, it's invalid Schnorr and hence fails
				# VerifySignature, and then raises NULLFAIL.
				assert_raises_rpc_error(-26, RPC_NULLFAIL_ERROR,
				node.sendrawtransaction, ToHex(fakedtxspend))
				# And it can't be mined either...
				block(15)
				update_block(15, [fakedtxspend])
				yield rejected(RejectResult(16, b'blk-bad-inputs'))
				# Rewind bad block
				tip(5556)

				# The multisig throws a different error now
				assert_raises_rpc_error(-26, RPC_BAD_MULTISIG_ERROR,
				node.sendrawtransaction, ToHex(multisigtxspend))
				# And it still can't be mined
				block(16)
				update_block(16, [multisigtxspend])
				yield rejected(RejectResult(16, b'blk-bad-inputs'))
				# Rewind bad block
				tip(5556)

				# The Schnorr transaction is now valid
				spend_tx_id = send_transaction_to_mempool(txspend)
				# It can also be mined
				block(21)
				update_block(21, [txspend, ecdsatxspend])
				yield accepted()
				# (we mined the ecdsa tx too)
				assert spend_tx_id not in set(node.getrawmempool())
				assert ecdsa_tx_id not in set(node.getrawmempool())

				# Ok, now we check if a reorg work properly accross the activation.
				postforkblockid = node.getbestblockhash()
				node.invalidateblock(postforkblockid)
				# txes popped back into mempool
				assert spend_tx_id in set(node.getrawmempool())
				assert ecdsa_tx_id in set(node.getrawmempool())

				# Deactivating upgrade.
				forkblockid = node.getbestblockhash()
				node.invalidateblock(forkblockid)
				# This should kick out the Schnorr sig, but not the valid ECDSA sig.
				assert spend_tx_id not in set(node.getrawmempool())
				assert ecdsa_tx_id in set(node.getrawmempool())

				# Check that we also do it properly on deeper reorg.
				node.reconsiderblock(forkblockid)
				node.reconsiderblock(postforkblockid)
				node.invalidateblock(forkblockid)
				assert spend_tx_id not in set(node.getrawmempool())
				assert ecdsa_tx_id in set(node.getrawmempool())


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