1  #!/usr/bin/env python3
  2  # Copyright (c) 2014-2022 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 the invalidateblock RPC."""
  6  
  7  from test_framework.test_framework import BitcoinTestFramework
  8  from test_framework.address import ADDRESS_BCRT1_UNSPENDABLE_DESCRIPTOR
  9  from test_framework.blocktools import (
 10      create_block,
 11      create_coinbase,
 12  )
 13  from test_framework.util import (
 14      assert_equal,
 15      assert_raises_rpc_error,
 16  )
 17  
 18  
 19  class InvalidateTest(BitcoinTestFramework):
 20      def set_test_params(self):
 21          self.setup_clean_chain = True
 22          self.num_nodes = 3
 23  
 24      def setup_network(self):
 25          self.setup_nodes()
 26  
 27      def run_test(self):
 28          self.log.info("Make sure we repopulate setBlockIndexCandidates after InvalidateBlock:")
 29          self.log.info("Mine 4 blocks on Node 0")
 30          self.generate(self.nodes[0], 4, sync_fun=self.no_op)
 31          assert_equal(self.nodes[0].getblockcount(), 4)
 32          besthash_n0 = self.nodes[0].getbestblockhash()
 33  
 34          self.log.info("Mine competing 6 blocks on Node 1")
 35          self.generate(self.nodes[1], 6, sync_fun=self.no_op)
 36          assert_equal(self.nodes[1].getblockcount(), 6)
 37  
 38          self.log.info("Connect nodes to force a reorg")
 39          self.connect_nodes(0, 1)
 40          self.sync_blocks(self.nodes[0:2])
 41          assert_equal(self.nodes[0].getblockcount(), 6)
 42  
 43          # Add a header to the tip of node 0 without submitting the block. This shouldn't
 44          # affect results since this chain will be invalidated next.
 45          tip = self.nodes[0].getbestblockhash()
 46          block_time = self.nodes[0].getblock(self.nodes[0].getbestblockhash())['time'] + 1
 47          block = create_block(int(tip, 16), create_coinbase(self.nodes[0].getblockcount()), block_time, version=4)
 48          block.solve()
 49          self.nodes[0].submitheader(block.serialize().hex())
 50          assert_equal(self.nodes[0].getblockchaininfo()["headers"], self.nodes[0].getblockchaininfo()["blocks"] + 1)
 51  
 52          self.log.info("Invalidate block 2 on node 0 and verify we reorg to node 0's original chain")
 53          badhash = self.nodes[1].getblockhash(2)
 54          self.nodes[0].invalidateblock(badhash)
 55          assert_equal(self.nodes[0].getblockcount(), 4)
 56          assert_equal(self.nodes[0].getbestblockhash(), besthash_n0)
 57          # Should report consistent blockchain info
 58          assert_equal(self.nodes[0].getblockchaininfo()["headers"], self.nodes[0].getblockchaininfo()["blocks"])
 59  
 60          self.log.info("Reconsider block 6 on node 0 again and verify that the best header is set correctly")
 61          self.nodes[0].reconsiderblock(tip)
 62          assert_equal(self.nodes[0].getblockchaininfo()["headers"], self.nodes[0].getblockchaininfo()["blocks"] + 1)
 63  
 64          self.log.info("Invalidate block 2 on node 0 and verify we reorg to node 0's original chain again")
 65          self.nodes[0].invalidateblock(badhash)
 66          assert_equal(self.nodes[0].getblockcount(), 4)
 67          assert_equal(self.nodes[0].getbestblockhash(), besthash_n0)
 68          assert_equal(self.nodes[0].getblockchaininfo()["headers"], self.nodes[0].getblockchaininfo()["blocks"])
 69  
 70          self.log.info("Make sure we won't reorg to a lower work chain:")
 71          self.connect_nodes(1, 2)
 72          self.log.info("Sync node 2 to node 1 so both have 6 blocks")
 73          self.sync_blocks(self.nodes[1:3])
 74          assert_equal(self.nodes[2].getblockcount(), 6)
 75          self.log.info("Invalidate block 5 on node 1 so its tip is now at 4")
 76          self.nodes[1].invalidateblock(self.nodes[1].getblockhash(5))
 77          assert_equal(self.nodes[1].getblockcount(), 4)
 78          self.log.info("Invalidate block 3 on node 2, so its tip is now 2")
 79          self.nodes[2].invalidateblock(self.nodes[2].getblockhash(3))
 80          assert_equal(self.nodes[2].getblockcount(), 2)
 81          self.log.info("..and then mine a block")
 82          self.generate(self.nodes[2], 1, sync_fun=self.no_op)
 83          self.log.info("Verify all nodes are at the right height")
 84          self.wait_until(lambda: self.nodes[2].getblockcount() == 3, timeout=5)
 85          self.wait_until(lambda: self.nodes[0].getblockcount() == 4, timeout=5)
 86          self.wait_until(lambda: self.nodes[1].getblockcount() == 4, timeout=5)
 87  
 88          self.log.info("Verify that ancestors can become chain tip candidates when we reconsider blocks")
 89          # Invalidate node0's current chain (1' -> 2' -> 3' -> 4') so that we don't reorg back to it in this test
 90          badhash = self.nodes[0].getblockhash(1)
 91          self.nodes[0].invalidateblock(badhash)
 92          # Reconsider the tip so that node0's chain becomes this chain again : 1 -> 2 -> 3 -> 4 -> 5 -> 6 -> header 7
 93          self.nodes[0].reconsiderblock(tip)
 94          blockhash_3 = self.nodes[0].getblockhash(3)
 95          blockhash_4 = self.nodes[0].getblockhash(4)
 96          blockhash_6 = self.nodes[0].getblockhash(6)
 97          assert_equal(self.nodes[0].getbestblockhash(), blockhash_6)
 98  
 99          # Invalidate block 4 so that chain becomes : 1 -> 2 -> 3
100          self.nodes[0].invalidateblock(blockhash_4)
101          assert_equal(self.nodes[0].getbestblockhash(), blockhash_3)
102          assert_equal(self.nodes[0].getblockchaininfo()['blocks'], 3)
103          assert_equal(self.nodes[0].getblockchaininfo()['headers'], 3)
104  
105          # Reconsider the header
106          self.nodes[0].reconsiderblock(block.hash_hex)
107          # Since header doesn't have block data, it can't be chain tip
108          # Check if it's possible for an ancestor (with block data) to be the chain tip
109          assert_equal(self.nodes[0].getbestblockhash(), blockhash_6)
110          assert_equal(self.nodes[0].getblockchaininfo()['blocks'], 6)
111          assert_equal(self.nodes[0].getblockchaininfo()['headers'], 7)
112  
113          self.log.info("Verify that we reconsider all ancestors as well")
114          blocks = self.generatetodescriptor(self.nodes[1], 10, ADDRESS_BCRT1_UNSPENDABLE_DESCRIPTOR, sync_fun=self.no_op)
115          assert_equal(self.nodes[1].getbestblockhash(), blocks[-1])
116          # Invalidate the two blocks at the tip
117          self.nodes[1].invalidateblock(blocks[-1])
118          self.nodes[1].invalidateblock(blocks[-2])
119          assert_equal(self.nodes[1].getbestblockhash(), blocks[-3])
120          # Reconsider only the previous tip
121          self.nodes[1].reconsiderblock(blocks[-1])
122          # Should be back at the tip by now
123          assert_equal(self.nodes[1].getbestblockhash(), blocks[-1])
124  
125          self.log.info("Verify that we reconsider all descendants")
126          blocks = self.generatetodescriptor(self.nodes[1], 10, ADDRESS_BCRT1_UNSPENDABLE_DESCRIPTOR, sync_fun=self.no_op)
127          assert_equal(self.nodes[1].getbestblockhash(), blocks[-1])
128          # Invalidate the two blocks at the tip
129          self.nodes[1].invalidateblock(blocks[-2])
130          self.nodes[1].invalidateblock(blocks[-4])
131          assert_equal(self.nodes[1].getbestblockhash(), blocks[-5])
132          # Reconsider only the previous tip
133          self.nodes[1].reconsiderblock(blocks[-4])
134          # Should be back at the tip by now
135          assert_equal(self.nodes[1].getbestblockhash(), blocks[-1])
136          # Should report consistent blockchain info
137          assert_equal(self.nodes[1].getblockchaininfo()["headers"], self.nodes[1].getblockchaininfo()["blocks"])
138  
139          self.log.info("Verify that invalidating an unknown block throws an error")
140          assert_raises_rpc_error(-5, "Block not found", self.nodes[1].invalidateblock, "00" * 32)
141          assert_equal(self.nodes[1].getbestblockhash(), blocks[-1])
142  
143  
144  if __name__ == '__main__':
145      InvalidateTest(__file__).main()