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, ssl, errno, os
  21 import threading, traceback, sys, time, json, Queue
  22 import socks
  23
  24 from version import ELECTRUM_VERSION, PROTOCOL_VERSION
  25 from util import print_error, print_msg
  26
  27
  28 DEFAULT_TIMEOUT = 5
  29 proxy_modes = ['socks4', 'socks5', 'http']
  30
  31
  32 class Interface(threading.Thread):
  33
  34
  35     def __init__(self, config=None):
  36
  37         if config is None:
  38             from simple_config import SimpleConfig
  39             config = SimpleConfig()
  40
  41         threading.Thread.__init__(self)
  42         self.daemon = True
  43         self.config = config
  44         self.connect_event = threading.Event()
  45
  46         self.subscriptions = {}
  47         self.lock = threading.Lock()
  48
  49         self.rtime = 0
  50         self.bytes_received = 0
  51         self.is_connected = False
  52         self.poll_interval = 1
  53
  54         #json
  55         self.message_id = 0
  56         self.unanswered_requests = {}
  57         self.pending_transactions_for_notifications= []
  58
  59         # parse server
  60         s = config.get('server')
  61         host, port, protocol = s.split(':')
  62         port = int(port)
  63         if protocol not in 'ghst':
  64             raise BaseException('Unknown protocol: %s'%protocol)
  65
  66         self.host = host
  67         self.port = port
  68         self.protocol = protocol
  69         self.use_ssl = ( protocol in 'sg' )
  70         self.proxy = self.parse_proxy_options(config.get('proxy'))
  71         self.server = host + ':%d:%s'%(port, protocol)
  72
  73
  74
  75     def queue_json_response(self, c):
  76
  77         # uncomment to debug
  78         # print_error( "<--",c )
  79
  80         msg_id = c.get('id')
  81         error = c.get('error')
  82
  83         if error:
  84             print_error("received error:", c)
  85             if msg_id is not None:
  86                 with self.lock:
  87                     method, params, callback = self.unanswered_requests.pop(msg_id)
  88                 callback(self,{'method':method, 'params':params, 'error':error, 'id':msg_id})
  89
  90             return
  91
  92         if msg_id is not None:
  93             with self.lock:
  94                 method, params, callback = self.unanswered_requests.pop(msg_id)
  95             result = c.get('result')
  96
  97         else:
  98             # notification
  99             method = c.get('method')
 100             params = c.get('params')
 101
 102             if method == 'blockchain.numblocks.subscribe':
 103                 result = params[0]
 104                 params = []
 105
 106             elif method == 'blockchain.headers.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             with self.lock:
 116                 for k,v in self.subscriptions.items():
 117                     if (method, params) in v:
 118                         callback = k
 119                         break
 120                 else:
 121                     print_error( "received unexpected notification", method, params)
 122                     print_error( self.subscriptions )
 123                     return
 124
 125
 126         callback(self, {'method':method, 'params':params, 'result':result, 'id':msg_id})
 127
 128
 129     def on_version(self, i, result):
 130         self.server_version = result
 131
 132
 133     def init_http(self, host, port, proxy=None, use_ssl=True):
 134         self.session_id = None
 135         self.is_connected = True
 136         self.connection_msg = ('https' if self.use_ssl else 'http') + '://%s:%d'%( self.host, self.port )
 137         try:
 138             self.poll()
 139         except:
 140             print_error("http init session failed")
 141             self.is_connected = False
 142             return
 143
 144         if self.session_id:
 145             print_error('http session:',self.session_id)
 146             self.is_connected = True
 147         else:
 148             self.is_connected = False
 149
 150     def run_http(self):
 151         self.is_connected = True
 152         while self.is_connected:
 153             try:
 154                 if self.session_id:
 155                     self.poll()
 156                 time.sleep(self.poll_interval)
 157             except socket.gaierror:
 158                 break
 159             except socket.error:
 160                 break
 161             except:
 162                 traceback.print_exc(file=sys.stdout)
 163                 break
 164
 165         self.is_connected = False
 166
 167
 168     def poll(self):
 169         self.send([])
 170
 171
 172     def send_http(self, messages, callback):
 173         import urllib2, json, time, cookielib
 174         print_error( "send_http", messages )
 175
 176         if self.proxy:
 177             socks.setdefaultproxy(proxy_modes.index(self.proxy["mode"]) + 1, 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, callback
 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
 201         headers = {'content-type': 'application/json'}
 202         if self.session_id:
 203             headers['cookie'] = 'SESSION=%s'%self.session_id
 204
 205         try:
 206             req = urllib2.Request(self.connection_msg, data_json, headers)
 207             response_stream = urllib2.urlopen(req, timeout=DEFAULT_TIMEOUT)
 208         except:
 209             return
 210
 211         for index, cookie in enumerate(cj):
 212             if cookie.name=='SESSION':
 213                 self.session_id = cookie.value
 214
 215         response = response_stream.read()
 216         self.bytes_received += len(response)
 217         if response:
 218             response = json.loads( response )
 219             if type(response) is not type([]):
 220                 self.queue_json_response(response)
 221             else:
 222                 for item in response:
 223                     self.queue_json_response(item)
 224
 225         if response:
 226             self.poll_interval = 1
 227         else:
 228             if self.poll_interval < 15:
 229                 self.poll_interval += 1
 230         #print self.poll_interval, response
 231
 232         self.rtime = time.time() - t1
 233         self.is_connected = True
 234
 235
 236
 237
 238     def start_tcp(self):
 239
 240         if self.use_ssl:
 241             cert_path = os.path.join( self.config.get('path'), 'certs', self.host)
 242             if not os.path.exists(cert_path):
 243                 dir_path = os.path.join( self.config.get('path'), 'certs')
 244                 if not os.path.exists(dir_path):
 245                     os.mkdir(dir_path)
 246                 try:
 247                     cert = ssl.get_server_certificate((self.host, self.port))
 248                 except:
 249                     print_error("failed to connect", self.host, self.port)
 250                     return
 251
 252                 with open(cert_path,"w") as f:
 253                     f.write(cert)
 254
 255         self.connection_msg = "%s:%d"%(self.host, self.port)
 256
 257
 258         if self.proxy is None:
 259             s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
 260         else:
 261             self.connection_msg += " using proxy %s:%s:%s"%(self.proxy.get('mode'), self.proxy.get('host'), self.proxy.get('port'))
 262             s = socks.socksocket()
 263             s.setproxy(proxy_modes.index(self.proxy["mode"]) + 1, self.proxy["host"], int(self.proxy["port"]) )
 264
 265         s.settimeout(2)
 266         s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
 267
 268         try:
 269             s.connect(( self.host.encode('ascii'), int(self.port)))
 270         except:
 271             print_error("failed to connect", self.host, self.port)
 272             return
 273
 274         if self.use_ssl:
 275             try:
 276                 s = ssl.wrap_socket(s,
 277                                     ssl_version=ssl.PROTOCOL_SSLv3,
 278                                     cert_reqs=ssl.CERT_REQUIRED,
 279                                     ca_certs=cert_path,
 280                                     do_handshake_on_connect=True)
 281             except ssl.SSLError, e:
 282                 print_error("SSL error:", self.host, e)
 283                 return
 284             except:
 285                 traceback.print_exc(file=sys.stdout)
 286                 print_error("wrap_socket failed", self.host)
 287                 return
 288
 289         # hostname verification (disabled)
 290         if self.use_ssl and False:
 291             from backports.ssl_match_hostname import match_hostname, CertificateError
 292             try:
 293                 match_hostname(s.getpeercert(), self.host)
 294                 print_error("hostname matches", self.host)
 295             except CertificateError, ce:
 296                 print_error("hostname does not match", self.host, s.getpeercert())
 297                 return
 298
 299         s.settimeout(60)
 300         self.s = s
 301         self.is_connected = True
 302         print_error("connected to", self.host, self.port)
 303
 304
 305     def run_tcp(self):
 306         try:
 307             #if self.use_ssl: self.s.do_handshake()
 308             out = ''
 309             while self.is_connected:
 310                 try:
 311                     timeout = False
 312                     msg = self.s.recv(1024)
 313                 except socket.timeout:
 314                     timeout = True
 315                 except ssl.SSLError:
 316                     timeout = True
 317                 except socket.error, err:
 318                     if err.errno in [11, 10035]:
 319                         print_error("socket errno", err.errno)
 320                         time.sleep(0.1)
 321                         continue
 322                     else:
 323                         traceback.print_exc(file=sys.stdout)
 324                         raise
 325
 326                 if timeout:
 327                     # ping the server with server.version, as a real ping does not exist yet
 328                     self.send([('server.version', [ELECTRUM_VERSION, PROTOCOL_VERSION])], self.on_version)
 329                     continue
 330
 331                 out += msg
 332                 self.bytes_received += len(msg)
 333                 if msg == '':
 334                     self.is_connected = False
 335
 336                 while True:
 337                     s = out.find('\n')
 338                     if s==-1: break
 339                     c = out[0:s]
 340                     out = out[s+1:]
 341                     c = json.loads(c)
 342                     self.queue_json_response(c)
 343
 344         except:
 345             traceback.print_exc(file=sys.stdout)
 346
 347         self.is_connected = False
 348
 349
 350     def send_tcp(self, messages, callback):
 351         """return the ids of the requests that we sent"""
 352         out = ''
 353         ids = []
 354         for m in messages:
 355             method, params = m
 356             request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } )
 357             self.unanswered_requests[self.message_id] = method, params, callback
 358             ids.append(self.message_id)
 359             # uncomment to debug
 360             # print "-->", request
 361             self.message_id += 1
 362             out += request + '\n'
 363         while out:
 364             try:
 365                 sent = self.s.send( out )
 366                 out = out[sent:]
 367             except socket.error,e:
 368                 if e[0] in (errno.EWOULDBLOCK,errno.EAGAIN):
 369                     print_error( "EAGAIN: retrying")
 370                     time.sleep(0.1)
 371                     continue
 372                 else:
 373                     traceback.print_exc(file=sys.stdout)
 374                     # this happens when we get disconnected
 375                     print_error( "Not connected, cannot send" )
 376                     return None
 377         return ids
 378
 379
 380
 381
 382
 383     def start_interface(self):
 384
 385         if self.protocol in 'st':
 386             self.start_tcp()
 387         elif self.protocol in 'gh':
 388             self.start_http()
 389
 390         self.connect_event.set()
 391
 392
 393
 394     def stop_subscriptions(self):
 395         for callback in self.subscriptions.keys():
 396             callback(self, None)
 397         self.subscriptions = {}
 398
 399
 400     def send(self, messages, callback):
 401
 402         sub = []
 403         for message in messages:
 404             m, v = message
 405             if m[-10:] == '.subscribe':
 406                 sub.append(message)
 407
 408         if sub:
 409             with self.lock:
 410                 if self.subscriptions.get(callback) is None:
 411                     self.subscriptions[callback] = []
 412                 for message in sub:
 413                     if message not in self.subscriptions[callback]:
 414                         self.subscriptions[callback].append(message)
 415
 416         if not self.is_connected:
 417             return
 418
 419         if self.protocol in 'st':
 420             with self.lock:
 421                 out = self.send_tcp(messages, callback)
 422         else:
 423             # do not use lock, http is synchronous
 424             out = self.send_http(messages, callback)
 425
 426         return out
 427
 428
 429     def parse_proxy_options(self, s):
 430         if type(s) == type({}): return s  # fixme: type should be fixed
 431         if type(s) != type(""): return None
 432         if s.lower() == 'none': return None
 433         proxy = { "mode":"socks5", "host":"localhost" }
 434         args = s.split(':')
 435         n = 0
 436         if proxy_modes.count(args[n]) == 1:
 437             proxy["mode"] = args[n]
 438             n += 1
 439         if len(args) > n:
 440             proxy["host"] = args[n]
 441             n += 1
 442         if len(args) > n:
 443             proxy["port"] = args[n]
 444         else:
 445             proxy["port"] = "8080" if proxy["mode"] == "http" else "1080"
 446         return proxy
 447
 448
 449
 450     def stop(self):
 451         if self.is_connected and self.protocol in 'st' and self.s:
 452             self.s.shutdown(socket.SHUT_RDWR)
 453             self.s.close()
 454
 455
 456     def is_up_to_date(self):
 457         return self.unanswered_requests == {}
 458
 459
 460     def synchronous_get(self, requests, timeout=100000000):
 461         # todo: use generators, unanswered_requests should be a list of arrays...
 462         queue = Queue.Queue()
 463         ids = self.send(requests, lambda i,r: queue.put(r))
 464         id2 = ids[:]
 465         res = {}
 466         while ids:
 467             r = queue.get(True, timeout)
 468             _id = r.get('id')
 469             if _id in ids:
 470                 ids.remove(_id)
 471                 res[_id] = r.get('result')
 472         out = []
 473         for _id in id2:
 474             out.append(res[_id])
 475         return out
 476
 477
 478     def start(self, queue):
 479         self.queue = queue
 480         threading.Thread.start(self)
 481
 482
 483     def run(self):
 484         self.start_interface()
 485         if self.is_connected:
 486             self.send([('server.version', [ELECTRUM_VERSION, PROTOCOL_VERSION])], self.on_version)
 487             self.change_status()
 488             self.run_tcp() if self.protocol in 'st' else self.run_http()
 489         self.change_status()
 490
 491
 492     def change_status(self):
 493         #print "change status", self.server, self.is_connected
 494         self.queue.put(self)
 495
 496
 497
 498 if __name__ == "__main__":
 499
 500     q = Queue.Queue()
 501     i = Interface({'server':'btc.it-zone.org:50002:s', 'path':'/extra/key/wallet', 'verbose':True})
 502     i.start(q)
 503     time.sleep(1)
 504     exit()
 505
 506