Differential D13876 Diff 40250 test/functional/abc-replay-protection.py

Changeset View

Standalone View

test/functional/abc-replay-protection.py

Show All 29 Lines
)		)
from test_framework.test_framework import BitcoinTestFramework		from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, assert_raises_rpc_error		from test_framework.util import assert_equal, assert_raises_rpc_error

# far into the future		# far into the future
REPLAY_PROTECTION_START_TIME = 2000000000		REPLAY_PROTECTION_START_TIME = 2000000000

# Error due to invalid signature		# Error due to invalid signature
RPC_INVALID_SIGNATURE_ERROR = "mandatory-script-verify-flag-failed (Signature must be zero for failed CHECK(MULTI)SIG operation)"		RPC_INVALID_SIGNATURE_ERROR = (
		"mandatory-script-verify-flag-failed (Signature must be zero for failed"
		" CHECK(MULTI)SIG operation)"
		)


class PreviousSpendableOutput(object):		class PreviousSpendableOutput(object):

def __init__(self, tx=CTransaction(), n=-1):		def __init__(self, tx=CTransaction(), n=-1):
self.tx = tx		self.tx = tx
self.n = n		self.n = n


class ReplayProtectionTest(BitcoinTestFramework):		class ReplayProtectionTest(BitcoinTestFramework):

def set_test_params(self):		def set_test_params(self):
self.num_nodes = 1		self.num_nodes = 1
self.setup_clean_chain = True		self.setup_clean_chain = True
self.block_heights = {}		self.block_heights = {}
self.tip = None		self.tip = None
self.blocks = {}		self.blocks = {}
self.extra_args = [['-whitelist=noban@127.0.0.1',		self.extra_args = [
		[
		"-whitelist=noban@127.0.0.1",
f"-replayprotectionactivationtime={REPLAY_PROTECTION_START_TIME}",		f"-replayprotectionactivationtime={REPLAY_PROTECTION_START_TIME}",
"-acceptnonstdtxn=1"]]		"-acceptnonstdtxn=1",
		]
		]

def next_block(self, number):		def next_block(self, number):
if self.tip is None:		if self.tip is None:
base_block_hash = self.genesis_hash		base_block_hash = self.genesis_hash
block_time = int(time.time()) + 1		block_time = int(time.time()) + 1
else:		else:
base_block_hash = self.tip.sha256		base_block_hash = self.tip.sha256
block_time = self.tip.nTime + 1		block_time = self.tip.nTime + 1
Show All 39 Lines	def run_test(self):
block.vtx.extend(new_transactions)		block.vtx.extend(new_transactions)
old_sha256 = block.sha256		old_sha256 = block.sha256
make_conform_to_ctor(block)		make_conform_to_ctor(block)
block.hashMerkleRoot = block.calc_merkle_root()		block.hashMerkleRoot = block.calc_merkle_root()
block.solve()		block.solve()
# Update the internal state just like in next_block		# Update the internal state just like in next_block
self.tip = block		self.tip = block
if block.sha256 != old_sha256:		if block.sha256 != old_sha256:
self.block_heights[		self.block_heights[block.sha256] = self.block_heights[old_sha256]
block.sha256] = self.block_heights[old_sha256]
del self.block_heights[old_sha256]		del self.block_heights[old_sha256]
self.blocks[block_number] = block		self.blocks[block_number] = block
return block		return block

# shorthand		# shorthand
block = self.next_block		block = self.next_block

# Create a new block		# Create a new block
Show All 19 Lines	def run_test(self):
private_key.generate()		private_key.generate()
public_key = private_key.get_pubkey().get_bytes()		public_key = private_key.get_pubkey().get_bytes()

# This is a little handier to use than the version in blocktools.py		# This is a little handier to use than the version in blocktools.py
def create_fund_and_spend_tx(spend, forkvalue=0):		def create_fund_and_spend_tx(spend, forkvalue=0):
# Fund transaction		# Fund transaction
script = CScript([public_key, OP_CHECKSIG])		script = CScript([public_key, OP_CHECKSIG])
txfund = create_tx_with_script(		txfund = create_tx_with_script(
spend.tx, spend.n, b'', amount=50 * COIN - 1000, script_pub_key=script)		spend.tx, spend.n, b"", amount=50 * COIN - 1000, script_pub_key=script
		)
txfund.rehash()		txfund.rehash()

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

# Sign the transaction		# Sign the transaction
sighashtype = (forkvalue << 8) \| SIGHASH_ALL \| SIGHASH_FORKID		sighashtype = (forkvalue << 8) \| SIGHASH_ALL \| SIGHASH_FORKID
sighash = SignatureHashForkId(		sighash = SignatureHashForkId(
script, txspend, 0, sighashtype, 50 * COIN - 1000)		script, txspend, 0, sighashtype, 50 * COIN - 1000
sig = private_key.sign_ecdsa(sighash) + \		)
bytes(bytearray([SIGHASH_ALL \| SIGHASH_FORKID]))		sig = private_key.sign_ecdsa(sighash) + bytes(
		bytearray([SIGHASH_ALL \| SIGHASH_FORKID])
		)
txspend.vin[0].scriptSig = CScript([sig])		txspend.vin[0].scriptSig = CScript([sig])
txspend.rehash()		txspend.rehash()

return [txfund, txspend]		return [txfund, txspend]

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

# Before the fork, no replay protection required to get in the mempool.		# Before the fork, no replay protection required to get in the mempool.
txns = create_fund_and_spend_tx(out[0])		txns = create_fund_and_spend_tx(out[0])
send_transaction_to_mempool(txns[0])		send_transaction_to_mempool(txns[0])
send_transaction_to_mempool(txns[1])		send_transaction_to_mempool(txns[1])

# And txns get mined in a block properly.		# And txns get mined in a block properly.
block(1)		block(1)
update_block(1, txns)		update_block(1, txns)
peer.send_blocks_and_test([self.tip], node)		peer.send_blocks_and_test([self.tip], node)

# Replay protected transactions are rejected.		# Replay protected transactions are rejected.
replay_txns = create_fund_and_spend_tx(out[1], 0xffdead)		replay_txns = create_fund_and_spend_tx(out[1], 0xFFDEAD)
send_transaction_to_mempool(replay_txns[0])		send_transaction_to_mempool(replay_txns[0])
assert_raises_rpc_error(-26, RPC_INVALID_SIGNATURE_ERROR,		assert_raises_rpc_error(
node.sendrawtransaction, ToHex(replay_txns[1]))		-26,
		RPC_INVALID_SIGNATURE_ERROR,
		node.sendrawtransaction,
		ToHex(replay_txns[1]),
		)

# And block containing them are rejected as well.		# And block containing them are rejected as well.
block(2)		block(2)
update_block(2, replay_txns)		update_block(2, replay_txns)
peer.send_blocks_and_test(		peer.send_blocks_and_test(
[self.tip], node, success=False, reject_reason='blk-bad-inputs')		[self.tip], node, success=False, reject_reason="blk-bad-inputs"
		)

# Rewind bad block		# Rewind bad block
self.set_tip(1)		self.set_tip(1)

# Create a block that would activate the replay protection.		# Create a block that would activate the replay protection.
bfork = block(5555)		bfork = block(5555)
bfork.nTime = REPLAY_PROTECTION_START_TIME - 1		bfork.nTime = REPLAY_PROTECTION_START_TIME - 1
update_block(5555, [])		update_block(5555, [])
peer.send_blocks_and_test([self.tip], node)		peer.send_blocks_and_test([self.tip], node)

activation_blocks = []		activation_blocks = []
for i in range(5):		for i in range(5):
block(5100 + i)		block(5100 + i)
activation_blocks.append(self.tip)		activation_blocks.append(self.tip)
peer.send_blocks_and_test(activation_blocks, node)		peer.send_blocks_and_test(activation_blocks, node)

# Check we are just before the activation time		# Check we are just before the activation time
assert_equal(		assert_equal(
node.getblockchaininfo()['mediantime'],		node.getblockchaininfo()["mediantime"], REPLAY_PROTECTION_START_TIME - 1
REPLAY_PROTECTION_START_TIME - 1)		)

# We are just before the fork, replay protected txns still are rejected		# We are just before the fork, replay protected txns still are rejected
assert_raises_rpc_error(-26, RPC_INVALID_SIGNATURE_ERROR,		assert_raises_rpc_error(
node.sendrawtransaction, ToHex(replay_txns[1]))		-26,
		RPC_INVALID_SIGNATURE_ERROR,
		node.sendrawtransaction,
		ToHex(replay_txns[1]),
		)

block(3)		block(3)
update_block(3, replay_txns)		update_block(3, replay_txns)
peer.send_blocks_and_test(		peer.send_blocks_and_test(
[self.tip], node, success=False, reject_reason='blk-bad-inputs')		[self.tip], node, success=False, reject_reason="blk-bad-inputs"
		)

# Rewind bad block		# Rewind bad block
self.set_tip(5104)		self.set_tip(5104)

# Send some non replay protected txns in the mempool to check		# Send some non replay protected txns in the mempool to check
# they get cleaned at activation.		# they get cleaned at activation.
txns = create_fund_and_spend_tx(out[2])		txns = create_fund_and_spend_tx(out[2])
send_transaction_to_mempool(txns[0])		send_transaction_to_mempool(txns[0])
tx_id = send_transaction_to_mempool(txns[1])		tx_id = send_transaction_to_mempool(txns[1])

# Activate the replay protection		# Activate the replay protection
block(5556)		block(5556)
peer.send_blocks_and_test([self.tip], node)		peer.send_blocks_and_test([self.tip], node)

# Check we just activated the replay protection		# Check we just activated the replay protection
assert_equal(		assert_equal(
node.getblockchaininfo()['mediantime'],		node.getblockchaininfo()["mediantime"], REPLAY_PROTECTION_START_TIME
REPLAY_PROTECTION_START_TIME)		)

# Non replay protected transactions are not valid anymore,		# Non replay protected transactions are not valid anymore,
# so they should be removed from the mempool.		# so they should be removed from the mempool.
assert tx_id not in set(node.getrawmempool())		assert tx_id not in set(node.getrawmempool())

# Good old transactions are now invalid.		# Good old transactions are now invalid.
send_transaction_to_mempool(txns[0])		send_transaction_to_mempool(txns[0])
assert_raises_rpc_error(-26, RPC_INVALID_SIGNATURE_ERROR,		assert_raises_rpc_error(
node.sendrawtransaction, ToHex(txns[1]))		-26, RPC_INVALID_SIGNATURE_ERROR, node.sendrawtransaction, ToHex(txns[1])
		)

# They also cannot be mined		# They also cannot be mined
block(4)		block(4)
update_block(4, txns)		update_block(4, txns)
peer.send_blocks_and_test(		peer.send_blocks_and_test(
[self.tip], node, success=False, reject_reason='blk-bad-inputs')		[self.tip], node, success=False, reject_reason="blk-bad-inputs"
		)

# Rewind bad block		# Rewind bad block
self.set_tip(5556)		self.set_tip(5556)

# The replay protected transaction is now valid		# The replay protected transaction is now valid
replay_tx0_id = send_transaction_to_mempool(replay_txns[0])		replay_tx0_id = send_transaction_to_mempool(replay_txns[0])
replay_tx1_id = send_transaction_to_mempool(replay_txns[1])		replay_tx1_id = send_transaction_to_mempool(replay_txns[1])

# Make sure the transaction are ready to be mined.		# Make sure the transaction are ready to be mined.
tmpl = node.getblocktemplate()		tmpl = node.getblocktemplate()

found_id0 = False		found_id0 = False
found_id1 = False		found_id1 = False

for txn in tmpl['transactions']:		for txn in tmpl["transactions"]:
txid = txn['txid']		txid = txn["txid"]
if txid == replay_tx0_id:		if txid == replay_tx0_id:
found_id0 = True		found_id0 = True
elif txid == replay_tx1_id:		elif txid == replay_tx1_id:
found_id1 = True		found_id1 = True

assert found_id0 and found_id1		assert found_id0 and found_id1

# And the mempool is still in good shape.		# And the mempool is still in good shape.
Show All 22 Lines	def run_test(self):
# Check that we also do it properly on deeper reorg.		# Check that we also do it properly on deeper reorg.
node.reconsiderblock(forkblockid)		node.reconsiderblock(forkblockid)
node.reconsiderblock(postforkblockid)		node.reconsiderblock(postforkblockid)
node.invalidateblock(forkblockid)		node.invalidateblock(forkblockid)
assert replay_tx0_id in set(node.getrawmempool())		assert replay_tx0_id in set(node.getrawmempool())
assert replay_tx1_id not in set(node.getrawmempool())		assert replay_tx1_id not in set(node.getrawmempool())


if __name__ == '__main__':		if __name__ == "__main__":
ReplayProtectionTest().main()		ReplayProtectionTest().main()