p2p_invalid_block.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2015-present The Bitcoin Core developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 """Test node responses to invalid blocks.
6
7 In this test we connect to one node over p2p, and test block requests:
8 1) Valid blocks should be requested and become chain tip.
9 2) Invalid block with duplicated transaction should be re-requested.
10 3) Invalid block with bad coinbase value should be rejected and not
11 re-requested.
12 4) Invalid block due to future timestamp is later accepted when that timestamp
13 becomes valid.
14 """
15 import copy
16 import time
17
18 from test_framework.blocktools import (
19 MAX_FUTURE_BLOCK_TIME,
20 create_block,
21 create_coinbase,
22 create_tx_with_script,
23 )
24 from test_framework.messages import COIN
25 from test_framework.p2p import P2PDataStore
26 from test_framework.script import OP_TRUE
27 from test_framework.test_framework import BitcoinTestFramework
28 from test_framework.util import (
29 assert_equal,
30 assert_not_equal,
31 )
32
33
34 class InvalidBlockRequestTest(BitcoinTestFramework):
35 def set_test_params(self):
36 self.num_nodes = 1
37 self.setup_clean_chain = True
38 # whitelist peers to speed up tx relay / mempool sync
39 self.noban_tx_relay = True
40
41 def run_test(self):
42 # Add p2p connection to node0
43 node = self.nodes[0] # convenience reference to the node
44 peer = node.add_p2p_connection(P2PDataStore())
45
46 best_block = node.getblock(node.getbestblockhash())
47 tip = int(node.getbestblockhash(), 16)
48 height = best_block["height"] + 1
49 block_time = best_block["time"] + 1
50
51 self.log.info("Create a new block with an anyone-can-spend coinbase")
52
53 block = create_block(tip, height=height, ntime=block_time)
54 block.solve()
55 # Save the coinbase for later
56 block1 = block
57 peer.send_blocks_and_test([block1], node, success=True)
58
59 self.log.info("Mature the block.")
60 self.generatetoaddress(node, 100, node.get_deterministic_priv_key().address)
61
62 best_block = node.getblock(node.getbestblockhash())
63 tip = int(node.getbestblockhash(), 16)
64 height = best_block["height"] + 1
65 block_time = best_block["time"] + 1
66
67 # Use merkle-root malleability to generate an invalid block with
68 # same blockheader (CVE-2012-2459).
69 # Manufacture a block with 3 transactions (coinbase, spend of prior
70 # coinbase, spend of that spend). Duplicate the 3rd transaction to
71 # leave merkle root and blockheader unchanged but invalidate the block.
72 # For more information on merkle-root malleability see src/consensus/merkle.cpp.
73 self.log.info("Test merkle root malleability.")
74
75 tx1 = create_tx_with_script(block1.vtx[0], 0, script_sig=bytes([OP_TRUE]), amount=50 * COIN)
76 tx2 = create_tx_with_script(tx1, 0, script_sig=bytes([OP_TRUE]), amount=50 * COIN)
77 block2 = create_block(tip, height=height, ntime=block_time, txlist=[tx1, tx2])
78 block_time += 1
79 block2.solve()
80 orig_hash = block2.hash_int
81 block2_orig = copy.deepcopy(block2)
82
83 # Mutate block 2
84 block2.vtx.append(tx2)
85 assert_equal(block2.hashMerkleRoot, block2.calc_merkle_root())
86 assert_equal(orig_hash, block2.hash_int)
87 assert_not_equal(block2_orig.vtx, block2.vtx)
88
89 peer.send_blocks_and_test([block2], node, success=False, reject_reason='bad-txns-duplicate')
90
91 # Check transactions for duplicate inputs (CVE-2018-17144)
92 self.log.info("Test duplicate input block.")
93
94 block2_dup = copy.deepcopy(block2_orig)
95 block2_dup.vtx[2].vin.append(block2_dup.vtx[2].vin[0])
96 block2_dup.hashMerkleRoot = block2_dup.calc_merkle_root()
97 block2_dup.solve()
98 peer.send_blocks_and_test([block2_dup], node, success=False, reject_reason='bad-txns-inputs-duplicate')
99
100 self.log.info("Test very broken block.")
101
102 block3 = create_block(tip, create_coinbase(height, nValue=100), ntime=block_time)
103 block_time += 1
104 block3.solve()
105
106 peer.send_blocks_and_test([block3], node, success=False, reject_reason='bad-cb-amount')
107
108
109 # Complete testing of CVE-2012-2459 by sending the original block.
110 # It should be accepted even though it has the same hash as the mutated one.
111
112 self.log.info("Test accepting original block after rejecting its mutated version.")
113 peer.send_blocks_and_test([block2_orig], node, success=True, timeout=5)
114
115 # Update tip info
116 height += 1
117 block_time += 1
118 tip = block2_orig.hash_int
119
120 # Complete testing of CVE-2018-17144, by checking for the inflation bug.
121 # Create a block that spends the output of a tx in a previous block.
122 tx3 = create_tx_with_script(tx2, 0, script_sig=bytes([OP_TRUE]), amount=50 * COIN)
123 tx3.vin.append(tx3.vin[0]) # Duplicates input
124 block4 = create_block(tip, height=height, ntime=block_time, txlist=[tx3])
125 block4.solve()
126 self.log.info("Test inflation by duplicating input")
127 peer.send_blocks_and_test([block4], node, success=False, reject_reason='bad-txns-inputs-duplicate')
128
129 self.log.info("Test accepting identical block after rejecting it due to a future timestamp.")
130 t = int(time.time())
131 node.setmocktime(t)
132 # Set block time +1 second past max future validity
133 block = create_block(tip, height=height, ntime=t + MAX_FUTURE_BLOCK_TIME + 1)
134 block.solve()
135 # Need force_send because the block will get rejected without a getdata otherwise
136 peer.send_blocks_and_test([block], node, force_send=True, success=False, reject_reason='time-too-new')
137 node.setmocktime(t + 1)
138 peer.send_blocks_and_test([block], node, success=True)
139
140
141 if __name__ == '__main__':
142 InvalidBlockRequestTest(__file__).main()
143