lib/interface.py

   1 #!/usr/bin/env python
   2 #
   3 # Electrum - lightweight Bitcoin client
   4 # Copyright (C) 2011 thomasv@gitorious
   5 #
   6 # This program is free software: you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation, either version 3 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14 # GNU General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
  18
  19
  20 import random, socket, ast, re
  21 import threading, traceback, sys, time, json, Queue
  22
  23 from version import ELECTRUM_VERSION
  24 from util import print_error
  25
  26
  27 DEFAULT_TIMEOUT = 5
  28 DEFAULT_SERVERS = [ 'electrum.novit.ro:50001:t',
  29                     'electrum.bytesized-hosting.com:50001:t']  # list of default servers
  30
  31 proxy_modes = ['socks4', 'socks5', 'http']
  32
  33 def pick_random_server():
  34     print "using random server"
  35     return random.choice( DEFAULT_SERVERS )
  36
  37
  38 def parse_proxy_options(s):
  39     if s.lower() == 'none': return None
  40     proxy = { "mode":"socks5", "host":"localhost" }
  41     args = s.split(':')
  42     n = 0
  43     if proxy_modes.count(args[n]) == 1:
  44         proxy["mode"] = args[n]
  45         n += 1
  46     if len(args) > n:
  47         proxy["host"] = args[n]
  48         n += 1
  49     if len(args) > n:
  50         proxy["port"] = args[n]
  51     else:
  52         proxy["port"] = "8080" if proxy["mode"] == "http" else "1080"
  53     return proxy
  54
  55
  56
  57
  58 class InterfaceAncestor(threading.Thread):
  59
  60     def __init__(self, host, port, proxy=None):
  61         threading.Thread.__init__(self)
  62         self.daemon = True
  63         self.host = host
  64         self.port = port
  65         self.proxy = proxy
  66
  67         self.servers = [] # actual list from IRC
  68         self.rtime = 0
  69         self.bytes_received = 0
  70
  71         self.is_connected = True
  72         self.poll_interval = 1
  73
  74         #json
  75         self.message_id = 0
  76         self.responses = Queue.Queue()
  77         self.unanswered_requests = {}
  78
  79     def init_socket(self):
  80         pass
  81
  82     def poke(self):
  83         # push a fake response so that the getting thread exits its loop
  84         self.responses.put(None)
  85
  86     def queue_json_response(self, c):
  87
  88         # uncomment to debug
  89         # print "<--",c
  90
  91         msg_id = c.get('id')
  92         error = c.get('error')
  93
  94         if error:
  95             print "received error:", c
  96             return
  97
  98         if msg_id is not None:
  99             method, params = self.unanswered_requests.pop(msg_id)
 100             result = c.get('result')
 101         else:
 102             # notification
 103             method = c.get('method')
 104             params = c.get('params')
 105
 106             if method == 'blockchain.numblocks.subscribe':
 107                 result = params[0]
 108                 params = []
 109
 110             elif method == 'blockchain.address.subscribe':
 111                 addr = params[0]
 112                 result = params[1]
 113                 params = [addr]
 114
 115         self.responses.put({'method':method, 'params':params, 'result':result})
 116
 117
 118
 119     def subscribe(self, addresses):
 120         messages = []
 121         for addr in addresses:
 122             messages.append(('blockchain.address.subscribe', [addr]))
 123         self.send(messages)
 124
 125
 126
 127
 128 class PollingInterface(InterfaceAncestor):
 129     """ non-persistent connection. synchronous calls"""
 130
 131     def __init__(self, host, port, proxy=None):
 132         InterfaceAncestor.__init__(self, host, port, proxy)
 133         self.session_id = None
 134
 135     def get_history(self, address):
 136         self.send([('blockchain.address.get_history', [address] )])
 137
 138     def poll(self):
 139         pass
 140
 141     def run(self):
 142         self.is_connected = True
 143         while self.is_connected:
 144             try:
 145                 if self.session_id:
 146                     self.poll()
 147                 time.sleep(self.poll_interval)
 148             except socket.gaierror:
 149                 break
 150             except socket.error:
 151                 break
 152             except:
 153                 traceback.print_exc(file=sys.stdout)
 154                 break
 155
 156         self.is_connected = False
 157         self.poke()
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167 class HttpStratumInterface(PollingInterface):
 168
 169     def poll(self):
 170         self.send([])
 171
 172     def send(self, messages):
 173         import urllib2, json, time, cookielib
 174
 175         if self.proxy:
 176             import socks
 177             socks.setdefaultproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) )
 178             socks.wrapmodule(urllib2)
 179
 180         cj = cookielib.CookieJar()
 181         opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
 182         urllib2.install_opener(opener)
 183
 184         t1 = time.time()
 185
 186         data = []
 187         for m in messages:
 188             method, params = m
 189             if type(params) != type([]): params = [params]
 190             data.append( { 'method':method, 'id':self.message_id, 'params':params } )
 191             self.unanswered_requests[self.message_id] = method, params
 192             self.message_id += 1
 193
 194         if data:
 195             data_json = json.dumps(data)
 196         else:
 197             # poll with GET
 198             data_json = None
 199
 200         host = 'http://%s:%d'%( self.host, self.port )
 201         headers = {'content-type': 'application/json'}
 202         if self.session_id:
 203             headers['cookie'] = 'SESSION=%s'%self.session_id
 204
 205         req = urllib2.Request(host, data_json, headers)
 206         response_stream = urllib2.urlopen(req)
 207
 208         for index, cookie in enumerate(cj):
 209             if cookie.name=='SESSION':
 210                 self.session_id = cookie.value
 211
 212         response = response_stream.read()
 213         self.bytes_received += len(response)
 214         if response:
 215             response = json.loads( response )
 216             if type(response) is not type([]):
 217                 self.queue_json_response(response)
 218             else:
 219                 for item in response:
 220                     self.queue_json_response(item)
 221
 222         if response:
 223             self.poll_interval = 1
 224         else:
 225             if self.poll_interval < 15:
 226                 self.poll_interval += 1
 227         #print self.poll_interval, response
 228
 229         self.rtime = time.time() - t1
 230         self.is_connected = True
 231
 232
 233
 234
 235 class TcpStratumInterface(InterfaceAncestor):
 236     """json-rpc over persistent TCP connection, asynchronous"""
 237
 238     def __init__(self, host, port, proxy=None):
 239         InterfaceAncestor.__init__(self, host, port, proxy)
 240
 241     def init_socket(self):
 242         global proxy_modes
 243         connection_msg = "%s:%d"%(self.host,self.port)
 244         if self.proxy is None:
 245             self.s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
 246         else:
 247             connection_msg += " using proxy %s:%s:%s"%(self.proxy.get('mode'), self.proxy.get('host'), self.proxy.get('port'))
 248             import socks
 249             self.s = socks.socksocket()
 250             self.s.setproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) )
 251         self.s.settimeout(60)
 252         self.s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
 253         try:
 254             self.s.connect(( self.host.encode('ascii'), int(self.port)))
 255             self.is_connected = True
 256             self.send([('server.version', [ELECTRUM_VERSION])])
 257             print "Connected to " + connection_msg
 258         except:
 259             self.is_connected = False
 260             print_error("Failed to connect" + connection_msg)
 261
 262     def run(self):
 263         try:
 264             out = ''
 265             while self.is_connected:
 266                 try: msg = self.s.recv(1024)
 267                 except socket.timeout:
 268                     # ping the server with server.version, as a real ping does not exist yet
 269                     self.send([('server.version', [ELECTRUM_VERSION])])
 270                     continue
 271                 out += msg
 272                 self.bytes_received += len(msg)
 273                 if msg == '':
 274                     self.is_connected = False
 275                     print "Disconnected."
 276
 277                 while True:
 278                     s = out.find('\n')
 279                     if s==-1: break
 280                     c = out[0:s]
 281                     out = out[s+1:]
 282                     c = json.loads(c)
 283                     self.queue_json_response(c)
 284
 285         except:
 286             traceback.print_exc(file=sys.stdout)
 287
 288         self.is_connected = False
 289         print "Poking"
 290         self.poke()
 291
 292     def send(self, messages):
 293         out = ''
 294         for m in messages:
 295             method, params = m
 296             request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } )
 297             self.unanswered_requests[self.message_id] = method, params
 298
 299             # uncomment to debug
 300             # print "-->",request
 301
 302             self.message_id += 1
 303             out += request + '\n'
 304
 305         while out:
 306             sent = self.s.send( out )
 307             out = out[sent:]
 308
 309     def get_history(self, addr):
 310         self.send([('blockchain.address.get_history', [addr])])
 311
 312
 313
 314 class Interface(TcpStratumInterface, HttpStratumInterface):
 315
 316     def __init__(self, config):
 317
 318         try:
 319             s = config.get('server')
 320             host, port, protocol = s.split(':')
 321             port = int(port)
 322         except:
 323             s = pick_random_server()
 324             host, port, protocol = s.split(':')
 325             port = int(port)
 326
 327         proxy = config.get('proxy')
 328         self.server = host + ':%d:%s'%(port, protocol)
 329
 330         #print protocol, host, port
 331         if protocol == 't':
 332             TcpStratumInterface.__init__(self, host, port, proxy)
 333         elif protocol == 'h':
 334             HttpStratumInterface.__init__(self, host, port, proxy)
 335         else:
 336             print_error("Error: Unknown protocol")
 337             TcpStratumInterface.__init__(self, host, port, proxy)
 338
 339
 340     def set_server(self, server, proxy=None):
 341         # raise an error if the format isnt correct
 342         a,b,c = server.split(':')
 343         b = int(b)
 344         assert c in ['t', 'h']
 345         # set the server
 346         if server != self.server or proxy != self.proxy:
 347             print "changing server:", server, proxy
 348             self.server = server
 349             self.proxy = proxy
 350             self.is_connected = False  # this exits the polling loop
 351             self.poke()
 352
 353
 354
 355
 356
 357
 358 class WalletSynchronizer(threading.Thread):
 359
 360     def __init__(self, wallet, config, loop=False, servers_loaded_callback=None):
 361         threading.Thread.__init__(self)
 362         self.daemon = True
 363         self.wallet = wallet
 364         self.loop = loop
 365         self.config = config
 366         self.init_interface()
 367         self.servers_loaded_callback = servers_loaded_callback
 368
 369     def init_interface(self):
 370         self.interface = Interface(self.config)
 371         self.wallet.interface = self.interface
 372
 373     def handle_response(self, r):
 374         if r is None:
 375             return
 376
 377         method = r['method']
 378         params = r['params']
 379         result = r['result']
 380
 381         if method == 'server.banner':
 382             self.wallet.banner = result
 383             self.wallet.was_updated = True
 384
 385         elif method == 'server.peers.subscribe':
 386             servers = []
 387             for item in result:
 388                 s = []
 389                 host = item[1]
 390                 ports = []
 391                 version = None
 392                 if len(item) > 2:
 393                     for v in item[2]:
 394                         if re.match("[th]\d+", v):
 395                             ports.append((v[0], v[1:]))
 396                         if re.match("v(.?)+", v):
 397                             version = v[1:]
 398                 if ports and version:
 399                     servers.append((host, ports))
 400             self.interface.servers = servers
 401             # servers_loaded_callback is None for commands, but should
 402             # NEVER be None when using the GUI.
 403             if self.servers_loaded_callback is not None:
 404                 self.servers_loaded_callback()
 405
 406         elif method == 'blockchain.address.subscribe':
 407             addr = params[0]
 408             self.wallet.receive_status_callback(addr, result)
 409
 410         elif method == 'blockchain.address.get_history':
 411             addr = params[0]
 412             self.wallet.receive_history_callback(addr, result)
 413             self.wallet.was_updated = True
 414
 415         elif method == 'blockchain.transaction.broadcast':
 416             self.wallet.tx_result = result
 417             self.wallet.tx_event.set()
 418
 419         elif method == 'blockchain.numblocks.subscribe':
 420             self.wallet.blocks = result
 421             self.wallet.was_updated = True
 422
 423         elif method == 'server.version':
 424             pass
 425
 426         else:
 427             print_error("Error: Unknown message:" + method + ", " + params + ", " + result)
 428
 429
 430     def start_interface(self):
 431         self.interface.init_socket()
 432         self.interface.start()
 433         if self.interface.is_connected:
 434             self.wallet.start_session(self.interface)
 435
 436
 437
 438     def run(self):
 439         import socket, time
 440         self.start_interface()
 441         while True:
 442             while self.interface.is_connected:
 443                 new_addresses = self.wallet.synchronize()
 444                 if new_addresses:
 445                     self.interface.subscribe(new_addresses)
 446
 447                 if self.wallet.is_up_to_date():
 448                     if not self.wallet.up_to_date:
 449                         self.wallet.up_to_date = True
 450                         self.wallet.was_updated = True
 451                         self.wallet.up_to_date_event.set()
 452                 else:
 453                     if self.wallet.up_to_date:
 454                         self.wallet.up_to_date = False
 455                         self.wallet.was_updated = True
 456
 457                 if self.wallet.was_updated:
 458                     self.wallet.trigger_callbacks()
 459                     self.wallet.was_updated = False
 460
 461                 response = self.interface.responses.get()
 462                 self.handle_response(response)
 463
 464             self.wallet.trigger_callbacks()
 465             if self.loop:
 466                 time.sleep(5)
 467                 # Server has been changed. Copy callback for new interface.
 468                 self.proxy = self.interface.proxy
 469                 self.init_interface()
 470                 self.start_interface()
 471                 continue
 472             else:
 473                 break
 474
 475
 476