fix get_link_requests()
[sfa.git] / 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 #can we get rid of that ?
25 from sfa.plc.plcsfaapi import PlcSfaApi
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             self.api = PlcSfaApi(peer_cert = peer_cert, 
99                               interface = self.server.interface, 
100                               key_file = self.server.key_file, 
101                               cert_file = self.server.cert_file,
102                               cache = self.cache)
103             # get arguments
104             request = self.rfile.read(int(self.headers["content-length"]))
105             remote_addr = (remote_ip, remote_port) = self.connection.getpeername()
106             self.api.remote_addr = remote_addr            
107             response = self.api.handle(remote_addr, request, self.server.method_map)
108         except Exception, fault:
109             # This should only happen if the module is buggy
110             # internal error, report as HTTP server error
111             logger.log_exc("server.do_POST")
112             response = self.api.prepare_response(fault)
113             #self.send_response(500)
114             #self.end_headers()
115        
116         # got a valid response
117         self.send_response(200)
118         self.send_header("Content-type", "text/xml")
119         self.send_header("Content-length", str(len(response)))
120         self.end_headers()
121         self.wfile.write(response)
122
123         # shut down the connection
124         self.wfile.flush()
125         self.connection.shutdown() # Modified here!
126
127 ##
128 # Taken from the web (XXX find reference). Implements an HTTPS xmlrpc server
129 class SecureXMLRPCServer(BaseHTTPServer.HTTPServer,SimpleXMLRPCServer.SimpleXMLRPCDispatcher):
130     def __init__(self, server_address, HandlerClass, key_file, cert_file, logRequests=True):
131         """Secure XML-RPC server.
132
133         It it very similar to SimpleXMLRPCServer but it uses HTTPS for transporting XML data.
134         """
135         logger.debug("SecureXMLRPCServer.__init__, server_address=%s, cert_file=%s"%(server_address,cert_file))
136         self.logRequests = logRequests
137         self.interface = None
138         self.key_file = key_file
139         self.cert_file = cert_file
140         self.method_map = {}
141         # add cache to the request handler
142         HandlerClass.cache = Cache()
143         #for compatibility with python 2.4 (centos53)
144         if sys.version_info < (2, 5):
145             SimpleXMLRPCServer.SimpleXMLRPCDispatcher.__init__(self)
146         else:
147            SimpleXMLRPCServer.SimpleXMLRPCDispatcher.__init__(self, True, None)
148         SocketServer.BaseServer.__init__(self, server_address, HandlerClass)
149         ctx = SSL.Context(SSL.SSLv23_METHOD)
150         ctx.use_privatekey_file(key_file)        
151         ctx.use_certificate_file(cert_file)
152         # If you wanted to verify certs against known CAs.. this is how you would do it
153         #ctx.load_verify_locations('/etc/sfa/trusted_roots/plc.gpo.gid')
154         config = Config()
155         trusted_cert_files = TrustedRoots(config.get_trustedroots_dir()).get_file_list()
156         for cert_file in trusted_cert_files:
157             ctx.load_verify_locations(cert_file)
158         ctx.set_verify(SSL.VERIFY_PEER | SSL.VERIFY_FAIL_IF_NO_PEER_CERT, verify_callback)
159         ctx.set_verify_depth(5)
160         ctx.set_app_data(self)
161         self.socket = SSL.Connection(ctx, socket.socket(self.address_family,
162                                                         self.socket_type))
163         self.server_bind()
164         self.server_activate()
165
166     # _dispatch
167     #
168     # Convert an exception on the server to a full stack trace and send it to
169     # the client.
170
171     def _dispatch(self, method, params):
172         logger.debug("SecureXMLRPCServer._dispatch, method=%s"%method)
173         try:
174             return SimpleXMLRPCServer.SimpleXMLRPCDispatcher._dispatch(self, method, params)
175         except:
176             # can't use format_exc() as it is not available in jython yet
177             # (even in trunk).
178             type, value, tb = sys.exc_info()
179             raise xmlrpclib.Fault(1,''.join(traceback.format_exception(type, value, tb)))
180
181     # override this one from the python 2.7 code
182     # originally defined in class TCPServer
183     def shutdown_request(self, request):
184         """Called to shutdown and close an individual request."""
185         # ---------- 
186         # the std python 2.7 code just attempts a request.shutdown(socket.SHUT_WR)
187         # this works fine with regular sockets
188         # However we are dealing with an instance of OpenSSL.SSL.Connection instead
189         # This one only supports shutdown(), and in addition this does not
190         # always perform as expected
191         # ---------- std python 2.7 code
192         try:
193             #explicitly shutdown.  socket.close() merely releases
194             #the socket and waits for GC to perform the actual close.
195             request.shutdown(socket.SHUT_WR)
196         except socket.error:
197             pass #some platforms may raise ENOTCONN here
198         # ----------
199         except TypeError:
200             # we are dealing with an OpenSSL.Connection object, 
201             # try to shut it down but never mind if that fails
202             try: request.shutdown()
203             except: pass
204         # ----------
205         self.close_request(request)
206
207 ## From Active State code: http://code.activestate.com/recipes/574454/
208 # This is intended as a drop-in replacement for the ThreadingMixIn class in 
209 # module SocketServer of the standard lib. Instead of spawning a new thread 
210 # for each request, requests are processed by of pool of reusable threads.
211 class ThreadPoolMixIn(SocketServer.ThreadingMixIn):
212     """
213     use a thread pool instead of a new thread on every request
214     """
215     # XX TODO: Make this configurable
216     # config = Config()
217     # numThreads = config.SFA_SERVER_NUM_THREADS
218     numThreads = 25
219     allow_reuse_address = True  # seems to fix socket.error on server restart
220
221     def serve_forever(self):
222         """
223         Handle one request at a time until doomsday.
224         """
225         # set up the threadpool
226         self.requests = Queue()
227
228         for x in range(self.numThreads):
229             t = threading.Thread(target = self.process_request_thread)
230             t.setDaemon(1)
231             t.start()
232
233         # server main loop
234         while True:
235             self.handle_request()
236             
237         self.server_close()
238
239     
240     def process_request_thread(self):
241         """
242         obtain request from queue instead of directly from server socket
243         """
244         while True:
245             SocketServer.ThreadingMixIn.process_request_thread(self, *self.requests.get())
246
247     
248     def handle_request(self):
249         """
250         simply collect requests and put them on the queue for the workers.
251         """
252         try:
253             request, client_address = self.get_request()
254         except socket.error:
255             return
256         if self.verify_request(request, client_address):
257             self.requests.put((request, client_address))
258
259 class ThreadedServer(ThreadPoolMixIn, SecureXMLRPCServer):
260     pass