mercurial/sslutil.py

36764:7a25f6cfebe8

def _hostsettings(ui, hostname):
    """Obtain security settings for a hostname.

    Returns a dict of settings relevant to that hostname.
    """
    s = {
        # Whether we should attempt to load default/available CA certs
        # if an explicit ``cafile`` is not defined.
        'allowloaddefaultcerts': True,
        # List of 2-tuple of (hash algorithm, hash).

        # internal config: hostsecurity.disabletls10warning
        if not ui.configbool('hostsecurity', 'disabletls10warning'):
            ui.warn(_('warning: connecting to %s using legacy security '
                      'technology (TLS 1.0); see '
                      'https://mercurial-scm.org/wiki/SecureConnections for '
                      'more info\n') % hostname)
        defaultprotocol = 'tls1.0'

    key = 'minimumprotocol'
    protocol = ui.config('hostsecurity', key, defaultprotocol)
    validateprotocol(protocol, key)

    key = '%s:minimumprotocol' % hostname
    protocol = ui.config('hostsecurity', key, protocol)
    validateprotocol(protocol, key)

    # If --insecure is used, we allow the use of TLS 1.0 despite config options.
    # We always print a "connection security to %s is disabled..." message when

        protocol = 'tls1.0'

    s['protocol'], s['ctxoptions'], s['protocolui'] = protocolsettings(protocol)

    ciphers = ui.config('hostsecurity', 'ciphers')
    ciphers = ui.config('hostsecurity', '%s:ciphers' % hostname, ciphers)
    s['ciphers'] = ciphers

    # Look for fingerprints in [hostsecurity] section. Value is a list
    # of <alg>:<fingerprint> strings.
    fingerprints = ui.configlist('hostsecurity', '%s:fingerprints' % hostname)
    for fingerprint in fingerprints:
        if not (fingerprint.startswith(('sha1:', 'sha256:', 'sha512:'))):
            raise error.Abort(_('invalid fingerprint for %s: %s') % (
                                hostname, fingerprint),
                              hint=_('must begin with "sha1:", "sha256:", '
                                     'or "sha512:"'))

        alg, fingerprint = fingerprint.split(':', 1)
        fingerprint = fingerprint.replace(':', '').lower()
        s['certfingerprints'].append((alg, fingerprint))

    # Fingerprints from [hostfingerprints] are always SHA-1.
    for fingerprint in ui.configlist('hostfingerprints', hostname):
        fingerprint = fingerprint.replace(':', '').lower()
        s['certfingerprints'].append(('sha1', fingerprint))
        s['legacyfingerprint'] = True

    # If a host cert fingerprint is defined, it is the only thing that

        s['allowloaddefaultcerts'] = False

    # If both fingerprints and a per-host ca file are specified, issue a warning
    # because users should not be surprised about what security is or isn't
    # being performed.
    cafile = ui.config('hostsecurity', '%s:verifycertsfile' % hostname)
    if s['certfingerprints'] and cafile:
        ui.warn(_('(hostsecurity.%s:verifycertsfile ignored when host '
                  'fingerprints defined; using host fingerprints for '
                  'verification)\n') % hostname)

    # Try to hook up CA certificate validation unless something above
    # makes it not necessary.
    if s['verifymode'] is None:
        # Look at per-host ca file first.
        if cafile:
            cafile = util.expandpath(cafile)
            if not os.path.exists(cafile):
                raise error.Abort(_('path specified by %s does not exist: %s') %
                                  ('hostsecurity.%s:verifycertsfile' % hostname,
                                   cafile))
            s['cafile'] = cafile
        else:
            # Find global certificates file in config.
            cafile = ui.config('web', 'cacerts')

            if len(e.args) == 1: # pypy has different SSLError args
                msg = e.args[0]
            else:
                msg = e.args[1]
            raise error.Abort(_('error loading CA file %s: %s') % (
                              settings['cafile'], msg),
                              hint=_('file is empty or malformed?'))
        caloaded = True
    elif settings['allowloaddefaultcerts']:
        # This is a no-op on old Python.
        sslcontext.load_default_certs()

    the value in ``dn``.
    """
    pats = []
    if not dn:
        return False

    pieces = dn.split(r'.')
    leftmost = pieces[0]
    remainder = pieces[1:]
    wildcards = leftmost.count('*')
    if wildcards > maxwildcards:
        raise wildcarderror(

        if key == 'DNS':
            try:
                if _dnsnamematch(value, hostname):
                    return
            except wildcarderror as e:
                return e.args[0]

            dnsnames.append(value)

    if not dnsnames:
        # The subject is only checked when there is no DNS in subjectAltName.
        for sub in cert.get('subject', []):
            for key, value in sub:
                # According to RFC 2818 the most specific Common Name must
                # be used.
                if key == 'commonName':
                    # 'subject' entries are unicode.
                    try:
                        value = value.encode('ascii')
                    except UnicodeEncodeError:
                        return _('IDN in certificate not supported')

                    try:
                        if _dnsnamematch(value, hostname):
                            return
                    except wildcarderror as e:
                        return e.args[0]

                    dnsnames.append(value)

    if len(dnsnames) > 1:
        return _('certificate is for %s') % ', '.join(dnsnames)

def validatesocket(sock):
    """Validate a socket meets security requirements.

    The passed socket must have been created with ``wrapsocket()``.
    """
    host = sock._hgstate['hostname']
    ui = sock._hgstate['ui']
    settings = sock._hgstate['settings']

    try:
        peercert = sock.getpeercert(True)

            hint=_('see https://mercurial-scm.org/wiki/SecureConnections for '
                   'how to configure Mercurial to avoid this error or set '
                   'hostsecurity.%s:fingerprints=%s to trust this server') %
                   (host, nicefingerprint))

    msg = _verifycert(peercert2, host)
    if msg:
        raise error.Abort(_('%s certificate error: %s') % (host, msg),
                         hint=_('set hostsecurity.%s:certfingerprints=%s '
                                'config setting or use --insecure to connect '
                                'insecurely') %

36765:424994a0adfd

def _hostsettings(ui, hostname):
    """Obtain security settings for a hostname.

    Returns a dict of settings relevant to that hostname.
    """
    bhostname = pycompat.bytesurl(hostname)
    s = {
        # Whether we should attempt to load default/available CA certs
        # if an explicit ``cafile`` is not defined.
        'allowloaddefaultcerts': True,
        # List of 2-tuple of (hash algorithm, hash).

        # internal config: hostsecurity.disabletls10warning
        if not ui.configbool('hostsecurity', 'disabletls10warning'):
            ui.warn(_('warning: connecting to %s using legacy security '
                      'technology (TLS 1.0); see '
                      'https://mercurial-scm.org/wiki/SecureConnections for '
                      'more info\n') % bhostname)
        defaultprotocol = 'tls1.0'

    key = 'minimumprotocol'
    protocol = ui.config('hostsecurity', key, defaultprotocol)
    validateprotocol(protocol, key)

    key = '%s:minimumprotocol' % bhostname
    protocol = ui.config('hostsecurity', key, protocol)
    validateprotocol(protocol, key)

    # If --insecure is used, we allow the use of TLS 1.0 despite config options.
    # We always print a "connection security to %s is disabled..." message when

        protocol = 'tls1.0'

    s['protocol'], s['ctxoptions'], s['protocolui'] = protocolsettings(protocol)

    ciphers = ui.config('hostsecurity', 'ciphers')
    ciphers = ui.config('hostsecurity', '%s:ciphers' % bhostname, ciphers)
    s['ciphers'] = ciphers

    # Look for fingerprints in [hostsecurity] section. Value is a list
    # of <alg>:<fingerprint> strings.
    fingerprints = ui.configlist('hostsecurity', '%s:fingerprints' % bhostname)
    for fingerprint in fingerprints:
        if not (fingerprint.startswith(('sha1:', 'sha256:', 'sha512:'))):
            raise error.Abort(_('invalid fingerprint for %s: %s') % (
                                bhostname, fingerprint),
                              hint=_('must begin with "sha1:", "sha256:", '
                                     'or "sha512:"'))

        alg, fingerprint = fingerprint.split(':', 1)
        fingerprint = fingerprint.replace(':', '').lower()
        s['certfingerprints'].append((alg, fingerprint))

    # Fingerprints from [hostfingerprints] are always SHA-1.
    for fingerprint in ui.configlist('hostfingerprints', bhostname):
        fingerprint = fingerprint.replace(':', '').lower()
        s['certfingerprints'].append(('sha1', fingerprint))
        s['legacyfingerprint'] = True

    # If a host cert fingerprint is defined, it is the only thing that

        s['allowloaddefaultcerts'] = False

    # If both fingerprints and a per-host ca file are specified, issue a warning
    # because users should not be surprised about what security is or isn't
    # being performed.
    cafile = ui.config('hostsecurity', '%s:verifycertsfile' % bhostname)
    if s['certfingerprints'] and cafile:
        ui.warn(_('(hostsecurity.%s:verifycertsfile ignored when host '
                  'fingerprints defined; using host fingerprints for '
                  'verification)\n') % bhostname)

    # Try to hook up CA certificate validation unless something above
    # makes it not necessary.
    if s['verifymode'] is None:
        # Look at per-host ca file first.
        if cafile:
            cafile = util.expandpath(cafile)
            if not os.path.exists(cafile):
                raise error.Abort(_('path specified by %s does not exist: %s') %
                                  ('hostsecurity.%s:verifycertsfile' % (
                                      bhostname,), cafile))
            s['cafile'] = cafile
        else:
            # Find global certificates file in config.
            cafile = ui.config('web', 'cacerts')

            if len(e.args) == 1: # pypy has different SSLError args
                msg = e.args[0]
            else:
                msg = e.args[1]
            raise error.Abort(_('error loading CA file %s: %s') % (
                              settings['cafile'], util.forcebytestr(msg)),
                              hint=_('file is empty or malformed?'))
        caloaded = True
    elif settings['allowloaddefaultcerts']:
        # This is a no-op on old Python.
        sslcontext.load_default_certs()

    the value in ``dn``.
    """
    pats = []
    if not dn:
        return False
    dn = pycompat.bytesurl(dn)
    hostname = pycompat.bytesurl(hostname)

    pieces = dn.split('.')
    leftmost = pieces[0]
    remainder = pieces[1:]
    wildcards = leftmost.count('*')
    if wildcards > maxwildcards:
        raise wildcarderror(

        if key == 'DNS':
            try:
                if _dnsnamematch(value, hostname):
                    return
            except wildcarderror as e:
                return util.forcebytestr(e.args[0])

            dnsnames.append(value)

    if not dnsnames:
        # The subject is only checked when there is no DNS in subjectAltName.
        for sub in cert.get(r'subject', []):
            for key, value in sub:
                # According to RFC 2818 the most specific Common Name must
                # be used.
                if key == r'commonName':
                    # 'subject' entries are unicode.
                    try:
                        value = value.encode('ascii')
                    except UnicodeEncodeError:
                        return _('IDN in certificate not supported')

                    try:
                        if _dnsnamematch(value, hostname):
                            return
                    except wildcarderror as e:
                        return util.forcebytestr(e.args[0])

                    dnsnames.append(value)

    if len(dnsnames) > 1:
        return _('certificate is for %s') % ', '.join(dnsnames)

def validatesocket(sock):
    """Validate a socket meets security requirements.

    The passed socket must have been created with ``wrapsocket()``.
    """
    shost = sock._hgstate['hostname']
    host = pycompat.bytesurl(shost)
    ui = sock._hgstate['ui']
    settings = sock._hgstate['settings']

    try:
        peercert = sock.getpeercert(True)

            hint=_('see https://mercurial-scm.org/wiki/SecureConnections for '
                   'how to configure Mercurial to avoid this error or set '
                   'hostsecurity.%s:fingerprints=%s to trust this server') %
                   (host, nicefingerprint))

    msg = _verifycert(peercert2, shost)
    if msg:
        raise error.Abort(_('%s certificate error: %s') % (host, msg),
                         hint=_('set hostsecurity.%s:certfingerprints=%s '
                                'config setting or use --insecure to connect '
                                'insecurely') %