2 # Copyright(C) 2012 thomasv@gitorious
4 # This program is free software: you can redistribute it and/or modify
5 # it under the terms of the GNU Affero General Public License as
6 # published by the Free Software Foundation, either version 3 of the
7 # License, or (at your option) any later version.
9 # This program is distributed in the hope that it will be useful, but
10 # WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12 # Affero General Public License for more details.
14 # You should have received a copy of the GNU Affero General Public
15 # License along with this program. If not, see
16 # <http://www.gnu.org/licenses/agpl.html>.
20 * server should check and return bitcoind status..
21 * improve txpoint sorting
22 * command to check cache
24 mempool transactions do not need to be added to the database; it slows it down
28 import time, json, socket, operator, thread, ast, sys,re
32 from json import dumps, loads
36 config = ConfigParser.ConfigParser()
37 # set some defaults, which will be overwritten by the config file
38 config.add_section('server')
39 config.set('server','banner', 'Welcome to Electrum!')
40 config.set('server', 'host', 'localhost')
41 config.set('server', 'port', '50000')
42 config.set('server', 'password', '')
43 config.set('server', 'irc', 'yes')
44 config.set('server', 'ircname', 'Electrum server')
45 config.add_section('database')
46 config.set('database', 'type', 'psycopg2')
47 config.set('database', 'database', 'abe')
50 f = open('/etc/electrum.conf','r')
54 print "Could not read electrum.conf. I will use the default values."
57 f = open('/etc/electrum.banner','r')
58 config.set('server','banner', f.read())
64 password = config.get('server','password')
70 sessions_sub_numblocks = {} # sessions that have subscribed to the service
72 m_sessions = [{}] # served by http
76 wallets = {} # for ultra-light clients such as bccapi
78 from Queue import Queue
80 output_queue = Queue()
81 address_queue = Queue()
91 return ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(N))
95 def cmd_stop(_,__,pw):
101 return 'wrong password'
103 def cmd_load(_,__,pw):
105 return repr( len(sessions) )
107 return 'wrong password'
113 def modified_addresses(session):
116 addresses = session['addresses']
117 session['last_time'] = time.time()
120 for addr in addresses:
121 status = get_address_status( addr )
122 msg_id, last_status = addresses.get( addr )
123 if last_status != status:
124 addresses[addr] = msg_id, status
127 t2 = time.time() - t1
128 #if t2 > 10: print "high load:", session_id, "%d/%d"%(k,len(addresses)), t2
129 return ret, addresses
132 def poll_session(session_id):
134 session = sessions.get(session_id)
136 print time.asctime(), "session not found", session_id
139 ret, addresses = modified_addresses(session)
140 if ret: sessions[session_id]['addresses'] = addresses
141 return repr( (block_number,ret))
144 def poll_session_json(session_id, message_id):
145 session = m_sessions[0].get(session_id)
147 raise BaseException("session not found %s"%session_id)
150 ret, addresses = modified_addresses(session)
152 m_sessions[0][session_id]['addresses'] = addresses
154 msg_id, status = addresses[addr]
155 out.append( { 'id':msg_id, 'result':status } )
157 msg_id, last_nb = session.get('numblocks')
159 if last_nb != block_number:
160 m_sessions[0][session_id]['numblocks'] = msg_id, block_number
161 out.append( {'id':msg_id, 'result':block_number} )
166 def do_update_address(addr):
167 # an address was involved in a transaction; we check if it was subscribed to in a session
168 # the address can be subscribed in several sessions; the cache should ensure that we don't do redundant requests
170 for session_id in sessions.keys():
171 session = sessions[session_id]
172 if session.get('type') != 'persistent': continue
173 addresses = session['addresses'].keys()
175 if addr in addresses:
176 status = get_address_status( addr )
177 message_id, last_status = session['addresses'][addr]
178 if last_status != status:
179 #print "sending new status for %s:"%addr, status
180 send_status(session_id,message_id,addr,status)
181 sessions[session_id]['addresses'][addr] = (message_id,status)
183 def get_address_status(addr):
184 # get address status, i.e. the last block for that address.
185 tx_points = store.get_history(addr)
189 lastpoint = tx_points[-1]
190 status = lastpoint['blk_hash']
191 # this is a temporary hack; move it up once old clients have disappeared
192 if status == 'mempool': # and session['version'] != "old":
193 status = status + ':%d'% len(tx_points)
197 def send_numblocks(session_id):
198 message_id = sessions_sub_numblocks[session_id]
199 out = json.dumps( {'id':message_id, 'result':block_number} )
200 output_queue.put((session_id, out))
202 def send_status(session_id, message_id, address, status):
203 out = json.dumps( { 'id':message_id, 'result':status } )
204 output_queue.put((session_id, out))
206 def address_get_history_json(_,message_id,address):
207 return store.get_history(address)
209 def subscribe_to_numblocks(session_id, message_id):
210 sessions_sub_numblocks[session_id] = message_id
211 send_numblocks(session_id)
213 def subscribe_to_numblocks_json(session_id, message_id):
215 m_sessions[0][session_id]['numblocks'] = message_id,block_number
218 def subscribe_to_address(session_id, message_id, address):
219 status = get_address_status(address)
220 sessions[session_id]['addresses'][address] = (message_id, status)
221 sessions[session_id]['last_time'] = time.time()
222 send_status(session_id, message_id, address, status)
224 def add_address_to_session_json(session_id, message_id, address):
226 sessions = m_sessions[0]
227 status = get_address_status(address)
228 sessions[session_id]['addresses'][address] = (message_id, status)
229 sessions[session_id]['last_time'] = time.time()
230 m_sessions[0] = sessions
233 def add_address_to_session(session_id, address):
234 status = get_address_status(address)
235 sessions[session_id]['addresses'][addr] = ("", status)
236 sessions[session_id]['last_time'] = time.time()
239 def new_session(version, addresses):
240 session_id = random_string(10)
241 sessions[session_id] = { 'addresses':{}, 'version':version }
243 sessions[session_id]['addresses'][a] = ('','')
244 out = repr( (session_id, config.get('server','banner').replace('\\n','\n') ) )
245 sessions[session_id]['last_time'] = time.time()
249 def client_version_json(session_id, _, version):
251 sessions = m_sessions[0]
252 sessions[session_id]['version'] = version
253 m_sessions[0] = sessions
255 def create_session_json(_, __):
256 sessions = m_sessions[0]
257 session_id = random_string(10)
258 print "creating session", session_id
259 sessions[session_id] = { 'addresses':{}, 'numblocks':('','') }
260 sessions[session_id]['last_time'] = time.time()
261 m_sessions[0] = sessions
266 def get_banner(_,__):
267 return config.get('server','banner').replace('\\n','\n')
269 def update_session(session_id,addresses):
270 """deprecated in 0.42"""
271 sessions[session_id]['addresses'] = {}
273 sessions[session_id]['addresses'][a] = ''
274 sessions[session_id]['last_time'] = time.time()
277 def native_server_thread():
278 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
279 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
280 s.bind((config.get('server','host'), config.getint('server','port')))
283 conn, addr = s.accept()
285 thread.start_new_thread(native_client_thread, (addr, conn,))
287 # can't start new thread if there is no memory..
288 traceback.print_exc(file=sys.stdout)
291 def native_client_thread(ipaddr,conn):
292 #print "client thread", ipaddr
302 msg = msg.split('#', 1)[0]
305 cmd, data = ast.literal_eval(msg)
307 print "syntax error", repr(msg), ipaddr
311 out = do_command(cmd, data, ipaddr)
313 #print ipaddr, cmd, len(out)
317 print "error, could not send"
324 return time.strftime("[%d/%m/%Y-%H:%M:%S]")
326 # used by the native handler
327 def do_command(cmd, data, ipaddr):
330 out = "%d"%block_number
332 elif cmd in ['session','new_session']:
335 addresses = ast.literal_eval(data)
338 version, addresses = ast.literal_eval(data)
339 if version[0]=="0": version = "v" + version
343 print timestr(), "new session", ipaddr, addresses[0] if addresses else addresses, len(addresses), version
344 out = new_session(version, addresses)
346 elif cmd=='address.subscribe':
348 session_id, addr = ast.literal_eval(data)
350 traceback.print_exc(file=sys.stdout)
352 out = add_address_to_session(session_id,addr)
354 elif cmd=='update_session':
356 session_id, addresses = ast.literal_eval(data)
358 traceback.print_exc(file=sys.stdout)
360 print timestr(), "update session", ipaddr, addresses[0] if addresses else addresses, len(addresses)
361 out = update_session(session_id,addresses)
364 out = poll_session(data)
369 out = repr( store.get_history( address ) )
372 out = cmd_load(None,None,data)
375 out = store.send_tx(data)
376 print timestr(), "sent tx:", ipaddr, out
382 out = repr(peer_list.values())
391 ####################################################################
393 def tcp_server_thread():
394 thread.start_new_thread(process_input_queue, ())
395 thread.start_new_thread(process_output_queue, ())
397 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
398 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
399 s.bind((config.get('server','host'), 50001))
402 conn, addr = s.accept()
404 thread.start_new_thread(tcp_client_thread, (addr, conn,))
406 # can't start new thread if there is no memory..
407 traceback.print_exc(file=sys.stdout)
410 def close_session(session_id):
411 #print "lost connection", session_id
412 sessions.pop(session_id)
413 if session_id in sessions_sub_numblocks:
414 sessions_sub_numblocks.pop(session_id)
417 # one thread per client. put requests in a queue.
418 def tcp_client_thread(ipaddr,conn):
419 """ use a persistent connection. put commands in a queue."""
421 print timestr(), "TCP session", ipaddr
424 session_id = random_string(10)
425 sessions[session_id] = { 'conn':conn, 'addresses':{}, 'version':'unknown', 'type':'persistent' }
436 close_session(session_id)
449 close_session(session_id)
454 print "json error", repr(c)
457 message_id = c.get('id')
458 method = c.get('method')
459 params = c.get('params')
461 print "syntax error", repr(c), ipaddr
465 input_queue.put((session_id, message_id, method, params))
469 # read commands from the input queue. perform requests, etc. this should be called from the main thread.
470 def process_input_queue():
472 session_id, message_id, method, data = input_queue.get()
473 if session_id not in sessions.keys():
476 if method == 'address.subscribe':
478 subscribe_to_address(session_id,message_id,address)
479 elif method == 'numblocks.subscribe':
480 subscribe_to_numblocks(session_id,message_id)
481 elif method == 'client.version':
482 sessions[session_id]['version'] = data[0]
483 elif method == 'server.banner':
484 out = { 'result':config.get('server','banner').replace('\\n','\n') }
485 elif method == 'server.peers':
486 out = { 'result':peer_list.values() }
487 elif method == 'address.get_history':
489 out = { 'result':store.get_history( address ) }
490 elif method == 'transaction.broadcast':
491 postdata = dumps({"method": 'importtransaction', 'params': [data], 'id':'jsonrpc'})
492 txo = urllib.urlopen(bitcoind_url, postdata).read()
493 print "sent tx:", txo
494 out = json.loads(txo)
496 print "unknown command", method
498 out['id'] = message_id
499 out = json.dumps( out )
500 output_queue.put((session_id, out))
502 # this is a separate thread
503 def process_output_queue():
505 session_id, out = output_queue.get()
506 session = sessions.get(session_id)
509 conn = session.get('conn')
512 close_session(session_id)
517 ####################################################################
522 def clean_session_thread():
526 for k,s in sessions.items():
527 if s.get('type') == 'persistent': continue
531 print "lost session", k
536 NICK = 'E_'+random_string(10)
540 s.connect(('irc.freenode.net', 6667))
541 s.send('USER electrum 0 * :'+config.get('server','host')+' '+config.get('server','ircname')+'\n')
542 s.send('NICK '+NICK+'\n')
543 s.send('JOIN #electrum\n')
544 sf = s.makefile('r', 0)
548 line = line.rstrip('\r\n')
551 s.send('PONG '+line[1]+'\n')
552 elif '353' in line: # answer to /names
553 k = line.index('353')
554 for item in line[k+1:]:
555 if item[0:2] == 'E_':
556 s.send('WHO %s\n'%item)
557 elif '352' in line: # answer to /who
558 # warning: this is a horrible hack which apparently works
559 k = line.index('352')
561 ip = socket.gethostbyname(ip)
564 peer_list[name] = (ip,host)
565 if time.time() - t > 5*60:
566 s.send('NAMES #electrum\n')
570 traceback.print_exc(file=sys.stdout)
576 def get_peers_json(_,__):
577 return peer_list.values()
579 def http_server_thread():
580 # see http://code.google.com/p/jsonrpclib/
581 from SocketServer import ThreadingMixIn
582 from StratumJSONRPCServer import StratumJSONRPCServer
583 class StratumThreadedJSONRPCServer(ThreadingMixIn, StratumJSONRPCServer): pass
584 server = StratumThreadedJSONRPCServer(( config.get('server','host'), 8081))
585 server.register_function(get_peers_json, 'server.peers')
586 server.register_function(cmd_stop, 'stop')
587 server.register_function(cmd_load, 'load')
588 server.register_function(get_banner, 'server.banner')
589 server.register_function(lambda a,b,c: store.send_tx(c), 'transaction.broadcast')
590 server.register_function(address_get_history_json, 'address.get_history')
591 server.register_function(add_address_to_session_json, 'address.subscribe')
592 server.register_function(subscribe_to_numblocks_json, 'numblocks.subscribe')
593 server.register_function(client_version_json, 'client.version')
594 server.register_function(create_session_json, 'session.create') # internal message (not part of protocol)
595 server.register_function(poll_session_json, 'session.poll') # internal message (not part of protocol)
596 server.serve_forever()
602 if __name__ == '__main__':
606 server = jsonrpclib.Server('http://%s:8081'%config.get('server','host'))
609 out = server.load(password)
611 out = server.server.peers()
613 out = server.stop(password)
614 elif cmd == 'clear_cache':
615 out = server.clear_cache(password)
616 elif cmd == 'get_cache':
617 out = server.get_cache(password,sys.argv[2])
619 out = server.address.get_history(sys.argv[2])
621 out = server.transaction.broadcast(sys.argv[2])
623 out = server.numblocks.subscribe()
625 out = "Unknown command: '%s'" % cmd
632 store = db.MyStore(config,address_queue)
634 # supported protocols
635 thread.start_new_thread(native_server_thread, ())
636 thread.start_new_thread(tcp_server_thread, ())
637 thread.start_new_thread(http_server_thread, ())
638 thread.start_new_thread(clean_session_thread, ())
640 if (config.get('server','irc') == 'yes' ):
641 thread.start_new_thread(irc_thread, ())
643 print "starting Electrum server"
647 block_number = store.main_iteration()
649 if block_number != old_block_number:
650 old_block_number = block_number
651 for session_id in sessions_sub_numblocks.keys():
652 send_numblocks(session_id)
656 addr = address_queue.get(False)
659 do_update_address(addr)
662 print "server stopped"