Suds over https with cert Suds over https with cert python python

Suds over https with cert


It sounds like you want to authenticate using a client certificate, not a server certificate as was stated in some of the comments. I had the same issue and was able to write a custom transport for SUDS. Here's the code that works for me.

You'll need your certificates in PEM format for this to work; OpenSSL can easily perform this conversion, though I don't remember the exact syntax.

import urllib2, httplib, socketfrom suds.client import Clientfrom suds.transport.http import HttpTransport, Reply, TransportErrorclass HTTPSClientAuthHandler(urllib2.HTTPSHandler):    def __init__(self, key, cert):        urllib2.HTTPSHandler.__init__(self)        self.key = key        self.cert = cert    def https_open(self, req):        #Rather than pass in a reference to a connection class, we pass in        # a reference to a function which, for all intents and purposes,        # will behave as a constructor        return self.do_open(self.getConnection, req)    def getConnection(self, host, timeout=300):        return httplib.HTTPSConnection(host,                                       key_file=self.key,                                       cert_file=self.cert)class HTTPSClientCertTransport(HttpTransport):    def __init__(self, key, cert, *args, **kwargs):        HttpTransport.__init__(self, *args, **kwargs)        self.key = key        self.cert = cert    def u2open(self, u2request):        """        Open a connection.        @param u2request: A urllib2 request.        @type u2request: urllib2.Requet.        @return: The opened file-like urllib2 object.        @rtype: fp        """        tm = self.options.timeout        url = urllib2.build_opener(HTTPSClientAuthHandler(self.key, self.cert))        if self.u2ver() < 2.6:            socket.setdefaulttimeout(tm)            return url.open(u2request)        else:            return url.open(u2request, timeout=tm)# These lines enable debug logging; remove them once everything works.import logginglogging.basicConfig(level=logging.INFO)logging.getLogger('suds.client').setLevel(logging.DEBUG)logging.getLogger('suds.transport').setLevel(logging.DEBUG)c = Client('https://YOUR_URL_HERE',    transport = HTTPSClientCertTransport('PRIVATE_KEY.pem',                                         'CERTIFICATE_CHAIN.pem'))print c


Another workaround is to use requests library as transport which has better support for ssl. This is what I'm using now to access SOAP services through https using suds:-

import requestsfrom suds.transport.http import HttpAuthenticatedfrom suds.transport import Reply, TransportErrorclass RequestsTransport(HttpAuthenticated):    def __init__(self, **kwargs):        self.cert = kwargs.pop('cert', None)        # super won't work because not using new style class        HttpAuthenticated.__init__(self, **kwargs)    def send(self, request):        self.addcredentials(request)        resp = requests.post(request.url, data=request.message,                             headers=request.headers, cert=self.cert)        result = Reply(resp.status_code, resp.headers, resp.content)        return result

And then you can instantiate suds client as:-

headers = {"Content-TYpe" : "text/xml;charset=UTF-8",           "SOAPAction" : ""}t = RequestsTransport(cert='/path/to/cert', **credentials)client = Client(wsdl_uri, location=send_url, headers=headers,                transport=t))

Update

We're now using Zeep, which use requests underneath.


Based on @k4ml answer, I've only added the open() which allows to fetch the WSDL using the certificate.

This method should fix the suds.transport.TransportError: HTTP Error 403: Forbidden when trying to fetch a WSDL (at Client creation) served behind a HTTPS service.

import requestsfrom suds.transport.http import HttpAuthenticatedfrom suds.transport import Reply, TransportErrorclass RequestsTransport(HttpAuthenticated):    def __init__(self, **kwargs):        self.cert = kwargs.pop('cert', None)        # super won't work because not using new style class        HttpAuthenticated.__init__(self, **kwargs)    def open(self, request):        """        Fetches the WSDL using cert.        """        self.addcredentials(request)        resp = requests.get(request.url, data=request.message,                             headers=request.headers, cert=self.cert)        result = io.StringIO(resp.content.decode('utf-8'))        return result    def send(self, request):        """        Posts to service using cert.        """        self.addcredentials(request)        resp = requests.post(request.url, data=request.message,                             headers=request.headers, cert=self.cert)        result = Reply(resp.status_code, resp.headers, resp.content)        return result

Side note, I've also made a suggested edit to k4ml's answer, but it can take ages before it gets approved.