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