typing: disable module attribute warnings for properly conditionalized code

Differential Revision: https://phab.mercurial-scm.org/D10208

mercurial/dispatch.py

@@ -1,1375 +1,1379 b''
 # dispatch.py - command dispatching for mercurial
 #
 # Copyright 2005-2007 Matt Mackall <[e-mail]>
 #
 # 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, print_function
 
 import errno
 import getopt
 import io
 import os
 import pdb
 import re
 import signal
 import sys
 import traceback
 
 
 from .i18n import _
 from .pycompat import getattr
 
 from hgdemandimport import tracing
 
 from . import (
     cmdutil,
     color,
     commands,
     demandimport,
     encoding,
     error,
     extensions,
     fancyopts,
     help,
     hg,
     hook,
     localrepo,
     profiling,
     pycompat,
     rcutil,
     registrar,
     requirements as requirementsmod,
     scmutil,
     ui as uimod,
     util,
     vfs,
 )
 
 from .utils import (
     procutil,
     stringutil,
 )
 
 
 class request(object):
     def __init__(
         self,
         args,
         ui=None,
         repo=None,
         fin=None,
         fout=None,
         ferr=None,
         fmsg=None,
         prereposetups=None,
     ):
         self.args = args
         self.ui = ui
         self.repo = repo
 
         # input/output/error streams
         self.fin = fin
         self.fout = fout
         self.ferr = ferr
         # separate stream for status/error messages
         self.fmsg = fmsg
 
         # remember options pre-parsed by _earlyparseopts()
         self.earlyoptions = {}
 
         # reposetups which run before extensions, useful for chg to pre-fill
         # low-level repo state (for example, changelog) before extensions.
         self.prereposetups = prereposetups or []
 
         # store the parsed and canonical command
         self.canonical_command = None
 
     def _runexithandlers(self):
         exc = None
         handlers = self.ui._exithandlers
         try:
             while handlers:
                 func, args, kwargs = handlers.pop()
                 try:
                     func(*args, **kwargs)
                 except:  # re-raises below
                     if exc is None:
                         exc = sys.exc_info()[1]
                     self.ui.warnnoi18n(b'error in exit handlers:\n')
                     self.ui.traceback(force=True)
         finally:
             if exc is not None:
                 raise exc
 
 
 def _flushstdio(ui, err):
     status = None
     # In all cases we try to flush stdio streams.
     if util.safehasattr(ui, b'fout'):
         assert ui is not None  # help pytype
         assert ui.fout is not None  # help pytype
         try:
             ui.fout.flush()
         except IOError as e:
             err = e
             status = -1
 
     if util.safehasattr(ui, b'ferr'):
         assert ui is not None  # help pytype
         assert ui.ferr is not None  # help pytype
         try:
             if err is not None and err.errno != errno.EPIPE:
                 ui.ferr.write(
                     b'abort: %s\n' % encoding.strtolocal(err.strerror)
                 )
             ui.ferr.flush()
         # There's not much we can do about an I/O error here. So (possibly)
         # change the status code and move on.
         except IOError:
             status = -1
 
     return status
 
 
 def run():
     """run the command in sys.argv"""
     try:
         initstdio()
         with tracing.log('parse args into request'):
             req = request(pycompat.sysargv[1:])
 
         status = dispatch(req)
         _silencestdio()
     except KeyboardInterrupt:
         # Catch early/late KeyboardInterrupt as last ditch. Here nothing will
         # be printed to console to avoid another IOError/KeyboardInterrupt.
         status = -1
     sys.exit(status & 255)
 
 
 if pycompat.ispy3:
 
     def initstdio():
         # stdio streams on Python 3 are io.TextIOWrapper instances proxying another
         # buffer. These streams will normalize \n to \r\n by default. Mercurial's
         # preferred mechanism for writing output (ui.write()) uses io.BufferedWriter
         # instances, which write to the underlying stdio file descriptor in binary
         # mode. ui.write() uses \n for line endings and no line ending normalization
         # is attempted through this interface. This "just works," even if the system
         # preferred line ending is not \n.
         #
         # But some parts of Mercurial (e.g. hooks) can still send data to sys.stdout
         # and sys.stderr. They will inherit the line ending normalization settings,
         # potentially causing e.g. \r\n to be emitted. Since emitting \n should
         # "just work," here we change the sys.* streams to disable line ending
         # normalization, ensuring compatibility with our ui type.
 
         if sys.stdout is not None:
             # write_through is new in Python 3.7.
             kwargs = {
                 "newline": "\n",
                 "line_buffering": sys.stdout.line_buffering,
             }
             if util.safehasattr(sys.stdout, "write_through"):
+                # pytype: disable=attribute-error
                 kwargs["write_through"] = sys.stdout.write_through
+                # pytype: enable=attribute-error
             sys.stdout = io.TextIOWrapper(
                 sys.stdout.buffer,
                 sys.stdout.encoding,
                 sys.stdout.errors,
                 **kwargs
             )
 
         if sys.stderr is not None:
             kwargs = {
                 "newline": "\n",
                 "line_buffering": sys.stderr.line_buffering,
             }
             if util.safehasattr(sys.stderr, "write_through"):
+                # pytype: disable=attribute-error
                 kwargs["write_through"] = sys.stderr.write_through
+                # pytype: enable=attribute-error
             sys.stderr = io.TextIOWrapper(
                 sys.stderr.buffer,
                 sys.stderr.encoding,
                 sys.stderr.errors,
                 **kwargs
             )
 
         if sys.stdin is not None:
             # No write_through on read-only stream.
             sys.stdin = io.TextIOWrapper(
                 sys.stdin.buffer,
                 sys.stdin.encoding,
                 sys.stdin.errors,
                 # None is universal newlines mode.
                 newline=None,
                 line_buffering=sys.stdin.line_buffering,
             )
 
     def _silencestdio():
         for fp in (sys.stdout, sys.stderr):
             if fp is None:
                 continue
             # Check if the file is okay
             try:
                 fp.flush()
                 continue
             except IOError:
                 pass
             # Otherwise mark it as closed to silence "Exception ignored in"
             # message emitted by the interpreter finalizer.
             try:
                 fp.close()
             except IOError:
                 pass
 
 
 else:
 
     def initstdio():
         for fp in (sys.stdin, sys.stdout, sys.stderr):
             procutil.setbinary(fp)
 
     def _silencestdio():
         pass
 
 
 def _formatargs(args):
     return b' '.join(procutil.shellquote(a) for a in args)
 
 
 def dispatch(req):
     """run the command specified in req.args; returns an integer status code"""
     err = None
     try:
         status = _rundispatch(req)
     except error.StdioError as e:
         err = e
         status = -1
 
     ret = _flushstdio(req.ui, err)
     if ret:
         status = ret
     return status
 
 
 def _rundispatch(req):
     with tracing.log('dispatch._rundispatch'):
         if req.ferr:
             ferr = req.ferr
         elif req.ui:
             ferr = req.ui.ferr
         else:
             ferr = procutil.stderr
 
         try:
             if not req.ui:
                 req.ui = uimod.ui.load()
             req.earlyoptions.update(_earlyparseopts(req.ui, req.args))
             if req.earlyoptions[b'traceback']:
                 req.ui.setconfig(b'ui', b'traceback', b'on', b'--traceback')
 
             # set ui streams from the request
             if req.fin:
                 req.ui.fin = req.fin
             if req.fout:
                 req.ui.fout = req.fout
             if req.ferr:
                 req.ui.ferr = req.ferr
             if req.fmsg:
                 req.ui.fmsg = req.fmsg
         except error.Abort as inst:
             ferr.write(inst.format())
             return -1
 
         msg = _formatargs(req.args)
         starttime = util.timer()
         ret = 1  # default of Python exit code on unhandled exception
         try:
             ret = _runcatch(req) or 0
         except error.ProgrammingError as inst:
             req.ui.error(_(b'** ProgrammingError: %s\n') % inst)
             if inst.hint:
                 req.ui.error(_(b'** (%s)\n') % inst.hint)
             raise
         except KeyboardInterrupt as inst:
             try:
                 if isinstance(inst, error.SignalInterrupt):
                     msg = _(b"killed!\n")
                 else:
                     msg = _(b"interrupted!\n")
                 req.ui.error(msg)
             except error.SignalInterrupt:
                 # maybe pager would quit without consuming all the output, and
                 # SIGPIPE was raised. we cannot print anything in this case.
                 pass
             except IOError as inst:
                 if inst.errno != errno.EPIPE:
                     raise
             ret = -1
         finally:
             duration = util.timer() - starttime
             req.ui.flush()  # record blocked times
             if req.ui.logblockedtimes:
                 req.ui._blockedtimes[b'command_duration'] = duration * 1000
                 req.ui.log(
                     b'uiblocked',
                     b'ui blocked ms\n',
                     **pycompat.strkwargs(req.ui._blockedtimes)
                 )
             return_code = ret & 255
             req.ui.log(
                 b"commandfinish",
                 b"%s exited %d after %0.2f seconds\n",
                 msg,
                 return_code,
                 duration,
                 return_code=return_code,
                 duration=duration,
                 canonical_command=req.canonical_command,
             )
             try:
                 req._runexithandlers()
             except:  # exiting, so no re-raises
                 ret = ret or -1
             # do flush again since ui.log() and exit handlers may write to ui
             req.ui.flush()
         return ret
 
 
 def _runcatch(req):
     with tracing.log('dispatch._runcatch'):
 
         def catchterm(*args):
             raise error.SignalInterrupt
 
         ui = req.ui
         try:
             for name in b'SIGBREAK', b'SIGHUP', b'SIGTERM':
                 num = getattr(signal, name, None)
                 if num:
                     signal.signal(num, catchterm)
         except ValueError:
             pass  # happens if called in a thread
 
         def _runcatchfunc():
             realcmd = None
             try:
                 cmdargs = fancyopts.fancyopts(
                     req.args[:], commands.globalopts, {}
                 )
                 cmd = cmdargs[0]
                 aliases, entry = cmdutil.findcmd(cmd, commands.table, False)
                 realcmd = aliases[0]
             except (
                 error.UnknownCommand,
                 error.AmbiguousCommand,
                 IndexError,
                 getopt.GetoptError,
             ):
                 # Don't handle this here. We know the command is
                 # invalid, but all we're worried about for now is that
                 # it's not a command that server operators expect to
                 # be safe to offer to users in a sandbox.
                 pass
             if realcmd == b'serve' and b'--stdio' in cmdargs:
                 # We want to constrain 'hg serve --stdio' instances pretty
                 # closely, as many shared-ssh access tools want to grant
                 # access to run *only* 'hg -R $repo serve --stdio'. We
                 # restrict to exactly that set of arguments, and prohibit
                 # any repo name that starts with '--' to prevent
                 # shenanigans wherein a user does something like pass
                 # --debugger or --config=ui.debugger=1 as a repo
                 # name. This used to actually run the debugger.
                 if (
                     len(req.args) != 4
                     or req.args[0] != b'-R'
                     or req.args[1].startswith(b'--')
                     or req.args[2] != b'serve'
                     or req.args[3] != b'--stdio'
                 ):
                     raise error.Abort(
                         _(b'potentially unsafe serve --stdio invocation: %s')
                         % (stringutil.pprint(req.args),)
                     )
 
             try:
                 debugger = b'pdb'
                 debugtrace = {b'pdb': pdb.set_trace}
                 debugmortem = {b'pdb': pdb.post_mortem}
 
                 # read --config before doing anything else
                 # (e.g. to change trust settings for reading .hg/hgrc)
                 cfgs = _parseconfig(req.ui, req.earlyoptions[b'config'])
 
                 if req.repo:
                     # copy configs that were passed on the cmdline (--config) to
                     # the repo ui
                     for sec, name, val in cfgs:
                         req.repo.ui.setconfig(
                             sec, name, val, source=b'--config'
                         )
 
                 # developer config: ui.debugger
                 debugger = ui.config(b"ui", b"debugger")
                 debugmod = pdb
                 if not debugger or ui.plain():
                     # if we are in HGPLAIN mode, then disable custom debugging
                     debugger = b'pdb'
                 elif req.earlyoptions[b'debugger']:
                     # This import can be slow for fancy debuggers, so only
                     # do it when absolutely necessary, i.e. when actual
                     # debugging has been requested
                     with demandimport.deactivated():
                         try:
                             debugmod = __import__(debugger)
                         except ImportError:
                             pass  # Leave debugmod = pdb
 
                 debugtrace[debugger] = debugmod.set_trace
                 debugmortem[debugger] = debugmod.post_mortem
 
                 # enter the debugger before command execution
                 if req.earlyoptions[b'debugger']:
                     ui.warn(
                         _(
                             b"entering debugger - "
                             b"type c to continue starting hg or h for help\n"
                         )
                     )
 
                     if (
                         debugger != b'pdb'
                         and debugtrace[debugger] == debugtrace[b'pdb']
                     ):
                         ui.warn(
                             _(
                                 b"%s debugger specified "
                                 b"but its module was not found\n"
                             )
                             % debugger
                         )
                     with demandimport.deactivated():
                         debugtrace[debugger]()
                 try:
                     return _dispatch(req)
                 finally:
                     ui.flush()
             except:  # re-raises
                 # enter the debugger when we hit an exception
                 if req.earlyoptions[b'debugger']:
                     traceback.print_exc()
                     debugmortem[debugger](sys.exc_info()[2])
                 raise
 
         return _callcatch(ui, _runcatchfunc)
 
 
 def _callcatch(ui, func):
     """like scmutil.callcatch but handles more high-level exceptions about
     config parsing and commands. besides, use handlecommandexception to handle
     uncaught exceptions.
     """
     detailed_exit_code = -1
     try:
         return scmutil.callcatch(ui, func)
     except error.AmbiguousCommand as inst:
         detailed_exit_code = 10
         ui.warn(
             _(b"hg: command '%s' is ambiguous:\n    %s\n")
             % (inst.prefix, b" ".join(inst.matches))
         )
     except error.CommandError as inst:
         detailed_exit_code = 10
         if inst.command:
             ui.pager(b'help')
             msgbytes = pycompat.bytestr(inst.message)
             ui.warn(_(b"hg %s: %s\n") % (inst.command, msgbytes))
             commands.help_(ui, inst.command, full=False, command=True)
         else:
             ui.warn(_(b"hg: %s\n") % inst.message)
             ui.warn(_(b"(use 'hg help -v' for a list of global options)\n"))
     except error.UnknownCommand as inst:
         detailed_exit_code = 10
         nocmdmsg = _(b"hg: unknown command '%s'\n") % inst.command
         try:
             # check if the command is in a disabled extension
             # (but don't check for extensions themselves)
             formatted = help.formattedhelp(
                 ui, commands, inst.command, unknowncmd=True
             )
             ui.warn(nocmdmsg)
             ui.write(formatted)
         except (error.UnknownCommand, error.Abort):
             suggested = False
             if inst.all_commands:
                 sim = error.getsimilar(inst.all_commands, inst.command)
                 if sim:
                     ui.warn(nocmdmsg)
                     ui.warn(b"(%s)\n" % error.similarity_hint(sim))
                     suggested = True
             if not suggested:
                 ui.warn(nocmdmsg)
                 ui.warn(_(b"(use 'hg help' for a list of commands)\n"))
     except IOError:
         raise
     except KeyboardInterrupt:
         raise
     except:  # probably re-raises
         if not handlecommandexception(ui):
             raise
 
     if ui.configbool(b'ui', b'detailed-exit-code'):
         return detailed_exit_code
     else:
         return -1
 
 
 def aliasargs(fn, givenargs):
     args = []
     # only care about alias 'args', ignore 'args' set by extensions.wrapfunction
     if not util.safehasattr(fn, b'_origfunc'):
         args = getattr(fn, 'args', args)
     if args:
         cmd = b' '.join(map(procutil.shellquote, args))
 
         nums = []
 
         def replacer(m):
             num = int(m.group(1)) - 1
             nums.append(num)
             if num < len(givenargs):
                 return givenargs[num]
             raise error.InputError(_(b'too few arguments for command alias'))
 
         cmd = re.sub(br'\$(\d+|\$)', replacer, cmd)
         givenargs = [x for i, x in enumerate(givenargs) if i not in nums]
         args = pycompat.shlexsplit(cmd)
     return args + givenargs
 
 
 def aliasinterpolate(name, args, cmd):
     """interpolate args into cmd for shell aliases
 
     This also handles $0, $@ and "$@".
     """
     # util.interpolate can't deal with "$@" (with quotes) because it's only
     # built to match prefix + patterns.
     replacemap = {b'$%d' % (i + 1): arg for i, arg in enumerate(args)}
     replacemap[b'$0'] = name
     replacemap[b'$$'] = b'$'
     replacemap[b'$@'] = b' '.join(args)
     # Typical Unix shells interpolate "$@" (with quotes) as all the positional
     # parameters, separated out into words. Emulate the same behavior here by
     # quoting the arguments individually. POSIX shells will then typically
     # tokenize each argument into exactly one word.
     replacemap[b'"$@"'] = b' '.join(procutil.shellquote(arg) for arg in args)
     # escape '\$' for regex
     regex = b'|'.join(replacemap.keys()).replace(b'$', br'\$')
     r = re.compile(regex)
     return r.sub(lambda x: replacemap[x.group()], cmd)
 
 
 class cmdalias(object):
     def __init__(self, ui, name, definition, cmdtable, source):
         self.name = self.cmd = name
         self.cmdname = b''
         self.definition = definition
         self.fn = None
         self.givenargs = []
         self.opts = []
         self.help = b''
         self.badalias = None
         self.unknowncmd = False
         self.source = source
 
         try:
             aliases, entry = cmdutil.findcmd(self.name, cmdtable)
             for alias, e in pycompat.iteritems(cmdtable):
                 if e is entry:
                     self.cmd = alias
                     break
             self.shadows = True
         except error.UnknownCommand:
             self.shadows = False
 
         if not self.definition:
             self.badalias = _(b"no definition for alias '%s'") % self.name
             return
 
         if self.definition.startswith(b'!'):
             shdef = self.definition[1:]
             self.shell = True
 
             def fn(ui, *args):
                 env = {b'HG_ARGS': b' '.join((self.name,) + args)}
 
                 def _checkvar(m):
                     if m.groups()[0] == b'$':
                         return m.group()
                     elif int(m.groups()[0]) <= len(args):
                         return m.group()
                     else:
                         ui.debug(
                             b"No argument found for substitution "
                             b"of %i variable in alias '%s' definition.\n"
                             % (int(m.groups()[0]), self.name)
                         )
                         return b''
 
                 cmd = re.sub(br'\$(\d+|\$)', _checkvar, shdef)
                 cmd = aliasinterpolate(self.name, args, cmd)
                 return ui.system(
                     cmd, environ=env, blockedtag=b'alias_%s' % self.name
                 )
 
             self.fn = fn
             self.alias = True
             self._populatehelp(ui, name, shdef, self.fn)
             return
 
         try:
             args = pycompat.shlexsplit(self.definition)
         except ValueError as inst:
             self.badalias = _(b"error in definition for alias '%s': %s") % (
                 self.name,
                 stringutil.forcebytestr(inst),
             )
             return
         earlyopts, args = _earlysplitopts(args)
         if earlyopts:
             self.badalias = _(
                 b"error in definition for alias '%s': %s may "
                 b"only be given on the command line"
             ) % (self.name, b'/'.join(pycompat.ziplist(*earlyopts)[0]))
             return
         self.cmdname = cmd = args.pop(0)
         self.givenargs = args
 
         try:
             tableentry = cmdutil.findcmd(cmd, cmdtable, False)[1]
             if len(tableentry) > 2:
                 self.fn, self.opts, cmdhelp = tableentry
             else:
                 self.fn, self.opts = tableentry
                 cmdhelp = None
 
             self.alias = True
             self._populatehelp(ui, name, cmd, self.fn, cmdhelp)
 
         except error.UnknownCommand:
             self.badalias = _(
                 b"alias '%s' resolves to unknown command '%s'"
             ) % (
                 self.name,
                 cmd,
             )
             self.unknowncmd = True
         except error.AmbiguousCommand:
             self.badalias = _(
                 b"alias '%s' resolves to ambiguous command '%s'"
             ) % (
                 self.name,
                 cmd,
             )
 
     def _populatehelp(self, ui, name, cmd, fn, defaulthelp=None):
         # confine strings to be passed to i18n.gettext()
         cfg = {}
         for k in (b'doc', b'help', b'category'):
             v = ui.config(b'alias', b'%s:%s' % (name, k), None)
             if v is None:
                 continue
             if not encoding.isasciistr(v):
                 self.badalias = _(
                     b"non-ASCII character in alias definition '%s:%s'"
                 ) % (name, k)
                 return
             cfg[k] = v
 
         self.help = cfg.get(b'help', defaulthelp or b'')
         if self.help and self.help.startswith(b"hg " + cmd):
             # drop prefix in old-style help lines so hg shows the alias
             self.help = self.help[4 + len(cmd) :]
 
         self.owndoc = b'doc' in cfg
         doc = cfg.get(b'doc', pycompat.getdoc(fn))
         if doc is not None:
             doc = pycompat.sysstr(doc)
         self.__doc__ = doc
 
         self.helpcategory = cfg.get(
             b'category', registrar.command.CATEGORY_NONE
         )
 
     @property
     def args(self):
         args = pycompat.maplist(util.expandpath, self.givenargs)
         return aliasargs(self.fn, args)
 
     def __getattr__(self, name):
         adefaults = {
             'norepo': True,
             'intents': set(),
             'optionalrepo': False,
             'inferrepo': False,
         }
         if name not in adefaults:
             raise AttributeError(name)
         if self.badalias or util.safehasattr(self, b'shell'):
             return adefaults[name]
         return getattr(self.fn, name)
 
     def __call__(self, ui, *args, **opts):
         if self.badalias:
             hint = None
             if self.unknowncmd:
                 try:
                     # check if the command is in a disabled extension
                     cmd, ext = extensions.disabledcmd(ui, self.cmdname)[:2]
                     hint = _(b"'%s' is provided by '%s' extension") % (cmd, ext)
                 except error.UnknownCommand:
                     pass
             raise error.ConfigError(self.badalias, hint=hint)
         if self.shadows:
             ui.debug(
                 b"alias '%s' shadows command '%s'\n" % (self.name, self.cmdname)
             )
 
         ui.log(
             b'commandalias',
             b"alias '%s' expands to '%s'\n",
             self.name,
             self.definition,
         )
         if util.safehasattr(self, b'shell'):
             return self.fn(ui, *args, **opts)
         else:
             try:
                 return util.checksignature(self.fn)(ui, *args, **opts)
             except error.SignatureError:
                 args = b' '.join([self.cmdname] + self.args)
                 ui.debug(b"alias '%s' expands to '%s'\n" % (self.name, args))
                 raise
 
 
 class lazyaliasentry(object):
     """like a typical command entry (func, opts, help), but is lazy"""
 
     def __init__(self, ui, name, definition, cmdtable, source):
         self.ui = ui
         self.name = name
         self.definition = definition
         self.cmdtable = cmdtable.copy()
         self.source = source
         self.alias = True
 
     @util.propertycache
     def _aliasdef(self):
         return cmdalias(
             self.ui, self.name, self.definition, self.cmdtable, self.source
         )
 
     def __getitem__(self, n):
         aliasdef = self._aliasdef
         if n == 0:
             return aliasdef
         elif n == 1:
             return aliasdef.opts
         elif n == 2:
             return aliasdef.help
         else:
             raise IndexError
 
     def __iter__(self):
         for i in range(3):
             yield self[i]
 
     def __len__(self):
         return 3
 
 
 def addaliases(ui, cmdtable):
     # aliases are processed after extensions have been loaded, so they
     # may use extension commands. Aliases can also use other alias definitions,
     # but only if they have been defined prior to the current definition.
     for alias, definition in ui.configitems(b'alias', ignoresub=True):
         try:
             if cmdtable[alias].definition == definition:
                 continue
         except (KeyError, AttributeError):
             # definition might not exist or it might not be a cmdalias
             pass
 
         source = ui.configsource(b'alias', alias)
         entry = lazyaliasentry(ui, alias, definition, cmdtable, source)
         cmdtable[alias] = entry
 
 
 def _parse(ui, args):
     options = {}
     cmdoptions = {}
 
     try:
         args = fancyopts.fancyopts(args, commands.globalopts, options)
     except getopt.GetoptError as inst:
         raise error.CommandError(None, stringutil.forcebytestr(inst))
 
     if args:
         cmd, args = args[0], args[1:]
         aliases, entry = cmdutil.findcmd(
             cmd, commands.table, ui.configbool(b"ui", b"strict")
         )
         cmd = aliases[0]
         args = aliasargs(entry[0], args)
         defaults = ui.config(b"defaults", cmd)
         if defaults:
             args = (
                 pycompat.maplist(util.expandpath, pycompat.shlexsplit(defaults))
                 + args
             )
         c = list(entry[1])
     else:
         cmd = None
         c = []
 
     # combine global options into local
     for o in commands.globalopts:
         c.append((o[0], o[1], options[o[1]], o[3]))
 
     try:
         args = fancyopts.fancyopts(args, c, cmdoptions, gnu=True)
     except getopt.GetoptError as inst:
         raise error.CommandError(cmd, stringutil.forcebytestr(inst))
 
     # separate global options back out
     for o in commands.globalopts:
         n = o[1]
         options[n] = cmdoptions[n]
         del cmdoptions[n]
 
     return (cmd, cmd and entry[0] or None, args, options, cmdoptions)
 
 
 def _parseconfig(ui, config):
     """parse the --config options from the command line"""
     configs = []
 
     for cfg in config:
         try:
             name, value = [cfgelem.strip() for cfgelem in cfg.split(b'=', 1)]
             section, name = name.split(b'.', 1)
             if not section or not name:
                 raise IndexError
             ui.setconfig(section, name, value, b'--config')
             configs.append((section, name, value))
         except (IndexError, ValueError):
             raise error.InputError(
                 _(
                     b'malformed --config option: %r '
                     b'(use --config section.name=value)'
                 )
                 % pycompat.bytestr(cfg)
             )
 
     return configs
 
 
 def _earlyparseopts(ui, args):
     options = {}
     fancyopts.fancyopts(
         args,
         commands.globalopts,
         options,
         gnu=not ui.plain(b'strictflags'),
         early=True,
         optaliases={b'repository': [b'repo']},
     )
     return options
 
 
 def _earlysplitopts(args):
     """Split args into a list of possible early options and remainder args"""
     shortoptions = b'R:'
     # TODO: perhaps 'debugger' should be included
     longoptions = [b'cwd=', b'repository=', b'repo=', b'config=']
     return fancyopts.earlygetopt(
         args, shortoptions, longoptions, gnu=True, keepsep=True
     )
 
 
 def runcommand(lui, repo, cmd, fullargs, ui, options, d, cmdpats, cmdoptions):
     # run pre-hook, and abort if it fails
     hook.hook(
         lui,
         repo,
         b"pre-%s" % cmd,
         True,
         args=b" ".join(fullargs),
         pats=cmdpats,
         opts=cmdoptions,
     )
     try:
         ret = _runcommand(ui, options, cmd, d)
         # run post-hook, passing command result
         hook.hook(
             lui,
             repo,
             b"post-%s" % cmd,
             False,
             args=b" ".join(fullargs),
             result=ret,
             pats=cmdpats,
             opts=cmdoptions,
         )
     except Exception:
         # run failure hook and re-raise
         hook.hook(
             lui,
             repo,
             b"fail-%s" % cmd,
             False,
             args=b" ".join(fullargs),
             pats=cmdpats,
             opts=cmdoptions,
         )
         raise
     return ret
 
 
 def _readsharedsourceconfig(ui, path):
     """if the current repository is shared one, this tries to read
     .hg/hgrc of shared source if we are in share-safe mode
 
     Config read is loaded into the ui object passed
 
     This should be called before reading .hg/hgrc or the main repo
     as that overrides config set in shared source"""
     try:
         with open(os.path.join(path, b".hg", b"requires"), "rb") as fp:
             requirements = set(fp.read().splitlines())
             if not (
                 requirementsmod.SHARESAFE_REQUIREMENT in requirements
                 and requirementsmod.SHARED_REQUIREMENT in requirements
             ):
                 return
             hgvfs = vfs.vfs(os.path.join(path, b".hg"))
             sharedvfs = localrepo._getsharedvfs(hgvfs, requirements)
             root = sharedvfs.base
             ui.readconfig(sharedvfs.join(b"hgrc"), root)
     except IOError:
         pass
 
 
 def _getlocal(ui, rpath, wd=None):
     """Return (path, local ui object) for the given target path.
 
     Takes paths in [cwd]/.hg/hgrc into account."
     """
     if wd is None:
         try:
             wd = encoding.getcwd()
         except OSError as e:
             raise error.Abort(
                 _(b"error getting current working directory: %s")
                 % encoding.strtolocal(e.strerror)
             )
 
     path = cmdutil.findrepo(wd) or b""
     if not path:
         lui = ui
     else:
         lui = ui.copy()
         if rcutil.use_repo_hgrc():
             _readsharedsourceconfig(lui, path)
             lui.readconfig(os.path.join(path, b".hg", b"hgrc"), path)
             lui.readconfig(os.path.join(path, b".hg", b"hgrc-not-shared"), path)
 
     if rpath:
         path = lui.expandpath(rpath)
         lui = ui.copy()
         if rcutil.use_repo_hgrc():
             _readsharedsourceconfig(lui, path)
             lui.readconfig(os.path.join(path, b".hg", b"hgrc"), path)
             lui.readconfig(os.path.join(path, b".hg", b"hgrc-not-shared"), path)
 
     return path, lui
 
 
 def _checkshellalias(lui, ui, args):
     """Return the function to run the shell alias, if it is required"""
     options = {}
 
     try:
         args = fancyopts.fancyopts(args, commands.globalopts, options)
     except getopt.GetoptError:
         return
 
     if not args:
         return
 
     cmdtable = commands.table
 
     cmd = args[0]
     try:
         strict = ui.configbool(b"ui", b"strict")
         aliases, entry = cmdutil.findcmd(cmd, cmdtable, strict)
     except (error.AmbiguousCommand, error.UnknownCommand):
         return
 
     cmd = aliases[0]
     fn = entry[0]
 
     if cmd and util.safehasattr(fn, b'shell'):
         # shell alias shouldn't receive early options which are consumed by hg
         _earlyopts, args = _earlysplitopts(args)
         d = lambda: fn(ui, *args[1:])
         return lambda: runcommand(
             lui, None, cmd, args[:1], ui, options, d, [], {}
         )
 
 
 def _dispatch(req):
     args = req.args
     ui = req.ui
 
     # check for cwd
     cwd = req.earlyoptions[b'cwd']
     if cwd:
         os.chdir(cwd)
 
     rpath = req.earlyoptions[b'repository']
     path, lui = _getlocal(ui, rpath)
 
     uis = {ui, lui}
 
     if req.repo:
         uis.add(req.repo.ui)
 
     if (
         req.earlyoptions[b'verbose']
         or req.earlyoptions[b'debug']
         or req.earlyoptions[b'quiet']
     ):
         for opt in (b'verbose', b'debug', b'quiet'):
             val = pycompat.bytestr(bool(req.earlyoptions[opt]))
             for ui_ in uis:
                 ui_.setconfig(b'ui', opt, val, b'--' + opt)
 
     if req.earlyoptions[b'profile']:
         for ui_ in uis:
             ui_.setconfig(b'profiling', b'enabled', b'true', b'--profile')
 
     profile = lui.configbool(b'profiling', b'enabled')
     with profiling.profile(lui, enabled=profile) as profiler:
         # Configure extensions in phases: uisetup, extsetup, cmdtable, and
         # reposetup
         extensions.loadall(lui)
         # Propagate any changes to lui.__class__ by extensions
         ui.__class__ = lui.__class__
 
         # (uisetup and extsetup are handled in extensions.loadall)
 
         # (reposetup is handled in hg.repository)
 
         addaliases(lui, commands.table)
 
         # All aliases and commands are completely defined, now.
         # Check abbreviation/ambiguity of shell alias.
         shellaliasfn = _checkshellalias(lui, ui, args)
         if shellaliasfn:
             # no additional configs will be set, set up the ui instances
             for ui_ in uis:
                 extensions.populateui(ui_)
             return shellaliasfn()
 
         # check for fallback encoding
         fallback = lui.config(b'ui', b'fallbackencoding')
         if fallback:
             encoding.fallbackencoding = fallback
 
         fullargs = args
         cmd, func, args, options, cmdoptions = _parse(lui, args)
 
         # store the canonical command name in request object for later access
         req.canonical_command = cmd
 
         if options[b"config"] != req.earlyoptions[b"config"]:
             raise error.InputError(_(b"option --config may not be abbreviated"))
         if options[b"cwd"] != req.earlyoptions[b"cwd"]:
             raise error.InputError(_(b"option --cwd may not be abbreviated"))
         if options[b"repository"] != req.earlyoptions[b"repository"]:
             raise error.InputError(
                 _(
                     b"option -R has to be separated from other options (e.g. not "
                     b"-qR) and --repository may only be abbreviated as --repo"
                 )
             )
         if options[b"debugger"] != req.earlyoptions[b"debugger"]:
             raise error.InputError(
                 _(b"option --debugger may not be abbreviated")
             )
         # don't validate --profile/--traceback, which can be enabled from now
 
         if options[b"encoding"]:
             encoding.encoding = options[b"encoding"]
         if options[b"encodingmode"]:
             encoding.encodingmode = options[b"encodingmode"]
         if options[b"time"]:
 
             def get_times():
                 t = os.times()
                 if t[4] == 0.0:
                     # Windows leaves this as zero, so use time.perf_counter()
                     t = (t[0], t[1], t[2], t[3], util.timer())
                 return t
 
             s = get_times()
 
             def print_time():
                 t = get_times()
                 ui.warn(
                     _(b"time: real %.3f secs (user %.3f+%.3f sys %.3f+%.3f)\n")
                     % (
                         t[4] - s[4],
                         t[0] - s[0],
                         t[2] - s[2],
                         t[1] - s[1],
                         t[3] - s[3],
                     )
                 )
 
             ui.atexit(print_time)
         if options[b"profile"]:
             profiler.start()
 
         # if abbreviated version of this were used, take them in account, now
         if options[b'verbose'] or options[b'debug'] or options[b'quiet']:
             for opt in (b'verbose', b'debug', b'quiet'):
                 if options[opt] == req.earlyoptions[opt]:
                     continue
                 val = pycompat.bytestr(bool(options[opt]))
                 for ui_ in uis:
                     ui_.setconfig(b'ui', opt, val, b'--' + opt)
 
         if options[b'traceback']:
             for ui_ in uis:
                 ui_.setconfig(b'ui', b'traceback', b'on', b'--traceback')
 
         if options[b'noninteractive']:
             for ui_ in uis:
                 ui_.setconfig(b'ui', b'interactive', b'off', b'-y')
 
         if cmdoptions.get(b'insecure', False):
             for ui_ in uis:
                 ui_.insecureconnections = True
 
         # setup color handling before pager, because setting up pager
         # might cause incorrect console information
         coloropt = options[b'color']
         for ui_ in uis:
             if coloropt:
                 ui_.setconfig(b'ui', b'color', coloropt, b'--color')
             color.setup(ui_)
 
         if stringutil.parsebool(options[b'pager']):
             # ui.pager() expects 'internal-always-' prefix in this case
             ui.pager(b'internal-always-' + cmd)
         elif options[b'pager'] != b'auto':
             for ui_ in uis:
                 ui_.disablepager()
 
         # configs are fully loaded, set up the ui instances
         for ui_ in uis:
             extensions.populateui(ui_)
 
         if options[b'version']:
             return commands.version_(ui)
         if options[b'help']:
             return commands.help_(ui, cmd, command=cmd is not None)
         elif not cmd:
             return commands.help_(ui, b'shortlist')
 
         repo = None
         cmdpats = args[:]
         assert func is not None  # help out pytype
         if not func.norepo:
             # use the repo from the request only if we don't have -R
             if not rpath and not cwd:
                 repo = req.repo
 
             if repo:
                 # set the descriptors of the repo ui to those of ui
                 repo.ui.fin = ui.fin
                 repo.ui.fout = ui.fout
                 repo.ui.ferr = ui.ferr
                 repo.ui.fmsg = ui.fmsg
             else:
                 try:
                     repo = hg.repository(
                         ui,
                         path=path,
                         presetupfuncs=req.prereposetups,
                         intents=func.intents,
                     )
                     if not repo.local():
                         raise error.InputError(
                             _(b"repository '%s' is not local") % path
                         )
                     repo.ui.setconfig(
                         b"bundle", b"mainreporoot", repo.root, b'repo'
                     )
                 except error.RequirementError:
                     raise
                 except error.RepoError:
                     if rpath:  # invalid -R path
                         raise
                     if not func.optionalrepo:
                         if func.inferrepo and args and not path:
                             # try to infer -R from command args
                             repos = pycompat.maplist(cmdutil.findrepo, args)
                             guess = repos[0]
                             if guess and repos.count(guess) == len(repos):
                                 req.args = [b'--repository', guess] + fullargs
                                 req.earlyoptions[b'repository'] = guess
                                 return _dispatch(req)
                         if not path:
                             raise error.InputError(
                                 _(
                                     b"no repository found in"
                                     b" '%s' (.hg not found)"
                                 )
                                 % encoding.getcwd()
                             )
                         raise
             if repo:
                 ui = repo.ui
                 if options[b'hidden']:
                     repo = repo.unfiltered()
             args.insert(0, repo)
         elif rpath:
             ui.warn(_(b"warning: --repository ignored\n"))
 
         msg = _formatargs(fullargs)
         ui.log(b"command", b'%s\n', msg)
         strcmdopt = pycompat.strkwargs(cmdoptions)
         d = lambda: util.checksignature(func)(ui, *args, **strcmdopt)
         try:
             return runcommand(
                 lui, repo, cmd, fullargs, ui, options, d, cmdpats, cmdoptions
             )
         finally:
             if repo and repo != req.repo:
                 repo.close()
 
 
 def _runcommand(ui, options, cmd, cmdfunc):
     """Run a command function, possibly with profiling enabled."""
     try:
         with tracing.log("Running %s command" % cmd):
             return cmdfunc()
     except error.SignatureError:
         raise error.CommandError(cmd, _(b'invalid arguments'))
 
 
 def _exceptionwarning(ui):
     """Produce a warning message for the current active exception"""
 
     # For compatibility checking, we discard the portion of the hg
     # version after the + on the assumption that if a "normal
     # user" is running a build with a + in it the packager
     # probably built from fairly close to a tag and anyone with a
     # 'make local' copy of hg (where the version number can be out
     # of date) will be clueful enough to notice the implausible
     # version number and try updating.
     ct = util.versiontuple(n=2)
     worst = None, ct, b'', b''
     if ui.config(b'ui', b'supportcontact') is None:
         for name, mod in extensions.extensions():
             # 'testedwith' should be bytes, but not all extensions are ported
             # to py3 and we don't want UnicodeException because of that.
             testedwith = stringutil.forcebytestr(
                 getattr(mod, 'testedwith', b'')
             )
             version = extensions.moduleversion(mod)
             report = getattr(mod, 'buglink', _(b'the extension author.'))
             if not testedwith.strip():
                 # We found an untested extension. It's likely the culprit.
                 worst = name, b'unknown', report, version
                 break
 
             # Never blame on extensions bundled with Mercurial.
             if extensions.ismoduleinternal(mod):
                 continue
 
             tested = [util.versiontuple(t, 2) for t in testedwith.split()]
             if ct in tested:
                 continue
 
             lower = [t for t in tested if t < ct]
             nearest = max(lower or tested)
             if worst[0] is None or nearest < worst[1]:
                 worst = name, nearest, report, version
     if worst[0] is not None:
         name, testedwith, report, version = worst
         if not isinstance(testedwith, (bytes, str)):
             testedwith = b'.'.join(
                 [stringutil.forcebytestr(c) for c in testedwith]
             )
         extver = version or _(b"(version N/A)")
         warning = _(
             b'** Unknown exception encountered with '
             b'possibly-broken third-party extension "%s" %s\n'
             b'** which supports versions %s of Mercurial.\n'
             b'** Please disable "%s" and try your action again.\n'
             b'** If that fixes the bug please report it to %s\n'
         ) % (name, extver, testedwith, name, stringutil.forcebytestr(report))
     else:
         bugtracker = ui.config(b'ui', b'supportcontact')
         if bugtracker is None:
             bugtracker = _(b"https://mercurial-scm.org/wiki/BugTracker")
         warning = (
             _(
                 b"** unknown exception encountered, "
                 b"please report by visiting\n** "
             )
             + bugtracker
             + b'\n'
         )
     sysversion = pycompat.sysbytes(sys.version).replace(b'\n', b'')
 
     def ext_with_ver(x):
         ext = x[0]
         ver = extensions.moduleversion(x[1])
         if ver:
             ext += b' ' + ver
         return ext
 
     warning += (
         (_(b"** Python %s\n") % sysversion)
         + (_(b"** Mercurial Distributed SCM (version %s)\n") % util.version())
         + (
             _(b"** Extensions loaded: %s\n")
             % b", ".join(
                 [ext_with_ver(x) for x in sorted(extensions.extensions())]
             )
         )
     )
     return warning
 
 
 def handlecommandexception(ui):
     """Produce a warning message for broken commands
 
     Called when handling an exception; the exception is reraised if
     this function returns False, ignored otherwise.
     """
     warning = _exceptionwarning(ui)
     ui.log(
         b"commandexception",
         b"%s\n%s\n",
         warning,
         pycompat.sysbytes(traceback.format_exc()),
     )
     ui.warn(warning)
     return False  # re-raise the exception

mercurial/pure/osutil.py

@@ -1,305 +1,305 b''
 # osutil.py - pure Python version of osutil.c
 #
 #  Copyright 2009 Matt Mackall <[e-mail]> and others
 #
 # 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, division
 
 import ctypes
 import ctypes.util
 import os
 import socket
 import stat as statmod
 
 from ..pycompat import getattr
 from .. import (
     encoding,
     pycompat,
 )
 
 
 def _mode_to_kind(mode):
     if statmod.S_ISREG(mode):
         return statmod.S_IFREG
     if statmod.S_ISDIR(mode):
         return statmod.S_IFDIR
     if statmod.S_ISLNK(mode):
         return statmod.S_IFLNK
     if statmod.S_ISBLK(mode):
         return statmod.S_IFBLK
     if statmod.S_ISCHR(mode):
         return statmod.S_IFCHR
     if statmod.S_ISFIFO(mode):
         return statmod.S_IFIFO
     if statmod.S_ISSOCK(mode):
         return statmod.S_IFSOCK
     return mode
 
 
 def listdir(path, stat=False, skip=None):
     """listdir(path, stat=False) -> list_of_tuples
 
     Return a sorted list containing information about the entries
     in the directory.
 
     If stat is True, each element is a 3-tuple:
 
       (name, type, stat object)
 
     Otherwise, each element is a 2-tuple:
 
       (name, type)
     """
     result = []
     prefix = path
     if not prefix.endswith(pycompat.ossep):
         prefix += pycompat.ossep
     names = os.listdir(path)
     names.sort()
     for fn in names:
         st = os.lstat(prefix + fn)
         if fn == skip and statmod.S_ISDIR(st.st_mode):
             return []
         if stat:
             result.append((fn, _mode_to_kind(st.st_mode), st))
         else:
             result.append((fn, _mode_to_kind(st.st_mode)))
     return result
 
 
 if not pycompat.iswindows:
     posixfile = open
 
     _SCM_RIGHTS = 0x01
     _socklen_t = ctypes.c_uint
 
     if pycompat.sysplatform.startswith(b'linux'):
         # socket.h says "the type should be socklen_t but the definition of
         # the kernel is incompatible with this."
         _cmsg_len_t = ctypes.c_size_t
         _msg_controllen_t = ctypes.c_size_t
         _msg_iovlen_t = ctypes.c_size_t
     else:
         _cmsg_len_t = _socklen_t
         _msg_controllen_t = _socklen_t
         _msg_iovlen_t = ctypes.c_int
 
     class _iovec(ctypes.Structure):
         _fields_ = [
             (u'iov_base', ctypes.c_void_p),
             (u'iov_len', ctypes.c_size_t),
         ]
 
     class _msghdr(ctypes.Structure):
         _fields_ = [
             (u'msg_name', ctypes.c_void_p),
             (u'msg_namelen', _socklen_t),
             (u'msg_iov', ctypes.POINTER(_iovec)),
             (u'msg_iovlen', _msg_iovlen_t),
             (u'msg_control', ctypes.c_void_p),
             (u'msg_controllen', _msg_controllen_t),
             (u'msg_flags', ctypes.c_int),
         ]
 
     class _cmsghdr(ctypes.Structure):
         _fields_ = [
             (u'cmsg_len', _cmsg_len_t),
             (u'cmsg_level', ctypes.c_int),
             (u'cmsg_type', ctypes.c_int),
             (u'cmsg_data', ctypes.c_ubyte * 0),
         ]
 
     _libc = ctypes.CDLL(ctypes.util.find_library(u'c'), use_errno=True)
     _recvmsg = getattr(_libc, 'recvmsg', None)
     if _recvmsg:
         _recvmsg.restype = getattr(ctypes, 'c_ssize_t', ctypes.c_long)
         _recvmsg.argtypes = (
             ctypes.c_int,
             ctypes.POINTER(_msghdr),
             ctypes.c_int,
         )
     else:
         # recvmsg isn't always provided by libc; such systems are unsupported
         def _recvmsg(sockfd, msg, flags):
             raise NotImplementedError(b'unsupported platform')
 
     def _CMSG_FIRSTHDR(msgh):
         if msgh.msg_controllen < ctypes.sizeof(_cmsghdr):
             return
         cmsgptr = ctypes.cast(msgh.msg_control, ctypes.POINTER(_cmsghdr))
         return cmsgptr.contents
 
     # The pure version is less portable than the native version because the
     # handling of socket ancillary data heavily depends on C preprocessor.
     # Also, some length fields are wrongly typed in Linux kernel.
     def recvfds(sockfd):
         """receive list of file descriptors via socket"""
         dummy = (ctypes.c_ubyte * 1)()
         iov = _iovec(ctypes.cast(dummy, ctypes.c_void_p), ctypes.sizeof(dummy))
         cbuf = ctypes.create_string_buffer(256)
         msgh = _msghdr(
             None,
             0,
             ctypes.pointer(iov),
             1,
             ctypes.cast(cbuf, ctypes.c_void_p),
             ctypes.sizeof(cbuf),
             0,
         )
         r = _recvmsg(sockfd, ctypes.byref(msgh), 0)
         if r < 0:
             e = ctypes.get_errno()
             raise OSError(e, os.strerror(e))
         # assumes that the first cmsg has fds because it isn't easy to write
         # portable CMSG_NXTHDR() with ctypes.
         cmsg = _CMSG_FIRSTHDR(msgh)
         if not cmsg:
             return []
         if (
             cmsg.cmsg_level != socket.SOL_SOCKET
             or cmsg.cmsg_type != _SCM_RIGHTS
         ):
             return []
         rfds = ctypes.cast(cmsg.cmsg_data, ctypes.POINTER(ctypes.c_int))
         rfdscount = (
             cmsg.cmsg_len - _cmsghdr.cmsg_data.offset
         ) // ctypes.sizeof(ctypes.c_int)
         return [rfds[i] for i in pycompat.xrange(rfdscount)]
 
 
 else:
     import msvcrt
 
-    _kernel32 = ctypes.windll.kernel32
+    _kernel32 = ctypes.windll.kernel32  # pytype: disable=module-attr
 
     _DWORD = ctypes.c_ulong
     _LPCSTR = _LPSTR = ctypes.c_char_p
     _HANDLE = ctypes.c_void_p
 
     _INVALID_HANDLE_VALUE = _HANDLE(-1).value
 
     # CreateFile
     _FILE_SHARE_READ = 0x00000001
     _FILE_SHARE_WRITE = 0x00000002
     _FILE_SHARE_DELETE = 0x00000004
 
     _CREATE_ALWAYS = 2
     _OPEN_EXISTING = 3
     _OPEN_ALWAYS = 4
 
     _GENERIC_READ = 0x80000000
     _GENERIC_WRITE = 0x40000000
 
     _FILE_ATTRIBUTE_NORMAL = 0x80
 
     # open_osfhandle flags
     _O_RDONLY = 0x0000
     _O_RDWR = 0x0002
     _O_APPEND = 0x0008
 
     _O_TEXT = 0x4000
     _O_BINARY = 0x8000
 
     # types of parameters of C functions used (required by pypy)
 
     _kernel32.CreateFileA.argtypes = [
         _LPCSTR,
         _DWORD,
         _DWORD,
         ctypes.c_void_p,
         _DWORD,
         _DWORD,
         _HANDLE,
     ]
     _kernel32.CreateFileA.restype = _HANDLE
 
     def _raiseioerror(name):
-        err = ctypes.WinError()
+        err = ctypes.WinError()  # pytype: disable=module-attr
         raise IOError(
             err.errno, '%s: %s' % (encoding.strfromlocal(name), err.strerror)
         )
 
     class posixfile(object):
         """a file object aiming for POSIX-like semantics
 
         CPython's open() returns a file that was opened *without* setting the
         _FILE_SHARE_DELETE flag, which causes rename and unlink to abort.
         This even happens if any hardlinked copy of the file is in open state.
         We set _FILE_SHARE_DELETE here, so files opened with posixfile can be
         renamed and deleted while they are held open.
         Note that if a file opened with posixfile is unlinked, the file
         remains but cannot be opened again or be recreated under the same name,
         until all reading processes have closed the file."""
 
         def __init__(self, name, mode=b'r', bufsize=-1):
             if b'b' in mode:
                 flags = _O_BINARY
             else:
                 flags = _O_TEXT
 
             m0 = mode[0:1]
             if m0 == b'r' and b'+' not in mode:
                 flags |= _O_RDONLY
                 access = _GENERIC_READ
             else:
                 # work around http://support.microsoft.com/kb/899149 and
                 # set _O_RDWR for 'w' and 'a', even if mode has no '+'
                 flags |= _O_RDWR
                 access = _GENERIC_READ | _GENERIC_WRITE
 
             if m0 == b'r':
                 creation = _OPEN_EXISTING
             elif m0 == b'w':
                 creation = _CREATE_ALWAYS
             elif m0 == b'a':
                 creation = _OPEN_ALWAYS
                 flags |= _O_APPEND
             else:
                 raise ValueError("invalid mode: %s" % pycompat.sysstr(mode))
 
             fh = _kernel32.CreateFileA(
                 name,
                 access,
                 _FILE_SHARE_READ | _FILE_SHARE_WRITE | _FILE_SHARE_DELETE,
                 None,
                 creation,
                 _FILE_ATTRIBUTE_NORMAL,
                 None,
             )
             if fh == _INVALID_HANDLE_VALUE:
                 _raiseioerror(name)
 
-            fd = msvcrt.open_osfhandle(fh, flags)
+            fd = msvcrt.open_osfhandle(fh, flags)  # pytype: disable=module-attr
             if fd == -1:
                 _kernel32.CloseHandle(fh)
                 _raiseioerror(name)
 
             f = os.fdopen(fd, pycompat.sysstr(mode), bufsize)
             # unfortunately, f.name is '<fdopen>' at this point -- so we store
             # the name on this wrapper. We cannot just assign to f.name,
             # because that attribute is read-only.
             object.__setattr__(self, 'name', name)
             object.__setattr__(self, '_file', f)
 
         def __iter__(self):
             return self._file
 
         def __getattr__(self, name):
             return getattr(self._file, name)
 
         def __setattr__(self, name, value):
             """mimics the read-only attributes of Python file objects
             by raising 'TypeError: readonly attribute' if someone tries:
               f = posixfile('foo.txt')
               f.name = 'bla'
             """
             return self._file.__setattr__(name, value)
 
         def __enter__(self):
             self._file.__enter__()
             return self
 
         def __exit__(self, exc_type, exc_value, exc_tb):
             return self._file.__exit__(exc_type, exc_value, exc_tb)

mercurial/sslutil.py

@@ -1,836 +1,838 b''
 # sslutil.py - SSL handling for mercurial
 #
 # Copyright 2005, 2006, 2007, 2008 Matt Mackall <[e-mail]>
 # Copyright 2006, 2007 Alexis S. L. Carvalho <[e-mail]>
 # Copyright 2006 Vadim Gelfer <[e-mail]>
 #
 # 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 .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)
 
     # 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'minimumprotocol'] = 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'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.
     #
     # 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'])
     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']:
         # This is a no-op on old Python.
         sslcontext.load_default_certs()
         caloaded = True
     else:
         caloaded = False
 
     try:
         sslsocket = sslcontext.wrap_socket(sock, server_hostname=serverhostname)
     except ssl.SSLError as e:
         # If we're doing certificate verification and no CA certs are loaded,
         # that is almost certainly the reason why verification failed. Provide
         # a hint to the user.
         # The exception handler is here to handle bugs around cert attributes:
         # https://bugs.python.org/issue20916#msg213479.  (See issues5313.)
         # When the main 20916 bug occurs, 'sslcontext.get_ca_certs()' is a
         # non-empty list, but the following conditional is otherwise True.
         try:
             if (
                 caloaded
                 and settings[b'verifymode'] == ssl.CERT_REQUIRED
                 and not sslcontext.get_ca_certs()
             ):
                 ui.warn(
                     _(
                         b'(an attempt was made to load CA certificates but '
                         b'none were loaded; see '
                         b'https://mercurial-scm.org/wiki/SecureConnections '
                         b'for how to configure Mercurial to avoid this '
                         b'error)\n'
                     )
                 )
         except ssl.SSLError:
             pass
 
         # Try to print more helpful error messages for known failures.
         if util.safehasattr(e, b'reason'):
             # This error occurs when the client and server don't share a
             # common/supported SSL/TLS protocol. We've disabled SSLv2 and SSLv3
             # outright. Hopefully the reason for this error is that we require
             # TLS 1.1+ and the server only supports TLS 1.0. Whatever the
             # reason, try to emit an actionable warning.
             if e.reason == 'UNSUPPORTED_PROTOCOL':
                 # We attempted TLS 1.0+.
                 if settings[b'minimumprotocol'] == b'tls1.0':
                     # We support more than just TLS 1.0+. If this happens,
                     # the likely scenario is either the client or the server
                     # is really old. (e.g. server doesn't support TLS 1.0+ or
                     # client doesn't support modern TLS versions introduced
                     # several years from when this comment was written).
                     if supportedprotocols != {b'tls1.0'}:
                         ui.warn(
                             _(
                                 b'(could not communicate with %s using security '
                                 b'protocols %s; if you are using a modern Mercurial '
                                 b'version, consider contacting the operator of this '
                                 b'server; see '
                                 b'https://mercurial-scm.org/wiki/SecureConnections '
                                 b'for more info)\n'
                             )
                             % (
                                 pycompat.bytesurl(serverhostname),
                                 b', '.join(sorted(supportedprotocols)),
                             )
                         )
                     else:
                         ui.warn(
                             _(
                                 b'(could not communicate with %s using TLS 1.0; the '
                                 b'likely cause of this is the server no longer '
                                 b'supports TLS 1.0 because it has known security '
                                 b'vulnerabilities; see '
                                 b'https://mercurial-scm.org/wiki/SecureConnections '
                                 b'for more info)\n'
                             )
                             % pycompat.bytesurl(serverhostname)
                         )
                 else:
                     # We attempted TLS 1.1+. We can only get here if the client
                     # supports the configured protocol. So the likely reason is
                     # the client wants better security than the server can
                     # offer.
                     ui.warn(
                         _(
                             b'(could not negotiate a common security protocol (%s+) '
                             b'with %s; the likely cause is Mercurial is configured '
                             b'to be more secure than the server can support)\n'
                         )
                         % (
                             settings[b'minimumprotocol'],
                             pycompat.bytesurl(serverhostname),
                         )
                     )
                     ui.warn(
                         _(
                             b'(consider contacting the operator of this '
                             b'server and ask them to support modern TLS '
                             b'protocol versions; or, set '
                             b'hostsecurity.%s:minimumprotocol=tls1.0 to allow '
                             b'use of legacy, less secure protocols when '
                             b'communicating with this server)\n'
                         )
                         % pycompat.bytesurl(serverhostname)
                     )
                     ui.warn(
                         _(
                             b'(see https://mercurial-scm.org/wiki/SecureConnections '
                             b'for more info)\n'
                         )
                     )
 
             elif e.reason == 'CERTIFICATE_VERIFY_FAILED' and pycompat.iswindows:
 
                 ui.warn(
                     _(
                         b'(the full certificate chain may not be available '
                         b'locally; see "hg help debugssl")\n'
                     )
                 )
         raise
 
     # check if wrap_socket failed silently because socket had been
     # closed
     # - see http://bugs.python.org/issue13721
     if not sslsocket.cipher():
         raise error.SecurityError(_(b'ssl connection failed'))
 
     sslsocket._hgstate = {
         b'caloaded': caloaded,
         b'hostname': serverhostname,
         b'settings': settings,
         b'ui': ui,
     }
 
     return sslsocket
 
 
 def wrapserversocket(
     sock, ui, certfile=None, keyfile=None, cafile=None, requireclientcert=False
 ):
     """Wrap a socket for use by servers.
 
     ``certfile`` and ``keyfile`` specify the files containing the certificate's
     public and private keys, respectively. Both keys can be defined in the same
     file via ``certfile`` (the private key must come first in the file).
 
     ``cafile`` defines the path to certificate authorities.
 
     ``requireclientcert`` specifies whether to require client certificates.
 
     Typically ``cafile`` is only defined if ``requireclientcert`` is true.
     """
     # This function is not used much by core Mercurial, so the error messaging
     # doesn't have to be as detailed as for wrapsocket().
     for f in (certfile, keyfile, cafile):
         if f and not os.path.exists(f):
             raise error.Abort(
                 _(b'referenced certificate file (%s) does not exist') % f
             )
 
     # Despite its name, PROTOCOL_SSLv23 selects the highest protocol that both
     # ends support, including TLS protocols. commonssloptions() restricts the
     # set of allowed protocols.
     protocol = ssl.PROTOCOL_SSLv23
     options = commonssloptions(b'tls1.0')
 
     # This config option is intended for use in tests only. It is a giant
     # footgun to kill security. Don't define it.
     exactprotocol = ui.config(b'devel', b'serverexactprotocol')
     if exactprotocol == b'tls1.0':
         if b'tls1.0' not in supportedprotocols:
             raise error.Abort(_(b'TLS 1.0 not supported by this Python'))
         protocol = ssl.PROTOCOL_TLSv1
     elif exactprotocol == b'tls1.1':
         if b'tls1.1' not in supportedprotocols:
             raise error.Abort(_(b'TLS 1.1 not supported by this Python'))
         protocol = ssl.PROTOCOL_TLSv1_1
     elif exactprotocol == b'tls1.2':
         if b'tls1.2' not in supportedprotocols:
             raise error.Abort(_(b'TLS 1.2 not supported by this Python'))
         protocol = ssl.PROTOCOL_TLSv1_2
     elif exactprotocol:
         raise error.Abort(
             _(b'invalid value for serverexactprotocol: %s') % exactprotocol
         )
 
     # We /could/ use create_default_context() here since it doesn't load
     # CAs when configured for client auth. However, it is hard-coded to
     # use ssl.PROTOCOL_SSLv23 which may not be appropriate here.
     sslcontext = ssl.SSLContext(protocol)
     sslcontext.options |= options
 
     # Improve forward secrecy.
     sslcontext.options |= getattr(ssl, 'OP_SINGLE_DH_USE', 0)
     sslcontext.options |= getattr(ssl, 'OP_SINGLE_ECDH_USE', 0)
 
     # Use the list of more secure ciphers if found in the ssl module.
     if util.safehasattr(ssl, b'_RESTRICTED_SERVER_CIPHERS'):
         sslcontext.options |= getattr(ssl, 'OP_CIPHER_SERVER_PREFERENCE', 0)
+        # pytype: disable=module-attr
         sslcontext.set_ciphers(ssl._RESTRICTED_SERVER_CIPHERS)
+        # pytype: enable=module-attr
 
     if requireclientcert:
         sslcontext.verify_mode = ssl.CERT_REQUIRED
     else:
         sslcontext.verify_mode = ssl.CERT_NONE
 
     if certfile or keyfile:
         sslcontext.load_cert_chain(certfile=certfile, keyfile=keyfile)
 
     if cafile:
         sslcontext.load_verify_locations(cafile=cafile)
 
     return sslcontext.wrap_socket(sock, server_side=True)
 
 
 class wildcarderror(Exception):
     """Represents an error parsing wildcards in DNS name."""
 
 
 def _dnsnamematch(dn, hostname, maxwildcards=1):
     """Match DNS names according RFC 6125 section 6.4.3.
 
     This code is effectively copied from CPython's ssl._dnsname_match.
 
     Returns a bool indicating whether the expected hostname matches
     the value in ``dn``.
     """
     pats = []
     if not dn:
         return False
     dn = pycompat.bytesurl(dn)
     hostname = pycompat.bytesurl(hostname)
 
     pieces = dn.split(b'.')
     leftmost = pieces[0]
     remainder = pieces[1:]
     wildcards = leftmost.count(b'*')
     if wildcards > maxwildcards:
         raise wildcarderror(
             _(b'too many wildcards in certificate DNS name: %s') % dn
         )
 
     # speed up common case w/o wildcards
     if not wildcards:
         return dn.lower() == hostname.lower()
 
     # RFC 6125, section 6.4.3, subitem 1.
     # The client SHOULD NOT attempt to match a presented identifier in which
     # the wildcard character comprises a label other than the left-most label.
     if leftmost == b'*':
         # When '*' is a fragment by itself, it matches a non-empty dotless
         # fragment.
         pats.append(b'[^.]+')
     elif leftmost.startswith(b'xn--') or hostname.startswith(b'xn--'):
         # RFC 6125, section 6.4.3, subitem 3.
         # The client SHOULD NOT attempt to match a presented identifier
         # where the wildcard character is embedded within an A-label or
         # U-label of an internationalized domain name.
         pats.append(stringutil.reescape(leftmost))
     else:
         # Otherwise, '*' matches any dotless string, e.g. www*
         pats.append(stringutil.reescape(leftmost).replace(br'\*', b'[^.]*'))
 
     # add the remaining fragments, ignore any wildcards
     for frag in remainder:
         pats.append(stringutil.reescape(frag))
 
     pat = re.compile(br'\A' + br'\.'.join(pats) + br'\Z', re.IGNORECASE)
     return pat.match(hostname) is not None
 
 
 def _verifycert(cert, hostname):
     """Verify that cert (in socket.getpeercert() format) matches hostname.
     CRLs is not handled.
 
     Returns error message if any problems are found and None on success.
     """
     if not cert:
         return _(b'no certificate received')
 
     dnsnames = []
     san = cert.get('subjectAltName', [])
     for key, value in san:
         if key == 'DNS':
             try:
                 if _dnsnamematch(value, hostname):
                     return
             except wildcarderror as e:
                 return stringutil.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('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 _(b'IDN in certificate not supported')
 
                     try:
                         if _dnsnamematch(value, hostname):
                             return
                     except wildcarderror as e:
                         return stringutil.forcebytestr(e.args[0])
 
                     dnsnames.append(value)
 
     dnsnames = [pycompat.bytesurl(d) for d in dnsnames]
     if len(dnsnames) > 1:
         return _(b'certificate is for %s') % b', '.join(dnsnames)
     elif len(dnsnames) == 1:
         return _(b'certificate is for %s') % dnsnames[0]
     else:
         return _(b'no commonName or subjectAltName found in certificate')
 
 
 def _plainapplepython():
     """return true if this seems to be a pure Apple Python that
     * is unfrozen and presumably has the whole mercurial module in the file
       system
     * presumably is an Apple Python that uses Apple OpenSSL which has patches
       for using system certificate store CAs in addition to the provided
       cacerts file
     """
     if (
         not pycompat.isdarwin
         or resourceutil.mainfrozen()
         or not pycompat.sysexecutable
     ):
         return False
     exe = os.path.realpath(pycompat.sysexecutable).lower()
     return exe.startswith(b'/usr/bin/python') or exe.startswith(
         b'/system/library/frameworks/python.framework/'
     )
 
 
 def _defaultcacerts(ui):
     """return path to default CA certificates or None.
 
     It is assumed this function is called when the returned certificates
     file will actually be used to validate connections. Therefore this
     function may print warnings or debug messages assuming this usage.
 
     We don't print a message when the Python is able to load default
     CA certs because this scenario is detected at socket connect time.
     """
     # The "certifi" Python package provides certificates. If it is installed
     # and usable, assume the user intends it to be used and use it.
     try:
         import certifi
 
         certs = certifi.where()
         if os.path.exists(certs):
             ui.debug(b'using ca certificates from certifi\n')
             return pycompat.fsencode(certs)
     except (ImportError, AttributeError):
         pass
 
     # Apple's OpenSSL has patches that allow a specially constructed certificate
     # to load the system CA store. If we're running on Apple Python, use this
     # trick.
     if _plainapplepython():
         dummycert = os.path.join(
             os.path.dirname(pycompat.fsencode(__file__)), b'dummycert.pem'
         )
         if os.path.exists(dummycert):
             return dummycert
 
     return None
 
 
 def validatesocket(sock):
     """Validate a socket meets security requirements.
 
     The passed socket must have been created with ``wrapsocket()``.
     """
     shost = sock._hgstate[b'hostname']
     host = pycompat.bytesurl(shost)
     ui = sock._hgstate[b'ui']
     settings = sock._hgstate[b'settings']
 
     try:
         peercert = sock.getpeercert(True)
         peercert2 = sock.getpeercert()
     except AttributeError:
         raise error.SecurityError(_(b'%s ssl connection error') % host)
 
     if not peercert:
         raise error.SecurityError(
             _(b'%s certificate error: no certificate received') % host
         )
 
     if settings[b'disablecertverification']:
         # We don't print the certificate fingerprint because it shouldn't
         # be necessary: if the user requested certificate verification be
         # disabled, they presumably already saw a message about the inability
         # to verify the certificate and this message would have printed the
         # fingerprint. So printing the fingerprint here adds little to no
         # value.
         ui.warn(
             _(
                 b'warning: connection security to %s is disabled per current '
                 b'settings; communication is susceptible to eavesdropping '
                 b'and tampering\n'
             )
             % host
         )
         return
 
     # If a certificate fingerprint is pinned, use it and only it to
     # validate the remote cert.
     peerfingerprints = {
         b'sha1': hex(hashutil.sha1(peercert).digest()),
         b'sha256': hex(hashlib.sha256(peercert).digest()),
         b'sha512': hex(hashlib.sha512(peercert).digest()),
     }
 
     def fmtfingerprint(s):
         return b':'.join([s[x : x + 2] for x in range(0, len(s), 2)])
 
     nicefingerprint = b'sha256:%s' % fmtfingerprint(peerfingerprints[b'sha256'])
 
     if settings[b'certfingerprints']:
         for hash, fingerprint in settings[b'certfingerprints']:
             if peerfingerprints[hash].lower() == fingerprint:
                 ui.debug(
                     b'%s certificate matched fingerprint %s:%s\n'
                     % (host, hash, fmtfingerprint(fingerprint))
                 )
                 if settings[b'legacyfingerprint']:
                     ui.warn(
                         _(
                             b'(SHA-1 fingerprint for %s found in legacy '
                             b'[hostfingerprints] section; '
                             b'if you trust this fingerprint, remove the old '
                             b'SHA-1 fingerprint from [hostfingerprints] and '
                             b'add the following entry to the new '
                             b'[hostsecurity] section: %s:fingerprints=%s)\n'
                         )
                         % (host, host, nicefingerprint)
                     )
                 return
 
         # Pinned fingerprint didn't match. This is a fatal error.
         if settings[b'legacyfingerprint']:
             section = b'hostfingerprint'
             nice = fmtfingerprint(peerfingerprints[b'sha1'])
         else:
             section = b'hostsecurity'
             nice = b'%s:%s' % (hash, fmtfingerprint(peerfingerprints[hash]))
         raise error.SecurityError(
             _(b'certificate for %s has unexpected fingerprint %s')
             % (host, nice),
             hint=_(b'check %s configuration') % section,
         )
 
     # Security is enabled but no CAs are loaded. We can't establish trust
     # for the cert so abort.
     if not sock._hgstate[b'caloaded']:
         raise error.SecurityError(
             _(
                 b'unable to verify security of %s (no loaded CA certificates); '
                 b'refusing to connect'
             )
             % host,
             hint=_(
                 b'see https://mercurial-scm.org/wiki/SecureConnections for '
                 b'how to configure Mercurial to avoid this error or set '
                 b'hostsecurity.%s:fingerprints=%s to trust this server'
             )
             % (host, nicefingerprint),
         )
 
     msg = _verifycert(peercert2, shost)
     if msg:
         raise error.SecurityError(
             _(b'%s certificate error: %s') % (host, msg),
             hint=_(
                 b'set hostsecurity.%s:certfingerprints=%s '
                 b'config setting or use --insecure to connect '
                 b'insecurely'
             )
             % (host, nicefingerprint),
         )