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