comparison mercurial/sslutil.py @ 29258:6315c1e14f75

sslutil: introduce a function for determining host-specific settings

This patch marks the beginning of a series that introduces a new,
more configurable, per-host security settings mechanism. Currently,
we have global settings (like web.cacerts and the --insecure argument).
We also have per-host settings via [hostfingerprints].

Global security settings are good for defaults, but they don't
provide the amount of control often wanted. For example, an
organization may want to require a particular CA is used for a
particular hostname.

[hostfingerprints] is nice. But it currently assumes SHA-1.
Furthermore, there is no obvious place to put additional per-host
settings.

Subsequent patches will be introducing new mechanisms for defining
security settings, some on a per-host basis. This commits starts
the transition to that world by introducing the _hostsettings
function. It takes a ui and hostname and returns a dict of security
settings. Currently, it limits itself to returning host fingerprint
info.

We foreshadow the future support of non-SHA1 hashing algorithms
for verifying the host fingerprint by making the "certfingerprints"
key a list of tuples instead of a list of hashes.

We add this dict to the hgstate property on the socket and use it
during socket validation for checking fingerprints. There should be
no change in behavior.

29253:9da137faaa9c

            if self._supportsciphers:
                args['ciphers'] = self._ciphers

            return ssl.wrap_socket(socket, **args)

def _determinecertoptions(ui, host):
    """Determine certificate options for a connections.

    Returns a tuple of (cert_reqs, ca_certs).

        raise error.Abort(_('ssl connection failed'))

    sslsocket._hgstate = {
        'caloaded': caloaded,
        'hostname': serverhostname,
        'ui': ui,
    }

    return sslsocket

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

    try:
        peercert = sock.getpeercert(True)
        peercert2 = sock.getpeercert()
    except AttributeError:

        raise error.Abort(_('%s certificate error: '
                           'no certificate received') % host)

    # If a certificate fingerprint is pinned, use it and only it to
    # validate the remote cert.
    hostfingerprints = ui.configlist('hostfingerprints', host)
    peerfingerprint = util.sha1(peercert).hexdigest()
    nicefingerprint = ":".join([peerfingerprint[x:x + 2]
        for x in xrange(0, len(peerfingerprint), 2)])
    if hostfingerprints:
        fingerprintmatch = False
        for hostfingerprint in hostfingerprints:
            if peerfingerprint.lower() == \
                    hostfingerprint.replace(':', '').lower():
                fingerprintmatch = True
                break
        if not fingerprintmatch:
            raise error.Abort(_('certificate for %s has unexpected '
                               'fingerprint %s') % (host, nicefingerprint),

29258:6315c1e14f75

            if self._supportsciphers:
                args['ciphers'] = self._ciphers

            return ssl.wrap_socket(socket, **args)

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

    Returns a dict of settings relevant to that hostname.
    """
    s = {
        # List of 2-tuple of (hash algorithm, hash).
        'certfingerprints': [],
    }

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

    return s

def _determinecertoptions(ui, host):
    """Determine certificate options for a connections.

    Returns a tuple of (cert_reqs, ca_certs).

        raise error.Abort(_('ssl connection failed'))

    sslsocket._hgstate = {
        'caloaded': caloaded,
        'hostname': serverhostname,
        'settings': _hostsettings(ui, serverhostname),
        'ui': ui,
    }

    return sslsocket

    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)
        peercert2 = sock.getpeercert()
    except AttributeError:

        raise error.Abort(_('%s certificate error: '
                           'no certificate received') % host)

    # If a certificate fingerprint is pinned, use it and only it to
    # validate the remote cert.
    peerfingerprint = util.sha1(peercert).hexdigest()
    nicefingerprint = ":".join([peerfingerprint[x:x + 2]
        for x in xrange(0, len(peerfingerprint), 2)])
    if settings['certfingerprints']:
        fingerprintmatch = False
        for hash, fingerprint in settings['certfingerprints']:
            if peerfingerprint.lower() == fingerprint:
                fingerprintmatch = True
                break
        if not fingerprintmatch:
            raise error.Abort(_('certificate for %s has unexpected '
                               'fingerprint %s') % (host, nicefingerprint),