# sslutil.py - SSL handling for mercurial

#

# Copyright 2005, 2006, 2007, 2008 Matt Mackall <mpm@selenic.com>

# Copyright 2006, 2007 Alexis S. L. Carvalho <alexis@cecm.usp.br>

# Copyright 2006 Vadim Gelfer <vadim.gelfer@gmail.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

from __future__ import absolute_import

import hashlib

import os

import re

import ssl

from .i18n import _

from .pycompat import getattr

from . import (

    encoding,

    error,

    node,

    pycompat,

    util,

)

from .utils import (

    hashutil,

    resourceutil,

    stringutil,

)

# Python 2.7.9+ overhauled the built-in SSL/TLS features of Python. It added

# support for TLS 1.1, TLS 1.2, SNI, system CA stores, etc. These features are

# all exposed via the "ssl" module.

#

# We require in setup.py the presence of ssl.SSLContext, which indicates modern

# SSL/TLS support.

configprotocols = {

    b'tls1.0',

    b'tls1.1',

    b'tls1.2',

}

hassni = getattr(ssl, 'HAS_SNI', False)

# TLS 1.1 and 1.2 may not be supported if the OpenSSL Python is compiled

# against doesn't support them.

# FIXME: Since CPython commit 6e8cda91d92da72800d891b2fc2073ecbc134d98

# individual TLS versions can be turned on and off, and the

# ssl.PROTOCOL_TLSv1_* constants are always defined.

# This means that, on unusual configurations, the following dict may contain

# too many entries. A proper fix would be to check ssl.HAS_TLSv* where

# available (Python 3.7+). Before that, this module should be proofed against

# all possible combinations.

supportedprotocols = {b'tls1.0'}

if util.safehasattr(ssl, b'PROTOCOL_TLSv1_1'):

    supportedprotocols.add(b'tls1.1')

if util.safehasattr(ssl, b'PROTOCOL_TLSv1_2'):

    supportedprotocols.add(b'tls1.2')

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.

        b'allowloaddefaultcerts': True,

        # List of 2-tuple of (hash algorithm, hash).

        b'certfingerprints': [],

        # Path to file containing concatenated CA certs. Used by

        # SSLContext.load_verify_locations().

        b'cafile': None,

        # Whether certificate verification should be disabled.

        b'disablecertverification': False,

        # Whether the legacy [hostfingerprints] section has data for this host.

        b'legacyfingerprint': False,

        # PROTOCOL_* constant to use for SSLContext.__init__.

        b'protocol': None,

        # String representation of minimum protocol to be used for UI

        # presentation.

        b'minimumprotocolui': None,

        # ssl.CERT_* constant used by SSLContext.verify_mode.

        b'verifymode': None,

        # Defines extra ssl.OP* bitwise options to set.

        b'ctxoptions': None,

        # OpenSSL Cipher List to use (instead of default).

        b'ciphers': None,

    }

    # Allow minimum TLS protocol to be specified in the config.

    def validateprotocol(protocol, key):

        if protocol not in configprotocols:

            raise error.Abort(

                _(b'unsupported protocol from hostsecurity.%s: %s')

                % (key, protocol),

                hint=_(b'valid protocols: %s')

                % b' '.join(sorted(configprotocols)),

            )

    # We default to TLS 1.1+ where we can because TLS 1.0 has known

    # vulnerabilities (like BEAST and POODLE). We allow users to downgrade to

    # TLS 1.0+ via config options in case a legacy server is encountered.

    if supportedprotocols - {b'tls1.0'}:

        defaultminimumprotocol = b'tls1.1'

    else:

        # Let people know they are borderline secure.

        # We don't document this config option because we want people to see

        # the bold warnings on the web site.

        # internal config: hostsecurity.disabletls10warning

        if not ui.configbool(b'hostsecurity', b'disabletls10warning'):

            ui.warn(

                _(

                    b'warning: connecting to %s using legacy security '

                    b'technology (TLS 1.0); see '

                    b'https://mercurial-scm.org/wiki/SecureConnections for '

                    b'more info\n'

                )

                % bhostname

            )

        defaultminimumprotocol = b'tls1.0'

    key = b'minimumprotocol'

    minimumprotocol = ui.config(b'hostsecurity', key, defaultminimumprotocol)

    validateprotocol(minimumprotocol, key)

    key = b'%s:minimumprotocol' % bhostname

    minimumprotocol = ui.config(b'hostsecurity', key, minimumprotocol)

    validateprotocol(minimumprotocol, 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

    # --insecure is used. So no need to print anything more here.

    if ui.insecureconnections:

        minimumprotocol = b'tls1.0'

    s[b'minimumprotocolui'] = minimumprotocol

    s[b'protocol'], s[b'ctxoptions'] = protocolsettings(minimumprotocol)

    ciphers = ui.config(b'hostsecurity', b'ciphers')

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

    s[b'ciphers'] = ciphers

    # Look for fingerprints in [hostsecurity] section. Value is a list

    # of <alg>:<fingerprint> strings.

    fingerprints = ui.configlist(

        b'hostsecurity', b'%s:fingerprints' % bhostname

    )

    for fingerprint in fingerprints:

        if not (fingerprint.startswith((b'sha1:', b'sha256:', b'sha512:'))):

            raise error.Abort(

                _(b'invalid fingerprint for %s: %s') % (bhostname, fingerprint),

                hint=_(b'must begin with "sha1:", "sha256:", or "sha512:"'),

            )

        alg, fingerprint = fingerprint.split(b':', 1)

        fingerprint = fingerprint.replace(b':', b'').lower()

        s[b'certfingerprints'].append((alg, fingerprint))

    # Fingerprints from [hostfingerprints] are always SHA-1.

    for fingerprint in ui.configlist(b'hostfingerprints', bhostname):

        fingerprint = fingerprint.replace(b':', b'').lower()

        s[b'certfingerprints'].append((b'sha1', fingerprint))

        s[b'legacyfingerprint'] = True

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

    # matters. No need to validate CA certs.

    if s[b'certfingerprints']:

        s[b'verifymode'] = ssl.CERT_NONE

        s[b'allowloaddefaultcerts'] = False

    # If --insecure is used, don't take CAs into consideration.

    elif ui.insecureconnections:

        s[b'disablecertverification'] = True

        s[b'verifymode'] = ssl.CERT_NONE

        s[b'allowloaddefaultcerts'] = False

    if ui.configbool(b'devel', b'disableloaddefaultcerts'):

        s[b'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(b'hostsecurity', b'%s:verifycertsfile' % bhostname)

    if s[b'certfingerprints'] and cafile:

        ui.warn(

            _(

                b'(hostsecurity.%s:verifycertsfile ignored when host '

                b'fingerprints defined; using host fingerprints for '

                b'verification)\n'

            )

            % bhostname

        )

    # Try to hook up CA certificate validation unless something above

    # makes it not necessary.

    if s[b'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(

                    _(b'path specified by %s does not exist: %s')

                    % (

                        b'hostsecurity.%s:verifycertsfile' % (bhostname,),

                        cafile,

                    )

                )

            s[b'cafile'] = cafile

        else:

            # Find global certificates file in config.

            cafile = ui.config(b'web', b'cacerts')

            if cafile:

                cafile = util.expandpath(cafile)

                if not os.path.exists(cafile):

                    raise error.Abort(

                        _(b'could not find web.cacerts: %s') % cafile

                    )

            elif s[b'allowloaddefaultcerts']:

                # CAs not defined in config. Try to find system bundles.

                cafile = _defaultcacerts(ui)

                if cafile:

                    ui.debug(b'using %s for CA file\n' % cafile)

            s[b'cafile'] = cafile

        # Require certificate validation if CA certs are being loaded and

        # verification hasn't been disabled above.

        if cafile or s[b'allowloaddefaultcerts']:

            s[b'verifymode'] = ssl.CERT_REQUIRED

        else:

            # At this point we don't have a fingerprint, aren't being

            # explicitly insecure, and can't load CA certs. Connecting

            # is insecure. We allow the connection and abort during

            # validation (once we have the fingerprint to print to the

            # user).

            s[b'verifymode'] = ssl.CERT_NONE

    assert s[b'protocol'] is not None

    assert s[b'ctxoptions'] is not None

    assert s[b'verifymode'] is not None

    return s

def protocolsettings(minimumprotocol):

    """Resolve the protocol for a config value.

    Returns a tuple of (protocol, options) which are values used by SSLContext.

    """

    if minimumprotocol not in configprotocols:

        raise ValueError(b'protocol value not supported: %s' % minimumprotocol)

    # Despite its name, PROTOCOL_SSLv23 selects the highest protocol

    # that both ends support, including TLS protocols. On legacy stacks,

    # the highest it likely goes is TLS 1.0. On modern stacks, it can

    # support TLS 1.2.

    #

    # The PROTOCOL_TLSv* constants select a specific TLS version

    # only (as opposed to multiple versions). So the method for

    # supporting multiple TLS versions is to use PROTOCOL_SSLv23 and

    # disable protocols via SSLContext.options and OP_NO_* constants.

    if supportedprotocols == {b'tls1.0'}:

        if minimumprotocol != b'tls1.0':

            raise error.Abort(

                _(b'current Python does not support protocol setting %s')

                % minimumprotocol,

                hint=_(

                    b'upgrade Python or disable setting since '

                    b'only TLS 1.0 is supported'

                ),

            )

        return ssl.PROTOCOL_TLSv1, 0

    # SSLv2 and SSLv3 are broken. We ban them outright.

    options = ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3

    if minimumprotocol == b'tls1.0':

        # Defaults above are to use TLS 1.0+

        pass

    elif minimumprotocol == b'tls1.1':

        options |= ssl.OP_NO_TLSv1

    elif minimumprotocol == b'tls1.2':

        options |= ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1

    else:

        raise error.Abort(_(b'this should not happen'))

    # Prevent CRIME.

    # There is no guarantee this attribute is defined on the module.

    options |= getattr(ssl, 'OP_NO_COMPRESSION', 0)

    return ssl.PROTOCOL_SSLv23, options

def wrapsocket(sock, keyfile, certfile, ui, serverhostname=None):

    """Add SSL/TLS to a socket.

    This is a glorified wrapper for ``ssl.wrap_socket()``. It makes sane

    choices based on what security options are available.

    In addition to the arguments supported by ``ssl.wrap_socket``, we allow

    the following additional arguments:

    * serverhostname - The expected hostname of the remote server. If the

      server (and client) support SNI, this tells the server which certificate

      to use.

    """

    if not serverhostname:

        raise error.Abort(_(b'serverhostname argument is required'))

    if b'SSLKEYLOGFILE' in encoding.environ:

        try:

            import sslkeylog

            sslkeylog.set_keylog(

                pycompat.fsdecode(encoding.environ[b'SSLKEYLOGFILE'])

            )

            ui.warnnoi18n(

                b'sslkeylog enabled by SSLKEYLOGFILE environment variable\n'

            )

        except ImportError:

            ui.warnnoi18n(

                b'sslkeylog module missing, '

                b'but SSLKEYLOGFILE set in environment\n'

            )

    for f in (keyfile, certfile):

        if f and not os.path.exists(f):

            raise error.Abort(

                _(b'certificate file (%s) does not exist; cannot connect to %s')

                % (f, pycompat.bytesurl(serverhostname)),

                hint=_(

                    b'restore missing file or fix references '

                    b'in Mercurial config'

                ),

            )

    settings = _hostsettings(ui, serverhostname)

    # We can't use ssl.create_default_context() because it calls

    # load_default_certs() unless CA arguments are passed to it. We want to

    # have explicit control over CA loading because implicitly loading

    # CAs may undermine the user's intent. For example, a user may define a CA

    # bundle with a specific CA cert removed. If the system/default CA bundle

    # is loaded and contains that removed CA, you've just undone the user's

    # choice.

    sslcontext = ssl.SSLContext(settings[b'protocol'])

    sslcontext.options |= settings[b'ctxoptions']

    sslcontext.verify_mode = settings[b'verifymode']

    if settings[b'ciphers']:

        try:

            sslcontext.set_ciphers(pycompat.sysstr(settings[b'ciphers']))

        except ssl.SSLError as e:

            raise error.Abort(

                _(b'could not set ciphers: %s')

                % stringutil.forcebytestr(e.args[0]),

                hint=_(b'change cipher string (%s) in config')

                % settings[b'ciphers'],

            )

    if certfile is not None:

        def password():

            f = keyfile or certfile

            return ui.getpass(_(b'passphrase for %s: ') % f, b'')

        sslcontext.load_cert_chain(certfile, keyfile, password)

    if settings[b'cafile'] is not None:

        try:

            sslcontext.load_verify_locations(cafile=settings[b'cafile'])

        except ssl.SSLError as e:

            if len(e.args) == 1:  # pypy has different SSLError args

                msg = e.args[0]

            else:

                msg = e.args[1]

            raise error.Abort(

                _(b'error loading CA file %s: %s')

                % (settings[b'cafile'], stringutil.forcebytestr(msg)),

                hint=_(b'file is empty or malformed?'),

            )

        caloaded = True