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 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() + '\n' + 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         if service_name not in GROUPS['web-services']:
 239             raise Exception('AccessDenied')
 240         try:
 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             try:
 335                 self.server.socket.shutdown(
 336                     hasattr(socket, 'SHUT_RDWR') and socket.SHUT_RDWR or 2)
 337             except socket.error, e:
 338                 if e.errno != 57: raise
 339                 # OSX, socket shutdowns both sides if any side closes it
 340                 # causing an error 57 'Socket is not connected' on shutdown
 341                 # of the other side (or something), see
 342                 # http://bugs.python.org/issue4397
 343                 Logger().notifyChannel(
 344                     'server', LOG_DEBUG,
 345                     '"%s" when shutting down server socket, '
 346                     'this is normal under OS X'%e)
 347         self.server.socket.close()
 348
 349     def run(self):
 350         self.server.register_introspection_functions()
 351
 352         self.running = True
 353         while self.running:
 354             self.server.handle_request()
 355         return True
 356
 357         # If the server need to be run recursively
 358         #
 359         #signal.signal(signal.SIGALRM, self.my_handler)
 360         #signal.alarm(6)
 361         #while True:
 362         #   self.server.handle_request()
 363         #signal.alarm(0)          # Disable the alarm
 364
 365 import tiny_socket
 366 class TinySocketClientThread(threading.Thread, OpenERPDispatcher):
 367     def __init__(self, sock, threads):
 368         threading.Thread.__init__(self)
 369         self.sock = sock
 370         self.threads = threads
 371
 372     def run(self):
 373         import select
 374         self.running = True
 375         try:
 376             ts = tiny_socket.mysocket(self.sock)
 377         except:
 378             self.sock.close()
 379             self.threads.remove(self)
 380             return False
 381         while self.running:
 382             try:
 383                 msg = ts.myreceive()
 384             except:
 385                 self.sock.close()
 386                 self.threads.remove(self)
 387                 return False
 388             try:
 389                 result = self.dispatch(msg[0], msg[1], msg[2:])
 390                 ts.mysend(result)
 391             except OpenERPDispatcherException, e:
 392                 new_e = Exception(tools.exception_to_unicode(e.exception)) # avoid problems of pickeling
 393                 ts.mysend(new_e, exception=True, traceback=e.traceback)
 394
 395             self.sock.close()
 396             self.threads.remove(self)
 397             return True
 398
 399     def stop(self):
 400         self.running = False
 401
 402
 403 class TinySocketServerThread(threading.Thread):
 404     def __init__(self, interface, port, secure=False):
 405         threading.Thread.__init__(self)
 406         self.__port = port
 407         self.__interface = interface
 408         self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 409         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 410         self.socket.bind((self.__interface, self.__port))
 411         self.socket.listen(5)
 412         self.threads = []
 413
 414     def run(self):
 415         import select
 416         try:
 417             self.running = True
 418             while self.running:
 419                 (clientsocket, address) = self.socket.accept()
 420                 ct = TinySocketClientThread(clientsocket, self.threads)
 421                 self.threads.append(ct)
 422                 ct.start()
 423             self.socket.close()
 424         except Exception, e:
 425             self.socket.close()
 426             return False
 427
 428     def stop(self):
 429         self.running = False
 430         for t in self.threads:
 431             t.stop()
 432         try:
 433             if hasattr(socket, 'SHUT_RDWR'):
 434                 self.socket.shutdown(socket.SHUT_RDWR)
 435             else:
 436                 self.socket.shutdown(2)
 437             self.socket.close()
 438         except:
 439             return False
 440
 441 # vim:expandtab:smartindent:tabstop=4:softtabstop=4:shiftwidth=4: