bin/netsvc.py

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