entrouvert
/
glasnost
Archived
17
0
Fork 0
Code Issues Pull Requests Releases Activity
This repository has been archived on 2023-02-21. You can view files and clone it, but cannot push or open issues or pull requests.
glasnost/shared/server/xmlrpcServer.py

658 lines
23 KiB
Python
Raw Blame History

# -*- coding: iso-8859-15 -*-
# Glasnost
# By: Odile Bénassy <obenassy@entrouvert.com>
# Romain Chantereau <rchantereau@entrouvert.com>
# Nicolas Clapiès <nclapies@easter-eggs.org>
# Pierre-Antoine Dejace <padejace@entrouvert.be>
# Thierry Dulieu <tdulieu@easter-eggs.com>
# Florent Monnier <monnier@codelutin.com>
# Cédric Musso <cmusso@easter-eggs.org>
# Frédéric Péters <fpeters@entrouvert.be>
# Benjamin Poussin <poussin@codelutin.com>
# Emmanuel Raviart <eraviart@entrouvert.com>
# Sébastien Régnier <regnier@codelutin.com>
# Emmanuel Saracco <esaracco@easter-eggs.com>
#
# Copyright (C) 2000, 2001 Easter-eggs & Emmanuel Raviart
# Copyright (C) 2002 Odile Bénassy, Code Lutin, Thierry Dulieu, Easter-eggs,
# Entr'ouvert, Frédéric Péters, Benjamin Poussin, Emmanuel Raviart,
# Emmanuel Saracco & Théridion
# Copyright (C) 2003 Odile Bénassy, Romain Chantereau, Nicolas Clapiès,
# Code Lutin, Pierre-Antoine Dejace, Thierry Dulieu, Easter-eggs,
# Entr'ouvert, Florent Monnier, Cédric Musso, Ouvaton, Frédéric Péters,
# Benjamin Poussin, Rodolphe Quiédeville, Emmanuel Raviart, Sébastien
# Régnier, Emmanuel Saracco, Théridion & Vecam
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
# This file:
#
# Contains works derived from xmlrpcServer.py (an xmlrpc server framework for
# python), Copyright (C) 2001 by Ken McIvor.
# xmlrpcServer.py is availabe at http://mrken.net/code/xml-rpc/xmlrpcServer.py
#
# Contains works derived from xmlrpcserver.py (a simple XML-RPC server for
# Python), Copyright (C) 1999 by Fredrik Lundh, Copyright (C) 1999 by Secret
# Labs AB. xmlrpcserver.py is available as part of version 0.9.8 of
# PythonWare's XML-RPC library, found at
# http://www.pythonware.com/downloads/index.htm
#
# Contains works derived from xmlrpc_registry.py (A method registry for use with
# xmlrpclib), Copyright (C) 2001 by Eric Kidd (all rights reserved).
# xmlrpc_registry.py is available at
# http://xmlrpc-c.sourceforge.net/hacks/xmlrpc_registry.py
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. The name of the author may not be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
# SUCH DAMAGE.
__version__ = '$Revision$'[11:-2]
import BaseHTTPServer
import marshal
import os
import socket
import SocketServer
import sys
import threading
import time
import traceback
import xmlrpclib
try:
from M2Crypto import SSL
except ImportError:
SSL = None
import glasnost.common.context as context
import glasnost.common.faults as faults
from glasnost.proxy.DispatcherProxy import callServer
from glasnost.proxy.tools import getProxyForServerRole
# Some type names for use in method signatures.
INT = 'int'
BOOLEAN = 'boolean'
DOUBLE = 'double'
STRING = 'string'
DATETIME = 'dateTime.iso8601'
BASE64 = 'base64'
ARRAY = 'array'
STRUCT = 'struct'
# Some error codes, borrowed from xmlrpc-c.
INTERNAL_ERROR = -500
TYPE_ERROR = -501
INDEX_ERROR = -502
PARSE_ERROR = -503
NETWORK_ERROR = -504
TIMEOUT_ERROR = -505
NO_SUCH_METHOD_ERROR = -506
REQUEST_REFUSED_ERROR = -507
INTROSPECTION_DISABLED_ERROR = -508
LIMIT_EXCEEDED_ERROR = -509
INVALID_UTF8_ERROR = -510
class NotificationThread(threading.Thread):
def __init__(self, subscriberId, callBack, params):
threading.Thread.__init__(self)
self.subscriberId = subscriberId
self.callBack = callBack
self.params = params
def run(self):
try:
callServer(self.subscriberId, self.callBack, self.params)
except:
pass # notification failed. Who should we warn?
#
# Servers.
#
class ServerMixin:
__server_version = 'RpcServer/' + __version__
__python_version = 'Python ' + sys.version.split()[0]
_allowIntrospection = 1
monthName = [None,
'Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',
'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
def __init__ (self):
self._methods = {}
self._wrappedMethods = {}
self._wrapper = None
self._signatures = {}
self._help = {}
self._capabilities = {}
self._default_method = None
self._subscriptions_before = {}
self._subscriptions_after = {}
self.log_message('%s starting up', self.version_string())
if self._allowIntrospection:
state = 'enabled'
else:
state = 'disabled'
self.log_message('Introspection is %s', state)
self._install_system_methods()
def _install_system_methods(self):
self.add_method('system.listMethods',
self.system_listMethods,
[[ARRAY]])
self.add_method('system.methodSignature',
self.system_methodSignature,
[[ARRAY, STRING]])
self.add_method('system.methodHelp',
self.system_methodHelp,
[[STRING, STRING]])
self.add_method('system.multicall',
self.system_multicall,
[[ARRAY, ARRAY]])
self.add_method('system.getCapabilities',
self.system_getCapabilities,
[[STRUCT]])
def _introspection_check(self):
if not self._allowIntrospection:
raise xmlrpclib.Fault(INTROSPECTION_DISABLED_ERROR,
("Introspection has been disabled on this server."))
def _no_such_method(self, name):
raise xmlrpclib.Fault(NO_SUCH_METHOD_ERROR, str(name))
def add_capability(self, name, specUrl, specVersion):
self._capabilities[name] = {
'specUrl': str(specUrl),
'specVersion': int(specVersion),
}
self.log_message('add_capability %s %s %s', name, specUrl,
specVersion)
def add_method(self, name, method, signature = None, help = None):
"""Add a method to the Xml Rpc methods dictionnary.
Keyword argument:
=================
*name*:
The public method name string.
*method*:
The internal Xml Rpc handler method name string.
*signature*:
+ A signature is an array of types. The first of these types is the
return type of the method, the rest are parameters.
+ Because multiple signatures (ie. overloading) is permitted, a
list of signatures rather than a singleton is prefered.
+ Default None.
*help*:
+ The documentation string describing the use of that method.
+ The documentation string may contain HTML markup.
+ Default None.
"""
if help is None:
help = ''
if signature is None:
signature = 'undef'
self._methods[name] = method
self._signatures[name] = signature
self._help[name] = help
self.log_message('add_method %s', name)
def add_subscription(self, when, virtualServerId, functionName,
subscriberId, callBackName):
if when == 'after':
subscriptions = self._subscriptions_after
else:
subscriptions = self._subscriptions_before
listSubscriptions = subscriptions.setdefault(virtualServerId, [])
listSubscriptions.append( (functionName, subscriberId, callBackName) )
def add_wrappedMethod(self, name, method, signature = None, help = None):
"""Add a wrapped method to the Xml Rpc methods dictionnary.
This is a slight modification of the add_method method.
"""
if help is None:
help = ''
if signature is None:
signature = 'undef'
self._wrappedMethods[name] = method
self._signatures[name] = signature
self._help[name] = help
self.log_message('add_wrappedMethod %s', name)
def dispatch_call(self, name, params, isDirectCall = 0):
useWrapper = 0
if self._methods.has_key(name):
method = self._methods[name]
elif self._wrappedMethods.has_key(name):
method = self._wrappedMethods[name]
useWrapper = 1
else:
method = self._default_method
if method == None:
self._no_such_method(name)
try:
if params:
self.send_before_notify(params[0], name, params)
if not isDirectCall:
context.initFromOther(self.baseContext)
currentContext = context.get()
if useWrapper and self._wrapper is not None:
result = self._wrapper(method, params, isDirectCall)
else:
result = apply(method, params)
if not isDirectCall:
assert currentContext == context.get()
if params:
self.send_after_notify(params[0], name, result)
return result
except faults.Fault:
self.log_error(
'error for request %s(%s) on %s', name, params, method)
traceback.print_exc()
raise
except:
self.log_error(
'error for request %s(%s) on %s', name, params, method)
traceback.print_exc()
raise faults.UnknownServerException()
def log_date_time_string(self):
now = time.time()
year, month, day, hh, mm, ss, x, y, z = time.localtime(now)
s = "%02d/%3s/%04d %02d:%02d:%02d" % (
day, self.monthName[month], year, hh, mm, ss)
return s
def log_error(self, format, *args):
self.write_error("%s [%s] %s" % (self._address,
self.log_date_time_string(), format % args))
def log_exception(self, method, info):
tb = traceback.extract_tb(info[2])
self.log_error(
'Exception %s "%s" %s %s %s',
method,
traceback.format_exception_only(info[0], info[1])[0].strip(),
tb[-1][0], tb[-1][1], tb[-1][2])
def log_fault(self, method, fault):
self.log_error('Fault %s %s "%s"', method, fault.faultCode,
fault.faultString)
def log_message(self, format, *args):
self.write_message("%s [%s] %s" % (self._address, \
self.log_date_time_string(), format % args))
def remove_subscription(self, when, virtualServerId, functionName,
subscriberId, callBackName):
if when == 'any':
self.remove_subscription(
'before', virtualServerId, functionName, subscriberId,
callBackName)
self.remove_subscription(
'after', virtualServerId, functionName, subscriberId,
callBackName)
return
if when == 'after':
subscriptions = self._subscriptions_after
else:
subscriptions = self._subscriptions_before
listSubscriptions = subscriptions.setdefault(virtualServerId, [])
subscriptions[virtualServerId] = [
x
for x in listSubscriptions
if x[1] != subscriberId or (
(x[0] != functionName and functionName != '_ALL_') \
and (x[2] != callBackName and callBackName != 'any'))]
def send_after_notify(self, virtualServerId, functionName, result):
if functionName.startswith('system.') or type(virtualServerId) != '':
return
subscriptions = self._subscriptions_after
for name, subscriber, callBack in subscriptions.setdefault(
virtualServerId, []):
if name not in (functionName, '_ALL_'):
continue
self.send_notify(
subscriber, callBack, [virtualServerId, functionName, result])
def send_before_notify(self, virtualServerId, functionName, params):
if functionName.startswith('system.') or type(virtualServerId) != '':
return
subscriptions = self._subscriptions_before
for name, subscriber, callBack in subscriptions.setdefault(
virtualServerId, []):
if name not in (functionName, '_ALL_'):
continue
self.send_notify(
subscriber, callBack, [virtualServerId, functionName, params])
def send_notify(self, subscriberId, callBack, params):
notifThread = NotificationThread(subscriberId, callBack, params)
notifThread.start()
def set_default_method(self, method):
self._default_method = method
def set_wrapper(self, method):
self._wrapper = method
def system_getCapabilities(self):
self._introspection_check()
return self._capabilities
def system_listMethods(self):
self._introspection_check()
return self._methods.keys() + self._wrappedMethods.keys()
def system_methodHelp(self, name):
self._introspection_check()
if self._help.has_key(name):
return self._help[name]
else:
self._no_such_method(name)
def system_methodSignature(self, name):
self._introspection_check()
if self._signatures.has_key(name):
return self._signatures[name]
else:
self._no_such_method(name)
def system_multicall(self, calls):
results = []
for call in calls:
# XXX: individual faults and exceptions are not logged
# probable solution: move exception handling from do_POST to
# dispatch_call
try:
name = call['methodName']
params = call['params']
if name == 'system.multicall':
errmsg = "Recursive system.multicall forbidden"
raise xmlrpclib.Fault(REQUEST_REFUSED_ERROR, errmsg)
print ' multicall, calling', name
result = [self.dispatch_call(name, params, isDirectCall = 1)]
except xmlrpclib.Fault, fault:
result = {'faultCode': fault.faultCode,
'faultString': fault.faultString}
except:
info = sys.exc_info()
errmsg = "%s:%s" % (info[0], info[1])
result = {'faultCode': 1, 'faultString': errmsg}
results.append(result)
return results
def version_string(self):
return self.__server_version + ' (' + self.__python_version + ')'
def write_error(self, message):
sys.stderr.write("ERROR: %s\n" % message)
def write_message(self, message):
sys.stderr.write("LOG: %s\n" % message)
class FastServer(ServerMixin, SocketServer.UnixStreamServer):
__server_version = 'FastServer/' + __version__
def __init__(self, serverAddress):
socketPath = '/tmp/.glasnost:%s' % serverAddress[1]
try:
os.unlink(socketPath)
except OSError:
pass
SocketServer.UnixStreamServer.__init__(
self, socketPath, FastRequestHandler)
os.chmod(socketPath, 0777)
self._address = socketPath
ServerMixin.__init__(self)
def serve_forever(self):
self.log_message('Awaiting Fast connections')
SocketServer.UnixStreamServer.serve_forever(self)
self.log_message('Finished serving Fast connections')
class FastForkingServer(SocketServer.ForkingMixIn, FastServer):
pass
class FastThreadingServer(SocketServer.ThreadingMixIn, FastServer):
pass
class XmlRpcServer(ServerMixin, SocketServer.TCPServer):
__server_version = 'XmlRpcServer/' + __version__
def __init__(self, serverAddress):
# Try five time to obtain the socket.
i = 0
while 1:
try:
SocketServer.TCPServer.__init__(self, serverAddress,
XmlRpcRequestHandler)
break
except socket.error, e:
if e.args[0] == 98:
i += 1
if i == 5:
raise
time.sleep(2)
continue
raise
if serverAddress[0] == '':
host = socket.gethostbyaddr(socket.gethostname())[0]
else:
host = serverAddress[0]
self._address = '%s:%s' % (host, serverAddress[1])
ServerMixin.__init__(self)
def serve_forever(self):
self.log_message('Awaiting XML-RPC connections')
SocketServer.TCPServer.serve_forever(self)
self.log_message('Finished serving XML-RPC connections')
class XmlRpcThreadingServer(SocketServer.ThreadingMixIn, XmlRpcServer):
pass
class XmlRpcForkingServer(SocketServer.ForkingMixIn, XmlRpcServer):
pass
if SSL:
class XmlRpcSslServer(ServerMixin, SSL.SSLServer):
__server_version = 'XmlRpcSslServer/' + __version__
def __init__(self, serverAddress, sslContext):
# Try five time to obtain the socket.
i = 0
while 1:
try:
SSL.SSLServer.__init__(self, serverAddress,
XmlRpcRequestHandler, sslContext)
break
except socket.error, e:
if e.args[0] == 98:
i += 1
if i == 5:
raise
time.sleep(2)
continue
raise
if serverAddress[0] == '':
host = socket.gethostbyaddr(socket.gethostname())[0]
else:
host = serverAddress[0]
self._address = '%s:%s' % (host, serverAddress[1])
ServerMixin.__init__(self)
def finish(self):
print 'Finishing request'
self.request.set_shutdown(
SSL.SSL_RECEIVED_SHUTDOWN | SSL.SSL_SENT_SHUTDOWN)
print 'Almost done'
self.request.close()
print 'Done'
def serve_forever(self):
self.log_message('Awaiting XML-RPC SSL connections')
SSL.SSLServer.serve_forever(self)
self.log_message('Finished serving XML-RPC SSL connections')
class XmlRpcSslThreadingServer(SocketServer.ThreadingMixIn, XmlRpcSslServer):
pass
class XmlRpcSslForkingServer(SocketServer.ForkingMixIn, XmlRpcSslServer):
pass
#
# Request Handlers.
#
class FastRequestHandler(SocketServer.StreamRequestHandler):
def __init__(self, request, client_address, server):
self.server = server
SocketServer.StreamRequestHandler.__init__(
self, request, client_address, server)
def handle(self):
bytes = self.rfile.readline()
data = marshal.loads(self.rfile.read(int(bytes)))
params, method = data['params'], data['methodName']
self.server.log_message('Begin of request: %s', method)
try:
response = self.server.dispatch_call(method, params)
if response is None:
response = 0
if type(response) != type(()):
response = (response,)
except xmlrpclib.Fault, e:
response = [e.faultCode, e.faultString]
except:
info = sys.exc_info()
e = xmlrpclib.Fault(
1,
"%s" % traceback.format_exception_only(
info[0], info[1])[0].strip())
response = [e.faultCode, e.faultString]
response = marshal.dumps(response)
self.wfile.write('%11d\n' % len(response))
self.wfile.write(response)
self.wfile.flush()
#self.wfile.close()
self.server.log_message('End of request: %s', method)
class XmlRpcRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
def __init__(self, request, client_address, server):
self.server = server
BaseHTTPServer.BaseHTTPRequestHandler.__init__(
self, request, client_address, server)
def do_POST(self):
data = self.rfile.read(int(self.headers["content-length"]))
params, method = xmlrpclib.loads(data)
self.server.log_message('Begin of request: %s', method)
try:
response = self.server.dispatch_call(method, params)
if response is None:
response = 0
if type(response) != type(()):
response = (response,)
except xmlrpclib.Fault, fault:
self.server.log_fault(method, fault)
response = xmlrpclib.dumps(fault)
except:
info = sys.exc_info()
self.server.log_exception(method, info)
response = xmlrpclib.dumps(xmlrpclib.Fault(1, "%s" %
traceback.format_exception_only(info[0], info[1])[0].strip()))
else:
try:
response = xmlrpclib.dumps( response, methodresponse=1)
except: # TODO: tighter check
info = sys.exc_info()
self.server.log_exception(method, info)
response = xmlrpclib.dumps(xmlrpclib.Fault(
1,
"%s" % traceback.format_exception_only(
info[0], info[1])[0].strip()))
self.server.log_message('End of request: %s', method)
self.send_response(200)
self.send_header("Content-type", "text/xml")
self.send_header("Content-length", str(len(response)))
self.end_headers()
self.wfile.write(response)
self.wfile.flush()
self.connection.shutdown(1)
View Git Blame Copy Permalink