This repository has been archived by the owner on Oct 21, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 9
/
nanocast.py
653 lines (575 loc) · 33.1 KB
/
nanocast.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
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
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
# -*- coding: utf-8 -*-
import tornado.ioloop
import tornado.web
import tornado.websocket
import tornado.httpclient
import tornado.httpserver
import tornado.gen
import json
import logging
import os
import redis
import socket
import ssl
import sys
import time
import uuid
from tornado.ioloop import IOLoop
from tornado.websocket import websocket_connect
from logging.handlers import WatchedFileHandler
from bitstring import BitArray
# future use for caching blocks
# rblock = redis.StrictRedis(host='localhost', port=6379, db=0)
# future use for pending blocks for accounts, cached work
# racct = redis.StrictRedis(host='localhost', port=6379, db=1)
rdata = redis.StrictRedis(host='localhost', port=6379, db=2) # used for price data and subscriber uuid info
# get environment
rpc_url = os.getenv('NANO_RPC_URL', 'http://127.0.0.1:7076') # use env, else default to localhost rpc port
websocket_url = os.getenv('NANO_WS_URL', 'ws://127.0.0.1:7078') # use env, else default to localhost websocket port
work_url = os.getenv('NANO_WORK_URL', rpc_url) # use env, else default to rpc
socket_port = os.getenv('NANO_SOCKET_PORT', 443)
cert_dir = os.getenv('NANO_CERT_DIR') # use /home/username instead of /home/username/
cert_key_file = os.getenv('NANO_KEY_FILE') # TLS certificate private key
cert_crt_file = os.getenv('NANO_CRT_FILE') # full TLS certificate bundle
# whitelisted commands, disallow anything used for local node-based wallet as we may be using multiple back ends
allowed_rpc_actions = ["account_balance", "account_block_count", "account_check", "account_info", "account_history",
"account_representative", "account_subscribe", "account_weight", "accounts_balances",
"accounts_frontiers", "accounts_pending", "available_supply", "block", "block_hash",
"block_create", "blocks", "blocks_info", "block_account", "block_count", "block_count_type",
"chain", "delegators", "delegators_count", "frontiers", "frontier_count", "history",
"key_expand", "process", "representatives", "republish", "peers", "version", "pending",
"pending_exists", "price_data", "work_generate"]
# all currencies polled on CMC
currency_list = ["BTC", "AUD", "BRL", "CAD", "CHF", "CLP", "CNY", "CZK", "DKK", "EUR", "GBP", "HKD", "HUF", "IDR",
"ILS", "INR", "JPY", "KRW", "MXN", "MYR", "NOK", "NZD", "PHP", "PKR", "PLN", "RUB", "SEK", "SGD",
"THB", "TRY", "TWD", "USD", "ZAR"]
# ephemeral data
clients = {} # store websocket sessions
subscriptions = {} # store subscription ids
sub_pref_cur = {} # store currency subscription preferences [change to use redis next]
conn_count = {} # track number of open connections per IP
mesg_last = {} # track time of last message from IP
active_messages = set() # track messages in-flight - combats duplicate requests while one is active
# track work requests active, eliminate client requesting multiples on the
# same hash (drops work server efficiency as it hasnt had time to cache yet, this way it doesnt queue)
active_work = set()
def address_decode(address):
# Given a string containing an XRB/NANO address, confirm validity and provide resulting hex address
if address[:4] == 'xrb_' or address[:5] == 'nano_':
account_map = "13456789abcdefghijkmnopqrstuwxyz" # each index = binary value, account_lookup[0] == '1'
account_lookup = {}
for i in range(0, 32): # populate lookup index with prebuilt bitarrays ready to append
account_lookup[account_map[i]] = BitArray(uint=i, length=5)
data = address.split('_')[1]
acrop_key = data[:-8] # we want everything after 'xrb_' or 'nano_' but before the 8-char checksum
acrop_check = data[-8:] # extract checksum
# convert base-32 (5-bit) values to byte string by appending each 5-bit value to the bitstring,
# essentially bitshifting << 5 and then adding the 5-bit value.
number_l = BitArray()
for x in range(0, len(acrop_key)):
number_l.append(account_lookup[acrop_key[x]])
number_l = number_l[4:] # reduce from 260 to 256 bit (upper 4 bits are never used as account is a uint256)
check_l = BitArray()
for x in range(0, len(acrop_check)):
check_l.append(account_lookup[acrop_check[x]])
check_l.byteswap() # reverse byte order to match hashing format
result = number_l.hex.upper()
return result
return False
# strip whitespace, conform to string output
def strclean(instr):
if type(instr) is str:
return ' '.join(instr.split())
elif type(instr) is bytes:
return ' '.join(instr.decode('utf-8').split())
@tornado.gen.coroutine
def send_prices():
# global active_work
# active_work = set()
# empty out this set periodically, to ensure clients dont somehow get stuck when an error causes their
# work not to return
if len(clients):
print('[' + str(int(time.time())) + '] Pushing price data to ' + str(len(clients)) + ' subscribers...')
logging.info('pushing price data to ' + str(len(clients)) + ' connections')
btc = float(rdata.hget("prices", "coinmarketcap:nano-btc").decode('utf-8'))
for client in clients:
try:
try:
currency = sub_pref_cur[client]
except:
currency = 'usd'
price = float(rdata.hget("prices", "coinmarketcap:nano-" + currency.lower()).decode('utf-8'))
clients[client].write_message(
'{"currency":"' + currency.lower() + '","price":' + str(price) + ',"btc":' + str(btc) + '}')
except:
print(' > Error pushing prices for client ' + client)
logging.error('error pushing prices for client;' + handler.request.remote_ip + ';' + client)
@tornado.gen.coroutine
def rpc_request(http_client, body):
response = yield http_client.fetch(rpc_url, method='POST', body=body)
raise tornado.gen.Return(response)
@tornado.gen.coroutine
def rpc_defer(handler, message):
rpc = tornado.httpclient.AsyncHTTPClient()
response = yield rpc_request(rpc, message)
logging.info('rpc request return code;' + str(response.code))
if response.error:
logging.error('rpc defer request failure;' + str(
response.error) + ';' + rpc_url + ';' + message + ';' + handler.request.remote_ip + ';' + handler.id)
reply = "rpc defer error"
else:
logging.info('rpc defer response sent;' + str(
strclean(response.body)) + ';' + rpc_url + ';' + handler.request.remote_ip + ';' + handler.id)
reply = response.body
handler.write_message(reply)
# Since someone might get cute and attempt to spam users with low-value transactions in an effort to deny them the
# ability to receive, we will take the performance hit for them and pull all pending block data. Then we will sort by
# most valuable to least valuable. Finally, to save the client processing burden and give the server room to breathe,
# we return only the top 10.
@tornado.gen.coroutine
def pending_defer(handler, request):
rpc = tornado.httpclient.AsyncHTTPClient()
requested = json.loads(request)
response = yield rpc_request(rpc, request)
if response.error:
logging.error('pending defer request failure;' + str(
response.error) + ';' + rpc_url + ';' + message + ';' + handler.request.remote_ip + ';' + handler.id)
reply = "pending defer error"
else:
data = json.loads(response.body.decode('ascii'))
# sort dict keys by amount value within, descending
newlist = sorted(data['blocks'], key=lambda x: (int(data['blocks'][x]['amount'])), reverse=True)
# only provide the first 10
newlist = newlist[:10]
# build a new json structure
if len(newlist) > 0:
newdict = {"blocks": {}}
for x in newlist:
newdict['blocks'][x] = data['blocks'][x]
else:
# returning {} as the value for blocks causes issues with clients, RPC provides "", lets do the same.
newdict = {
"blocks": ""}
reply = json.dumps(newdict)
logging.info('pending defer response sent;' + str(
strclean(reply)) + ';' + rpc_url + ';' + handler.request.remote_ip + ';' + handler.id)
# return to client
handler.write_message(reply)
# Server-side check for any incidental mixups due to race conditions or misunderstanding protocol.
# Check blocks submitted for processing to ensure the user or client has not accidentally created a send to an unknown
# address due to balance miscalculation leading to the state block being interpreted as a send rather than a receive.
@tornado.gen.coroutine
def process_defer(handler, block):
rpc = tornado.httpclient.AsyncHTTPClient()
# check for receive race condition
# if block['type'] == 'state' and block['previous'] and block['balance'] and block['link']:
if block['type'] == 'state' and {'previous', 'balance', 'link'} <= set(block):
try:
prev_response = yield rpc_request(rpc, json.dumps({
'action': 'blocks_info',
'hashes': [block['previous']],
'balance': 'true'
}))
prev_response = json.loads(prev_response.body.decode('ascii'))
try:
prev_block = json.loads(prev_response['blocks'][block['previous']]['contents'])
if prev_block['type'] != 'state' and ('balance' in prev_block):
prev_balance = int(prev_block['balance'], 16)
elif prev_block['type'] != 'state' and ('balance' not in prev_block):
prev_balance = int(prev_response['blocks'][block['previous']]['balance'])
else:
prev_balance = int(prev_block['balance'])
if int(block['balance']) < prev_balance:
link_hash = block['link']
if link_hash.startswith('xrb_') or link_hash.startswith('nano_'):
link_hash = address_decode(link_hash)
# this is a send
link_response = yield rpc_request(rpc, json.dumps({
'action': 'block',
'hash': link_hash
}))
link_response = json.loads(link_response.body.decode('ascii'))
# print('link_response',link_response)
if 'error' not in link_response and 'contents' in link_response:
logging.error(
'rpc process receive race condition detected;' + handler.request.remote_ip +
';' + handler.id + ';User-Agent:' + str(handler.request.headers.get('User-Agent')))
handler.write_message('{"error":"receive race condition detected"}')
return
except:
# no contents, means an error was returned for previous block. no action needed
if 'error' not in prev_response:
exc_type, exc_obj, exc_tb = sys.exc_info()
print('exception', exc_type, exc_obj, exc_tb.tb_lineno)
print('prev_response', prev_response)
# print('prev_block',prev_block)
pass
except:
logging.error('rpc process receive race condition exception;' + str(
sys.exc_info()) + ';' + handler.request.remote_ip + ';' + handler.id + ';User-Agent:' + str(
handler.request.headers.get('User-Agent')))
pass
response = yield rpc_defer(handler, json.dumps({
'action': 'process',
'block': json.dumps(block)
}))
@tornado.gen.coroutine
def work_request(http_client, body):
response = yield http_client.fetch(work_url, method='POST', body=body)
raise tornado.gen.Return(response)
@tornado.gen.coroutine
def work_defer(handler, message):
request = json.loads(message)
if request['hash'] in active_work:
logging.error('work already requested;' + handler.request.remote_ip + ';' + handler.id)
return
else:
active_work.add(request['hash'])
try:
rpc = tornado.httpclient.AsyncHTTPClient()
response = yield work_request(rpc, message)
logging.info('work request return code;' + str(response.code))
if response.error:
logging.error('work defer error;' + handler.request.remote_ip + ';' + handler.id)
handler.write_message("work defer error")
else:
logging.info('work defer response sent:;' + str(
strclean(response.body)) + ';' + handler.request.remote_ip + ';' + handler.id)
handler.write_message(response.body)
active_work.remove(request['hash'])
except:
logging.error(
'work defer exception;' + str(sys.exc_info()) + ';' + handler.request.remote_ip + ';' + handler.id)
active_work.remove(request['hash'])
@tornado.gen.coroutine
def rpc_subscribe(handler, account, currency):
logging.info('subscribing;' + handler.request.remote_ip + ';' + handler.id)
rpc = tornado.httpclient.AsyncHTTPClient()
message = '{\"action\":\"account_info",\"account\":\"' + account + '\",\"pending\":true,\"representative\":true}'
logging.info('sending request;' + message + ';' + handler.request.remote_ip + ';' + handler.id)
response = yield rpc_request(rpc, message)
if response.error:
logging.error('subscribe error;' + handler.request.remote_ip + ';' + handler.id)
handler.write_message('{"error":"subscribe error"}')
else:
subscriptions[account] = handler.id
rdata.hset(handler.id, "account", account)
sub_pref_cur[handler.id] = currency
rdata.hset(handler.id, "currency", currency)
rdata.hset(handler.id, "last-connect", float(time.time()))
info = json.loads(response.body)
info['uuid'] = handler.id
price_cur = rdata.hget("prices", "coinmarketcap:nano-" + sub_pref_cur[handler.id].lower()).decode('utf-8')
price_btc = rdata.hget("prices", "coinmarketcap:nano-btc").decode('utf-8')
info['currency'] = sub_pref_cur[handler.id].lower()
info['price'] = price_cur
info['btc'] = price_btc
info = json.dumps(info)
logging.info('subscribe response sent;' + str(
strclean(response.body)) + ';' + handler.request.remote_ip + ';' + handler.id)
handler.write_message(info)
@tornado.gen.coroutine
def rpc_reconnect(handler):
logging.info('reconnecting;' + handler.request.remote_ip + ';' + handler.id)
rpc = tornado.httpclient.AsyncHTTPClient()
try:
account = rdata.hget(handler.id, "account").decode('utf-8')
except:
logging.error(
'reconnect error, account not seen on this server before;' + handler.request.remote_ip + ';' + handler.id)
handler.write_message('{"error":"reconnect error","detail":"account not seen on this server before"}')
return
message = '{\"action\":\"account_info",\"account\":\"' + account + '\",\"pending\":true,\"representative\":true}'
logging.info('sending request;' + message + ';' + handler.request.remote_ip + ';' + handler.id)
response = yield rpc_request(rpc, message)
if response.error:
logging.error('reconnect error;' + handler.request.remote_ip + ';' + handler.id)
handler.write_message('{"error":"reconnect error"}')
else:
subscriptions[
account] = handler.id # may be an issue here if we are to allow multiple clients with the same address...
sub_pref_cur[handler.id] = rdata.hget(handler.id, "currency").decode('utf-8')
rdata.hset(handler.id, "last-connect", float(time.time()))
info = json.loads(response.body.decode('ascii'))
price_cur = rdata.hget("prices", "coinmarketcap:nano-" + sub_pref_cur[handler.id].lower()).decode('utf-8')
price_btc = rdata.hget("prices", "coinmarketcap:nano-btc").decode('utf-8')
info['currency'] = sub_pref_cur[handler.id].lower()
info['price'] = float(price_cur)
info['btc'] = float(price_btc)
info = json.dumps(info)
logging.info(
'reconnect response sent ' + str(len(info)) + 'bytes;' + handler.request.remote_ip + ';' + handler.id)
handler.write_message(info)
@tornado.gen.coroutine
def rpc_accountcheck(handler, account):
logging.info('checking account;' + handler.request.remote_ip + ';' + handler.id)
rpc = tornado.httpclient.AsyncHTTPClient()
message = '{\"action\":\"account_info",\"account\":\"' + account + '\"}'
logging.info('sending request;' + message + ';' + handler.request.remote_ip + ';' + handler.id)
response = yield rpc_request(rpc, message)
if response.error:
logging.error('account check error;' + handler.request.remote_ip + ';' + handler.id)
handler.write_message('{"error":"account check error"}')
else:
info = json.loads(response.body.decode('ascii'))
try:
if info['error'] == "Account not found":
info = '{"ready":false}'
except:
info = '{"ready":true}'
logging.info('account check response sent;' + handler.request.remote_ip + ';' + handler.id)
handler.write_message(info)
class WSHandler(tornado.websocket.WebSocketHandler):
def open(self):
self.id = str(uuid.uuid4())
clients[self.id] = self
logging.info('new connection;' + self.request.remote_ip + ';' + self.id + ';User-Agent:' + str(
self.request.headers.get('User-Agent')))
def on_message(self, message):
address = str(self.request.remote_ip)
now = int(round(time.time() * 1000))
if address in mesg_last:
if (now - mesg_last[address]) < 25:
logging.error('client messaging too quickly: ' + str(
now - mesg_last[address]) + 'ms;' + self.request.remote_ip + ';' + self.id + ';User-Agent:' + str(
self.request.headers.get('User-Agent')))
mesg_last[address] = now
logging.info('request;' + message + ';' + self.request.remote_ip + ';' + self.id)
if message not in active_messages:
active_messages.add(message)
else:
logging.error('request already active;' + message + ';' + self.request.remote_ip + ';' + self.id)
return
try:
nanocast_request = json.loads(message)
if nanocast_request['action'] in allowed_rpc_actions:
if 'request_id' in nanocast_request:
requestid = nanocast_request['request_id']
else:
requestid = None
# adjust counts so nobody can block the node with a huge request - disregard, we have three nodes to
# load balance
# if 'count' in nanocast_request:
# if (nanocast_request['count'] < 0) or (nanocast_request['count'] > 1000):
# nanocast_request['count'] = 1000
# logging.info('requested count is <0 or >1000, correcting to 1000;'+self.request.remote_ip+';'+self.id)
# rpc: account_subscribe
if nanocast_request['action'] == "account_subscribe":
# already subscribed, reconnect
if 'uuid' in nanocast_request:
del clients[self.id]
self.id = nanocast_request['uuid']
clients[self.id] = self
logging.info('reconnection request;' + self.request.remote_ip + ';' + self.id)
try:
if 'currency' in nanocast_request and nanocast_request['currency'] in currency_list:
currency = nanocast_request['currency']
sub_pref_cur[self.id] = currency
rdata.hset(self.id, "currency", currency)
else:
setting = rdata.hget(self.id, "currency")
if setting is not None:
sub_pref_cur[self.id] = setting
else:
sub_pref_cur[self.id] = 'usd'
rdata.hset(self.id, "currency", 'usd')
rpc_reconnect(self)
rdata.rpush("conntrack",
str(float(time.time())) + ":" + self.id + ":connect:" + self.request.remote_ip)
except Exception as e:
logging.error('reconnect error;' + str(e) + ';' + self.request.remote_ip + ';' + self.id)
reply = {'error': 'reconnect error', 'detail': str(e)}
if requestid is not None: reply['request_id'] = requestid
self.write_message(json.dumps(reply))
# new user, setup uuid(or use existing if available) and account info
else:
logging.info('subscription request;' + self.request.remote_ip + ';' + self.id)
try:
if 'currency' in nanocast_request and nanocast_request['currency'] in currency_list:
currency = nanocast_request['currency']
else:
currency = "usd"
rpc_subscribe(self, nanocast_request['account'], currency)
rdata.rpush("conntrack",
str(float(time.time())) + ":" + self.id + ":connect:" + self.request.remote_ip)
except Exception as e:
logging.error('subscribe error;' + str(e) + ';' + self.request.remote_ip + ';' + self.id)
reply = {'error': 'subscribe error', 'detail': str(e)}
if requestid is not None: reply['request_id'] = requestid
self.write_message(json.dumps(reply))
# rpc: price_data
elif nanocast_request['action'] == "price_data":
logging.info('price data request;' + self.request.remote_ip + ';' + self.id)
try:
if nanocast_request['currency'].upper() in currency_list:
try:
price = rdata.hget("prices",
"coinmarketcap:nano-" + nanocast_request['currency'].lower()).decode(
'utf-8')
self.write_message(
'{"currency":"' + nanocast_request['currency'].lower() + '","price":' + str(
price) + '}')
except:
logging.error(
'price data error, unable to get price;' + self.request.remote_ip + ';' + self.id)
self.write_message('{"error":"price data error - unable to get price"}')
else:
logging.error(
'price data error, unknown currency;' + self.request.remote_ip + ';' + self.id)
self.write_message('{"error":"unknown currency"}')
except Exception as e:
logging.error('price data error;' + str(e) + ';' + self.request.remote_ip + ';' + self.id)
self.write_message('{"error":"price data error","detail":"' + str(e) + '"}')
# rpc: account_check
elif nanocast_request['action'] == "account_check":
logging.info('account check request;' + self.request.remote_ip + ';' + self.id)
try:
rpc_accountcheck(self, nanocast_request['account'])
except Exception as e:
logging.error('account check error;' + str(e) + ';' + self.request.remote_ip + ';' + self.id)
self.write_message('{"error":"account check error","detail":"' + str(e) + '"}')
# rpc: work_generate
elif nanocast_request['action'] == "work_generate":
if self.request.headers.get('X-Client-Version') is None:
xcver = 0
else:
xcver = int(self.request.headers.get('X-Client-Version'))
# logging.debug(self.request.headers)
if (self.request.headers.get('User-Agent') != 'SwiftWebSocket') and (xcver < 28):
logging.error(
'work rpc denied;work disable;' + self.request.remote_ip + ';' + self.id + ';User-Agent:' + str(
self.request.headers.get('User-Agent')))
self.write_message(
'{"error":"work rpc denied","detail":"you are using an old client, please update"}')
else:
try:
work_defer(self, json.dumps(nanocast_request))
except Exception as e:
logging.error('work rpc error;' + str(
e) + ';' + self.request.remote_ip + ';' + self.id + ';User-Agent:' + str(
self.request.headers.get('User-Agent')))
self.write_message('{"error":"work rpc error","detail":"' + str(e) + '"}')
# rpc: process
elif nanocast_request['action'] == "process":
try:
process_defer(self, json.loads(nanocast_request['block']))
except Exception as e:
logging.error('process rpc error;' + str(
e) + ';' + self.request.remote_ip + ';' + self.id + ';User-Agent:' + str(
self.request.headers.get('User-Agent')))
self.write_message('{"error":"process rpc error","detail":"' + str(e) + '"}')
# rpc: pending
elif nanocast_request['action'] == "pending":
try:
pending_defer(self, json.dumps(nanocast_request))
except Exception as e:
logging.error('pending rpc error;' + str(
e) + ';' + self.request.remote_ip + ';' + self.id + ';User-Agent:' + str(
self.request.headers.get('User-Agent')))
self.write_message('{"error":"pending rpc error","detail":"' + str(e) + '"}')
elif nanocast_request['action'] == 'account_history':
if rdata.hget(self.id, "account") is None:
rdata.hset(self.id, "account", nanocast_request['account'])
try:
rpc_defer(self, json.dumps(nanocast_request))
except Exception as e:
logging.error('rpc error;' + str(e) + ';' + self.request.remote_ip + ';' + self.id)
self.write_message('{"error":"rpc error","detail":"' + str(e) + '"}')
# rpc: fallthrough and error catch
else:
try:
rpc_defer(self, json.dumps(nanocast_request))
except Exception as e:
logging.error('rpc error;' + str(e) + ';' + self.request.remote_ip + ';' + self.id)
self.write_message('{"error":"rpc error","detail":"' + str(e) + '"}')
else:
logging.error(
'rpc not allowed;' + nanocast_request['action'] + ';' + self.request.remote_ip + ';' + self.id)
self.write_message('{"error":"rpc command not allowed"}')
except Exception as e:
logging.error('uncaught error;' + str(e) + ';' + self.request.remote_ip + ';' + self.id)
self.write_message('{"error":"general error","detail":"' + str(e) + '"}')
active_messages.remove(message)
# cleanup when done, allow repeats after done processing the first
active_messages.remove(message)
def on_close(self):
logging.info('connection closed;' + self.request.remote_ip + ';' + self.id + ';User-Agent:' + str(
self.request.headers.get('User-Agent')))
rdata.rpush("conntrack", str(float(time.time())) + ":" + self.id + ":disconnect:" + self.request.remote_ip)
rdata.hset(self.id, "last-disconnect", float(time.time()))
if self.id in clients: del clients[self.id]
for account in subscriptions:
if subscriptions[account] == self.id:
del subscriptions[account]
break
class Client(object):
def __init__(self, url, timeout):
self.url = url
self.timeout = timeout
self.ioloop = IOLoop.instance()
self.ws = None
self.connect()
self.ioloop.start()
@tornado.gen.coroutine
def connect(self):
print("trying to connect")
try:
self.ws = yield websocket_connect(self.url)
except Exception as e:
print("connection error")
else:
print("connected")
self.ws.write_message(
json.dumps({'action': 'subscribe', 'topic': 'confirmation', 'ack': 'true', 'id': socket.gethostname()}))
self.run()
@tornado.gen.coroutine
def run(self):
while True:
msg = yield self.ws.read_message()
if msg is not None:
self.missed = 0
data = json.loads(msg)
try:
data['topic'] == 'confirmation'
if data['message']['block']['type'] == 'send':
target = data['message']['block']['destination']
if subscriptions.get(target):
print(" Pushing to client %s" % subscriptions[target])
logging.info(
'push to client;' + json.dumps(data['message']['block']) + ';' + subscriptions[target])
clients[subscriptions[target]].write_message(json.dumps(data['message']))
elif data['message']['block']['type'] == 'state':
link = data['message']['block']['link_as_account']
if subscriptions.get(link):
print(" Pushing to client %s" % subscriptions[link])
logging.info('push to client;' + json.dumps(data['message']) + ';' + subscriptions[link])
clients[subscriptions[link]].write_message(json.dumps(data['message']))
elif subscriptions.get(data['message']['account']):
print(" Pushing to client %s" % subscriptions[data['message']['account']])
logging.info('push to client;' + json.dumps(data['message']) + ';' + subscriptions[data['message']['account']])
clients[subscriptions[data['message']['account']]].write_message(json.dumps(data['message']))
except Exception as e:
# Sometimes we receive messages we dont understand. This is ok!!
logging.info(json.dumps(data) + ' message unhandled')
application = tornado.web.Application([
(r"/", WSHandler),
])
if __name__ == "__main__":
handler = WatchedFileHandler(os.environ.get("NANO_LOG_FILE", "nanocast.log"))
formatter = logging.Formatter("%(asctime)s;%(levelname)s;%(message)s", "%Y-%m-%d %H:%M:%S %z")
handler.setFormatter(formatter)
root = logging.getLogger()
root.setLevel(os.environ.get("NANO_LOG_LEVEL", "INFO"))
root.addHandler(handler)
print("[" + str(int(time.time())) + "] Starting NANOCast Server...")
logging.info('Starting NANOCast Server')
logging.getLogger('tornado.access').disabled = True
cert = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
print(os.path.join(cert_dir, cert_crt_file), os.path.join(cert_dir, cert_key_file))
cert.load_cert_chain(os.path.join(cert_dir, cert_crt_file), os.path.join(cert_dir, cert_key_file))
https_server = tornado.httpserver.HTTPServer(application, ssl_options=cert)
https_server.listen(socket_port)
# push latest price data to all subscribers every minute
tornado.ioloop.PeriodicCallback(send_prices, 60000).start()
client = Client(websocket_url, 5)
tornado.ioloop.IOLoop.instance().start()