Differential D2116 Diff 6030 test/functional/bip65-cltv-p2p.py

Changeset View

Standalone View

test/functional/bip65-cltv-p2p.py

#!/usr/bin/env python3		#!/usr/bin/env python3
# Copyright (c) 2015-2016 The Bitcoin Core developers		# Copyright (c) 2015-2016 The Bitcoin Core 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 BIP65 (CHECKLOCKTIMEVERIFY).		"""Test BIP65 (CHECKLOCKTIMEVERIFY).

Test that the CHECKLOCKTIMEVERIFY soft-fork activates at (regtest) block height		Test that the CHECKLOCKTIMEVERIFY soft-fork activates at (regtest) block height
1351.		1351.
"""		"""

from test_framework.test_framework import BitcoinTestFramework		from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import *		from test_framework.util import *
from test_framework.mininode import *		from test_framework.mininode import *
from test_framework.blocktools import create_coinbase, create_block		from test_framework.blocktools import create_coinbase, create_block
from test_framework.script import CScript, OP_1NEGATE, OP_CHECKLOCKTIMEVERIFY, OP_DROP, CScriptNum		from test_framework.script import CScript, CScriptNum, OP_1NEGATE, OP_CHECKLOCKTIMEVERIFY, OP_DROP, OP_TRUE
		from test_framework.txtools import pad_tx

CLTV_HEIGHT = 1351		CLTV_HEIGHT = 1351

# far in the future
MAGNETIC_ANOMALY_START_TIME = 2000000000

# Reject codes that we might receive in this test		# Reject codes that we might receive in this test
REJECT_INVALID = 16		REJECT_INVALID = 16
REJECT_OBSOLETE = 17		REJECT_OBSOLETE = 17
REJECT_NONSTANDARD = 64		REJECT_NONSTANDARD = 64


def cltv_invalidate(tx):		def cltv_lock_to_height(node, tx, height=-1):
'''Modify the signature in vin 0 of the tx to fail CLTV		'''Modify the scriptPubKey to add an OP_CHECKLOCKTIMEVERIFY

		This transforms the script to anyonecanpay (OP_TRUE) if the lock time
		condition is valid.

Prepends -1 CLTV DROP in the scriptSig itself.		Default height is -1 which leads CLTV to fail

TODO: test more ways that transactions using CLTV could be invalid (eg		TODO: test more ways that transactions using CLTV could be invalid (eg
locktime requirements fail, sequence time requirements fail, etc).		locktime requirements fail, sequence time requirements fail, etc).
'''		'''
tx.vin[0].scriptSig = CScript([OP_1NEGATE, OP_CHECKLOCKTIMEVERIFY, OP_DROP] +		height_op = OP_1NEGATE
list(CScript(tx.vin[0].scriptSig)))		if(height > 0):
tx.rehash()


def cltv_validate(node, tx, height):
'''Modify the signature in vin 0 of the tx to pass CLTV
Prepends <height> CLTV DROP in the scriptSig, and sets
the locktime to height'''
tx.vin[0].nSequence = 0		tx.vin[0].nSequence = 0
tx.nLockTime = height		tx.nLockTime = height
		height_op = CScriptNum(height)

		tx.vout[0].scriptPubKey = CScript(
		[height_op, OP_CHECKLOCKTIMEVERIFY, OP_DROP, OP_TRUE])
		tx.rehash()

# Need to re-sign, since nSequence and nLockTime changed
signed_result = node.signrawtransaction(ToHex(tx))		signed_result = node.signrawtransaction(ToHex(tx))
new_tx = CTransaction()
new_tx.deserialize(BytesIO(hex_str_to_bytes(signed_result['hex'])))

new_tx.vin[0].scriptSig = CScript([CScriptNum(height), OP_CHECKLOCKTIMEVERIFY, OP_DROP] +		new_tx = FromHex(CTransaction(), signed_result['hex'])
list(CScript(new_tx.vin[0].scriptSig)))		pad_tx(new_tx)
		new_tx.rehash()

return new_tx		return new_tx


def create_transaction(node, coinbase, to_address, amount):		def create_transaction(node, coinbase, to_address, amount):
from_txid = node.getblock(coinbase)['tx'][0]		from_txid = node.getblock(coinbase)['tx'][0]
inputs = [{"txid": from_txid, "vout": 0}]		inputs = [{"txid": from_txid, "vout": 0}]
outputs = {to_address: amount}		outputs = {to_address: amount}
rawtx = node.createrawtransaction(inputs, outputs)		rawtx = node.createrawtransaction(inputs, outputs)
signresult = node.signrawtransaction(rawtx)		signresult = node.signrawtransaction(rawtx)
tx = FromHex(CTransaction(), signresult['hex'])		tx = FromHex(CTransaction(), signresult['hex'])
return tx		return tx


class BIP65Test(BitcoinTestFramework):		class BIP65Test(BitcoinTestFramework):
def set_test_params(self):		def set_test_params(self):
self.num_nodes = 1		self.num_nodes = 1
self.extra_args = [		self.extra_args = [
['-promiscuousmempoolflags=1', '-whitelist=127.0.0.1', '-magneticanomalyactivationtime=%d' % MAGNETIC_ANOMALY_START_TIME]]		['-promiscuousmempoolflags=1', '-whitelist=127.0.0.1']]
self.setup_clean_chain = True		self.setup_clean_chain = True

def run_test(self):		def run_test(self):
node0 = NodeConnCB()		node0 = NodeConnCB()
connections = []		connections = []
connections.append(		connections.append(
NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0))		NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0))
node0.add_connection(connections[0])		node0.add_connection(connections[0])

# Start up network handling in another thread		# Start up network handling in another thread
NetworkThread().start()		NetworkThread().start()

# wait_for_verack ensures that the P2P connection is fully up.		# wait_for_verack ensures that the P2P connection is fully up.
node0.wait_for_verack()		node0.wait_for_verack()

self.log.info("Mining %d blocks", CLTV_HEIGHT - 2)		self.log.info("Mining %d blocks", CLTV_HEIGHT - 2)
self.coinbase_blocks = self.nodes[0].generate(CLTV_HEIGHT - 2)		self.coinbase_blocks = self.nodes[0].generate(CLTV_HEIGHT - 2)
self.nodeaddress = self.nodes[0].getnewaddress()		self.nodeaddress = self.nodes[0].getnewaddress()

self.log.info(		self.log.info(
"Test that an invalid-according-to-CLTV transaction can still appear in a block")		"Test that an invalid-according-to-CLTV transaction can still appear in a block")

spendtx = create_transaction(self.nodes[0], self.coinbase_blocks[0],		spendtx = create_transaction(self.nodes[0], self.coinbase_blocks[0],
self.nodeaddress, 50.0)		self.nodeaddress, 50.0)
cltv_invalidate(spendtx)		spendtx = cltv_lock_to_height(self.nodes[0], spendtx)

# Make sure the tx is valid		# Make sure the tx is valid
self.nodes[0].sendrawtransaction(ToHex(spendtx))		self.nodes[0].sendrawtransaction(ToHex(spendtx))

tip = self.nodes[0].getbestblockhash()		tip = self.nodes[0].getbestblockhash()
block_time = self.nodes[0].getblockheader(tip)['mediantime'] + 1		block_time = self.nodes[0].getblockheader(tip)['mediantime'] + 1
block = create_block(int(tip, 16), create_coinbase(		block = create_block(int(tip, 16), create_coinbase(
CLTV_HEIGHT - 1), block_time)		CLTV_HEIGHT - 1), block_time)
Show All 23 Lines	def run_test(self):
assert_equal(node0.last_message["reject"].data, block.sha256)		assert_equal(node0.last_message["reject"].data, block.sha256)
del node0.last_message["reject"]		del node0.last_message["reject"]

self.log.info(		self.log.info(
"Test that invalid-according-to-cltv transactions cannot appear in a block")		"Test that invalid-according-to-cltv transactions cannot appear in a block")
block.nVersion = 4		block.nVersion = 4

spendtx = create_transaction(self.nodes[0], self.coinbase_blocks[1],		spendtx = create_transaction(self.nodes[0], self.coinbase_blocks[1],
self.nodeaddress, 1.0)		self.nodeaddress, 49.99)
cltv_invalidate(spendtx)		spendtx = cltv_lock_to_height(self.nodes[0], spendtx)

# First we show that this tx is valid except for CLTV by getting it		# First we show that this tx is valid except for CLTV by getting it
# accepted to the mempool (which we can achieve with		# accepted to the mempool (which we can achieve with
# -promiscuousmempoolflags).		# -promiscuousmempoolflags).
node0.send_and_ping(msg_tx(spendtx))		node0.send_and_ping(msg_tx(spendtx))
assert spendtx.hash in self.nodes[0].getrawmempool()		assert spendtx.hash in self.nodes[0].getrawmempool()

# Now we verify that a block with this transaction is invalid.		# Now we verify that a block spending this transaction is invalid.
block.vtx.append(spendtx)		block.vtx.append(spendtx)
block.hashMerkleRoot = block.calc_merkle_root()		block.hashMerkleRoot = block.calc_merkle_root()
block.solve()		block.solve()

node0.send_and_ping(msg_block(block))		node0.send_and_ping(msg_block(block))
		# This block is still valid
		assert_equal(int(self.nodes[0].getbestblockhash(), 16), block.sha256)

		# But a block containing a transaction spending this utxo is not
		rawspendtx = self.nodes[0].decoderawtransaction(ToHex(spendtx))
		inputs = [{
		"txid": rawspendtx['txid'],
		"vout": rawspendtx['vout'][0]['n']
		}]
		output = {self.nodeaddress: 49.98}

		rejectedtx_raw = self.nodes[0].createrawtransaction(inputs, output)
		rejectedtx_signed = self.nodes[0].signrawtransaction(rejectedtx_raw)

		#!Couldn't complete signature due to CLTV
		assert(rejectedtx_signed['errors'][0]['error'] == 'Negative locktime')

		rejectedtx = FromHex(CTransaction(), rejectedtx_signed['hex'])
		pad_tx(rejectedtx)
		rejectedtx.rehash()

		tip = block.sha256
		block_time += 1
		block = create_block(tip, create_coinbase(CLTV_HEIGHT+1), block_time)
		block.nVersion = 4
		block.vtx.append(rejectedtx)
		block.hashMerkleRoot = block.calc_merkle_root()
		block.solve()

		node0.send_and_ping(msg_block(block))
		# This block is invalid
assert_equal(int(self.nodes[0].getbestblockhash(), 16), tip)		assert_equal(int(self.nodes[0].getbestblockhash(), 16), tip)

wait_until(lambda: "reject" in node0.last_message.keys(),		wait_until(lambda: "reject" in node0.last_message.keys(),
lock=mininode_lock)		lock=mininode_lock)
with mininode_lock:		with mininode_lock:
assert node0.last_message["reject"].code in [		assert node0.last_message["reject"].code in [
REJECT_INVALID, REJECT_NONSTANDARD]		REJECT_INVALID, REJECT_NONSTANDARD]
assert_equal(node0.last_message["reject"].data, block.sha256)		assert_equal(node0.last_message["reject"].data, block.sha256)
if node0.last_message["reject"].code == REJECT_INVALID:		if node0.last_message["reject"].code == REJECT_INVALID:
# Generic rejection when a block is invalid		# Generic rejection when a block is invalid
assert_equal(		assert_equal(
node0.last_message["reject"].reason, b'blk-bad-inputs')		node0.last_message["reject"].reason, b'blk-bad-inputs')
else:		else:
assert b'Negative locktime' in node0.last_message["reject"].reason		assert b'Negative locktime' in node0.last_message["reject"].reason

self.log.info(		self.log.info(
"Test that a version 4 block with a valid-according-to-CLTV transaction is accepted")		"Test that a version 4 block with a valid-according-to-CLTV transaction is accepted")
spendtx = cltv_validate(self.nodes[0], spendtx, CLTV_HEIGHT - 1)		spendtx = create_transaction(self.nodes[0], self.coinbase_blocks[2],
spendtx.rehash()		self.nodeaddress, 49.99)
		spendtx = cltv_lock_to_height(self.nodes[0], spendtx, CLTV_HEIGHT - 1)

		# Modify the transaction in the block to be valid against CLTV
block.vtx.pop(1)		block.vtx.pop(1)
block.vtx.append(spendtx)		block.vtx.append(spendtx)
block.hashMerkleRoot = block.calc_merkle_root()		block.hashMerkleRoot = block.calc_merkle_root()
block.solve()		block.solve()

node0.send_and_ping(msg_block(block))		node0.send_and_ping(msg_block(block))
		# This block is now valid
		assert_equal(int(self.nodes[0].getbestblockhash(), 16), block.sha256)

		# A block containing a transaction spending this utxo is also valid
		# Build this transaction
		rawspendtx = self.nodes[0].decoderawtransaction(ToHex(spendtx))
		inputs = [{
		"txid": rawspendtx['txid'],
		"vout": rawspendtx['vout'][0]['n'],
		"sequence": 0
		}]
		output = {self.nodeaddress: 49.98}

		validtx_raw = self.nodes[0].createrawtransaction(
		inputs, output, CLTV_HEIGHT)

		validtx = FromHex(CTransaction(), validtx_raw)

		# Signrawtransaction won't sign a non standard tx.
		# But the prevout being anyoncanpay, scriptsig can be left empty
		validtx.vin[0].scriptSig = CScript()
		pad_tx(validtx)
		validtx.rehash()

		tip = block.sha256
		block_time += 1
		block = create_block(tip, create_coinbase(CLTV_HEIGHT+3), block_time)
		block.nVersion = 4
		block.vtx.append(validtx)
		block.hashMerkleRoot = block.calc_merkle_root()
		block.solve()

		node0.send_and_ping(msg_block(block))
		# This block is valid
assert_equal(int(self.nodes[0].getbestblockhash(), 16), block.sha256)		assert_equal(int(self.nodes[0].getbestblockhash(), 16), block.sha256)


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