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