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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
|
import asyncio
import collections
import aiorpcx
from aiorpcx import RPCError
import electrum
from electrum.interface import ServerAddr, Interface, PaddedRSTransport
from electrum import util, blockchain
from electrum.util import OldTaskGroup, bfh
from electrum.logging import Logger
from electrum.simple_config import SimpleConfig
from electrum.transaction import Transaction
from . import ElectrumTestCase
class TestServerAddr(ElectrumTestCase):
def test_from_str(self):
self.assertEqual(ServerAddr(host="104.198.149.61", port=80, protocol="t"),
ServerAddr.from_str("104.198.149.61:80:t"))
self.assertEqual(ServerAddr(host="ecdsa.net", port=110, protocol="s"),
ServerAddr.from_str("ecdsa.net:110:s"))
self.assertEqual(ServerAddr(host="2400:6180:0:d1::86b:e001", port=50002, protocol="s"),
ServerAddr.from_str("[2400:6180:0:d1::86b:e001]:50002:s"))
self.assertEqual(ServerAddr(host="localhost", port=8080, protocol="s"),
ServerAddr.from_str("localhost:8080:s"))
def test_from_str_with_inference(self):
self.assertEqual(None, ServerAddr.from_str_with_inference("104.198.149.61"))
self.assertEqual(None, ServerAddr.from_str_with_inference("ecdsa.net"))
self.assertEqual(None, ServerAddr.from_str_with_inference("2400:6180:0:d1::86b:e001"))
self.assertEqual(None, ServerAddr.from_str_with_inference("[2400:6180:0:d1::86b:e001]"))
self.assertEqual(ServerAddr(host="104.198.149.61", port=80, protocol="s"),
ServerAddr.from_str_with_inference("104.198.149.61:80"))
self.assertEqual(ServerAddr(host="ecdsa.net", port=110, protocol="s"),
ServerAddr.from_str_with_inference("ecdsa.net:110"))
self.assertEqual(ServerAddr(host="2400:6180:0:d1::86b:e001", port=50002, protocol="s"),
ServerAddr.from_str_with_inference("[2400:6180:0:d1::86b:e001]:50002"))
self.assertEqual(ServerAddr(host="104.198.149.61", port=80, protocol="t"),
ServerAddr.from_str_with_inference("104.198.149.61:80:t"))
self.assertEqual(ServerAddr(host="ecdsa.net", port=110, protocol="s"),
ServerAddr.from_str_with_inference("ecdsa.net:110:s"))
self.assertEqual(ServerAddr(host="2400:6180:0:d1::86b:e001", port=50002, protocol="s"),
ServerAddr.from_str_with_inference("[2400:6180:0:d1::86b:e001]:50002:s"))
def test_to_friendly_name(self):
self.assertEqual("104.198.149.61:80:t",
ServerAddr(host="104.198.149.61", port=80, protocol="t").to_friendly_name())
self.assertEqual("ecdsa.net:110",
ServerAddr(host="ecdsa.net", port=110, protocol="s").to_friendly_name())
self.assertEqual("ecdsa.net:50001:t",
ServerAddr(host="ecdsa.net", port=50001, protocol="t").to_friendly_name())
self.assertEqual("[2400:6180:0:d1::86b:e001]:50002",
ServerAddr(host="2400:6180:0:d1::86b:e001", port=50002, protocol="s").to_friendly_name())
self.assertEqual("[2400:6180:0:d1::86b:e001]:50001:t",
ServerAddr(host="2400:6180:0:d1::86b:e001", port=50001, protocol="t").to_friendly_name())
class MockNetwork:
def __init__(self, *, config: SimpleConfig):
self.config = config
self.asyncio_loop = util.get_asyncio_loop()
self.taskgroup = OldTaskGroup()
blockchain.read_blockchains(self.config)
blockchain.init_headers_file_for_best_chain()
self.proxy = None
self.debug = True
self.bhi_lock = asyncio.Lock()
self.interface = None # type: Interface | None
async def connection_down(self, interface: Interface):
pass
def get_network_timeout_seconds(self, request_type) -> int:
return 10
def check_interface_against_healthy_spread_of_connected_servers(self, iface_to_check: Interface) -> bool:
return True
def update_fee_estimates(self, *, fee_est: dict[int, int] = None):
pass
async def switch_unwanted_fork_interface(self):
pass
async def switch_lagging_interface(self):
pass
# regtest chain:
BLOCK_HEADERS = {
0: bfh("0100000000000000000000000000000000000000000000000000000000000000000000003ba3edfd7a7b12b27ac72c3e67768f617fc81bc3888a51323a9fb8aa4b1e5e4adae5494dffff7f2002000000"),
1: bfh("0000002006226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f186c8dfd970a4545f79916bc1d75c9d00432f57c89209bf3bb115b7612848f509c25f45bffff7f2000000000"),
2: bfh("00000020686bdfc6a3db73d5d93e8c9663a720a26ecb1ef20eb05af11b36cdbc57c19f7ebf2cbf153013a1c54abaf70e95198fcef2f3059cc6b4d0f7e876808e7d24d11cc825f45bffff7f2000000000"),
3: bfh("00000020122baa14f3ef54985ae546d1611559e3f487bd2a0f46e8dbb52fbacc9e237972e71019d7feecd9b8596eca9a67032c5f4641b23b5d731dc393e37de7f9c2f299e725f45bffff7f2000000000"),
4: bfh("00000020f8016f7ef3a17d557afe05d4ea7ab6bde1b2247b7643896c1b63d43a1598b747a3586da94c71753f27c075f57f44faf913c31177a0957bbda42e7699e3a2141aed25f45bffff7f2001000000"),
5: bfh("000000201d589c6643c1d121d73b0573e5ee58ab575b8fdf16d507e7e915c5fbfbbfd05e7aee1d692d1615c3bdf52c291032144ce9e3b258a473c17c745047f3431ff8e2ee25f45bffff7f2000000000"),
6: bfh("00000020b833ed46eea01d4c980f59feee44a66aa1162748b6801029565d1466790c405c3a141ce635cbb1cd2b3a4fcdd0a3380517845ba41736c82a79cab535d31128066526f45bffff7f2001000000"),
7: bfh("00000020abe8e119d1877c9dc0dc502d1a253fb9a67967c57732d2f71ee0280e8381ff0a9690c2fe7c1a4450c74dc908fe94dd96c3b0637d51475e9e06a78e944a0c7fe28126f45bffff7f2000000000"),
8: bfh("000000202ce41d94eb70e1518bc1f72523f84a903f9705d967481e324876e1f8cf4d3452148be228a4c3f2061bafe7efdfc4a8d5a94759464b9b5c619994d45dfcaf49e1a126f45bffff7f2000000000"),
9: bfh("00000020552755b6c59f3d51e361d16281842a4e166007799665b5daed86a063dd89857415681cb2d00ff889193f6a68a93f5096aeb2d84ca0af6185a462555822552221a626f45bffff7f2000000000"),
10: bfh("00000020a13a491cbefc93cd1bb1938f19957e22a134faf14c7dee951c45533e2c750f239dc087fc977b06c24a69c682d1afd1020e6dc1f087571ccec66310a786e1548fab26f45bffff7f2000000000"),
11: bfh("00000020dbf3a9b55dfefbaf8b6e43a89cf833fa2e208bbc0c1c5d76c0d71b9e4a65337803b243756c25053253aeda309604363460a3911015929e68705bd89dff6fe064b026f45bffff7f2002000000"),
12: bfh("000000203d0932b3b0c78eccb39a595a28ae4a7c966388648d7783fd1305ec8d40d4fe5fd67cb902a7d807cee7676cb543feec3e053aa824d5dfb528d5b94f9760313d9db726f45bffff7f2001000000"),
}
_active_server_sessions = set()
def _get_active_server_session() -> 'ServerSession':
assert 1 == len(_active_server_sessions), len(_active_server_sessions)
return list(_active_server_sessions)[0]
class ServerSession(aiorpcx.RPCSession, Logger):
def __init__(self, *args, **kwargs):
aiorpcx.RPCSession.__init__(self, *args, **kwargs)
Logger.__init__(self)
self.logger.debug(f'connection from {self.remote_address()}')
self.cur_height = 6 # type: int # chain tip
self.txs = {
"bdae818ad3c1f261317738ae9284159bf54874356f186dbc7afd631dc1527fcb": bfh("020000000001010000000000000000000000000000000000000000000000000000000000000000ffffffff025100ffffffff0200f2052a010000001600140297bde2689a3c79ffe050583b62f86f2d9dae540000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf90120000000000000000000000000000000000000000000000000000000000000000000000000"),
} # type: dict[str, bytes]
self._method_counts = collections.defaultdict(int) # type: dict[str, int]
_active_server_sessions.add(self)
async def connection_lost(self):
await super().connection_lost()
self.logger.debug(f'{self.remote_address()} disconnected')
_active_server_sessions.discard(self)
async def handle_request(self, request):
handlers = {
'server.version': self._handle_server_version,
'blockchain.estimatefee': self._handle_estimatefee,
'blockchain.headers.subscribe': self._handle_headers_subscribe,
'blockchain.block.header': self._handle_block_header,
'blockchain.block.headers': self._handle_block_headers,
'blockchain.transaction.get': self._handle_transaction_get,
'blockchain.transaction.broadcast': self._handle_transaction_broadcast,
'server.ping': self._handle_ping,
}
handler = handlers.get(request.method)
self._method_counts[request.method] += 1
coro = aiorpcx.handler_invocation(handler, request)()
return await coro
async def _handle_server_version(self, client_name='', protocol_version=None):
return ['best_server_impl/0.1', '1.4']
async def _handle_estimatefee(self, number, mode=None):
return 1000
async def _handle_headers_subscribe(self):
return {'hex': BLOCK_HEADERS[self.cur_height].hex(), 'height': self.cur_height}
async def _handle_block_header(self, height):
return BLOCK_HEADERS[height].hex()
async def _handle_block_headers(self, start_height, count):
assert start_height <= self.cur_height, (start_height, self.cur_height)
last_height = min(start_height+count-1, self.cur_height) # [start_height, last_height]
count = last_height - start_height + 1
headers = b"".join(BLOCK_HEADERS[idx] for idx in range(start_height, last_height+1))
return {'hex': headers.hex(), 'count': count, 'max': 2016}
async def _handle_ping(self):
return None
async def _handle_transaction_get(self, tx_hash: str, verbose=False):
assert not verbose
rawtx = self.txs.get(tx_hash)
if rawtx is None:
DAEMON_ERROR = 2
raise RPCError(DAEMON_ERROR, f'daemon error: unknown txid={tx_hash}')
return rawtx.hex()
async def _handle_transaction_broadcast(self, raw_tx: str):
tx = Transaction(raw_tx)
self.txs[tx.txid()] = bfh(raw_tx)
return tx.txid()
class TestInterface(ElectrumTestCase):
REGTEST = True
def setUp(self):
super().setUp()
self.config = SimpleConfig({'electrum_path': self.electrum_path})
self._orig_WAIT_FOR_BUFFER_GROWTH_SECONDS = PaddedRSTransport.WAIT_FOR_BUFFER_GROWTH_SECONDS
PaddedRSTransport.WAIT_FOR_BUFFER_GROWTH_SECONDS = 0
def tearDown(self):
PaddedRSTransport.WAIT_FOR_BUFFER_GROWTH_SECONDS = self._orig_WAIT_FOR_BUFFER_GROWTH_SECONDS
super().tearDown()
async def asyncSetUp(self):
await super().asyncSetUp()
self._server: asyncio.base_events.Server = await aiorpcx.serve_rs(ServerSession, "127.0.0.1")
server_socket_addr = self._server.sockets[0].getsockname()
self._server_port = server_socket_addr[1]
self.network = MockNetwork(config=self.config)
async def asyncTearDown(self):
if self.network.interface:
await self.network.interface.close()
self._server.close()
await self._server.wait_closed()
await super().asyncTearDown()
async def _start_iface_and_wait_for_sync(self):
interface = Interface(network=self.network, server=ServerAddr(host="127.0.0.1", port=self._server_port, protocol="t"))
self.network.interface = interface
await util.wait_for2(interface.ready, 5)
await interface._blockchain_updated.wait()
return interface
async def test_client_syncs_headers_to_tip(self):
interface = await self._start_iface_and_wait_for_sync()
self.assertEqual(_get_active_server_session().cur_height, interface.tip)
self.assertFalse(interface.got_disconnected.is_set())
async def test_transaction_get(self):
interface = await self._start_iface_and_wait_for_sync()
# try requesting tx unknown to server:
with self.assertRaises(RPCError) as ctx:
await interface.get_transaction("deadbeef"*8)
self.assertTrue("unknown txid" in ctx.exception.message)
# try requesting known tx:
rawtx = await interface.get_transaction("bdae818ad3c1f261317738ae9284159bf54874356f186dbc7afd631dc1527fcb")
self.assertEqual(rawtx, _get_active_server_session().txs["bdae818ad3c1f261317738ae9284159bf54874356f186dbc7afd631dc1527fcb"].hex())
self.assertEqual(_get_active_server_session()._method_counts["blockchain.transaction.get"], 2)
async def test_transaction_broadcast(self):
interface = await self._start_iface_and_wait_for_sync()
rawtx1 = "020000000001010000000000000000000000000000000000000000000000000000000000000000ffffffff025200ffffffff0200f2052a010000001600140297bde2689a3c79ffe050583b62f86f2d9dae540000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf90120000000000000000000000000000000000000000000000000000000000000000000000000"
tx = Transaction(rawtx1)
# broadcast
await interface.broadcast_transaction(tx)
self.assertEqual(bfh(rawtx1), _get_active_server_session().txs.get(tx.txid()))
# now request tx.
# as we just broadcast this same tx, this will hit the client iface cache, and won't call the server.
self.assertEqual(_get_active_server_session()._method_counts["blockchain.transaction.get"], 0)
rawtx2 = await interface.get_transaction(tx.txid())
self.assertEqual(rawtx1, rawtx2)
self.assertEqual(_get_active_server_session()._method_counts["blockchain.transaction.get"], 0)
|
