( )⚙ D2687 sslutil: lots of unicode/bytes cleanup

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2687

sslutil: lots of unicode/bytes cleanup
ClosedPublic

Authored by durin42 on Mar 4 2018, 9:19 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG424994a0adfd: sslutil: lots of unicode/bytes cleanup
Summary

In general, we handle hostnames as bytes, except where Python forces
them to be unicodes.

This fixes all the tracebacks I was seeing in test-https.t, but
there's still some ECONNRESET weirdness that I can't hunt down...

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

durin42 created this revision.Mar 4 2018, 9:19 PM
Closed by commit rHG424994a0adfd: sslutil: lots of unicode/bytes cleanup (authored by durin42). · Explain WhyMar 6 2018, 4:38 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/sslutil.py (40 lines)

Diff 6665

mercurial/sslutil.py

return ssl.wrap_socket(socket, **args) return ssl.wrap_socket(socket, **args)
def _hostsettings(ui, hostname): def _hostsettings(ui, hostname):
"""Obtain security settings for a hostname. """Obtain security settings for a hostname.
Returns a dict of settings relevant to that hostname. Returns a dict of settings relevant to that hostname.
""" """
bhostname = pycompat.bytesurl(hostname)
s = { s = {
# Whether we should attempt to load default/available CA certs # Whether we should attempt to load default/available CA certs
# if an explicit ``cafile`` is not defined. # if an explicit ``cafile`` is not defined.
'allowloaddefaultcerts': True, 'allowloaddefaultcerts': True,
# List of 2-tuple of (hash algorithm, hash). # List of 2-tuple of (hash algorithm, hash).
'certfingerprints': [], 'certfingerprints': [],
# Path to file containing concatenated CA certs. Used by # Path to file containing concatenated CA certs. Used by
# SSLContext.load_verify_locations(). # SSLContext.load_verify_locations().
# Let people know they are borderline secure. # Let people know they are borderline secure.
# We don't document this config option because we want people to see # We don't document this config option because we want people to see
# the bold warnings on the web site. # the bold warnings on the web site.
# internal config: hostsecurity.disabletls10warning # internal config: hostsecurity.disabletls10warning
if not ui.configbool('hostsecurity', 'disabletls10warning'): if not ui.configbool('hostsecurity', 'disabletls10warning'):
ui.warn(_('warning: connecting to %s using legacy security ' ui.warn(_('warning: connecting to %s using legacy security '
'technology (TLS 1.0); see ' 'technology (TLS 1.0); see '
'https://mercurial-scm.org/wiki/SecureConnections for ' 'https://mercurial-scm.org/wiki/SecureConnections for '
'more info\n') % hostname) 'more info\n') % bhostname)
defaultprotocol = 'tls1.0' defaultprotocol = 'tls1.0'
key = 'minimumprotocol' key = 'minimumprotocol'
protocol = ui.config('hostsecurity', key, defaultprotocol) protocol = ui.config('hostsecurity', key, defaultprotocol)
validateprotocol(protocol, key) validateprotocol(protocol, key)
key = '%s:minimumprotocol' % hostname key = '%s:minimumprotocol' % bhostname
protocol = ui.config('hostsecurity', key, protocol) protocol = ui.config('hostsecurity', key, protocol)
validateprotocol(protocol, key) validateprotocol(protocol, key)
# If --insecure is used, we allow the use of TLS 1.0 despite config options. # 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 # We always print a "connection security to %s is disabled..." message when
# --insecure is used. So no need to print anything more here. # --insecure is used. So no need to print anything more here.
if ui.insecureconnections: if ui.insecureconnections:
protocol = 'tls1.0' protocol = 'tls1.0'
s['protocol'], s['ctxoptions'], s['protocolui'] = protocolsettings(protocol) s['protocol'], s['ctxoptions'], s['protocolui'] = protocolsettings(protocol)
ciphers = ui.config('hostsecurity', 'ciphers') ciphers = ui.config('hostsecurity', 'ciphers')
ciphers = ui.config('hostsecurity', '%s:ciphers' % hostname, ciphers) ciphers = ui.config('hostsecurity', '%s:ciphers' % bhostname, ciphers)
s['ciphers'] = ciphers s['ciphers'] = ciphers
# Look for fingerprints in [hostsecurity] section. Value is a list # Look for fingerprints in [hostsecurity] section. Value is a list
# of <alg>:<fingerprint> strings. # of <alg>:<fingerprint> strings.
fingerprints = ui.configlist('hostsecurity', '%s:fingerprints' % hostname) fingerprints = ui.configlist('hostsecurity', '%s:fingerprints' % bhostname)
for fingerprint in fingerprints: for fingerprint in fingerprints:
if not (fingerprint.startswith(('sha1:', 'sha256:', 'sha512:'))): if not (fingerprint.startswith(('sha1:', 'sha256:', 'sha512:'))):
raise error.Abort(_('invalid fingerprint for %s: %s') % ( raise error.Abort(_('invalid fingerprint for %s: %s') % (
hostname, fingerprint), bhostname, fingerprint),
hint=_('must begin with "sha1:", "sha256:", ' hint=_('must begin with "sha1:", "sha256:", '
'or "sha512:"')) 'or "sha512:"'))
alg, fingerprint = fingerprint.split(':', 1) alg, fingerprint = fingerprint.split(':', 1)
fingerprint = fingerprint.replace(':', '').lower() fingerprint = fingerprint.replace(':', '').lower()
s['certfingerprints'].append((alg, fingerprint)) s['certfingerprints'].append((alg, fingerprint))
# Fingerprints from [hostfingerprints] are always SHA-1. # Fingerprints from [hostfingerprints] are always SHA-1.
for fingerprint in ui.configlist('hostfingerprints', hostname): for fingerprint in ui.configlist('hostfingerprints', bhostname):
fingerprint = fingerprint.replace(':', '').lower() fingerprint = fingerprint.replace(':', '').lower()
s['certfingerprints'].append(('sha1', fingerprint)) s['certfingerprints'].append(('sha1', fingerprint))
s['legacyfingerprint'] = True s['legacyfingerprint'] = True
# If a host cert fingerprint is defined, it is the only thing that # If a host cert fingerprint is defined, it is the only thing that
# matters. No need to validate CA certs. # matters. No need to validate CA certs.
if s['certfingerprints']: if s['certfingerprints']:
s['verifymode'] = ssl.CERT_NONE s['verifymode'] = ssl.CERT_NONE
s['allowloaddefaultcerts'] = False s['allowloaddefaultcerts'] = False
# If --insecure is used, don't take CAs into consideration. # If --insecure is used, don't take CAs into consideration.
elif ui.insecureconnections: elif ui.insecureconnections:
s['disablecertverification'] = True s['disablecertverification'] = True
s['verifymode'] = ssl.CERT_NONE s['verifymode'] = ssl.CERT_NONE
s['allowloaddefaultcerts'] = False s['allowloaddefaultcerts'] = False
if ui.configbool('devel', 'disableloaddefaultcerts'): if ui.configbool('devel', 'disableloaddefaultcerts'):
s['allowloaddefaultcerts'] = False s['allowloaddefaultcerts'] = False
# If both fingerprints and a per-host ca file are specified, issue a warning # 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 # because users should not be surprised about what security is or isn't
# being performed. # being performed.
cafile = ui.config('hostsecurity', '%s:verifycertsfile' % hostname) cafile = ui.config('hostsecurity', '%s:verifycertsfile' % bhostname)
if s['certfingerprints'] and cafile: if s['certfingerprints'] and cafile:
ui.warn(_('(hostsecurity.%s:verifycertsfile ignored when host ' ui.warn(_('(hostsecurity.%s:verifycertsfile ignored when host '
'fingerprints defined; using host fingerprints for ' 'fingerprints defined; using host fingerprints for '
'verification)\n') % hostname) 'verification)\n') % bhostname)
# Try to hook up CA certificate validation unless something above # Try to hook up CA certificate validation unless something above
# makes it not necessary. # makes it not necessary.
if s['verifymode'] is None: if s['verifymode'] is None:
# Look at per-host ca file first. # Look at per-host ca file first.
if cafile: if cafile:
cafile = util.expandpath(cafile) cafile = util.expandpath(cafile)
if not os.path.exists(cafile): if not os.path.exists(cafile):
raise error.Abort(_('path specified by %s does not exist: %s') % raise error.Abort(_('path specified by %s does not exist: %s') %
('hostsecurity.%s:verifycertsfile' % hostname, ('hostsecurity.%s:verifycertsfile' % (
cafile)) bhostname,), cafile))
s['cafile'] = cafile s['cafile'] = cafile
else: else:
# Find global certificates file in config. # Find global certificates file in config.
cafile = ui.config('web', 'cacerts') cafile = ui.config('web', 'cacerts')
if cafile: if cafile:
cafile = util.expandpath(cafile) cafile = util.expandpath(cafile)
if not os.path.exists(cafile): if not os.path.exists(cafile):
try: try:
sslcontext.load_verify_locations(cafile=settings['cafile']) sslcontext.load_verify_locations(cafile=settings['cafile'])
except ssl.SSLError as e: except ssl.SSLError as e:
if len(e.args) == 1: # pypy has different SSLError args if len(e.args) == 1: # pypy has different SSLError args
msg = e.args[0] msg = e.args[0]
else: else:
msg = e.args[1] msg = e.args[1]
raise error.Abort(_('error loading CA file %s: %s') % ( raise error.Abort(_('error loading CA file %s: %s') % (
settings['cafile'], msg), settings['cafile'], util.forcebytestr(msg)),
hint=_('file is empty or malformed?')) hint=_('file is empty or malformed?'))
caloaded = True caloaded = True
elif settings['allowloaddefaultcerts']: elif settings['allowloaddefaultcerts']:
# This is a no-op on old Python. # This is a no-op on old Python.
sslcontext.load_default_certs() sslcontext.load_default_certs()
caloaded = True caloaded = True
else: else:
caloaded = False caloaded = False
This code is effectively copied from CPython's ssl._dnsname_match. This code is effectively copied from CPython's ssl._dnsname_match.
Returns a bool indicating whether the expected hostname matches Returns a bool indicating whether the expected hostname matches
the value in ``dn``. the value in ``dn``.
""" """
pats = [] pats = []
if not dn: if not dn:
return False return False
dn = pycompat.bytesurl(dn)
hostname = pycompat.bytesurl(hostname)
pieces = dn.split(r'.') pieces = dn.split('.')
leftmost = pieces[0] leftmost = pieces[0]
remainder = pieces[1:] remainder = pieces[1:]
wildcards = leftmost.count('*') wildcards = leftmost.count('*')
if wildcards > maxwildcards: if wildcards > maxwildcards:
raise wildcarderror( raise wildcarderror(
_('too many wildcards in certificate DNS name: %s') % dn) _('too many wildcards in certificate DNS name: %s') % dn)
# speed up common case w/o wildcards # speed up common case w/o wildcards
dnsnames = [] dnsnames = []
san = cert.get('subjectAltName', []) san = cert.get('subjectAltName', [])
for key, value in san: for key, value in san:
if key == 'DNS': if key == 'DNS':
try: try:
if _dnsnamematch(value, hostname): if _dnsnamematch(value, hostname):
return return
except wildcarderror as e: except wildcarderror as e:
return e.args[0] return util.forcebytestr(e.args[0])
dnsnames.append(value) dnsnames.append(value)
if not dnsnames: if not dnsnames:
# The subject is only checked when there is no DNS in subjectAltName. # The subject is only checked when there is no DNS in subjectAltName.
for sub in cert.get('subject', []): for sub in cert.get(r'subject', []):
for key, value in sub: for key, value in sub:
# According to RFC 2818 the most specific Common Name must # According to RFC 2818 the most specific Common Name must
# be used. # be used.
if key == 'commonName': if key == r'commonName':
# 'subject' entries are unicode. # 'subject' entries are unicode.
try: try:
value = value.encode('ascii') value = value.encode('ascii')
except UnicodeEncodeError: except UnicodeEncodeError:
return _('IDN in certificate not supported') return _('IDN in certificate not supported')
try: try:
if _dnsnamematch(value, hostname): if _dnsnamematch(value, hostname):
return return
except wildcarderror as e: except wildcarderror as e:
return e.args[0] return util.forcebytestr(e.args[0])
dnsnames.append(value) dnsnames.append(value)
if len(dnsnames) > 1: if len(dnsnames) > 1:
return _('certificate is for %s') % ', '.join(dnsnames) return _('certificate is for %s') % ', '.join(dnsnames)
elif len(dnsnames) == 1: elif len(dnsnames) == 1:
return _('certificate is for %s') % dnsnames[0] return _('certificate is for %s') % dnsnames[0]
else: else:
return None return None
def validatesocket(sock): def validatesocket(sock):
"""Validate a socket meets security requirements. """Validate a socket meets security requirements.
The passed socket must have been created with ``wrapsocket()``. The passed socket must have been created with ``wrapsocket()``.
""" """
host = sock._hgstate['hostname'] shost = sock._hgstate['hostname']
host = pycompat.bytesurl(shost)
ui = sock._hgstate['ui'] ui = sock._hgstate['ui']
settings = sock._hgstate['settings'] settings = sock._hgstate['settings']
try: try:
peercert = sock.getpeercert(True) peercert = sock.getpeercert(True)
peercert2 = sock.getpeercert() peercert2 = sock.getpeercert()
except AttributeError: except AttributeError:
raise error.Abort(_('%s ssl connection error') % host) raise error.Abort(_('%s ssl connection error') % host)
raise error.Abort( raise error.Abort(
_('unable to verify security of %s (no loaded CA certificates); ' _('unable to verify security of %s (no loaded CA certificates); '
'refusing to connect') % host, 'refusing to connect') % host,
hint=_('see https://mercurial-scm.org/wiki/SecureConnections for ' hint=_('see https://mercurial-scm.org/wiki/SecureConnections for '
'how to configure Mercurial to avoid this error or set ' 'how to configure Mercurial to avoid this error or set '
'hostsecurity.%s:fingerprints=%s to trust this server') % 'hostsecurity.%s:fingerprints=%s to trust this server') %
(host, nicefingerprint)) (host, nicefingerprint))
msg = _verifycert(peercert2, host) msg = _verifycert(peercert2, shost)
if msg: if msg:
raise error.Abort(_('%s certificate error: %s') % (host, msg), raise error.Abort(_('%s certificate error: %s') % (host, msg),
hint=_('set hostsecurity.%s:certfingerprints=%s ' hint=_('set hostsecurity.%s:certfingerprints=%s '
'config setting or use --insecure to connect ' 'config setting or use --insecure to connect '
'insecurely') % 'insecurely') %
(host, nicefingerprint)) (host, nicefingerprint))