bin/netsvc.py

   1 #!/usr/bin/python
   2 # -*- encoding: utf-8 -*-
   3 ##############################################################################
   4 #
   5 #    OpenERP, Open Source Management Solution
   6 #    Copyright (C) 2004-2009 Tiny SPRL (<http://tiny.be>). All Rights Reserved
   7 #    The refactoring about the OpenSSL support come from Tryton
   8 #    Copyright (C) 2007-2009 Cédric Krier.
   9 #    Copyright (C) 2007-2009 Bertrand Chenal.
  10 #    Copyright (C) 2008 B2CK SPRL.
  11 #
  12 #    This program is free software: you can redistribute it and/or modify
  13 #    it under the terms of the GNU General Public License as published by
  14 #    the Free Software Foundation, either version 3 of the License, or
  15 #    (at your option) any later version.
  16 #
  17 #    This program is distributed in the hope that it will be useful,
  18 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  19 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  20 #    GNU General Public License for more details.
  21 #
  22 #    You should have received a copy of the GNU General Public License
  23 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  24 #
  25 ##############################################################################
  26
  27
  28 import SimpleXMLRPCServer
  29 import SocketServer
  30 import logging
  31 import logging.handlers
  32 import os
  33 import signal
  34 import socket
  35 import sys
  36 import threading
  37 import time
  38 import xmlrpclib
  39 import release
  40
  41 SERVICES = {}
  42 GROUPS = {}
  43
  44 class Service(object):
  45     def __init__(self, name, audience=''):
  46         SERVICES[name] = self
  47         self.__name = name
  48         self._methods = {}
  49
  50     def joinGroup(self, name):
  51         GROUPS.setdefault(name, {})[self.__name] = self
  52
  53     def exportMethod(self, method):
  54         if callable(method):
  55             self._methods[method.__name__] = method
  56
  57     def abortResponse(self, error, description, origin, details):
  58         if not tools.config['debug_mode']:
  59             raise Exception("%s -- %s\n\n%s"%(origin, description, details))
  60         else:
  61             raise
  62
  63 class LocalService(Service):
  64     def __init__(self, name):
  65         self.__name = name
  66         try:
  67             self._service = SERVICES[name]
  68             for method_name, method_definition in self._service._methods.items():
  69                 setattr(self, method_name, method_definition)
  70         except KeyError, keyError:
  71             Logger().notifyChannel('module', LOG_ERROR, 'This service does not exists: %s' % (str(keyError),) )
  72             raise
  73     def __call__(self, method, *params):
  74         return getattr(self, method)(*params)
  75
  76 def service_exist(name):
  77     return SERVICES.get(name, False)
  78
  79 LOG_NOTSET = 'notset'
  80 LOG_DEBUG_RPC = 'debug_rpc'
  81 LOG_DEBUG = 'debug'
  82 LOG_INFO = 'info'
  83 LOG_WARNING = 'warn'
  84 LOG_ERROR = 'error'
  85 LOG_CRITICAL = 'critical'
  86
  87 # add new log level below DEBUG
  88 logging.DEBUG_RPC = logging.DEBUG - 1
  89
  90 def init_logger():
  91     import os
  92
  93     logger = logging.getLogger()
  94     # create a format for log messages and dates
  95     formatter = logging.Formatter('[%(asctime)s] %(levelname)s:%(name)s:%(message)s', '%Y-%m-%d %H:%M:%S')
  96
  97     if tools.config['syslog']:
  98         # SysLog Handler
  99         if os.name == 'nt':
 100             handler = logging.handlers.NTEventLogHandler("%s %s" %
 101                                                          (release.description,
 102                                                           release.version))
 103         else:
 104             handler = logging.handlers.SysLogHandler('/dev/log')
 105         formatter = logging.Formatter("%s %s" % (release.description, release.version) + ':%(levelname)s:%(name)s:%(message)s')
 106
 107     elif tools.config['logfile']:
 108         # LogFile Handler
 109         logf = tools.config['logfile']
 110         try:
 111             dirname = os.path.dirname(logf)
 112             if dirname and not os.path.isdir(dirname):
 113                 os.makedirs(dirname)
 114             handler = logging.handlers.TimedRotatingFileHandler(logf,'D',1,30)
 115         except Exception, ex:
 116             sys.stderr.write("ERROR: couldn't create the logfile directory\n")
 117             handler = logging.StreamHandler(sys.stdout)
 118     else:
 119         # Normal Handler on standard output
 120         handler = logging.StreamHandler(sys.stdout)
 121
 122
 123     # tell the handler to use this format
 124     handler.setFormatter(formatter)
 125
 126     # add the handler to the root logger
 127     logger.addHandler(handler)
 128     logger.setLevel(int(tools.config['log_level'] or '0'))
 129
 130     if (not isinstance(handler, logging.FileHandler)) and os.name != 'nt':
 131         # change color of level names
 132         # uses of ANSI color codes
 133         # see http://pueblo.sourceforge.net/doc/manual/ansi_color_codes.html
 134         # maybe use http://code.activestate.com/recipes/574451/
 135         colors = ['black', 'red', 'green', 'yellow', 'blue', 'magenta', 'cyan', 'white', None, 'default']
 136         foreground = lambda f: 30 + colors.index(f)
 137         background = lambda f: 40 + colors.index(f)
 138
 139         mapping = {
 140             'DEBUG_RPC': ('blue', 'white'),
 141             'DEBUG': ('blue', 'default'),
 142             'INFO': ('green', 'default'),
 143             'WARNING': ('yellow', 'default'),
 144             'ERROR': ('red', 'default'),
 145             'CRITICAL': ('white', 'red'),
 146         }
 147
 148         for level, (fg, bg) in mapping.items():
 149             msg = "\x1b[%dm\x1b[%dm%s\x1b[0m" % (foreground(fg), background(bg), level)
 150             logging.addLevelName(getattr(logging, level), msg)
 151
 152
 153 class Logger(object):
 154     def notifyChannel(self, name, level, msg):
 155         log = logging.getLogger(name)
 156
 157         if level == LOG_DEBUG_RPC and not hasattr(log, level):
 158             fct = lambda msg, *args, **kwargs: log.log(logging.DEBUG_RPC, msg, *args, **kwargs)
 159             setattr(log, LOG_DEBUG_RPC, fct)
 160
 161         level_method = getattr(log, level)
 162
 163         if isinstance(msg, Exception):
 164             msg = tools.exception_to_unicode(msg)
 165
 166         result = tools.ustr(msg).strip().split('\n')
 167         if len(result)>1:
 168             for idx, s in enumerate(result):
 169                 level_method('[%02d]: %s' % (idx+1, s,))
 170         elif result:
 171             level_method(result[0])
 172
 173     def shutdown(self):
 174         logging.shutdown()
 175
 176 import tools
 177 init_logger()
 178
 179 class Agent(object):
 180     _timers = []
 181     _logger = Logger()
 182
 183     def setAlarm(self, fn, dt, args=None, kwargs=None):
 184         if not args:
 185             args = []
 186         if not kwargs:
 187             kwargs = {}
 188         wait = dt - time.time()
 189         if wait > 0:
 190             self._logger.notifyChannel('timers', LOG_DEBUG, "Job scheduled in %s seconds for %s.%s" % (wait, fn.im_class.__name__, fn.func_name))
 191             timer = threading.Timer(wait, fn, args, kwargs)
 192             timer.start()
 193             self._timers.append(timer)
 194         for timer in self._timers[:]:
 195             if not timer.isAlive():
 196                 self._timers.remove(timer)
 197
 198     def quit(cls):
 199         for timer in cls._timers:
 200             timer.cancel()
 201     quit = classmethod(quit)
 202
 203
 204 import traceback
 205
 206 class xmlrpc(object):
 207     class RpcGateway(object):
 208         def __init__(self, name):
 209             self.name = name
 210
 211 class OpenERPDispatcherException(Exception):
 212     def __init__(self, exception, traceback):
 213         self.exception = exception
 214         self.traceback = traceback
 215
 216 class OpenERPDispatcher:
 217     def log(self, title, msg):
 218         from pprint import pformat
 219         Logger().notifyChannel('%s' % title, LOG_DEBUG_RPC, pformat(msg))
 220
 221     def dispatch(self, service_name, method, params):
 222         try:
 223             self.log('service', service_name)
 224             self.log('method', method)
 225             self.log('params', params)
 226             result = LocalService(service_name)(method, *params)
 227             self.log('result', result)
 228             return result
 229         except Exception, e:
 230             self.log('exception', tools.exception_to_unicode(e))
 231             if hasattr(e, 'traceback'):
 232                 tb = e.traceback
 233             else:
 234                 tb = sys.exc_info()
 235             tb_s = "".join(traceback.format_exception(*tb))
 236             if tools.config['debug_mode']:
 237                 import pdb
 238                 pdb.post_mortem(tb[2])
 239             raise OpenERPDispatcherException(e, tb_s)
 240
 241 class GenericXMLRPCRequestHandler(OpenERPDispatcher):
 242     def _dispatch(self, method, params):
 243         try:
 244             service_name = self.path.split("/")[-1]
 245             return self.dispatch(service_name, method, params)
 246         except OpenERPDispatcherException, e:
 247             raise xmlrpclib.Fault(tools.exception_to_unicode(e.exception), e.traceback)
 248
 249 class SSLSocket(object):
 250     def __init__(self, socket):
 251         if not hasattr(socket, 'sock_shutdown'):
 252             from OpenSSL import SSL
 253             ctx = SSL.Context(SSL.SSLv23_METHOD)
 254             ctx.use_privatekey_file(tools.config['secure_pkey_file'])
 255             ctx.use_certificate_file(tools.config['secure_cert_file'])
 256
 257             self.socket = SSL.Connection(ctx, socket)
 258         else:
 259             self.socket = socket
 260
 261     def shutdown(self, how):
 262         return self.socket.sock_shutdown(how)
 263
 264     def __getattr__(self, name):
 265         return getattr(self.socket, name)
 266
 267 class doesitgohere():
 268     def recv(self, bufsize):
 269         """ Another bugfix: SSL's recv() may raise
 270         recoverable exceptions, which simply need us to retry
 271         the call
 272         """
 273         while True:
 274             try:
 275                 return self.socket.recv(bufsize)
 276             except SSL.WantReadError:
 277                 pass
 278             except SSL.WantWriteError:
 279                 pass
 280
 281
 282 class SimpleXMLRPCRequestHandler(GenericXMLRPCRequestHandler, SimpleXMLRPCServer.SimpleXMLRPCRequestHandler):
 283     rpc_paths = map(lambda s: '/xmlrpc/%s' % s, SERVICES.keys())
 284
 285 class SecureXMLRPCRequestHandler(SimpleXMLRPCRequestHandler):
 286     def setup(self):
 287         self.connection = SSLSocket(self.request)
 288         self.rfile = socket._fileobject(self.request, "rb", self.rbufsize)
 289         self.wfile = socket._fileobject(self.request, "wb", self.wbufsize)
 290
 291 class SimpleThreadedXMLRPCServer(SocketServer.ThreadingMixIn, SimpleXMLRPCServer.SimpleXMLRPCServer):
 292     encoding = None
 293     allow_none = False
 294
 295     def server_bind(self):
 296         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 297         SimpleXMLRPCServer.SimpleXMLRPCServer.server_bind(self)
 298
 299     def handle_error(self, request, client_address):
 300         """ Override the error handler
 301         """
 302         import traceback
 303         Logger().notifyChannel("init", LOG_ERROR,"Server error in request from %s:\n%s" %
 304                 (client_address,traceback.format_exc()))
 305
 306 class SecureThreadedXMLRPCServer(SimpleThreadedXMLRPCServer):
 307     def __init__(self, server_address, HandlerClass, logRequests=1):
 308         SimpleThreadedXMLRPCServer.__init__(self, server_address, HandlerClass, logRequests)
 309         self.socket = SSLSocket(socket.socket(self.address_family, self.socket_type))
 310         self.server_bind()
 311         self.server_activate()
 312
 313     def handle_error(self, request, client_address):
 314         """ Override the error handler
 315         """
 316         import traceback
 317         e_type, e_value, e_traceback = sys.exc_info()
 318         Logger().notifyChannel("init", LOG_ERROR,"SSL Request handler error in request from %s: %s\n%s" %
 319                         (client_address,str(e_type),traceback.format_exc()))
 320
 321 class HttpDaemon(threading.Thread):
 322     def __init__(self, interface, port, secure=False):
 323         threading.Thread.__init__(self)
 324         self.__port = port
 325         self.__interface = interface
 326         self.secure = bool(secure)
 327         handler_class = (SimpleXMLRPCRequestHandler, SecureXMLRPCRequestHandler)[self.secure]
 328         server_class = (SimpleThreadedXMLRPCServer, SecureThreadedXMLRPCServer)[self.secure]
 329
 330         if self.secure:
 331             from OpenSSL.SSL import Error as SSLError
 332         else:
 333             class SSLError(Exception): pass
 334         try:
 335             self.server = server_class((interface, port), handler_class, 0)
 336         except SSLError, e:
 337             Logger().notifyChannel('xml-rpc-ssl', LOG_CRITICAL, "Can not load the certificate and/or the private key files")
 338             sys.exit(1)
 339         except Exception, e:
 340             Logger().notifyChannel('xml-rpc', LOG_CRITICAL, "Error occur when starting the server daemon: %s" % (e,))
 341             sys.exit(1)
 342
 343
 344     def attach(self, path, gw):
 345         pass
 346
 347     def stop(self):
 348         self.running = False
 349         if os.name != 'nt':
 350             self.server.socket.shutdown( hasattr(socket, 'SHUT_RDWR') and socket.SHUT_RDWR or 2 )
 351         self.server.socket.close()
 352
 353     def run(self):
 354         self.server.register_introspection_functions()
 355
 356         self.running = True
 357         while self.running:
 358             self.server.handle_request()
 359         return True
 360
 361         # If the server need to be run recursively
 362         #
 363         #signal.signal(signal.SIGALRM, self.my_handler)
 364         #signal.alarm(6)
 365         #while True:
 366         #   self.server.handle_request()
 367         #signal.alarm(0)          # Disable the alarm
 368
 369 import tiny_socket
 370 class TinySocketClientThread(threading.Thread, OpenERPDispatcher):
 371     def __init__(self, sock, threads):
 372         threading.Thread.__init__(self)
 373         self.sock = sock
 374         self.threads = threads
 375
 376     def run(self):
 377         import select
 378         self.running = True
 379         try:
 380             ts = tiny_socket.mysocket(self.sock)
 381         except:
 382             self.sock.close()
 383             self.threads.remove(self)
 384             return False
 385         while self.running:
 386             try:
 387                 msg = ts.myreceive()
 388             except:
 389                 self.sock.close()
 390                 self.threads.remove(self)
 391                 return False
 392             try:
 393                 result = self.dispatch(msg[0], msg[1], msg[2:])
 394                 ts.mysend(result)
 395             except OpenERPDispatcherException, e:
 396                 new_e = Exception(tools.exception_to_unicode(e.exception)) # avoid problems of pickeling
 397                 ts.mysend(new_e, exception=True, traceback=e.traceback)
 398
 399             self.sock.close()
 400             self.threads.remove(self)
 401             return True
 402
 403     def stop(self):
 404         self.running = False
 405
 406
 407 class TinySocketServerThread(threading.Thread):
 408     def __init__(self, interface, port, secure=False):
 409         threading.Thread.__init__(self)
 410         self.__port = port
 411         self.__interface = interface
 412         self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 413         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 414         self.socket.bind((self.__interface, self.__port))
 415         self.socket.listen(5)
 416         self.threads = []
 417
 418     def run(self):
 419         import select
 420         try:
 421             self.running = True
 422             while self.running:
 423                 (clientsocket, address) = self.socket.accept()
 424                 ct = TinySocketClientThread(clientsocket, self.threads)
 425                 self.threads.append(ct)
 426                 ct.start()
 427             self.socket.close()
 428         except Exception, e:
 429             self.socket.close()
 430             return False
 431
 432     def stop(self):
 433         self.running = False
 434         for t in self.threads:
 435             t.stop()
 436         try:
 437             if hasattr(socket, 'SHUT_RDWR'):
 438                 self.socket.shutdown(socket.SHUT_RDWR)
 439             else:
 440                 self.socket.shutdown(2)
 441             self.socket.close()
 442         except:
 443             return False
 444
 445 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: