3 # Electrum - lightweight Bitcoin client
4 # Copyright (C) 2011 thomasv@gitorious
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.
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.
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/>.
20 import random, socket, ast, re
21 import threading, traceback, sys, time, json, Queue
23 from version import ELECTRUM_VERSION
24 from util import print_error
27 DEFAULT_SERVERS = [ 'ecdsa.org:50001:t',
28 'electrum.novit.ro:50001:t',
29 'uncle-enzo.info:50001:t',
30 'electrum.bytesized-hosting.com:50001:t'] # list of default servers
32 proxy_modes = ['none', 'socks4', 'socks5', 'http' ]
34 def replace_keys(obj, old_key, new_key):
35 if isinstance(obj, dict):
37 obj[new_key] = obj[old_key]
39 for elem in obj.itervalues():
40 replace_keys(elem, old_key, new_key)
41 elif isinstance(obj, list):
43 replace_keys(elem, old_key, new_key)
46 replace_keys(d, 'blk_hash', 'block_hash')
47 replace_keys(d, 'pos', 'index')
48 replace_keys(d, 'nTime', 'timestamp')
49 replace_keys(d, 'is_in', 'is_input')
50 replace_keys(d, 'raw_scriptPubKey', 'raw_output_script')
52 def parse_proxy_options(s):
53 proxy = { "mode":"socks5", "host":"localhost" }
56 if proxy_modes.count(args[n]) == 1:
57 proxy["mode"] = args[n]
60 proxy["host"] = args[n]
63 proxy["port"] = args[n]
65 proxy["port"] = "8080" if proxy["mode"] == "http" else "1080"
68 class Interface(threading.Thread):
69 def __init__(self, host, port, debug_server, proxy):
70 threading.Thread.__init__(self)
76 self.servers = [] # actual list from IRC
78 self.bytes_received = 0
80 self.is_connected = True
81 self.poll_interval = 1
85 self.responses = Queue.Queue()
86 self.unanswered_requests = {}
88 self.debug_server = debug_server
90 def init_socket(self):
94 # push a fake response so that the getting thread exits its loop
95 self.responses.put(None)
97 def queue_json_response(self, c):
103 error = c.get('error')
106 print "received error:", c
109 if msg_id is not None:
110 method, params = self.unanswered_requests.pop(msg_id)
111 result = c.get('result')
114 method = c.get('method')
115 params = c.get('params')
117 if method == 'blockchain.numblocks.subscribe':
121 elif method == 'blockchain.address.subscribe':
126 self.responses.put({'method':method, 'params':params, 'result':result})
130 def subscribe(self, addresses):
132 for addr in addresses:
133 messages.append(('blockchain.address.subscribe', [addr]))
139 class PollingInterface(Interface):
140 """ non-persistent connection. synchronous calls"""
142 def __init__(self, host, port, debug_server, proxy):
143 Interface.__init__(self, host, port, debug_server, proxy)
144 self.session_id = None
145 self.debug_server = debug_server
147 def get_history(self, address):
148 self.send([('blockchain.address.get_history', [address] )])
152 #if is_new or wallet.remote_url:
153 # self.was_updated = True
154 # is_new = wallet.synchronize()
155 # wallet.update_tx_history()
162 self.is_connected = True
163 while self.is_connected:
167 time.sleep(self.poll_interval)
168 except socket.gaierror:
173 traceback.print_exc(file=sys.stdout)
176 self.is_connected = False
187 class HttpStratumInterface(PollingInterface):
192 def send(self, messages):
193 import urllib2, json, time, cookielib
195 if self.proxy["mode"] != "none":
197 socks.setdefaultproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) )
198 socks.wrapmodule(urllib2)
199 cj = cookielib.CookieJar()
200 opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))
201 urllib2.install_opener(opener)
208 if type(params) != type([]): params = [params]
209 data.append( { 'method':method, 'id':self.message_id, 'params':params } )
210 self.unanswered_requests[self.message_id] = method, params
214 data_json = json.dumps(data)
219 host = 'http://%s:%d'%( self.host, self.port )
220 headers = {'content-type': 'application/json'}
222 headers['cookie'] = 'SESSION=%s'%self.session_id
224 req = urllib2.Request(host, data_json, headers)
225 response_stream = urllib2.urlopen(req)
227 for index, cookie in enumerate(cj):
228 if cookie.name=='SESSION':
229 self.session_id = cookie.value
231 response = response_stream.read()
232 self.bytes_received += len(response)
234 response = json.loads( response )
235 if type(response) is not type([]):
236 self.queue_json_response(response)
238 for item in response:
239 self.queue_json_response(item)
242 self.poll_interval = 1
244 if self.poll_interval < 15:
245 self.poll_interval += 1
246 #print self.poll_interval, response
248 self.rtime = time.time() - t1
249 self.is_connected = True
254 class TcpStratumInterface(Interface):
255 """json-rpc over persistent TCP connection, asynchronous"""
257 def __init__(self, host, port, debug_server, proxy):
258 Interface.__init__(self, host, port, debug_server, proxy)
259 self.debug_server = debug_server
261 def init_socket(self):
263 if self.proxy["mode"] == "none":
264 self.s = socket.socket( socket.AF_INET, socket.SOCK_STREAM )
267 self.s = socks.socksocket()
268 print "Using Proxy", self.proxy
269 self.s.setproxy(proxy_modes.index(self.proxy["mode"]), self.proxy["host"], int(self.proxy["port"]) )
270 self.s.settimeout(60)
271 self.s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
273 self.s.connect(( self.host.encode('ascii'), int(self.port)))
274 self.is_connected = True
275 self.send([('server.version', [ELECTRUM_VERSION])])
276 print "Connected to %s:%d"%(self.host,self.port)
278 self.is_connected = False
279 print_error("Not connected")
284 while self.is_connected:
285 try: msg = self.s.recv(1024)
286 except socket.timeout:
287 # ping the server with server.version, as a real ping does not exist yet
288 self.send([('server.version', [ELECTRUM_VERSION])])
291 self.bytes_received += len(msg)
293 self.is_connected = False
294 print "Disconnected."
302 self.queue_json_response(c)
305 traceback.print_exc(file=sys.stdout)
307 self.is_connected = False
311 def send(self, messages):
315 request = json.dumps( { 'id':self.message_id, 'method':method, 'params':params } )
316 self.unanswered_requests[self.message_id] = method, params
318 if self.debug_server:
322 out += request + '\n'
325 sent = self.s.send( out )
328 def get_history(self, addr):
329 self.send([('blockchain.address.get_history', [addr])])
335 class WalletSynchronizer(threading.Thread):
337 def __init__(self, wallet, loop=False, servers_loaded_callback=None, proxy=None):
338 threading.Thread.__init__(self)
343 self.init_interface()
344 self.servers_loaded_callback = servers_loaded_callback
346 def init_interface(self):
348 host, port, protocol = self.wallet.server.split(':')
351 self.wallet.pick_random_server()
352 host, port, protocol = self.wallet.server.split(':')
355 #print protocol, host, port
357 InterfaceClass = TcpStratumInterface
358 elif protocol == 'h':
359 InterfaceClass = HttpStratumInterface
361 print_error("Error: Unknown protocol")
362 InterfaceClass = TcpStratumInterface
364 self.interface = InterfaceClass(host, port, self.wallet.debug_server, self.proxy)
365 self.wallet.interface = self.interface
367 def handle_response(self, r):
375 if method == 'server.banner':
376 self.wallet.banner = result
377 self.wallet.was_updated = True
379 elif method == 'server.peers.subscribe':
388 if re.match("[th]\d+", v):
389 ports.append((v[0], v[1:]))
390 if re.match("v(.?)+", v):
392 if ports and version:
393 servers.append((host, ports))
394 self.interface.servers = servers
395 # servers_loaded_callback is None for commands, but should
396 # NEVER be None when using the GUI.
397 if self.servers_loaded_callback is not None:
398 self.servers_loaded_callback()
400 elif method == 'blockchain.address.subscribe':
402 self.wallet.receive_status_callback(addr, result)
404 elif method == 'blockchain.address.get_history':
406 self.wallet.receive_history_callback(addr, result)
407 self.wallet.was_updated = True
409 elif method == 'blockchain.transaction.broadcast':
410 self.wallet.tx_result = result
411 self.wallet.tx_event.set()
413 elif method == 'blockchain.numblocks.subscribe':
414 self.wallet.blocks = result
415 self.wallet.was_updated = True
417 elif method == 'server.version':
421 print_error("Error: Unknown message:" + method + ", " + params + ", " + result)
424 def start_interface(self):
425 self.interface.init_socket()
426 self.interface.start()
427 if self.interface.is_connected:
428 self.wallet.start_session(self.interface)
434 self.start_interface()
436 while self.interface.is_connected:
437 new_addresses = self.wallet.synchronize()
439 self.interface.subscribe(new_addresses)
441 if self.wallet.is_up_to_date():
442 if not self.wallet.up_to_date:
443 self.wallet.up_to_date = True
444 self.wallet.was_updated = True
445 self.wallet.up_to_date_event.set()
447 if self.wallet.up_to_date:
448 self.wallet.up_to_date = False
449 self.wallet.was_updated = True
451 if self.wallet.was_updated:
452 self.wallet.trigger_callbacks()
453 self.wallet.was_updated = False
455 response = self.interface.responses.get()
456 self.handle_response(response)
458 self.wallet.trigger_callbacks()
461 # Server has been changed. Copy callback for new interface.
462 self.proxy = self.interface.proxy
463 self.init_interface()
464 self.start_interface()