#!/usr/bin/env python # # Electrum - lightweight Bitcoin client # Copyright (C) 2011 thomasv@gitorious # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import random, socket, ast, re import threading, traceback, sys, time, json, Queue from version import ELECTRUM_VERSION from util import print_error DEFAULT_TIMEOUT = 5 DEFAULT_SERVERS = [ 'electrum.novit.ro:50001:t', 'electrum.pdmc.net:50001:t', #'ecdsa.org:50002:s', 'electrum.bitcoins.sk:50001:t', 'uncle-enzo.info:50001:t', 'electrum.bytesized-hosting.com:50001:t', 'california.stratum.bitcoin.cz:50001:t', 'electrum.bitfoo.org:50001:t' ] proxy_modes = ['socks4', 'socks5', 'http'] def pick_random_server(): return random_choice( DEFAULT_SERVERS ) def pick_random_interface(): servers = DEFAULT_SERVERS while servers: server = random.choice( servers ) servers.remove(server) i = Interface({'server':server}) if i.is_connected: return i raise BaseException('no server available') class InterfaceAncestor(threading.Thread): def __init__(self, host, port, proxy=None, use_ssl=True): threading.Thread.__init__(self) self.daemon = True self.host = host self.port = port self.proxy = proxy self.use_ssl = use_ssl self.servers = [] # actual list from IRC self.rtime = 0 self.bytes_received = 0 self.is_connected = True self.poll_interval = 1 #json self.message_id = 0 self.responses = Queue.Queue() self.unanswered_requests = {} def poke(self): # push a fake response so that the getting thread exits its loop self.responses.put(None) def queue_json_response(self, c): # uncomment to debug # print "<--",c msg_id = c.get('id') error = c.get('error') if error: print "received error:", c return if msg_id is not None: method, params = self.unanswered_requests.pop(msg_id) result = c.get('result') else: # notification method = c.get('method') params = c.get('params') if method == 'blockchain.numblocks.subscribe': result = params[0] params = [] elif method == 'blockchain.address.subscribe': addr = params[0] result = params[1] params = [addr] self.responses.put({'method':method, 'params':params, 'result':result, 'id':msg_id}) def subscribe(self, addresses): messages = [] for addr in addresses: messages.append(('blockchain.address.subscribe', [addr])) self.send(messages) class HttpStratumInterface(InterfaceAncestor): """ non-persistent connection. synchronous calls""" def __init__(self, host, port, proxy=None, use_ssl=True): InterfaceAncestor.__init__(self, host, port, proxy, use_ssl) self.session_id = None self.connection_msg = ('https' if self.use_ssl else 'http') + '://%s:%d'%( self.host, self.port ) def get_history(self, address): self.send([('blockchain.address.get_history', [address] )]) def run(self): self.is_connected = True while self.is_connected: try: if self.session_id: self.poll() time.sleep(self.poll_interval) except socket.gaierror: break except socket.error: break except: traceback.print_exc(file=sys.stdout) break self.is_connected = False self.poke() def poll(self): self.send([]) def send(self, messages): import urllib2, json, time, cookielib if self.proxy: import socks socks.setdefaultproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) ) socks.wrapmodule(urllib2) cj = cookielib.CookieJar() opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj)) urllib2.install_opener(opener) t1 = time.time() data = [] for m in messages: method, params = m if type(params) != type([]): params = [params] data.append( { 'method':method, 'id':self.message_id, 'params':params } ) self.unanswered_requests[self.message_id] = method, params self.message_id += 1 if data: data_json = json.dumps(data) else: # poll with GET data_json = None headers = {'content-type': 'application/json'} if self.session_id: headers['cookie'] = 'SESSION=%s'%self.session_id req = urllib2.Request(self.connection_msg, data_json, headers) response_stream = urllib2.urlopen(req) for index, cookie in enumerate(cj): if cookie.name=='SESSION': self.session_id = cookie.value response = response_stream.read() self.bytes_received += len(response) if response: response = json.loads( response ) if type(response) is not type([]): self.queue_json_response(response) else: for item in response: self.queue_json_response(item) if response: self.poll_interval = 1 else: if self.poll_interval < 15: self.poll_interval += 1 #print self.poll_interval, response self.rtime = time.time() - t1 self.is_connected = True class TcpStratumInterface(InterfaceAncestor): """json-rpc over persistent TCP connection, asynchronous""" def __init__(self, host, port, proxy=None, use_ssl=True): InterfaceAncestor.__init__(self, host, port, proxy, use_ssl) self.init_socket() def init_socket(self): import ssl global proxy_modes self.connection_msg = "%s:%d"%(self.host,self.port) if self.proxy is None: s = socket.socket( socket.AF_INET, socket.SOCK_STREAM ) else: self.connection_msg += " using proxy %s:%s:%s"%(self.proxy.get('mode'), self.proxy.get('host'), self.proxy.get('port')) import socks s = socks.socksocket() s.setproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) ) if self.use_ssl: s = ssl.wrap_socket(s, ssl_version=ssl.PROTOCOL_SSLv23) s.settimeout(2) s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) try: s.connect(( self.host.encode('ascii'), int(self.port))) s.settimeout(60) self.s = s self.is_connected = True self.send([('server.version', [ELECTRUM_VERSION])]) except: self.is_connected = False self.s = None def run(self): try: out = '' while self.is_connected: try: msg = self.s.recv(1024) except socket.timeout: # ping the server with server.version, as a real ping does not exist yet self.send([('server.version', [ELECTRUM_VERSION])]) continue out += msg self.bytes_received += len(msg) if msg == '': self.is_connected = False print "Disconnected." while True: s = out.find('\n') if s==-1: break c = out[0:s] out = out[s+1:] c = json.loads(c) self.queue_json_response(c) except: traceback.print_exc(file=sys.stdout) self.is_connected = False print "Poking" self.poke() def send(self, messages): """return the ids of the requests that we sent""" out = '' ids = [] for m in messages: method, params = m request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } ) self.unanswered_requests[self.message_id] = method, params ids.append(self.message_id) # uncomment to debug # print "-->",request self.message_id += 1 out += request + '\n' while out: sent = self.s.send( out ) out = out[sent:] return ids def get_history(self, addr): self.send([('blockchain.address.get_history', [addr])]) class Interface(TcpStratumInterface, HttpStratumInterface): def __init__(self, config = None): if config is None: from simple_config import SimpleConfig config = SimpleConfig() s = config.get('server') host, port, protocol = s.split(':') port = int(port) self.protocol = protocol proxy = self.parse_proxy_options(config.get('proxy')) self.server = host + ':%d:%s'%(port, protocol) #print protocol, host, port if protocol in 'st': TcpStratumInterface.__init__(self, host, port, proxy, use_ssl=(protocol=='s')) elif protocol in 'gh': HttpStratumInterface.__init__(self, host, port, proxy, use_ssl=(protocol=='g')) else: raise BaseException('Unknown protocol: %s'%protocol) def run(self): if self.protocol in 'st': TcpStratumInterface.run(self) else: HttpStratumInterface.run(self) def send(self, messages): if self.protocol in 'st': return TcpStratumInterface.send(self, messages) else: return HttpStratumInterface.send(self, messages) def parse_proxy_options(self, s): if type(s) == type({}): return s # fixme: type should be fixed if type(s) != type(""): return None if s.lower() == 'none': return None proxy = { "mode":"socks5", "host":"localhost" } args = s.split(':') n = 0 if proxy_modes.count(args[n]) == 1: proxy["mode"] = args[n] n += 1 if len(args) > n: proxy["host"] = args[n] n += 1 if len(args) > n: proxy["port"] = args[n] else: proxy["port"] = "8080" if proxy["mode"] == "http" else "1080" return proxy def set_server(self, server, proxy=None): # raise an error if the format isnt correct a,b,c = server.split(':') b = int(b) assert c in 'stgh' # set the server if server != self.server or proxy != self.proxy: print "changing server:", server, proxy self.server = server self.proxy = proxy self.is_connected = False # this exits the polling loop self.poke() def is_up_to_date(self): return self.responses.empty() and not self.unanswered_requests def synchronous_get(self, requests, timeout=100000000): # todo: use generators, unanswered_requests should be a list of arrays... ids = self.send(requests) id2 = ids[:] res = {} while ids: r = self.responses.get(True, timeout) _id = r.get('id') if _id in ids: ids.remove(_id) res[_id] = r.get('result') out = [] for _id in id2: out.append(res[_id]) return out class WalletSynchronizer(threading.Thread): def __init__(self, wallet, config, loop=False, servers_loaded_callback=None): threading.Thread.__init__(self) self.daemon = True self.wallet = wallet self.loop = loop self.config = config self.init_interface() self.servers_loaded_callback = servers_loaded_callback def init_interface(self): if self.config.get('server'): self.interface = Interface(self.config) else: print "Using random server..." self.interface = pick_random_interface() if self.interface.is_connected: print "Connected to " + self.interface.connection_msg else: print_error("Failed to connect " + self.interface.connection_msg) self.wallet.interface = self.interface def handle_response(self, r): if r is None: return method = r['method'] params = r['params'] result = r['result'] if method == 'server.banner': self.wallet.banner = result self.wallet.was_updated = True elif method == 'server.peers.subscribe': servers = [] for item in result: s = [] host = item[1] ports = [] version = None if len(item) > 2: for v in item[2]: if re.match("[stgh]\d+", v): ports.append((v[0], v[1:])) if re.match("v(.?)+", v): version = v[1:] if ports and version: servers.append((host, ports)) self.interface.servers = servers # servers_loaded_callback is None for commands, but should # NEVER be None when using the GUI. if self.servers_loaded_callback is not None: self.servers_loaded_callback() elif method == 'blockchain.address.subscribe': addr = params[0] self.wallet.receive_status_callback(addr, result) elif method == 'blockchain.address.get_history': addr = params[0] self.wallet.receive_history_callback(addr, result) self.wallet.was_updated = True elif method == 'blockchain.transaction.broadcast': self.wallet.tx_result = result self.wallet.tx_event.set() elif method == 'blockchain.numblocks.subscribe': self.wallet.blocks = result self.wallet.was_updated = True elif method == 'server.version': pass else: print_error("Error: Unknown message:" + method + ", " + params + ", " + result) def start_interface(self): self.interface.start() if self.interface.is_connected: self.wallet.start_session(self.interface) def run(self): import socket, time self.start_interface() while True: while self.interface.is_connected: new_addresses = self.wallet.synchronize() if new_addresses: self.interface.subscribe(new_addresses) if self.interface.is_up_to_date(): if not self.wallet.up_to_date: self.wallet.up_to_date = True self.wallet.was_updated = True self.wallet.up_to_date_event.set() else: if self.wallet.up_to_date: self.wallet.up_to_date = False self.wallet.was_updated = True if self.wallet.was_updated: self.wallet.trigger_callbacks() self.wallet.was_updated = False response = self.interface.responses.get() self.handle_response(response) self.wallet.trigger_callbacks() if self.loop: time.sleep(5) # Server has been changed. Copy callback for new interface. self.proxy = self.interface.proxy self.init_interface() self.start_interface() continue else: break