import json
+import Queue as queue
import socket
+import select
import threading
import time
-import Queue as queue
+import traceback, sys
+
+from processor import Session, Dispatcher
+from utils import print_log
+
+
+READ_ONLY = select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR
+READ_WRITE = READ_ONLY | select.POLLOUT
+TIMEOUT = 100
-from processor import Session, Dispatcher, timestr
+import ssl
class TcpSession(Session):
- def __init__(self, connection, address):
- Session.__init__(self)
- self._connection = connection
- self.address = address[0]
- self.name = "TCP session"
+ def __init__(self, dispatcher, poller, connection, address, use_ssl, ssl_certfile, ssl_keyfile):
+ Session.__init__(self, dispatcher)
+ self.use_ssl = use_ssl
+ self.poller = poller
+ self.raw_connection = connection
+ if use_ssl:
+ import ssl
+ self._connection = ssl.wrap_socket(
+ connection,
+ server_side=True,
+ certfile=ssl_certfile,
+ keyfile=ssl_keyfile,
+ ssl_version=ssl.PROTOCOL_SSLv23,
+ do_handshake_on_connect=False)
+ else:
+ self._connection = connection
+
+ self.address = address[0] + ":%d"%address[1]
+ self.name = "TCP " if not use_ssl else "SSL "
+ self.timeout = 1000
+ self.dispatcher.add_session(self)
+ self.response_queue = queue.Queue()
+ self.message = ''
+ self.retry_msg = ''
+ self.handshake = not self.use_ssl
+
+
+ def check_do_handshake(self):
+ if self.handshake:
+ return
+ try:
+ self._connection.do_handshake()
+ except ssl.SSLError as err:
+ if err.args[0] == ssl.SSL_ERROR_WANT_READ:
+ return
+ elif err.args[0] == ssl.SSL_ERROR_WANT_WRITE:
+ self.poller.modify(self.raw_connection, READ_WRITE)
+ return
+ else:
+ raise
+
+ self.poller.modify(self.raw_connection, READ_ONLY)
+ self.handshake = True
+
+
def connection(self):
if self.stopped():
else:
return self._connection
- def stop(self):
+ def shutdown(self):
+ try:
+ self._connection.shutdown(socket.SHUT_RDWR)
+ except:
+ # print_log("problem shutting down", self.address)
+ # traceback.print_exc(file=sys.stdout)
+ pass
+
self._connection.close()
- #print "Terminating connection:", self.address
- with self.lock:
- self._stopped = True
def send_response(self, response):
- data = json.dumps(response) + "\n"
- # Possible race condition here by having session
- # close connection?
- # I assume Python connections are thread safe interfaces
try:
- connection = self.connection()
- while data:
- l = connection.send(data)
- data = data[l:]
+ msg = json.dumps(response) + '\n'
except:
- self.stop()
+ traceback.print_exc(file=sys.stdout)
+ return
+ self.response_queue.put(msg)
+ try:
+ self.poller.modify(self.raw_connection, READ_WRITE)
+ except:
+ traceback.print_exc(file=sys.stdout)
+ return
-class TcpClientRequestor(threading.Thread):
- def __init__(self, dispatcher, session):
- self.shared = dispatcher.shared
- self.dispatcher = dispatcher
- self.message = ""
- self.session = session
- threading.Thread.__init__(self)
- def run(self):
- while not self.shared.stopped():
- if not self.update():
- break
+ def parse_message(self):
- while self.parse():
- pass
+ message = self.message
+ self.time = time.time()
- def update(self):
- data = self.receive()
- if not data:
- # close_session
- self.session.stop()
+ raw_buffer = message.find('\n')
+ if raw_buffer == -1:
return False
- self.message += data
- return True
+ raw_command = message[0:raw_buffer].strip()
+ self.message = message[raw_buffer + 1:]
+ return raw_command
- def receive(self):
- try:
- return self.session.connection().recv(2048)
- except:
- return ''
- def parse(self):
- raw_buffer = self.message.find('\n')
- if raw_buffer == -1:
- return False
- raw_command = self.message[0:raw_buffer].strip()
- self.message = self.message[raw_buffer + 1:]
- if raw_command == 'quit':
- self.session.stop()
- return False
+
+class TcpServer(threading.Thread):
+
+ def __init__(self, dispatcher, host, port, use_ssl, ssl_certfile, ssl_keyfile):
+ self.shared = dispatcher.shared
+ self.dispatcher = dispatcher.request_dispatcher
+ threading.Thread.__init__(self)
+ self.daemon = True
+ self.host = host
+ self.port = port
+ self.lock = threading.Lock()
+ self.use_ssl = use_ssl
+ self.ssl_keyfile = ssl_keyfile
+ self.ssl_certfile = ssl_certfile
+
+ self.fd_to_session = {}
+ self.buffer_size = 4096
+
+
+
+
+
+ def handle_command(self, raw_command, session):
try:
command = json.loads(raw_command)
except:
- self.dispatcher.push_response({"error": "bad JSON", "request": raw_command})
+ session.send_response({"error": "bad JSON"})
return True
try:
method = command['method']
except KeyError:
# Return an error JSON in response.
- self.dispatcher.push_response({"error": "syntax error", "request": raw_command})
+ session.send_response({"error": "syntax error", "request": raw_command})
else:
- self.dispatcher.push_request(self.session,command)
+ self.dispatcher.push_request(session, command)
+ ## sleep a bit to prevent a single session from DOSing the queue
+ #time.sleep(0.01)
+
- return True
-class TcpServer(threading.Thread):
- def __init__(self, dispatcher, host, port):
- self.shared = dispatcher.shared
- self.dispatcher = dispatcher.request_dispatcher
- threading.Thread.__init__(self)
- self.daemon = True
- self.host = host
- self.port = port
- self.lock = threading.Lock()
def run(self):
- print "TCP server started."
- sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- sock.bind((self.host, self.port))
- sock.listen(1)
+
+ print_log( ("SSL" if self.use_ssl else "TCP") + " server started on port %d"%self.port)
+ server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ server.setblocking(0)
+ server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ server.bind((self.host, self.port))
+ server.listen(5)
+ server_fd = server.fileno()
+
+ poller = select.poll()
+ poller.register(server)
+
+ def stop_session(fd):
+ try:
+ # unregister before we close s
+ poller.unregister(fd)
+ except:
+ print_log("unregister error", fd)
+ traceback.print_exc(file=sys.stdout)
+
+ session = self.fd_to_session.pop(fd)
+ # this will close the socket
+ session.stop()
+
+
+ redo = []
+
while not self.shared.stopped():
- session = TcpSession(*sock.accept())
- client_req = TcpClientRequestor(self.dispatcher, session)
- client_req.start()
- self.dispatcher.add_session(session)
- self.dispatcher.collect_garbage()
+ if redo:
+ events = redo
+ redo = []
+ else:
+ events = poller.poll(TIMEOUT)
+
+ for fd, flag in events:
+
+ if fd != server_fd:
+ session = self.fd_to_session[fd]
+ s = session._connection
+ try:
+ session.check_do_handshake()
+ except:
+ stop_session(fd)
+ continue
+
+ # handle inputs
+ if flag & (select.POLLIN | select.POLLPRI):
+
+ if fd == server_fd:
+ connection, address = server.accept()
+ try:
+ session = TcpSession(self.dispatcher, poller, connection, address,
+ use_ssl=self.use_ssl, ssl_certfile=self.ssl_certfile, ssl_keyfile=self.ssl_keyfile)
+ except BaseException as e:
+ print_log("cannot start TCP session", str(e), address)
+ connection.close()
+ continue
+
+ connection = session._connection
+ connection.setblocking(0)
+ self.fd_to_session[ connection.fileno() ] = session
+ poller.register(connection, READ_ONLY)
+ try:
+ session.check_do_handshake()
+ except:
+ print_log( "handshake failure", address )
+ stop_session(connection.fileno())
+
+ continue
+
+ try:
+ data = s.recv(self.buffer_size)
+ except ssl.SSLError as x:
+ if x.args[0] == ssl.SSL_ERROR_WANT_READ:
+ pass
+ else:
+ print_log("SSL error", x)
+ continue
+ except socket.error as x:
+ # print_log("recv err", x)
+ stop_session(fd)
+ continue
+
+ if data:
+ if len(data) == self.buffer_size:
+ redo.append( (fd, flag) )
+
+ session.message += data
+ while True:
+ cmd = session.parse_message()
+ if not cmd:
+ break
+ if cmd == 'quit':
+ data = False
+ break
+ self.handle_command(cmd, session)
+
+ if not data:
+ stop_session(fd)
+ continue
+
+ elif flag & select.POLLHUP:
+ print_log('client hung up', address)
+ stop_session(fd)
+
+
+ elif flag & select.POLLOUT:
+ # Socket is ready to send data, if there is any to send.
+ if session.retry_msg:
+ next_msg = session.retry_msg
+ else:
+ try:
+ next_msg = session.response_queue.get_nowait()
+ except queue.Empty:
+ # No messages waiting so stop checking for writability.
+ poller.modify(s, READ_ONLY)
+ continue
+
+ try:
+ sent = s.send(next_msg)
+ except socket.error as x:
+ # print_log("recv err", x)
+ stop_session(fd)
+ continue
+
+ session.retry_msg = next_msg[sent:]
+ elif flag & select.POLLERR:
+ print_log('handling exceptional condition for', session.address)
+ stop_session(fd)
+ elif flag & select.POLLNVAL:
+ print_log('invalid request', session.address)
+ stop_session(fd)
git://git.novaco.in / electrum-server.git / blobdiff