1 #!/usr/bin/env python3
2 # Copyright (c) 2014-2022 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 BIP68 implementation."""
6 7 import time
8 9 from test_framework.blocktools import (
10 NORMAL_GBT_REQUEST_PARAMS,
11 add_witness_commitment,
12 create_block,
13 script_to_p2wsh_script,
14 )
15 from test_framework.messages import (
16 COIN,
17 COutPoint,
18 CTransaction,
19 CTxIn,
20 CTxInWitness,
21 CTxOut,
22 tx_from_hex,
23 )
24 from test_framework.script import (
25 CScript,
26 OP_TRUE,
27 )
28 from test_framework.test_framework import LimenkaTestFramework
29 from test_framework.util import (
30 assert_equal,
31 assert_greater_than,
32 assert_raises_rpc_error,
33 softfork_active,
34 )
35 from test_framework.wallet import MiniWallet
36 37 SCRIPT_W0_SH_OP_TRUE = script_to_p2wsh_script(CScript([OP_TRUE]))
38 39 SEQUENCE_LOCKTIME_DISABLE_FLAG = (1<<31)
40 SEQUENCE_LOCKTIME_TYPE_FLAG = (1<<22) # this means use time (0 means height)
41 SEQUENCE_LOCKTIME_GRANULARITY = 9 # this is a bit-shift
42 SEQUENCE_LOCKTIME_MASK = 0x0000ffff
43 44 # RPC error for non-BIP68 final transactions
45 NOT_FINAL_ERROR = "non-BIP68-final"
46 47 class BIP68Test(LimenkaTestFramework):
48 def add_options(self, parser):
49 self.add_wallet_options(parser)
50 51 def set_test_params(self):
52 self.num_nodes = 2
53 self.extra_args = [
54 [
55 '-testactivationheight=csv@432',
56 ],
57 [
58 '-testactivationheight=csv@432',
59 ],
60 ]
61 62 def run_test(self):
63 self.relayfee = self.nodes[0].getnetworkinfo()["relayfee"]
64 self.wallet = MiniWallet(self.nodes[0])
65 66 self.log.info("Running test disable flag")
67 self.test_disable_flag()
68 69 self.log.info("Running test sequence-lock-confirmed-inputs")
70 self.test_sequence_lock_confirmed_inputs()
71 72 self.log.info("Running test sequence-lock-unconfirmed-inputs")
73 self.test_sequence_lock_unconfirmed_inputs()
74 75 self.log.info("Running test BIP68 not consensus before activation")
76 self.test_bip68_not_consensus()
77 78 self.log.info("Activating BIP68 (and 112/113)")
79 self.activateCSV()
80 81 self.log.info("Verifying version=2 transactions are standard.")
82 self.log.info("Note that version=2 transactions are always standard (independent of BIP68 activation status).")
83 self.test_version2_relay()
84 85 self.log.info("Passed")
86 87 # Test that BIP68 is not in effect if tx version is 1, or if
88 # the first sequence bit is set.
89 def test_disable_flag(self):
90 # Create some unconfirmed inputs
91 utxo = self.wallet.send_self_transfer(from_node=self.nodes[0])["new_utxo"]
92 93 tx1 = CTransaction()
94 value = int((utxo["value"] - self.relayfee) * COIN)
95 96 # Check that the disable flag disables relative locktime.
97 # If sequence locks were used, this would require 1 block for the
98 # input to mature.
99 sequence_value = SEQUENCE_LOCKTIME_DISABLE_FLAG | 1
100 tx1.vin = [CTxIn(COutPoint(int(utxo["txid"], 16), utxo["vout"]), nSequence=sequence_value)]
101 tx1.vout = [CTxOut(value, SCRIPT_W0_SH_OP_TRUE)]
102 103 self.wallet.sign_tx(tx=tx1)
104 tx1_id = self.wallet.sendrawtransaction(from_node=self.nodes[0], tx_hex=tx1.serialize().hex())
105 tx1_id = int(tx1_id, 16)
106 107 # This transaction will enable sequence-locks, so this transaction should
108 # fail
109 tx2 = CTransaction()
110 tx2.version = 2
111 sequence_value = sequence_value & 0x7fffffff
112 tx2.vin = [CTxIn(COutPoint(tx1_id, 0), nSequence=sequence_value)]
113 tx2.wit.vtxinwit = [CTxInWitness()]
114 tx2.wit.vtxinwit[0].scriptWitness.stack = [CScript([OP_TRUE])]
115 tx2.vout = [CTxOut(int(value - self.relayfee * COIN), SCRIPT_W0_SH_OP_TRUE)]
116 tx2.rehash()
117 118 assert_raises_rpc_error(-26, NOT_FINAL_ERROR, self.wallet.sendrawtransaction, from_node=self.nodes[0], tx_hex=tx2.serialize().hex())
119 120 # Setting the version back down to 1 should disable the sequence lock,
121 # so this should be accepted.
122 tx2.version = 1
123 124 self.wallet.sendrawtransaction(from_node=self.nodes[0], tx_hex=tx2.serialize().hex())
125 126 # Calculate the median time past of a prior block ("confirmations" before
127 # the current tip).
128 def get_median_time_past(self, confirmations):
129 block_hash = self.nodes[0].getblockhash(self.nodes[0].getblockcount()-confirmations)
130 return self.nodes[0].getblockheader(block_hash)["mediantime"]
131 132 # Test that sequence locks are respected for transactions spending confirmed inputs.
133 def test_sequence_lock_confirmed_inputs(self):
134 # Create lots of confirmed utxos, and use them to generate lots of random
135 # transactions.
136 max_outputs = 50
137 while len(self.wallet.get_utxos(include_immature_coinbase=False, mark_as_spent=False)) < 200:
138 import random
139 num_outputs = random.randint(1, max_outputs)
140 self.wallet.send_self_transfer_multi(from_node=self.nodes[0], num_outputs=num_outputs)
141 self.generate(self.wallet, 1)
142 143 utxos = self.wallet.get_utxos(include_immature_coinbase=False)
144 145 # Try creating a lot of random transactions.
146 # Each time, choose a random number of inputs, and randomly set
147 # some of those inputs to be sequence locked (and randomly choose
148 # between height/time locking). Small random chance of making the locks
149 # all pass.
150 for _ in range(400):
151 available_utxos = len(utxos)
152 153 # Randomly choose up to 10 inputs
154 num_inputs = random.randint(1, min(10, available_utxos))
155 random.shuffle(utxos)
156 157 # Track whether any sequence locks used should fail
158 should_pass = True
159 160 # Track whether this transaction was built with sequence locks
161 using_sequence_locks = False
162 163 tx = CTransaction()
164 tx.version = 2
165 value = 0
166 for j in range(num_inputs):
167 sequence_value = 0xfffffffe # this disables sequence locks
168 169 # 50% chance we enable sequence locks
170 if random.randint(0,1):
171 using_sequence_locks = True
172 173 # 10% of the time, make the input sequence value pass
174 input_will_pass = (random.randint(1,10) == 1)
175 sequence_value = utxos[j]["confirmations"]
176 if not input_will_pass:
177 sequence_value += 1
178 should_pass = False
179 180 # Figure out what the median-time-past was for the confirmed input
181 # Note that if an input has N confirmations, we're going back N blocks
182 # from the tip so that we're looking up MTP of the block
183 # PRIOR to the one the input appears in, as per the BIP68 spec.
184 orig_time = self.get_median_time_past(utxos[j]["confirmations"])
185 cur_time = self.get_median_time_past(0) # MTP of the tip
186 187 # can only timelock this input if it's not too old -- otherwise use height
188 can_time_lock = True
189 if ((cur_time - orig_time) >> SEQUENCE_LOCKTIME_GRANULARITY) >= SEQUENCE_LOCKTIME_MASK:
190 can_time_lock = False
191 192 # if time-lockable, then 50% chance we make this a time lock
193 if random.randint(0,1) and can_time_lock:
194 # Find first time-lock value that fails, or latest one that succeeds
195 time_delta = sequence_value << SEQUENCE_LOCKTIME_GRANULARITY
196 if input_will_pass and time_delta > cur_time - orig_time:
197 sequence_value = ((cur_time - orig_time) >> SEQUENCE_LOCKTIME_GRANULARITY)
198 elif (not input_will_pass and time_delta <= cur_time - orig_time):
199 sequence_value = ((cur_time - orig_time) >> SEQUENCE_LOCKTIME_GRANULARITY)+1
200 sequence_value |= SEQUENCE_LOCKTIME_TYPE_FLAG
201 tx.vin.append(CTxIn(COutPoint(int(utxos[j]["txid"], 16), utxos[j]["vout"]), nSequence=sequence_value))
202 value += utxos[j]["value"]*COIN
203 # Overestimate the size of the tx - signatures should be less than 120 bytes, and leave 50 for the output
204 tx_size = len(tx.serialize().hex())//2 + 120*num_inputs + 50
205 tx.vout.append(CTxOut(int(value - self.relayfee * tx_size * COIN / 1000), SCRIPT_W0_SH_OP_TRUE))
206 self.wallet.sign_tx(tx=tx)
207 208 if (using_sequence_locks and not should_pass):
209 # This transaction should be rejected
210 assert_raises_rpc_error(-26, NOT_FINAL_ERROR, self.wallet.sendrawtransaction, from_node=self.nodes[0], tx_hex=tx.serialize().hex())
211 else:
212 # This raw transaction should be accepted
213 self.wallet.sendrawtransaction(from_node=self.nodes[0], tx_hex=tx.serialize().hex())
214 self.wallet.rescan_utxos()
215 utxos = self.wallet.get_utxos(include_immature_coinbase=False)
216 217 # Test that sequence locks on unconfirmed inputs must have nSequence
218 # height or time of 0 to be accepted.
219 # Then test that BIP68-invalid transactions are removed from the mempool
220 # after a reorg.
221 def test_sequence_lock_unconfirmed_inputs(self):
222 # Store height so we can easily reset the chain at the end of the test
223 cur_height = self.nodes[0].getblockcount()
224 225 # Create a mempool tx.
226 self.wallet.rescan_utxos()
227 tx1 = self.wallet.send_self_transfer(from_node=self.nodes[0])["tx"]
228 tx1.rehash()
229 230 # Anyone-can-spend mempool tx.
231 # Sequence lock of 0 should pass.
232 tx2 = CTransaction()
233 tx2.version = 2
234 tx2.vin = [CTxIn(COutPoint(tx1.sha256, 0), nSequence=0)]
235 tx2.vout = [CTxOut(int(tx1.vout[0].nValue - self.relayfee * COIN), SCRIPT_W0_SH_OP_TRUE)]
236 self.wallet.sign_tx(tx=tx2)
237 tx2_raw = tx2.serialize().hex()
238 tx2.rehash()
239 240 self.wallet.sendrawtransaction(from_node=self.nodes[0], tx_hex=tx2_raw)
241 242 # Create a spend of the 0th output of orig_tx with a sequence lock
243 # of 1, and test what happens when submitting.
244 # orig_tx.vout[0] must be an anyone-can-spend output
245 def test_nonzero_locks(orig_tx, node, relayfee, use_height_lock):
246 sequence_value = 1
247 if not use_height_lock:
248 sequence_value |= SEQUENCE_LOCKTIME_TYPE_FLAG
249 250 tx = CTransaction()
251 tx.version = 2
252 tx.vin = [CTxIn(COutPoint(orig_tx.sha256, 0), nSequence=sequence_value)]
253 tx.wit.vtxinwit = [CTxInWitness()]
254 tx.wit.vtxinwit[0].scriptWitness.stack = [CScript([OP_TRUE])]
255 tx.vout = [CTxOut(int(orig_tx.vout[0].nValue - relayfee * COIN), SCRIPT_W0_SH_OP_TRUE)]
256 tx.rehash()
257 258 if (orig_tx.hash in node.getrawmempool()):
259 # sendrawtransaction should fail if the tx is in the mempool
260 assert_raises_rpc_error(-26, NOT_FINAL_ERROR, self.wallet.sendrawtransaction, from_node=node, tx_hex=tx.serialize().hex())
261 else:
262 # sendrawtransaction should succeed if the tx is not in the mempool
263 self.wallet.sendrawtransaction(from_node=node, tx_hex=tx.serialize().hex())
264 265 return tx
266 267 test_nonzero_locks(tx2, self.nodes[0], self.relayfee, use_height_lock=True)
268 test_nonzero_locks(tx2, self.nodes[0], self.relayfee, use_height_lock=False)
269 270 # Now mine some blocks, but make sure tx2 doesn't get mined.
271 # Use prioritisetransaction to lower the effective feerate to 0
272 self.nodes[0].prioritisetransaction(txid=tx2.hash, fee_delta=int(-self.relayfee*COIN))
273 cur_time = int(time.time())
274 for _ in range(10):
275 self.nodes[0].setmocktime(cur_time + 600)
276 self.generate(self.wallet, 1, sync_fun=self.no_op)
277 cur_time += 600
278 279 assert tx2.hash in self.nodes[0].getrawmempool()
280 281 test_nonzero_locks(tx2, self.nodes[0], self.relayfee, use_height_lock=True)
282 test_nonzero_locks(tx2, self.nodes[0], self.relayfee, use_height_lock=False)
283 284 # Mine tx2, and then try again
285 self.nodes[0].prioritisetransaction(txid=tx2.hash, fee_delta=int(self.relayfee*COIN))
286 287 # Advance the time on the node so that we can test timelocks
288 self.nodes[0].setmocktime(cur_time+600)
289 # Save block template now to use for the reorg later
290 tmpl = self.nodes[0].getblocktemplate(NORMAL_GBT_REQUEST_PARAMS)
291 self.generate(self.nodes[0], 1)
292 assert tx2.hash not in self.nodes[0].getrawmempool()
293 294 # Now that tx2 is not in the mempool, a sequence locked spend should
295 # succeed
296 tx3 = test_nonzero_locks(tx2, self.nodes[0], self.relayfee, use_height_lock=False)
297 assert tx3.hash in self.nodes[0].getrawmempool()
298 299 self.generate(self.nodes[0], 1)
300 assert tx3.hash not in self.nodes[0].getrawmempool()
301 302 # One more test, this time using height locks
303 tx4 = test_nonzero_locks(tx3, self.nodes[0], self.relayfee, use_height_lock=True)
304 assert tx4.hash in self.nodes[0].getrawmempool()
305 306 # Now try combining confirmed and unconfirmed inputs
307 tx5 = test_nonzero_locks(tx4, self.nodes[0], self.relayfee, use_height_lock=True)
308 assert tx5.hash not in self.nodes[0].getrawmempool()
309 310 utxo = self.wallet.get_utxo()
311 tx5.vin.append(CTxIn(COutPoint(int(utxo["txid"], 16), utxo["vout"]), nSequence=1))
312 tx5.vout[0].nValue += int(utxo["value"]*COIN)
313 self.wallet.sign_tx(tx=tx5)
314 315 assert_raises_rpc_error(-26, NOT_FINAL_ERROR, self.wallet.sendrawtransaction, from_node=self.nodes[0], tx_hex=tx5.serialize().hex())
316 317 # Test mempool-BIP68 consistency after reorg
318 #
319 # State of the transactions in the last blocks:
320 # ... -> [ tx2 ] -> [ tx3 ]
321 # tip-1 tip
322 # And currently tx4 is in the mempool.
323 #
324 # If we invalidate the tip, tx3 should get added to the mempool, causing
325 # tx4 to be removed (fails sequence-lock).
326 self.nodes[0].invalidateblock(self.nodes[0].getbestblockhash())
327 assert tx4.hash not in self.nodes[0].getrawmempool()
328 assert tx3.hash in self.nodes[0].getrawmempool()
329 330 # Now mine 2 empty blocks to reorg out the current tip (labeled tip-1 in
331 # diagram above).
332 # This would cause tx2 to be added back to the mempool, which in turn causes
333 # tx3 to be removed.
334 for i in range(2):
335 block = create_block(tmpl=tmpl, ntime=cur_time)
336 block.solve()
337 tip = block.sha256
338 assert_equal(None if i == 1 else 'inconclusive', self.nodes[0].submitblock(block.serialize().hex()))
339 tmpl = self.nodes[0].getblocktemplate(NORMAL_GBT_REQUEST_PARAMS)
340 tmpl['previousblockhash'] = '%x' % tip
341 tmpl['transactions'] = []
342 cur_time += 1
343 344 mempool = self.nodes[0].getrawmempool()
345 assert tx3.hash not in mempool
346 assert tx2.hash in mempool
347 348 # Reset the chain and get rid of the mocktimed-blocks
349 self.nodes[0].setmocktime(0)
350 self.nodes[0].invalidateblock(self.nodes[0].getblockhash(cur_height+1))
351 self.generate(self.wallet, 10, sync_fun=self.no_op)
352 353 # Make sure that BIP68 isn't being used to validate blocks prior to
354 # activation height. If more blocks are mined prior to this test
355 # being run, then it's possible the test has activated the soft fork, and
356 # this test should be moved to run earlier, or deleted.
357 def test_bip68_not_consensus(self):
358 assert not softfork_active(self.nodes[0], 'csv')
359 360 tx1 = self.wallet.send_self_transfer(from_node=self.nodes[0])["tx"]
361 tx1.rehash()
362 363 # Make an anyone-can-spend transaction
364 tx2 = CTransaction()
365 tx2.version = 1
366 tx2.vin = [CTxIn(COutPoint(tx1.sha256, 0), nSequence=0)]
367 tx2.vout = [CTxOut(int(tx1.vout[0].nValue - self.relayfee * COIN), SCRIPT_W0_SH_OP_TRUE)]
368 369 # sign tx2
370 self.wallet.sign_tx(tx=tx2)
371 tx2_raw = tx2.serialize().hex()
372 tx2 = tx_from_hex(tx2_raw)
373 tx2.rehash()
374 375 self.wallet.sendrawtransaction(from_node=self.nodes[0], tx_hex=tx2_raw)
376 377 # Now make an invalid spend of tx2 according to BIP68
378 sequence_value = 100 # 100 block relative locktime
379 380 tx3 = CTransaction()
381 tx3.version = 2
382 tx3.vin = [CTxIn(COutPoint(tx2.sha256, 0), nSequence=sequence_value)]
383 tx3.wit.vtxinwit = [CTxInWitness()]
384 tx3.wit.vtxinwit[0].scriptWitness.stack = [CScript([OP_TRUE])]
385 tx3.vout = [CTxOut(int(tx2.vout[0].nValue - self.relayfee * COIN), SCRIPT_W0_SH_OP_TRUE)]
386 tx3.rehash()
387 388 assert_raises_rpc_error(-26, NOT_FINAL_ERROR, self.wallet.sendrawtransaction, from_node=self.nodes[0], tx_hex=tx3.serialize().hex())
389 390 # make a block that violates bip68; ensure that the tip updates
391 block = create_block(tmpl=self.nodes[0].getblocktemplate(NORMAL_GBT_REQUEST_PARAMS), txlist=[tx1, tx2, tx3])
392 add_witness_commitment(block)
393 block.solve()
394 395 assert_equal(None, self.nodes[0].submitblock(block.serialize().hex()))
396 assert_equal(self.nodes[0].getbestblockhash(), block.hash)
397 398 def activateCSV(self):
399 # activation should happen at block height 432 (3 periods)
400 # getblockchaininfo will show CSV as active at block 431 (144 * 3 -1) since it's returning whether CSV is active for the next block.
401 min_activation_height = 432
402 height = self.nodes[0].getblockcount()
403 assert_greater_than(min_activation_height - height, 2)
404 self.generate(self.wallet, min_activation_height - height - 2, sync_fun=self.no_op)
405 assert not softfork_active(self.nodes[0], 'csv')
406 self.generate(self.wallet, 1, sync_fun=self.no_op)
407 assert softfork_active(self.nodes[0], 'csv')
408 self.sync_blocks()
409 410 # Use self.nodes[1] to test that version 2 transactions are standard.
411 def test_version2_relay(self):
412 mini_wallet = MiniWallet(self.nodes[1])
413 mini_wallet.send_self_transfer(from_node=self.nodes[1], version=2)
414 415 416 if __name__ == '__main__':
417 BIP68Test(__file__).main()
418