#!/usr/bin/env python3 # Copyright (c) 2015 The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. # # Test replace-by-fee # import os import sys # Add python-bitcoinlib to module search path: sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), "python-bitcoinlib")) import unittest import bitcoin bitcoin.SelectParams('regtest') import bitcoin.rpc from bitcoin.core import * from bitcoin.core.script import * from bitcoin.wallet import * MAX_REPLACEMENT_LIMIT = 100 class Test_ReplaceByFee(unittest.TestCase): proxy = None @classmethod def setUpClass(cls): if cls.proxy is None: cls.proxy = bitcoin.rpc.Proxy() @classmethod def tearDownClass(cls): # Make sure mining works mempool_size = 1 while mempool_size: cls.proxy.call('generate',1) new_mempool_size = len(cls.proxy.getrawmempool()) # It's possible to get stuck in a loop here if the mempool has # transactions that can't be mined. assert(new_mempool_size != mempool_size) mempool_size = new_mempool_size def make_txout(self, amount, scriptPubKey=CScript([1])): """Create a txout with a given amount and scriptPubKey Mines coins as needed. """ fee = 1*COIN while self.proxy.getbalance() < amount + fee: self.proxy.call('generate', 100) addr = P2SHBitcoinAddress.from_redeemScript(CScript([])) txid = self.proxy.sendtoaddress(addr, amount + fee) tx1 = self.proxy.getrawtransaction(txid) i = None for i, txout in enumerate(tx1.vout): if txout.scriptPubKey == addr.to_scriptPubKey(): break assert i is not None tx2 = CTransaction([CTxIn(COutPoint(txid, i), CScript([1, CScript([])]), nSequence=0)], [CTxOut(amount, scriptPubKey)]) tx2_txid = self.proxy.sendrawtransaction(tx2, True) return COutPoint(tx2_txid, 0) def test_simple_doublespend(self): """Simple doublespend""" tx0_outpoint = self.make_txout(1.1*COIN) tx1a = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(1*COIN, CScript([b'a']))]) tx1a_txid = self.proxy.sendrawtransaction(tx1a, True) # Should fail because we haven't changed the fee tx1b = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(1*COIN, CScript([b'b']))]) try: tx1b_txid = self.proxy.sendrawtransaction(tx1b, True) except bitcoin.rpc.JSONRPCException as exp: self.assertEqual(exp.error['code'], -26) # insufficient fee else: self.fail() # Extra 0.1 BTC fee tx1b = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(0.9*COIN, CScript([b'b']))]) tx1b_txid = self.proxy.sendrawtransaction(tx1b, True) # tx1a is in fact replaced with self.assertRaises(IndexError): self.proxy.getrawtransaction(tx1a_txid) self.assertEqual(tx1b, self.proxy.getrawtransaction(tx1b_txid)) def test_doublespend_chain(self): """Doublespend of a long chain""" initial_nValue = 50*COIN tx0_outpoint = self.make_txout(initial_nValue) prevout = tx0_outpoint remaining_value = initial_nValue chain_txids = [] while remaining_value > 10*COIN: remaining_value -= 1*COIN tx = CTransaction([CTxIn(prevout, nSequence=0)], [CTxOut(remaining_value, CScript([1]))]) txid = self.proxy.sendrawtransaction(tx, True) chain_txids.append(txid) prevout = COutPoint(txid, 0) # Whether the double-spend is allowed is evaluated by including all # child fees - 40 BTC - so this attempt is rejected. dbl_tx = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(initial_nValue - 30*COIN, CScript([1]))]) try: self.proxy.sendrawtransaction(dbl_tx, True) except bitcoin.rpc.JSONRPCException as exp: self.assertEqual(exp.error['code'], -26) # insufficient fee else: self.fail() # Accepted with sufficient fee dbl_tx = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(1*COIN, CScript([1]))]) self.proxy.sendrawtransaction(dbl_tx, True) for doublespent_txid in chain_txids: with self.assertRaises(IndexError): self.proxy.getrawtransaction(doublespent_txid) def test_doublespend_tree(self): """Doublespend of a big tree of transactions""" initial_nValue = 50*COIN tx0_outpoint = self.make_txout(initial_nValue) def branch(prevout, initial_value, max_txs, *, tree_width=5, fee=0.0001*COIN, _total_txs=None): if _total_txs is None: _total_txs = [0] if _total_txs[0] >= max_txs: return txout_value = (initial_value - fee) // tree_width if txout_value < fee: return vout = [CTxOut(txout_value, CScript([i+1])) for i in range(tree_width)] tx = CTransaction([CTxIn(prevout, nSequence=0)], vout) self.assertTrue(len(tx.serialize()) < 100000) txid = self.proxy.sendrawtransaction(tx, True) yield tx _total_txs[0] += 1 for i, txout in enumerate(tx.vout): yield from branch(COutPoint(txid, i), txout_value, max_txs, tree_width=tree_width, fee=fee, _total_txs=_total_txs) fee = 0.0001*COIN n = MAX_REPLACEMENT_LIMIT tree_txs = list(branch(tx0_outpoint, initial_nValue, n, fee=fee)) self.assertEqual(len(tree_txs), n) # Attempt double-spend, will fail because too little fee paid dbl_tx = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(initial_nValue - fee*n, CScript([1]))]) try: self.proxy.sendrawtransaction(dbl_tx, True) except bitcoin.rpc.JSONRPCException as exp: self.assertEqual(exp.error['code'], -26) # insufficient fee else: self.fail() # 1 BTC fee is enough dbl_tx = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(initial_nValue - fee*n - 1*COIN, CScript([1]))]) self.proxy.sendrawtransaction(dbl_tx, True) for tx in tree_txs: with self.assertRaises(IndexError): self.proxy.getrawtransaction(tx.GetHash()) # Try again, but with more total transactions than the "max txs # double-spent at once" anti-DoS limit. for n in (MAX_REPLACEMENT_LIMIT, MAX_REPLACEMENT_LIMIT*2): fee = 0.0001*COIN tx0_outpoint = self.make_txout(initial_nValue) tree_txs = list(branch(tx0_outpoint, initial_nValue, n, fee=fee)) self.assertEqual(len(tree_txs), n) dbl_tx = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(initial_nValue - fee*n, CScript([1]))]) try: self.proxy.sendrawtransaction(dbl_tx, True) except bitcoin.rpc.JSONRPCException as exp: self.assertEqual(exp.error['code'], -26) else: self.fail() for tx in tree_txs: self.proxy.getrawtransaction(tx.GetHash()) def test_replacement_feeperkb(self): """Replacement requires fee-per-KB to be higher""" tx0_outpoint = self.make_txout(1.1*COIN) tx1a = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(1*COIN, CScript([b'a']))]) tx1a_txid = self.proxy.sendrawtransaction(tx1a, True) # Higher fee, but the fee per KB is much lower, so the replacement is # rejected. tx1b = CTransaction([CTxIn(tx0_outpoint, nSequence=0)], [CTxOut(0.001*COIN, CScript([b'a'*999000]))]) try: tx1b_txid = self.proxy.sendrawtransaction(tx1b, True) except bitcoin.rpc.JSONRPCException as exp: self.assertEqual(exp.error['code'], -26) # insufficient fee else: self.fail() def test_spends_of_conflicting_outputs(self): """Replacements that spend conflicting tx outputs are rejected""" utxo1 = self.make_txout(1.2*COIN) utxo2 = self.make_txout(3.0*COIN) tx1a = CTransaction([CTxIn(utxo1, nSequence=0)], [CTxOut(1.1*COIN, CScript([b'a']))]) tx1a_txid = self.proxy.sendrawtransaction(tx1a, True) # Direct spend an output of the transaction we're replacing. tx2 = CTransaction([CTxIn(utxo1, nSequence=0), CTxIn(utxo2, nSequence=0), CTxIn(COutPoint(tx1a_txid, 0), nSequence=0)], tx1a.vout) try: tx2_txid = self.proxy.sendrawtransaction(tx2, True) except bitcoin.rpc.JSONRPCException as exp: self.assertEqual(exp.error['code'], -26) else: self.fail() # Spend tx1a's output to test the indirect case. tx1b = CTransaction([CTxIn(COutPoint(tx1a_txid, 0), nSequence=0)], [CTxOut(1.0*COIN, CScript([b'a']))]) tx1b_txid = self.proxy.sendrawtransaction(tx1b, True) tx2 = CTransaction([CTxIn(utxo1, nSequence=0), CTxIn(utxo2, nSequence=0), CTxIn(COutPoint(tx1b_txid, 0))], tx1a.vout) try: tx2_txid = self.proxy.sendrawtransaction(tx2, True) except bitcoin.rpc.JSONRPCException as exp: self.assertEqual(exp.error['code'], -26) else: self.fail() if __name__ == '__main__': unittest.main()