source: fedd/federation/remote_service.py @ e8f2d4c

#!/usr/local/bin/python

import copy
import traceback

from socket import error as socket_error
from socket import sslerror
from socket import SHUT_RDWR

import M2Crypto.httpslib
from M2Crypto import SSL
from M2Crypto.m2xmlrpclib import SSL_Transport
from M2Crypto.SSL import SSLError
from M2Crypto.BIO import BIOError
from ZSI import ParseException, FaultException, SoapWriter

# Underlying SOAP comms use this and we need to catch their exceptions
import httplib

from proof import proof
from service_error import service_error
from xmlrpclib import ServerProxy, dumps, loads, Fault, Error, Binary
try:
    import fedd_services
    import fedd_internal_services
    service_port_name = 'getfeddPortType'
    internal_service_port_name = 'getfeddInternalPortType'
except ImportError:
    import fedd_client
    import fedd_internal_client
    fedd_services = fedd_client
    fedd_internal_services = fedd_internal_client
    service_port_name = 'getfeddPort'
    internal_service_port_name = 'getfedd_internal_port'
from util import fedd_ssl_context
from deter import fedid
import parse_detail

# Turn off the matching of hostname to certificate ID
SSL.Connection.clientPostConnectionCheck = None

# Used by the remote_service_base class.
def to_binary(o):
    """
    A function that converts an object into an xmlrpclib.Binary using
    either its internal packing method, or the standard Binary constructor.
    """
    pack = getattr(o, 'pack_xmlrpc', None)
    if callable(pack): return Binary(pack())
    else: return Binary(o)

# Classes that encapsulate the process of making and dealing with requests to
# WSDL-generated and XMLRPC remote accesses.  

class remote_service_base:
    """
    This invisible base class encapsulates the functions used to massage the
    dictionaries used to pass parameters into and out of the RPC formats.  It's
    mostly a container for the static methods to do that work, but defines some
    maps sued by sub classes on apply_to_tags
    """
    # A map used to convert fedid fields to fedid objects (when the field is
    # already a string)
    fedid_to_object = {'fedid': lambda x: fedid(bits=x)}
    # A map used by apply_to_tags to convert fedids from xmlrpclib.Binary
    # objects to fedid objects in one sweep.
    decap_fedids = {'fedid': lambda x: fedid(bits=x.data), 
            'credential': lambda x: x.data}
    # A map used to encapsulate fedids into xmlrpclib.Binary objects
    encap_fedids = {'fedid': to_binary, 'credential': to_binary}

    # fields that are never unicoded, because they represent non strings.
    do_not_unicode = set(['credential'])

    @staticmethod
    def pack_soap(container, name, contents):
        """
        Convert the dictionary in contents into a tree of ZSI classes.

        The holder classes are constructed from factories in container and
        assigned to either the element or attribute name.  This is used to
        recursively create the SOAP message.
        """
        if getattr(contents, "__iter__", None) != None:
            attr =getattr(container, "new_%s" % name, None)
            if attr: obj = attr()
            else:
                raise TypeError("%s does not have a new_%s attribute" % \
                        (container, name))
            for e, v in contents.iteritems():
                assign = getattr(obj, "set_element_%s" % e, None) or \
                        getattr(obj, "set_attribute_%s" % e, None)
                if isinstance(v, type(dict())):
                    assign(remote_service_base.pack_soap(obj, e, v))
                elif getattr(v, "__iter__", None) != None:
                    assign([ remote_service_base.pack_soap(obj, e, val ) \
                            for val in v])
                elif getattr(v, "pack_soap", None) != None:
                    assign(v.pack_soap())
                else:
                    assign(v)
            return obj
        else: return contents

    @staticmethod
    def unpack_soap(element):
        """
        Convert a tree of ZSI SOAP classes intro a hash.  The inverse of
        pack_soap

        Elements or elements that are empty are ignored.
        """
        methods = [ m for m in dir(element) \
                if m.startswith("get_element") or m.startswith("get_attribute")]
        if len(methods) > 0:
            rv = { }
            for m in methods:
                if m.startswith("get_element_"):
                    n = m.replace("get_element_","",1)
                else:
                    n = m.replace("get_attribute_", "", 1)
                sub = getattr(element, m)()
                if sub != None:
                    if isinstance(sub, basestring):
                        rv[n] = sub
                    elif getattr(sub, "__iter__", None) != None:
                        if len(sub) > 0: rv[n] = \
                                [remote_service_base.unpack_soap(e) \
                                    for e in sub]
                    else:
                        rv[n] = remote_service_base.unpack_soap(sub)
            return rv
        else: 
            return element

    @staticmethod
    def apply_to_tags(e, map):
        """
        Map is an iterable of ordered pairs (tuples) that map a key to a
        function.
        This function walks the given message and replaces any object with a
        key in the map with the result of applying that function to the object.
        """
        if isinstance(e, dict):
            for k in e.keys():
                if k in map:
                    fcn = map[k]
                    if isinstance(e[k], list):
                        e[k] = [ fcn(b) for b in e[k]]
                    else:
                        e[k] = fcn(e[k])
                elif isinstance(e[k], dict):
                    remote_service_base.apply_to_tags(e[k], map)
                elif isinstance(e[k], list):
                    for ee in e[k]:
                        remote_service_base.apply_to_tags(ee, map)
        # Other types end the recursion - they should be leaves
        return e

    @staticmethod
    def strip_unicode(obj):
        """Walk through a message and convert all strings to non-unicode
        strings"""
        if isinstance(obj, dict):
            for k in obj.keys():
                obj[k] = remote_service_base.strip_unicode(obj[k])
            return obj
        elif isinstance(obj, basestring) and not isinstance(obj, str):
            return str(obj)
        elif getattr(obj, "__iter__", None):
            return [ remote_service_base.strip_unicode(x) for x in obj]
        else:
            return obj

    @staticmethod
    def make_unicode(obj):
        """Walk through a message and convert all strings to unicode"""
        if isinstance(obj, dict):
            for k in obj.keys():
                if k not in remote_service_base.do_not_unicode:
                    obj[k] = remote_service_base.make_unicode(obj[k])
            return obj
        elif isinstance(obj, basestring) and not isinstance(obj, unicode):
            return unicode(obj)
        elif getattr(obj, "__iter__", None):
            return [ remote_service_base.make_unicode(x) for x in obj]
        else:
            return obj



class soap_handler(remote_service_base):
    """
    Encapsulate the handler code to unpack and pack SOAP requests and responses
    and call the given method.

    The code to decapsulate and encapsulate parameters encoded in SOAP is the
    same modulo a few parameters.  This is a functor that calls a fedd service
    trhough a soap interface.  The parameters are the typecode of the request
    parameters, the method to call (usually a bound instance of a method on a
    fedd service providing class), the constructor of a response packet and the
    name of the body element of that packet.  The handler takes a ParsedSoap
    object (the request) and returns an instance of the class created by
    constructor containing the response.  Failures of the constructor or badly
    created constructors will result in None being returned.
    """
    def __init__(self, service_name, method, typecode=None,
            constructor=None, body_name=None):
        self.method = method

        response_class_name = "%sResponseMessage" % service_name
        request_class_name = "%sRequestMessage" % service_name

        if body_name: self.body_name = body_name
        else: self.body_name = "%sResponseBody" % service_name

        if constructor: self.constructor = constructor
        else:
            self.constructor = self.get_class(response_class_name)
            if not self.constructor:
                raise service_error(service_error.internal,
                        "Cannot find class for %s" % response_class_name)

        if typecode: self.typecode = typecode
        else: 
            req = self.get_class(request_class_name)
            if req:
                self.typecode = req.typecode
            else:
                raise service_error(service_error.internal,
                        "Cannot find class for %s" % request_class_name)

            if not self.typecode:
                raise service_error(service_error.internal,
                        "Cannot get typecode for %s" % class_name)

    def get_class(self, class_name):
        return getattr(fedd_services, class_name, None) or \
                getattr(fedd_internal_services, class_name, None)

    def __call__(self, ps, fid):
        req = ps.Parse(self.typecode)
        # Convert the message to a dict with the fedid strings converted to
        # fedid objects
        req = self.apply_to_tags(self.unpack_soap(req), self.fedid_to_object)

        msg = self.method(req, fid)

        resp = self.constructor()
        set_element = getattr(resp, "set_element_%s" % self.body_name, None)
        if set_element and callable(set_element):
            try:
                set_element(self.pack_soap(resp, self.body_name, msg))
                return resp
            except (NameError, TypeError):
                return None
        else:
            return None

class xmlrpc_handler(remote_service_base):
    """
    Generate the handler code to unpack and pack XMLRPC requests and responses
    and call the given method.

    The code to marshall and unmarshall XMLRPC parameters to and from a fedd
    service is largely the same.  This helper creates such a handler.  The
    parameters are the method name, and the name of the body struct that
    contains the response.  A handler is created that takes the params response
    from an xmlrpclib.loads on the incoming rpc and a fedid and responds with
    a hash representing the struct ro be returned to the other side.  On error
    None is returned.  Fedid fields are decapsulated from binary and converted
    to fedid objects on input and encapsulated as Binaries on output.
    """
    def __init__(self, service_name, method):
        self.method = method
        self.body_name = "%sResponseBody" % service_name

    def __call__(self, params, fid):
        msg = None

        p = self.apply_to_tags(params[0], self.decap_fedids)
        try:
            msg = self.method(p, fid)
        except service_error, e:
            raise Fault(e.code, "%s: %s" % (e.code_string(), e.desc))
        if msg != None:
            return self.make_unicode(self.apply_to_tags(\
                    { self.body_name: msg }, self.encap_fedids))
        else:
            return None

class service_caller(remote_service_base):
    def __init__(self, service_name, request_message=None, 
            request_body_name=None, tracefile=None, strict=True,
            log=None, max_retries=None, fedd_encapsulation=True):
        self.service_name = service_name

        if getattr(fedd_services.feddBindingSOAP, service_name, None):
            self.locator = fedd_services.feddServiceLocator
            self.port_name = service_port_name
        elif getattr(fedd_internal_services.feddInternalBindingSOAP, 
                service_name, None):
            self.locator = fedd_internal_services.feddInternalServiceLocator
            self.port_name = internal_service_port_name

        if request_message: self.request_message = request_message
        else:
            request_message_name = "%sRequestMessage" % service_name
            self.request_message = \
                    getattr(fedd_services, request_message_name, None) or \
                    getattr(fedd_internal_services, request_message_name,
                            None)
            if not self.request_message and strict:
                raise service_error(service_error.internal,
                        "Cannot find class for %s" % request_message_name)

        if request_body_name is not None:
            self.request_body_name = request_body_name
        else: 
            self.request_body_name = "%sRequestBody" % service_name

        self.tracefile = tracefile
        self.__call__ = self.call_service
        if max_retries is not None: self.max_retries = max_retries 
        else: self.max_retries = 5
        self.log = log
        if not fedd_encapsulation:
            self.fedid_to_object = {}
            self.decap_fedids = {}
            self.encap_fedids = {}
            self.do_not_unicode = set()


    def serialize_soap(self, req):
        """
        Return a string containing the message that would be sent to call this
        service with the given request.
        """
        msg = self.request_message()
        set_element = getattr(msg, "set_element_%s" % self.request_body_name,
                None)
        if not set_element:
            raise service_error(service_error.internal,
                    "Cannot get element setting method for %s" % \
                            self.request_body_name)
        set_element(self.pack_soap(msg, self.request_body_name, req))
        sw = SoapWriter()
        sw.serialize(msg)
        return unicode(sw)

    def call_xmlrpc_service(self, url, req, cert_file=None, cert_pwd=None, 
            trusted_certs=None, context=None, tracefile=None):
        """Send an XMLRPC request.  """


        # If a context is given, use it.  Otherwise construct one from
        # components.  The construction shouldn't call out for passwords.
        if context:
            ctx = context
        else:
            try:
                ctx = fedd_ssl_context(cert_file, trusted_certs, 
                        password=cert_pwd)
            except SSL.SSLError, e:
                raise service_error(service_error.server_config,
                        "Certificates misconfigured: %s" % e)

        # Of all the dumbass things.  The XMLRPC library in use here won't
        # properly encode unicode strings, so we make a copy of req with
        # the unicode objects converted.  We also convert the url to a
        # basic string if it isn't one already.
        r = self.strip_unicode(copy.deepcopy(req))
        if self.request_body_name:
            r  = self.apply_to_tags(\
                    { self.request_body_name: r}, self.encap_fedids)
        else:
            r = self.apply_to_tags(r, self.encap_fedids)

        url = str(url)
        ok = False
        retries = 0

        while not ok and retries < self.max_retries:
            try:
                transport = SSL_Transport(ctx)
                port = ServerProxy(url, transport=transport)
                remote_method = getattr(port, self.service_name, None)
                resp = remote_method(r)
                ok = True
            except socket_error, e:
                raise service_error(service_error.connect, 
                        "Cannot connect to %s: %s" % (url, e[1]))
            except BIOError, e:
                if self.log:
                    self.log.warn("BIO error contacting %s: %s" % (url, e))
                retries += 1
            except sslerror, e:
                if self.log:
                    self.log.warn("SSL (socket) error contacting %s: %s" % 
                            (url, e))
                retries += 1
            except SSLError, e:
                if self.log:
                    self.log.warn("SSL error contacting %s: %s" % (url, e))
                retries += 1
            except httplib.HTTPException, e:
                if self.log:
                    self.log.warn("HTTP error contacting %s: %s" % (url, e))
                retries +=1
            except Fault, f:
                raise service_error(f.faultCode, f.faultString)
            except Error, e:
                raise service_error(service_error.protocol, 
                        "Remote XMLRPC Fault: %s" % e)

        if retries >= self.max_retries :
            raise service_error(service_error.connect, "Too many SSL failures")

        return self.apply_to_tags(resp, self.decap_fedids) 

    def hammer_port_shut(self, port):
        """
        If there is an ssl failure, the connection to the service often stays
        open and idle, confusing subsequent attempts to contact it (the symptom
        is a stuck connection).  This routine walks through the open port and
        hammers everything as closed as it can.  It is called a couple places
        in call_soap_service.  This is dirty code, walking the internals of a
        couple data structures, but without it, the daemon can lock up.
        """
        try:
            if port is not None:
                binding = getattr(port, 'binding', None)
                if binding is not None: connection = getattr(binding, 'h', None)
                else: connection = None

                if connection is not None:
                    csock = getattr(connection, 'sock', None)
                    if csock is not None:
                        csock.clear()
                        csock.close()
                        cssocket = getattr(csock, 'socket', None)
                        if cssocket is not None:
                            cssocket.shutdown(SHUT_RDWR)
                            cssocket.close()
                        del csock
                        connection.sock = None
                    httplib.HTTPConnection.close(connection)
                    if self.log:
                        self.log.debug("Closed connection with prejudice")

                del port
                port = None
                if self.log:
                    self.log.debug("Annihilated port")
        except Exception, e:
            port = None

        return port


    def call_soap_service(self, url, req, cert_file=None, cert_pwd=None,
            trusted_certs=None, context=None, tracefile=None):
        """
        Send req on to the real destination in dt and return the response

        Req is just the requestType object.  This function re-wraps it.  It
        also rethrows any faults.
        """

        tf = tracefile or self.tracefile or None

        if not self.request_body_name:
            raise service_error(service_error.internal, 
                    "Call to soap service without a configured request body");

        ok = False
        retries = 0
        port = None
        while not ok and retries < self.max_retries:
            try:
                # If this is a retry, close the request and annihilate the port
                port = self.hammer_port_shut(port)

                # Reconstruct the full request message
                msg = self.request_message()
                set_element = getattr(msg, "set_element_%s" % \
                        self.request_body_name,
                        None)
                if not set_element:
                    raise service_error(service_error.internal,
                            "Cannot get element setting method for %s" % \
                                    self.request_body_name)
                set_element(self.pack_soap(msg, self.request_body_name, req))
                # If a context is given, use it.  Otherwise construct one from
                # components.  The construction shouldn't call out for
                # passwords.
                if context:
                    if self.log:
                        self.log.debug("Context passed in to call_soap")
                    ctx = context
                else:
                    if self.log:
                        self.log.debug(
                                "Constructing context in call_soap: %s" % \
                                        cert_file)
                    try:
                        ctx = fedd_ssl_context(cert_file, trusted_certs, 
                                password=cert_pwd)
                    except SSL.SSLError, e:
                        if self.log:
                            self.log.debug("Certificate error: %s" % e)
                        raise service_error(service_error.server_config,
                                "Certificates misconfigured: %s" % e)
                loc = self.locator()
                get_port = getattr(loc, self.port_name, None)
                if not get_port:
                    raise service_error(service_error.internal, 
                            "Cannot get port %s from locator" % self.port_name)
                port = get_port(url,
                        transport=M2Crypto.httpslib.HTTPSConnection, 
                        transdict={ 'ssl_context' : ctx },
                        tracefile=tf)
                remote_method = getattr(port, self.service_name, None)
                if not remote_method:
                    raise service_error(service_error.internal,
                            "Cannot get service from SOAP port")

                fail_exc = None
                if self.log:
                    self.log.debug("Calling %s (retry %d)" % \
                            (self.service_name, retries))
                resp = remote_method(msg)
                ok = True
            except socket_error, e:
                self.hammer_port_shut(port)
                raise service_error(service_error.connect, 
                        "Cannot connect to %s: %s" % (url, e[1]))
            except BIOError, e:
                if self.log:
                    self.log.warn("BIO error contacting %s: %s" % (url, e))
                fail_exc = e
                retries += 1
            except sslerror, e:
                if self.log:
                    self.log.warn("SSL (socket) error contacting %s: %s" % 
                            (url, e))
                retries += 1
            except SSLError, e:
                if self.log:
                    self.log.warn("SSL error contacting %s: %s" % (url, e))
                fail_exc = e
                retries += 1
            except httplib.HTTPException, e:
                if self.log:
                    self.log.warn("HTTP error contacting %s: %s" % (url, e))
                fail_exc = e
                retries +=1
            except ParseException, e:
                port = self.hammer_port_shut(port)
                raise service_error(service_error.protocol,
                        "Bad format message (XMLRPC??): %s" % e)
            except FaultException, e:
                if self.log:
                    self.log.debug('SOAP Fault')
                port = self.hammer_port_shut(port)
                # If the method isn't implemented we get a FaultException
                # without a detail (which would be a FeddFault).  If that's the
                # case construct a service_error out of the SOAP fields of the
                # fault, if they're present.
                if e.fault.detail:
                    det = e.fault.detail[0]
                    ee = self.unpack_soap(det).get('FeddFaultBody', { })
                else:
                    ee = { 'code': service_error.internal, 
                            'desc': e.fault.string or "Something Weird" }
                if ee:
                    if 'proof' in ee: 
                        pl = [ proof.from_dict(p) for p in ee['proof']]
                    else: 
                        pl = None
                    raise service_error(ee.get('code', 'no code'), 
                            ee.get('desc','no desc'), proof=pl)
                else:
                    raise service_error(service_error.internal,
                            "Unexpected fault body")
            except:
                self.log.error('Something weird: %s' % traceback.format_exc())


        if retries >= self.max_retries and fail_exc and not ok:
            port = self.hammer_port_shut(port)
            raise service_error(service_error.connect, 
                    "Too many failures: %s" % fail_exc)

        # Unpack and convert fedids to objects
        r = self.apply_to_tags(self.unpack_soap(resp), self.fedid_to_object)

        #  Make sure all strings are unicode
        r = self.make_unicode(r)
        return r

    def call_service(self, url, req, cert_file=None, cert_pwd=None, 
        trusted_certs=None, context=None, tracefile=None):
        p_fault = None  # Any SOAP failure (sent unless XMLRPC works)
        resp = None
        try:
            # Try the SOAP request
            resp = self.call_soap_service(url, req, 
                    cert_file, cert_pwd, trusted_certs, context, tracefile)
            return resp
        except service_error, e:
            if e.code == service_error.protocol: p_fault = None
            else: raise
        except FaultException, f:
            p_fault = f.fault.detail[0]
                

        # If we could not get a valid SOAP response to the request above,
        # try the same address using XMLRPC and let any faults flow back
        # out.
        if p_fault == None:
            resp = self.call_xmlrpc_service(url, req, cert_file,
                    cert_pwd, trusted_certs, context, tracefile)
            return resp
        else:
            # Build the fault
            ee = unpack_soap(p_fault).get('FeddFaultBody', { })
            if ee:
                raise service_error(ee['code'], ee['desc'])
            else:
                raise service_error(service_error.internal,
                        "Unexpected fault body")