# sslutil.py - SSL handling for mercurial

#

# Copyright 2005, 2006, 2007, 2008 Olivia Mackall <olivia@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.

import hashlib

import os

import re

import ssl

import warnings

from .i18n import _

from .pycompat import getattr

from .node import hex

from . import (

    encoding,

    error,

    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)

# ssl.HAS_TLSv1* are preferred to check support but they were added in Python

# 3.7. Prior to CPython commit 6e8cda91d92da72800d891b2fc2073ecbc134d98

# (backported to the 3.7 branch), ssl.PROTOCOL_TLSv1_1 / ssl.PROTOCOL_TLSv1_2

# were defined only if compiled against a OpenSSL version with TLS 1.1 / 1.2

# support. At the mentioned commit, they were unconditionally defined.

supportedprotocols = set()

if getattr(ssl, 'HAS_TLSv1', util.safehasattr(ssl, 'PROTOCOL_TLSv1')):

    supportedprotocols.add(b'tls1.0')

if getattr(ssl, 'HAS_TLSv1_1', util.safehasattr(ssl, 'PROTOCOL_TLSv1_1')):

    supportedprotocols.add(b'tls1.1')

if getattr(ssl, 'HAS_TLSv1_2', util.safehasattr(ssl, '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,

        # String representation of minimum protocol to be used for UI

        # presentation.

        b'minimumprotocol': None,

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

        b'verifymode': 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+ 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.

    # setup.py checks that TLS 1.1 or TLS 1.2 is present, so the following

    # assert should not fail.

    assert supportedprotocols - {b'tls1.0'}

    defaultminimumprotocol = b'tls1.1'

    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)

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

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

    # 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'

        if not ciphers:

            ciphers = b'DEFAULT'

    s[b'minimumprotocol'] = minimumprotocol

    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()

        # pytype: disable=attribute-error

        # `s` is heterogeneous, but this entry is always a list of tuples

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

        # pytype: enable=attribute-error

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

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

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

        # pytype: disable=attribute-error

        # `s` is heterogeneous, but this entry is always a list of tuples

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

        # pytype: enable=attribute-error

        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'verifymode'] is not None

    return s

def commonssloptions(minimumprotocol):

    """Return SSLContext options common to servers and clients."""

    if minimumprotocol not in configprotocols:

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

    # 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 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  # pytype: disable=import-error

            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.

    if util.safehasattr(ssl, 'TLSVersion'):

        # python 3.7+

        sslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

        minimumprotocol = settings[b'minimumprotocol']

        if minimumprotocol == b'tls1.0':

            with warnings.catch_warnings():

                warnings.filterwarnings(

                    'ignore',

                    'ssl.TLSVersion.TLSv1 is deprecated',

                    DeprecationWarning,

                )

                sslcontext.minimum_version = ssl.TLSVersion.TLSv1

        elif minimumprotocol == b'tls1.1':

            with warnings.catch_warnings():

                warnings.filterwarnings(

                    'ignore',

                    'ssl.TLSVersion.TLSv1_1 is deprecated',

                    DeprecationWarning,

                )

                sslcontext.minimum_version = ssl.TLSVersion.TLSv1_1

        elif minimumprotocol == b'tls1.2':

            sslcontext.minimum_version = ssl.TLSVersion.TLSv1_2

        else:

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

        # Prevent CRIME.

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

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

    else:

        # Despite its name, PROTOCOL_SSLv23 selects the highest protocol that both

        # ends support, including TLS protocols. commonssloptions() restricts the

        # set of allowed protocols.

        sslcontext = ssl.SSLContext(ssl.PROTOCOL_SSLv23)

        sslcontext.options |= commonssloptions(settings[b'minimumprotocol'])

    # We check the hostname ourselves in _verifycert

    sslcontext.check_hostname = False

    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

    elif settings[b'allowloaddefaultcerts']: