Whether current host needs SSL or not.

        :param (str,int) host_and_port: the host/port pair to check, default current_host_and_port
517738
def __need_ssl(self, host_and_port=None):
        
        if not host_and_port:
            host_and_port = self.current_host_and_port

        return host_and_port in self.__ssl_params
801821
def on_connecting(self, host_and_port):
        
        log.info("connecting %s %s (x %s)", host_and_port[0], host_and_port[1], self.connections)
        self.connections += 1
157462
def ssl_context(self, verify=True, cert_reqs=None,
                    check_hostname=False, certfile=None, keyfile=None,
                    cafile=None, capath=None, cadata=None, **kw):
        
        assert ssl, 'SSL not supported'
        cafile = cafile or DEFAULT_CA_BUNDLE_PATH

        if verify is True:
            cert_reqs = ssl.CERT_REQUIRED
            check_hostname = True

        if isinstance(verify, str):
            cert_reqs = ssl.CERT_REQUIRED
            if os.path.isfile(verify):
                cafile = verify
            elif os.path.isdir(verify):
                capath = verify

        return ssl._create_unverified_context(cert_reqs=cert_reqs,
                                              check_hostname=check_hostname,
                                              certfile=certfile,
                                              keyfile=keyfile,
                                              cafile=cafile,
                                              capath=capath,
                                              cadata=cadata)
275249
def validate(self):
        
        if (self.scheme is None or self.scheme != '') \
          and (self.host is None or self.host == ''):
            return False
        return True
1020517
def verify_host_key_value(given)
      current_net_ssh = Net::SSH::Version::CURRENT
      new_value_version = Net::SSH::Version[5, 0, 0]
      if current_net_ssh >= new_value_version
        # 5.0+ style
        {
          # It's not a boolean anymore.
          'true' => :always,
          'false' => :never,
          true => :always,
          false => :never,
          # May be correct value, but strings from JSON config
          'always' => :always,
          'never' => :never,
          nil => :never,
        }.fetch(given, given)
      else
        # up to 4.2 style
        {
          'true' => true,
          'false' => false,
          nil => false,
        }.fetch(given, given)
      end
    end
1215329
def is_safe_url(url, host=None):
    
    if not url:
        return False
    netloc = urlparse.urlparse(url)[1]
    return not netloc or netloc == host
67004
def is_ssl_error(error=None):
    """"""
    if error is None:
        error = sys.exc_info()[1]
    from OpenSSL import SSL
    return isinstance(error, SSL.Error)
1081724
def check_ssl():
    """"""
    try:
        import ssl
    except:
        log.warning('Error importing SSL module', stack_info=True)
        print(SSL_ERROR_MESSAGE)
        sys.exit(1)
    else:
        log.info('SSL module is available')
        return ssl
1014869
def is_same_host(self, url):
        
        if url.startswith('/'):
            return True

        # TODO: Add optional support for socket.gethostbyname checking.
        scheme, host, port = get_host(url)

        host = _ipv6_host(host).lower()

        # Use explicit default port for comparison when none is given
        if self.port and not port:
            port = port_by_scheme.get(scheme)
        elif not self.port and port == port_by_scheme.get(scheme):
            port = None

        return (scheme, host, port) == (self.scheme, self.host, self.port)
1221903
def parse_host(hosts):
        
        hosts = EsParser._normalize_hosts(hosts)
        host = hosts[0]

        host_name = host[HostParsing.HOST]
        host_port = host[HostParsing.PORT]

        auth = None
        if HostParsing.HTTP_AUTH in host:
            http_auth = host[HostParsing.HTTP_AUTH]
            user_pass = http_auth.split(':')
            auth = (user_pass[0], user_pass[1])

        full_host = "{host}:{port}".format(host=host_name, port=host_port)
        if not host_name.startswith((HostParsing.HTTP + ':', HostParsing.HTTPS + ':')):
            scheme = HostParsing.HTTPS if host.get(HostParsing.USE_SSL) else HostParsing.HTTP
            full_host = "{scheme}://{full_host}".format(full_host=full_host, scheme=scheme)

        return full_host, auth
1345236
def _certificate_required(cls, hostname, port=XCLI_DEFAULT_PORT,
                              ca_certs=None, validate=None):
        '''
        
        '''
        if not ca_certs:
            return False

        xlog.debug("CONNECT SSL %s:%s, cert_file=%s",
                   hostname, port, ca_certs)
        certificate = ssl.get_server_certificate((hostname, port),
                                                 ca_certs=None)
        # handle XIV pre-defined certifications
        # if a validation function was given - we let the user check
        # the certificate himself, with the user's own validate function.
        # if the validate returned True - the user checked the cert
        # and we don't need check it, so we return false.
        if validate:
            return not validate(certificate)
        return True
1120378
def validate_host(cert, name):
    
    cn = None
    for t, v in cert.get_subject().get_components():
        if t == b'CN':
            cn = v
            break

    if cn == name:
        return True

    # checking SAN
    s_name = name.decode('ascii')
    for i in range(cert.get_extension_count()):
        ext = cert.get_extension(i)
        if ext.get_short_name() == b'subjectAltName':
            s = str(ext)
            # SANs are usually have form like: DNS:hostname
            if s.startswith('DNS:') and s[4:] == s_name:
                return True

    # TODO handle wildcards
    return False
932302
def port(self):
        
        return self._val.port or DEFAULT_PORTS.get(self._val.scheme)
777507
def proxy_bypass? host, port
    host = host.downcase
    host_port = [host, port].join ':'

    @no_proxy.each do |name|
      return true if host[-name.length, name.length] == name or
         host_port[-name.length, name.length] == name
    end

    false
  end
430474
def modem_configuration(self, host, port):
        
        request = {
            "command": "modem_configuration",
            "host": host,
            "port": port
        }
        status = self._check_command_response_status(request)
        return status
663979
def get_http_connection(self, host, is_secure):
        
        self.clean()
        with self.mutex:
            key = (host, is_secure)
            if key not in self.host_to_pool:
                return None
            return self.host_to_pool[key].get()
687052
def _create_socket(self):
        
        log.warning('No certificate check is performed for SSL connections')
        s = super(SSL, self)._create_socket()
        return wrap_socket(s)