sfa/server/threadedserver.py

   1 ##
   2 # This module implements a general-purpose server layer for sfa.
   3 # The same basic server should be usable on the registry, component, or
   4 # other interfaces.
   5 #
   6 # TODO: investigate ways to combine this with existing PLC server?
   7 ##
   8
   9 import sys
  10 import socket
  11 import traceback
  12 import threading
  13 from queue import Queue
  14 import socketserver
  15 import http.server
  16 import xmlrpc.server
  17 from OpenSSL import SSL
  18
  19 from sfa.util.sfalogging import logger
  20 from sfa.util.config import Config
  21 from sfa.util.cache import Cache
  22 from sfa.trust.certificate import Certificate
  23 from sfa.trust.trustedroots import TrustedRoots
  24 from sfa.util.py23 import xmlrpc_client
  25
  26 # don't hard code an api class anymore here
  27 from sfa.generic import Generic
  28
  29 ##
  30 # Verification callback for pyOpenSSL. We do our own checking of keys because
  31 # we have our own authentication spec. Thus we disable several of the normal
  32 # prohibitions that OpenSSL places on certificates
  33
  34
  35 def verify_callback(conn, x509, err, depth, preverify):
  36     # if the cert has been preverified, then it is ok
  37     if preverify:
  38         # print "  preverified"
  39         return 1
  40
  41     # the certificate verification done by openssl checks a number of things
  42     # that we aren't interested in, so we look out for those error messages
  43     # and ignore them
  44
  45     # XXX SMBAKER: I don't know what this error is, but it's being returned
  46     # xxx thierry: this most likely means the cert has a validity range in the future
  47     # by newer pl nodes.
  48     if err == 9:
  49         # print "  X509_V_ERR_CERT_NOT_YET_VALID"
  50         return 1
  51
  52     # allow self-signed certificates
  53     if err == 18:
  54         # print "  X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT"
  55         return 1
  56
  57     # allow certs that don't have an issuer
  58     if err == 20:
  59         # print "  X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY"
  60         return 1
  61
  62     # allow chained certs with self-signed roots
  63     if err == 19:
  64         return 1
  65
  66     # allow certs that are untrusted
  67     if err == 21:
  68         # print "  X509_V_ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE"
  69         return 1
  70
  71     # allow certs that are untrusted
  72     if err == 27:
  73         # print "  X509_V_ERR_CERT_UNTRUSTED"
  74         return 1
  75
  76     # ignore X509_V_ERR_CERT_SIGNATURE_FAILURE
  77     if err == 7:
  78         return 1
  79
  80     logger.debug("  error %s in verify_callback" % err)
  81
  82     return 0
  83
  84 ##
  85 # taken from the web (XXX find reference). Implements HTTPS xmlrpc request
  86 # handler
  87
  88
  89 class SecureXMLRpcRequestHandler(xmlrpc.server.SimpleXMLRPCRequestHandler):
  90     """Secure XML-RPC request handler class.
  91
  92     It it very similar to SimpleXMLRPCRequestHandler but it uses HTTPS for transporting XML data.
  93     """
  94
  95     def setup(self):
  96         self.connection = self.request
  97         self.rfile = socket._fileobject(self.request, "rb", self.rbufsize)
  98         self.wfile = socket._fileobject(self.request, "wb", self.wbufsize)
  99
 100     def do_POST(self):
 101         """Handles the HTTPS POST request.
 102
 103         It was copied out from SimpleXMLRPCServer.py and modified to shutdown
 104         the socket cleanly.
 105         """
 106         try:
 107             peer_cert = Certificate()
 108             peer_cert.load_from_pyopenssl_x509(
 109                 self.connection.get_peer_certificate())
 110             generic = Generic.the_flavour()
 111             self.api = generic.make_api(peer_cert=peer_cert,
 112                                         interface=self.server.interface,
 113                                         key_file=self.server.key_file,
 114                                         cert_file=self.server.cert_file,
 115                                         cache=self.cache)
 116             # logger.info("SecureXMLRpcRequestHandler.do_POST:")
 117             # logger.info("interface=%s"%self.server.interface)
 118             # logger.info("key_file=%s"%self.server.key_file)
 119             # logger.info("api=%s"%self.api)
 120             # logger.info("server=%s"%self.server)
 121             # logger.info("handler=%s"%self)
 122             # get arguments
 123             request = self.rfile.read(int(self.headers["content-length"]))
 124             remote_addr = (
 125                 remote_ip, remote_port) = self.connection.getpeername()
 126             self.api.remote_addr = remote_addr
 127             response = self.api.handle(
 128                 remote_addr, request, self.server.method_map)
 129         except Exception as fault:
 130             # This should only happen if the module is buggy
 131             # internal error, report as HTTP server error
 132             logger.log_exc("server.do_POST")
 133             response = self.api.prepare_response(fault)
 134             # self.send_response(500)
 135             # self.end_headers()
 136
 137         # avoid session/connection leaks : do this no matter what
 138         finally:
 139             self.send_response(200)
 140             self.send_header("Content-type", "text/xml")
 141             self.send_header("Content-length", str(len(response)))
 142             self.end_headers()
 143             self.wfile.write(response)
 144             self.wfile.flush()
 145             # close db connection
 146             self.api.close_dbsession()
 147             # shut down the connection
 148             self.connection.shutdown()  # Modified here!
 149
 150 ##
 151 # Taken from the web (XXX find reference). Implements an HTTPS xmlrpc server
 152
 153
 154 class SecureXMLRPCServer(http.server.HTTPServer, xmlrpc.server.SimpleXMLRPCDispatcher):
 155
 156     def __init__(self, server_address, HandlerClass, key_file, cert_file, logRequests=True):
 157         """
 158         Secure XML-RPC server.
 159
 160         It it very similar to SimpleXMLRPCServer but it uses HTTPS for transporting XML data.
 161         """
 162         logger.debug("SecureXMLRPCServer.__init__, server_address=%s, "
 163                      "cert_file=%s, key_file=%s" % (server_address, cert_file, key_file))
 164         self.logRequests = logRequests
 165         self.interface = None
 166         self.key_file = key_file
 167         self.cert_file = cert_file
 168         self.method_map = {}
 169         # add cache to the request handler
 170         HandlerClass.cache = Cache()
 171         # for compatibility with python 2.4 (centos53)
 172         if sys.version_info < (2, 5):
 173             xmlrpc.server.SimpleXMLRPCDispatcher.__init__(self)
 174         else:
 175             xmlrpc.server.SimpleXMLRPCDispatcher.__init__(
 176                 self, True, None)
 177         socketserver.BaseServer.__init__(self, server_address, HandlerClass)
 178         ctx = SSL.Context(SSL.SSLv23_METHOD)
 179         ctx.use_privatekey_file(key_file)
 180         ctx.use_certificate_file(cert_file)
 181         # If you wanted to verify certs against known CAs.. this is how you would do it
 182         # ctx.load_verify_locations('/etc/sfa/trusted_roots/plc.gpo.gid')
 183         config = Config()
 184         trusted_cert_files = TrustedRoots(
 185             config.get_trustedroots_dir()).get_file_list()
 186         for cert_file in trusted_cert_files:
 187             ctx.load_verify_locations(cert_file)
 188         ctx.set_verify(SSL.VERIFY_PEER |
 189                        SSL.VERIFY_FAIL_IF_NO_PEER_CERT, verify_callback)
 190         ctx.set_verify_depth(5)
 191         ctx.set_app_data(self)
 192         self.socket = SSL.Connection(ctx, socket.socket(self.address_family,
 193                                                         self.socket_type))
 194         self.server_bind()
 195         self.server_activate()
 196
 197     # _dispatch
 198     #
 199     # Convert an exception on the server to a full stack trace and send it to
 200     # the client.
 201
 202     def _dispatch(self, method, params):
 203         logger.debug("SecureXMLRPCServer._dispatch, method=%s" % method)
 204         try:
 205             return xmlrpc.server.SimpleXMLRPCDispatcher._dispatch(self, method, params)
 206         except:
 207             # can't use format_exc() as it is not available in jython yet
 208             # (even in trunk).
 209             type, value, tb = sys.exc_info()
 210             raise xmlrpc_client.Fault(1, ''.join(
 211                 traceback.format_exception(type, value, tb)))
 212
 213     # override this one from the python 2.7 code
 214     # originally defined in class TCPServer
 215     def shutdown_request(self, request):
 216         """Called to shutdown and close an individual request."""
 217         # ----------
 218         # the std python 2.7 code just attempts a request.shutdown(socket.SHUT_WR)
 219         # this works fine with regular sockets
 220         # However we are dealing with an instance of OpenSSL.SSL.Connection instead
 221         # This one only supports shutdown(), and in addition this does not
 222         # always perform as expected
 223         # ---------- std python 2.7 code
 224         try:
 225             # explicitly shutdown.  socket.close() merely releases
 226             # the socket and waits for GC to perform the actual close.
 227             request.shutdown(socket.SHUT_WR)
 228         except socket.error:
 229             pass  # some platforms may raise ENOTCONN here
 230         # ----------
 231         except TypeError:
 232             # we are dealing with an OpenSSL.Connection object,
 233             # try to shut it down but never mind if that fails
 234             try:
 235                 request.shutdown()
 236             except:
 237                 pass
 238         # ----------
 239         self.close_request(request)
 240
 241 # From Active State code: http://code.activestate.com/recipes/574454/
 242 # This is intended as a drop-in replacement for the ThreadingMixIn class in
 243 # module SocketServer of the standard lib. Instead of spawning a new thread
 244 # for each request, requests are processed by of pool of reusable threads.
 245
 246
 247 class ThreadPoolMixIn(socketserver.ThreadingMixIn):
 248     """
 249     use a thread pool instead of a new thread on every request
 250     """
 251     # XX TODO: Make this configurable
 252     # config = Config()
 253     # numThreads = config.SFA_SERVER_NUM_THREADS
 254     numThreads = 25
 255     allow_reuse_address = True  # seems to fix socket.error on server restart
 256
 257     def serve_forever(self):
 258         """
 259         Handle one request at a time until doomsday.
 260         """
 261         # set up the threadpool
 262         self.requests = Queue()
 263
 264         for x in range(self.numThreads):
 265             t = threading.Thread(target=self.process_request_thread)
 266             t.setDaemon(1)
 267             t.start()
 268
 269         # server main loop
 270         while True:
 271             self.handle_request()
 272
 273         self.server_close()
 274
 275     def process_request_thread(self):
 276         """
 277         obtain request from queue instead of directly from server socket
 278         """
 279         while True:
 280             socketserver.ThreadingMixIn.process_request_thread(
 281                 self, *self.requests.get())
 282
 283     def handle_request(self):
 284         """
 285         simply collect requests and put them on the queue for the workers.
 286         """
 287         try:
 288             request, client_address = self.get_request()
 289         except socket.error:
 290             return
 291         if self.verify_request(request, client_address):
 292             self.requests.put((request, client_address))
 293
 294
 295 class ThreadedServer(ThreadPoolMixIn, SecureXMLRPCServer):
 296     pass