Как сделать взаимную аутентификацию сертификата с httplib2

Я использую httplib2 для выполнения запроса от моего сервера до другого веб-сервиса. Мы хотим использовать взаимную аутентификацию сертификата. Я вижу, как использовать сертификат для исходящего соединения (h.set_certificate), но как я проверяю сертификат, используемый сервером ответа?

Этот билет, кажется, указывает, что httplib2 не делает этого самого и имеет только неопределенные предложения о том, где посмотреть.

Действительно ли это возможно? Я оказываюсь перед необходимостью взламывать на более низком уровне?

8
задан Ned Batchelder 2 March 2010 в 12:41
поделиться

1 ответ

Вот код, который мой коллега Дэйв Сен-Жермен написал для решения проблемы:

import ssl
import socket
from httplib2 import has_timeout
import httplib2
import socks


class CertificateValidationError(httplib2.HttpLib2Error):
    pass


def validating_sever_factory(ca_cert_file):
    # we need to define a closure here because we don't control
    # the arguments this class is instantiated with
    class ValidatingHTTPSConnection(httplib2.HTTPSConnectionWithTimeout):

        def connect(self):
            # begin copypasta from HTTPSConnectionWithTimeout
            "Connect to a host on a given (SSL) port."

            if self.proxy_info and self.proxy_info.isgood():
                sock = socks.socksocket(socket.AF_INET, socket.SOCK_STREAM)
                sock.setproxy(*self.proxy_info.astuple())
            else:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

            if has_timeout(self.timeout):
                sock.settimeout(self.timeout)
            sock.connect((self.host, self.port))
            # end copypasta


            try:
                self.sock = ssl.wrap_socket(sock,
                            self.key_file,
                            self.cert_file,
                            cert_reqs=ssl.CERT_REQUIRED,
                            ca_certs=ca_cert_file
                            )
            except ssl.SSLError:
                # we have to capture the exception here and raise later because 
                # httplib2 tries to ignore exceptions on connect
                import sys
                self._exc_info = sys.exc_info()
                raise
            else:
                self._exc_info = None

                # this might be redundant
                server_cert = self.sock.getpeercert()
                if not server_cert:
                    raise CertificateValidationError(repr(server_cert))

        def getresponse(self):
            if not self._exc_info:
                return httplib2.HTTPSConnectionWithTimeout.getresponse(self)
            else:
                raise self._exc_info[1], None, self._exc_info[2]
    return ValidatingHTTPSConnection


def do_request(url,
        method='GET',
        body=None,
        headers=None,
        keyfile=None,
        certfile=None,
        ca_certs=None,
        proxy_info=None,
        timeout=30):
    """
    makes an http/https request, with optional client certificate and server
    certificate verification.
    returns response, content
    """
    kwargs = {}
    h = httplib2.Http(proxy_info=proxy_info, timeout=timeout)
    is_ssl = url.startswith('https')
    if is_ssl and ca_certs:
        kwargs['connection_type'] = validating_sever_factory(ca_certs)

    if is_ssl and keyfile and certfile:
        h.add_certificate(keyfile, certfile, '')
    return h.request(url, method=method, body=body, headers=headers, **kwargs)
5
ответ дан 5 December 2019 в 20:15
поделиться
Другие вопросы по тегам:

Похожие вопросы: