通过证书在https上使用Suds

Question

我在Apache服务器上有一个SOAP服务，并且启用了SSL加密，使用suds的时候在没有SSL的情况下运行得很好。
我有客户端证书（my.crt和user.p12文件）。
我该如何配置suds客户端，让它能够通过HTTPS与服务正常工作呢？

如果没有证书，我看到的内容是

urllib2.URLError: <urlopen error [Errno 1] _ssl.c:499: error:14094410:SSL routines:SSL3_READ_BYTES:sslv3 alert handshake failure>

Answer 1

根据@k4ml的回答，我只添加了open()这个方法，这样就可以用证书来获取WSDL文件了。

这个方法应该能解决在创建客户端时，尝试获取一个通过HTTPS服务提供的WSDL时出现的suds.transport.TransportError: HTTP Error 403: Forbidden错误。

import requests
from suds.transport.http import HttpAuthenticated
from suds.transport import Reply, TransportError

class 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

顺便提一下，我也对k4ml的回答做了一些建议性的修改，不过这些修改可能需要很长时间才能被批准。

Answer 2

另一种解决办法是使用 requests 库来传输数据，因为它对 ssl 的支持更好。这就是我现在用来通过 https 访问 SOAP 服务的方法，使用的是 suds:-

import requests
from suds.transport.http import HttpAuthenticated
from suds.transport import Reply, TransportError

class 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

然后你可以这样创建 suds 客户端:-

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))

更新

我们现在使用的是 Zeep，它底层使用了 requests。

Answer 3

听起来你想用一个客户端证书来进行身份验证，而不是一些评论中提到的服务器证书。我之前也遇到过同样的问题，后来我为SUDS写了一个自定义的传输方式。下面是我用的代码。

要让这个工作，你需要把你的证书转换成PEM格式；使用OpenSSL可以很简单地完成这个转换，不过我不记得具体的命令了。

import urllib2, httplib, socket
from suds.client import Client
from suds.transport.http import HttpTransport, Reply, TransportError

class 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 logging
logging.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