python类PROTOCOL_SSLv3()的实例源码

def connect(self):
        """
        Connects to the Deluge instance
        """
        logger.info('Connecting to %s:%s' % (self.host, self.port))
        try:
            self._socket.connect((self.host, self.port))
        except ssl.SSLError as e:
            if e.reason != 'UNSUPPORTED_PROTOCOL' or not hasattr(ssl, 'PROTOCOL_SSLv3'):
                raise

            logger.warning('Was unable to ssl handshake, trying to force SSLv3 (insecure)')
            self._create_socket(ssl_version=ssl.PROTOCOL_SSLv3)
            self._socket.connect((self.host, self.port))

        logger.debug('Connected to Deluge, logging in')

        result = self.call('daemon.login', self.username, self.password)

        if type(result) == str:
            return result

        logger.debug('Logged in with value %r' % result)
        self.connected = True

def ssl_wrap_socket(self):

        # Allow sending of keep-alive messages - seems to prevent some servers
        # from closing SSL, leading to deadlocks.
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

        try:
            import ssl
            if self.ca_certs is not None:
                cert_reqs = ssl.CERT_REQUIRED
            else:
                cert_reqs = ssl.CERT_NONE

            if self.ssl_version == "tls1":
                ssl_version = ssl.PROTOCOL_TLSv1
            elif self.ssl_version == "ssl2":
                ssl_version = ssl.PROTOCOL_SSLv2
            elif self.ssl_version == "ssl3":
                ssl_version = ssl.PROTOCOL_SSLv3
            elif self.ssl_version == "ssl23" or self.ssl_version is None:
                ssl_version = ssl.PROTOCOL_SSLv23
            else:
                raise socket.sslerror("Invalid SSL version requested: %s", self.ssl_version)

            self.sock = ssl.wrap_socket(self.sock, self.keyfile, self.certfile, ca_certs=self.ca_certs, cert_reqs=cert_reqs, ssl_version=ssl_version)
            ssl_exc = ssl.SSLError
            self.read_fd = self.sock.fileno()
        except ImportError:
            # No ssl module, and socket.ssl has no fileno(), and does not allow certificate verification
            raise socket.sslerror("imaplib2 SSL mode does not work without ssl module")

        if self.cert_verify_cb is not None:
            cert_err = self.cert_verify_cb(self.sock.getpeercert(), self.host)
            if cert_err:
                raise ssl_exc(cert_err)

def testSslProtocolV3AndShouldFailDueToPoodle(self):
    http = httplib2.Http(ssl_version=ssl.PROTOCOL_SSLv3)
    urls = ['https://www.amazon.com',
            'https://www.apple.com',
            'https://www.twitter.com']
    for url in urls:
      if sys.version_info >= (2, 7):
        with self.assertRaises(httplib2.SSLHandshakeError):
          http.request(url)
        try:
          http.request(url)
        except httplib2.SSLHandshakeError as e:
          self.assertTrue('sslv3 alert handshake failure' in str(e))

def test_auth_ssl(self):
        try:
            self.client.ssl_version = ssl.PROTOCOL_SSLv3
            self.client.auth()
            self.assertRaises(ValueError, self.client.auth)
        finally:
            self.client.ssl_version = ssl.PROTOCOL_TLSv1

def test_ssl3_disabled(self):
        self.assertRaises(socket.error,
                          self._connect_socket,
                          ssl_version=ssl.PROTOCOL_SSLv3)

def test_auth_ssl(self):
        try:
            self.client.ssl_version = ssl.PROTOCOL_SSLv3
            self.client.auth()
            self.assertRaises(ValueError, self.client.auth)
        finally:
            self.client.ssl_version = ssl.PROTOCOL_TLSv1

def ssl_wrap_socket(self):

        # Allow sending of keep-alive messages - seems to prevent some servers
        # from closing SSL, leading to deadlocks.
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)

        try:
            import ssl
            if self.ca_certs is not None:
                cert_reqs = ssl.CERT_REQUIRED
            else:
                cert_reqs = ssl.CERT_NONE

            if self.ssl_version == "tls1":
                ssl_version = ssl.PROTOCOL_TLSv1
            elif self.ssl_version == "ssl2":
                ssl_version = ssl.PROTOCOL_SSLv2
            elif self.ssl_version == "ssl3":
                ssl_version = ssl.PROTOCOL_SSLv3
            elif self.ssl_version == "ssl23" or self.ssl_version is None:
                ssl_version = ssl.PROTOCOL_SSLv23
            else:
                raise socket.sslerror("Invalid SSL version requested: %s", self.ssl_version)

            self.sock = ssl.wrap_socket(self.sock, self.keyfile, self.certfile, ca_certs=self.ca_certs, cert_reqs=cert_reqs, ssl_version=ssl_version)
            ssl_exc = ssl.SSLError
            self.read_fd = self.sock.fileno()
        except ImportError:
            # No ssl module, and socket.ssl has no fileno(), and does not allow certificate verification
            raise socket.sslerror("imaplib2 SSL mode does not work without ssl module")

        if self.cert_verify_cb is not None:
            cert_err = self.cert_verify_cb(self.sock.getpeercert(), self.host)
            if cert_err:
                raise ssl_exc(cert_err)

def test_auth_ssl(self):
        try:
            self.client.ssl_version = ssl.PROTOCOL_SSLv3
            self.client.auth()
            self.assertRaises(ValueError, self.client.auth)
        finally:
            self.client.ssl_version = ssl.PROTOCOL_TLSv1

def try_protocol_combo(self, server_protocol, client_protocol):
        with self.server.lock:
            self.server.handler.ssl_version = server_protocol
        self.client.ssl_version = client_protocol
        self.client.close()
        self.client.connect(self.server.host, self.server.port)
        try:
            self.client.login()
        except (ssl.SSLError, socket.error):
            self.client.close()
        else:
            self.client.quit()

    # def test_ssl_version(self):
    #     protos = [ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv23,
    #               ssl.PROTOCOL_TLSv1]
    #     if hasattr(ssl, "PROTOCOL_SSLv2"):
    #         protos.append(ssl.PROTOCOL_SSLv2)
    #         for proto in protos:
    #             self.try_protocol_combo(ssl.PROTOCOL_SSLv2, proto)
    #     for proto in protos:
    #         self.try_protocol_combo(ssl.PROTOCOL_SSLv3, proto)
    #     for proto in protos:
    #         self.try_protocol_combo(ssl.PROTOCOL_SSLv23, proto)
    #     for proto in protos:
    #         self.try_protocol_combo(ssl.PROTOCOL_TLSv1, proto)

def test_auth_ssl(self):
        try:
            self.client.ssl_version = ssl.PROTOCOL_SSLv3
            self.client.auth()
            self.assertRaises(ValueError, self.client.auth)
        finally:
            self.client.ssl_version = ssl.PROTOCOL_TLSv1

def new_wrap_socket(*args, **kwargs):
    kwargs['ssl_version'] = ssl.PROTOCOL_SSLv3
    return orig_wrap_socket(*args, **kwargs)

def _get_socket(self, sslversion=ssl.PROTOCOL_TLSv1):
        """Sets up an https connection and do an HTTP/raw socket request

        :param sslversion: version of ssl session
        :raises: IloConnectionError, for connection failures
        :returns: ssl wrapped socket object
        """
        err = None
        sock = None
        try:
            for res in socket.getaddrinfo(
                    self.hostname, self.port, 0, socket.SOCK_STREAM):
                af, socktype, proto, canonname, sa = res
                try:
                    sock = socket.socket(af, socktype, proto)
                    sock.settimeout(self.timeout)
                    # Connecting to {self.hostname} at port {self.port}
                    sock.connect(sa)
                except socket.timeout:
                    if sock is not None:
                        sock.close()
                    err = exception.IloConnectionError(
                        "Timeout connecting to %(hostname)s:%(port)d"
                        % {'hostname': self.hostname, 'port': self.port})
                except socket.error:
                    if sock is not None:
                        sock.close()
                    e = sys.exc_info()[1]
                    err = exception.IloConnectionError(
                        "Error connecting to %(hostname)s:%(port)d : %(error)s"
                        % {'hostname': self.hostname, 'port': self.port,
                           'error': str(e)})
        except Exception:
            raise exception.IloConnectionError(
                "Unable to resolve %s" % self.hostname)

        if err is not None:
            raise err

        # wrapping the socket over ssl session
        try:
            return ssl.wrap_socket(sock, ssl_version=sslversion)
        except socket.sslerror:
            e = sys.exc_info()[1]
            msg = (getattr(e, 'reason', None) or
                   getattr(e, 'message', None))
            # Some older iLO s don't support TLSv1, retry with SSLv3
            if ('wrong version number' in msg) and (
                    sslversion == ssl.PROTOCOL_TLSv1):

                return self._get_socket(ssl.PROTOCOL_SSLv3)

            raise exception.IloConnectionError(
                "Cannot establish ssl session with %(hostname)s:%(port)d : "
                "%(error)s" % {'hostname': self.hostname, 'port': self.port,
                               'error': str(e)})

def _on_resolve(self, addrinfo):
        af, sockaddr = addrinfo[0]

        if self.parsed.scheme == "https":
            ssl_options = {}
            if self.request.validate_cert:
                ssl_options["cert_reqs"] = ssl.CERT_REQUIRED
            if self.request.ca_certs is not None:
                ssl_options["ca_certs"] = self.request.ca_certs
            else:
                ssl_options["ca_certs"] = _DEFAULT_CA_CERTS
            if self.request.client_key is not None:
                ssl_options["keyfile"] = self.request.client_key
            if self.request.client_cert is not None:
                ssl_options["certfile"] = self.request.client_cert

            # SSL interoperability is tricky.  We want to disable
            # SSLv2 for security reasons; it wasn't disabled by default
            # until openssl 1.0.  The best way to do this is to use
            # the SSL_OP_NO_SSLv2, but that wasn't exposed to python
            # until 3.2.  Python 2.7 adds the ciphers argument, which
            # can also be used to disable SSLv2.  As a last resort
            # on python 2.6, we set ssl_version to SSLv3.  This is
            # more narrow than we'd like since it also breaks
            # compatibility with servers configured for TLSv1 only,
            # but nearly all servers support SSLv3:
            # http://blog.ivanristic.com/2011/09/ssl-survey-protocol-support.html
            if sys.version_info >= (2, 7):
                ssl_options["ciphers"] = "DEFAULT:!SSLv2"
            else:
                # This is really only necessary for pre-1.0 versions
                # of openssl, but python 2.6 doesn't expose version
                # information.
                ssl_options["ssl_version"] = ssl.PROTOCOL_SSLv3

            self.stream = SSLIOStream(socket.socket(af),
                                      io_loop=self.io_loop,
                                      ssl_options=ssl_options,
                                      max_buffer_size=self.max_buffer_size)
        else:
            self.stream = IOStream(socket.socket(af),
                                   io_loop=self.io_loop,
                                   max_buffer_size=self.max_buffer_size)
        timeout = min(self.request.connect_timeout, self.request.request_timeout)
        if timeout:
            self._timeout = self.io_loop.add_timeout(
                self.start_time + timeout,
                stack_context.wrap(self._on_timeout))
        self.stream.set_close_callback(self._on_close)
        # ipv6 addresses are broken (in self.parsed.hostname) until
        # 2.7, here is correctly parsed value calculated in __init__
        self.stream.connect(sockaddr, self._on_connect,
                            server_hostname=self.parsed_hostname)