-
Notifications
You must be signed in to change notification settings - Fork 115
/
p2p-instantsend.py
executable file
·133 lines (119 loc) · 5.52 KB
/
p2p-instantsend.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
#!/usr/bin/env python3
# Copyright (c) 2018-2020 The Dash Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
from test_framework.mininode import *
from test_framework.test_framework import AxeTestFramework
from test_framework.util import isolate_node, sync_mempools, reconnect_isolated_node, assert_equal, \
assert_raises_rpc_error
'''
p2p-instantsend.py
Tests InstantSend functionality (prevent doublespend for unconfirmed transactions)
'''
class InstantSendTest(AxeTestFramework):
def set_test_params(self):
self.set_axe_test_params(7, 3, fast_dip3_enforcement=True)
# set sender, receiver, isolated nodes
self.isolated_idx = 1
self.receiver_idx = 2
self.sender_idx = 3
def run_test(self):
self.nodes[0].spork("SPORK_17_QUORUM_DKG_ENABLED", 0)
self.wait_for_sporks_same()
self.mine_quorum()
self.nodes[0].spork("SPORK_2_INSTANTSEND_ENABLED", 0)
self.nodes[0].spork("SPORK_3_INSTANTSEND_BLOCK_FILTERING", 0)
self.wait_for_sporks_same()
self.test_mempool_doublespend()
self.test_block_doublespend()
def test_block_doublespend(self):
sender = self.nodes[self.sender_idx]
receiver = self.nodes[self.receiver_idx]
isolated = self.nodes[self.isolated_idx]
# feed the sender with some balance
sender_addr = sender.getnewaddress()
self.nodes[0].sendtoaddress(sender_addr, 1)
self.bump_mocktime(1)
self.nodes[0].generate(2)
self.sync_all()
# create doublespending transaction, but don't relay it
dblspnd_tx = self.create_raw_tx(sender, isolated, 0.5, 1, 100)
# isolate one node from network
isolate_node(isolated)
# instantsend to receiver
receiver_addr = receiver.getnewaddress()
is_id = sender.sendtoaddress(receiver_addr, 0.9)
# wait for the transaction to propagate
connected_nodes = self.nodes.copy()
del connected_nodes[self.isolated_idx]
self.sync_mempools(connected_nodes)
for node in connected_nodes:
self.wait_for_instantlock(is_id, node)
# send doublespend transaction to isolated node
isolated.sendrawtransaction(dblspnd_tx['hex'])
# generate block on isolated node with doublespend transaction
self.bump_mocktime(1)
isolated.generate(1)
wrong_block = isolated.getbestblockhash()
# connect isolated block to network
reconnect_isolated_node(isolated, 0)
# check doublespend block is rejected by other nodes
timeout = 10
for i in range(0, self.num_nodes):
if i == self.isolated_idx:
continue
res = self.nodes[i].waitforblock(wrong_block, timeout)
assert (res['hash'] != wrong_block)
# wait for long time only for first node
timeout = 1
# send coins back to the controller node without waiting for confirmations
receiver.sendtoaddress(self.nodes[0].getnewaddress(), 0.9, "", "", True)
assert_equal(receiver.getwalletinfo()["balance"], 0)
# mine more blocks
# TODO: mine these blocks on an isolated node
self.bump_mocktime(1)
# make sure the above TX is on node0
self.sync_mempools([n for n in self.nodes if n is not isolated])
self.nodes[0].generate(2)
self.sync_all()
def test_mempool_doublespend(self):
sender = self.nodes[self.sender_idx]
receiver = self.nodes[self.receiver_idx]
isolated = self.nodes[self.isolated_idx]
# feed the sender with some balance
sender_addr = sender.getnewaddress()
self.nodes[0].sendtoaddress(sender_addr, 1)
self.bump_mocktime(1)
self.nodes[0].generate(2)
self.sync_all()
# create doublespending transaction, but don't relay it
dblspnd_tx = self.create_raw_tx(sender, isolated, 0.5, 1, 100)
dblspnd_txid = bytes_to_hex_str(hash256(hex_str_to_bytes(dblspnd_tx['hex']))[::-1])
# isolate one node from network
isolate_node(isolated)
# send doublespend transaction to isolated node
isolated.sendrawtransaction(dblspnd_tx['hex'])
# let isolated node rejoin the network
# The previously isolated node should NOT relay the doublespending TX
reconnect_isolated_node(isolated, 0)
for node in self.nodes:
if node is not isolated:
assert_raises_rpc_error(-5, "No such mempool or blockchain transaction", node.getrawtransaction, dblspnd_txid)
# instantsend to receiver. The previously isolated node should prune the doublespend TX and request the correct
# TX from other nodes.
receiver_addr = receiver.getnewaddress()
is_id = sender.sendtoaddress(receiver_addr, 0.9)
# wait for the transaction to propagate
self.sync_mempools()
for node in self.nodes:
self.wait_for_instantlock(is_id, node)
assert_raises_rpc_error(-5, "No such mempool or blockchain transaction", isolated.getrawtransaction, dblspnd_txid)
# send coins back to the controller node without waiting for confirmations
receiver.sendtoaddress(self.nodes[0].getnewaddress(), 0.9, "", "", True)
assert_equal(receiver.getwalletinfo()["balance"], 0)
# mine more blocks
self.bump_mocktime(1)
self.nodes[0].generate(2)
self.sync_all()
if __name__ == '__main__':
InstantSendTest().main()