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