prompt user for http authentication info

in interactive mode, mercurial now asks the user for the username and
password when the server requires it. the previous behavior was to fail
with an http 401.

based on patch from eric jaffe <[e-mail]>.

mercurial/httprepo.py

@@ -1,142 +1,169 b''
 # httprepo.py - HTTP repository proxy classes for mercurial
 #
 # Copyright 2005 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms
 # of the GNU General Public License, incorporated herein by reference.
 
 from node import *
 from remoterepo import *
 from i18n import gettext as _
 from demandload import *
 demandload(globals(), "hg os urllib urllib2 urlparse zlib util httplib")
 
+class passwordmgr(urllib2.HTTPPasswordMgr):
+    def __init__(self, ui):
+        urllib2.HTTPPasswordMgr.__init__(self)
+        self.ui = ui
+
+    def find_user_password(self, realm, authuri):
+        authinfo = urllib2.HTTPPasswordMgr.find_user_password(
+            self, realm, authuri)
+        if authinfo != (None, None):
+            return authinfo
+
+        self.ui.write(_("http authorization required\n"))
+        self.ui.status(_("realm: %s\n") % realm)
+        user = self.ui.prompt(_("user:"), default=None)
+        passwd = self.ui.getpass()
+
+        self.add_password(realm, authuri, user, passwd)
+        return (user, passwd)
+
 class httprepository(remoterepository):
     def __init__(self, ui, path):
         # fix missing / after hostname
         s = urlparse.urlsplit(path)
         partial = s[2]
         if not partial: partial = "/"
         self.url = urlparse.urlunsplit((s[0], s[1], partial, '', ''))
         self.ui = ui
         no_list = [ "localhost", "127.0.0.1" ]
         host = ui.config("http_proxy", "host")
         if host is None:
             host = os.environ.get("http_proxy")
         if host and host.startswith('http://'):
             host = host[7:]
         user = ui.config("http_proxy", "user")
         passwd = ui.config("http_proxy", "passwd")
         no = ui.config("http_proxy", "no")
         if no is None:
             no = os.environ.get("no_proxy")
         if no:
             no_list = no_list + no.split(",")
 
         no_proxy = 0
         for h in no_list:
             if (path.startswith("http://" + h + "/") or
                 path.startswith("http://" + h + ":") or
                 path == "http://" + h):
                 no_proxy = 1
 
         # Note: urllib2 takes proxy values from the environment and those will
         # take precedence
         for env in ["HTTP_PROXY", "http_proxy", "no_proxy"]:
             try:
                 if os.environ.has_key(env):
                     del os.environ[env]
             except OSError:
                 pass
 
         proxy_handler = urllib2.BaseHandler()
         if host and not no_proxy:
             proxy_handler = urllib2.ProxyHandler({"http" : "http://" + host})
 
-        authinfo = None
+        proxyauthinfo = None
         if user and passwd:
             passmgr = urllib2.HTTPPasswordMgrWithDefaultRealm()
             passmgr.add_password(None, host, user, passwd)
-            authinfo = urllib2.ProxyBasicAuthHandler(passmgr)
+            proxyauthinfo = urllib2.ProxyBasicAuthHandler(passmgr)
 
-        opener = urllib2.build_opener(proxy_handler, authinfo)
+        if ui.interactive:
+            passmgr = passwordmgr(ui)
+            opener = urllib2.build_opener(
+                proxy_handler, proxyauthinfo,
+                urllib2.HTTPBasicAuthHandler(passmgr),
+                urllib2.HTTPDigestAuthHandler(passmgr))
+        else:
+            opener = urllib2.build_opener(proxy_handler, proxyauthinfo)
+
         # 1.0 here is the _protocol_ version
         opener.addheaders = [('User-agent', 'mercurial/proto-1.0')]
         urllib2.install_opener(opener)
 
     def dev(self):
         return -1
 
     def lock(self):
         raise util.Abort(_('operation not supported over http'))
 
     def do_cmd(self, cmd, **args):
         self.ui.debug(_("sending %s command\n") % cmd)
         q = {"cmd": cmd}
         q.update(args)
         qs = urllib.urlencode(q)
         cu = "%s?%s" % (self.url, qs)
         resp = urllib2.urlopen(cu)
         proto = resp.headers['content-type']
 
         # accept old "text/plain" and "application/hg-changegroup" for now
         if not proto.startswith('application/mercurial') and \
                not proto.startswith('text/plain') and \
                not proto.startswith('application/hg-changegroup'):
             raise hg.RepoError(_("'%s' does not appear to be an hg repository") %
                                self.url)
 
         if proto.startswith('application/mercurial'):
             version = proto[22:]
             if float(version) > 0.1:
                 raise hg.RepoError(_("'%s' uses newer protocol %s") %
                                    (self.url, version))
 
         return resp
 
     def heads(self):
         d = self.do_cmd("heads").read()
         try:
             return map(bin, d[:-1].split(" "))
         except:
             self.ui.warn(_("unexpected response:\n") + d[:400] + "\n...\n")
             raise
 
     def branches(self, nodes):
         n = " ".join(map(hex, nodes))
         d = self.do_cmd("branches", nodes=n).read()
         try:
             br = [ tuple(map(bin, b.split(" "))) for b in d.splitlines() ]
             return br
         except:
             self.ui.warn(_("unexpected response:\n") + d[:400] + "\n...\n")
             raise
 
     def between(self, pairs):
         n = "\n".join(["-".join(map(hex, p)) for p in pairs])
         d = self.do_cmd("between", pairs=n).read()
         try:
             p = [ l and map(bin, l.split(" ")) or [] for l in d.splitlines() ]
             return p
         except:
             self.ui.warn(_("unexpected response:\n") + d[:400] + "\n...\n")
             raise
 
     def changegroup(self, nodes, kind):
         n = " ".join(map(hex, nodes))
         f = self.do_cmd("changegroup", roots=n)
         bytes = 0
 
         def zgenerator(f):
             zd = zlib.decompressobj()
             try:
                 for chnk in f:
                     yield zd.decompress(chnk)
             except httplib.HTTPException, inst:
                 raise IOError(None, _('connection ended unexpectedly'))
             yield zd.flush()
 
         return util.chunkbuffer(zgenerator(util.filechunkiter(f)))
 
 class httpsrepository(httprepository):
     pass

mercurial/ui.py

@@ -1,281 +1,284 b''
 # ui.py - user interface bits for mercurial
 #
 # Copyright 2005 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms
 # of the GNU General Public License, incorporated herein by reference.
 
 import ConfigParser
 from i18n import gettext as _
 from demandload import *
-demandload(globals(), "errno os re smtplib socket sys tempfile util")
+demandload(globals(), "errno getpass os re smtplib socket sys tempfile util")
 
 class ui(object):
     def __init__(self, verbose=False, debug=False, quiet=False,
                  interactive=True, traceback=False, parentui=None):
         self.overlay = {}
         if parentui is None:
             # this is the parent of all ui children
             self.parentui = None
             self.cdata = ConfigParser.SafeConfigParser()
             self.readconfig(util.rcpath())
 
             self.quiet = self.configbool("ui", "quiet")
             self.verbose = self.configbool("ui", "verbose")
             self.debugflag = self.configbool("ui", "debug")
             self.interactive = self.configbool("ui", "interactive", True)
             self.traceback = traceback
 
             self.updateopts(verbose, debug, quiet, interactive)
             self.diffcache = None
             self.header = []
             self.prev_header = []
             self.revlogopts = self.configrevlog()
         else:
             # parentui may point to an ui object which is already a child
             self.parentui = parentui.parentui or parentui
             parent_cdata = self.parentui.cdata
             self.cdata = ConfigParser.SafeConfigParser(parent_cdata.defaults())
             # make interpolation work
             for section in parent_cdata.sections():
                 self.cdata.add_section(section)
                 for name, value in parent_cdata.items(section, raw=True):
                     self.cdata.set(section, name, value)
 
     def __getattr__(self, key):
         return getattr(self.parentui, key)
 
     def updateopts(self, verbose=False, debug=False, quiet=False,
                    interactive=True, traceback=False):
         self.quiet = (self.quiet or quiet) and not verbose and not debug
         self.verbose = (self.verbose or verbose) or debug
         self.debugflag = (self.debugflag or debug)
         self.interactive = (self.interactive and interactive)
         self.traceback = self.traceback or traceback
 
     def readconfig(self, fn, root=None):
         if isinstance(fn, basestring):
             fn = [fn]
         for f in fn:
             try:
                 self.cdata.read(f)
             except ConfigParser.ParsingError, inst:
                 raise util.Abort(_("Failed to parse %s\n%s") % (f, inst))
         # translate paths relative to root (or home) into absolute paths
         if root is None:
             root = os.path.expanduser('~')
         for name, path in self.configitems("paths"):
             if path and path.find("://") == -1 and not os.path.isabs(path):
                 self.cdata.set("paths", name, os.path.join(root, path))
 
     def setconfig(self, section, name, val):
         self.overlay[(section, name)] = val
 
     def config(self, section, name, default=None):
         if self.overlay.has_key((section, name)):
             return self.overlay[(section, name)]
         if self.cdata.has_option(section, name):
             try:
                 return self.cdata.get(section, name)
             except ConfigParser.InterpolationError, inst:
                 raise util.Abort(_("Error in configuration:\n%s") % inst)
         if self.parentui is None:
             return default
         else:
             return self.parentui.config(section, name, default)
 
     def configbool(self, section, name, default=False):
         if self.overlay.has_key((section, name)):
             return self.overlay[(section, name)]
         if self.cdata.has_option(section, name):
             try:
                 return self.cdata.getboolean(section, name)
             except ConfigParser.InterpolationError, inst:
                 raise util.Abort(_("Error in configuration:\n%s") % inst)
         if self.parentui is None:
             return default
         else:
             return self.parentui.configbool(section, name, default)
 
     def configitems(self, section):
         items = {}
         if self.parentui is not None:
             items = dict(self.parentui.configitems(section))
         if self.cdata.has_section(section):
             try:
                 items.update(dict(self.cdata.items(section)))
             except ConfigParser.InterpolationError, inst:
                 raise util.Abort(_("Error in configuration:\n%s") % inst)
         x = items.items()
         x.sort()
         return x
 
     def walkconfig(self, seen=None):
         if seen is None:
             seen = {}
         for (section, name), value in self.overlay.iteritems():
             yield section, name, value
             seen[section, name] = 1
         for section in self.cdata.sections():
             for name, value in self.cdata.items(section):
                 if (section, name) in seen: continue
                 yield section, name, value.replace('\n', '\\n')
                 seen[section, name] = 1
         if self.parentui is not None:
             for parent in self.parentui.walkconfig(seen):
                 yield parent
 
     def extensions(self):
         return self.configitems("extensions")
 
     def hgignorefiles(self):
         result = []
         cfgitems = self.configitems("ui")
         for key, value in cfgitems:
             if key == 'ignore' or key.startswith('ignore.'):
                 path = os.path.expanduser(value)
                 result.append(path)
         return result
 
     def configrevlog(self):
         ret = {}
         for x in self.configitems("revlog"):
             k = x[0].lower()
             ret[k] = x[1]
         return ret
     def diffopts(self):
         if self.diffcache:
             return self.diffcache
         ret = { 'showfunc' : True, 'ignorews' : False}
         for x in self.configitems("diff"):
             k = x[0].lower()
             v = x[1]
             if v:
                 v = v.lower()
                 if v == 'true':
                     value = True
                 else:
                     value = False
                 ret[k] = value
         self.diffcache = ret
         return ret
 
     def username(self):
         """Return default username to be used in commits.
 
         Searched in this order: $HGUSER, [ui] section of hgrcs, $EMAIL
         and stop searching if one of these is set.
         Abort if found username is an empty string to force specifying
         the commit user elsewhere, e.g. with line option or repo hgrc.
         If not found, use $LOGNAME or $USERNAME +"@full.hostname".
         """
         user = os.environ.get("HGUSER")
         if user is None:
             user = self.config("ui", "username")
         if user is None:
             user = os.environ.get("EMAIL")
         if user is None:
             user = os.environ.get("LOGNAME") or os.environ.get("USERNAME")
             if user:
                 user = "%s@%s" % (user, socket.getfqdn())
         if not user:
             raise util.Abort(_("Please specify a username."))
         return user
 
     def shortuser(self, user):
         """Return a short representation of a user name or email address."""
         if not self.verbose: user = util.shortuser(user)
         return user
 
     def expandpath(self, loc):
         """Return repository location relative to cwd or from [paths]"""
         if loc.find("://") != -1 or os.path.exists(loc):
             return loc
 
         return self.config("paths", loc, loc)
 
     def write(self, *args):
         if self.header:
             if self.header != self.prev_header:
                 self.prev_header = self.header
                 self.write(*self.header)
             self.header = []
         for a in args:
             sys.stdout.write(str(a))
 
     def write_header(self, *args):
         for a in args:
             self.header.append(str(a))
 
     def write_err(self, *args):
         try:
             if not sys.stdout.closed: sys.stdout.flush()
             for a in args:
                 sys.stderr.write(str(a))
         except IOError, inst:
             if inst.errno != errno.EPIPE:
                 raise
 
     def flush(self):
         try: sys.stdout.flush()
         except: pass
         try: sys.stderr.flush()
         except: pass
 
     def readline(self):
         return sys.stdin.readline()[:-1]
-    def prompt(self, msg, pat, default="y"):
+    def prompt(self, msg, pat=None, default="y"):
         if not self.interactive: return default
         while 1:
             self.write(msg, " ")
             r = self.readline()
-            if re.match(pat, r):
+            if not pat or re.match(pat, r):
                 return r
             else:
                 self.write(_("unrecognized response\n"))
+    def getpass(self, prompt=None, default=None):
+        if not self.interactive: return default
+        return getpass.getpass(prompt or _('password: '))
     def status(self, *msg):
         if not self.quiet: self.write(*msg)
     def warn(self, *msg):
         self.write_err(*msg)
     def note(self, *msg):
         if self.verbose: self.write(*msg)
     def debug(self, *msg):
         if self.debugflag: self.write(*msg)
     def edit(self, text, user):
         (fd, name) = tempfile.mkstemp(prefix="hg-editor-", suffix=".txt",
                                       text=True)
         try:
             f = os.fdopen(fd, "w")
             f.write(text)
             f.close()
 
             editor = (os.environ.get("HGEDITOR") or
                     self.config("ui", "editor") or
                     os.environ.get("EDITOR", "vi"))
 
             util.system("%s \"%s\"" % (editor, name),
                         environ={'HGUSER': user},
                         onerr=util.Abort, errprefix=_("edit failed"))
 
             f = open(name)
             t = f.read()
             f.close()
             t = re.sub("(?m)^HG:.*\n", "", t)
         finally:
             os.unlink(name)
 
         return t
 
     def sendmail(self):
         s = smtplib.SMTP()
         s.connect(host = self.config('smtp', 'host', 'mail'),
                   port = int(self.config('smtp', 'port', 25)))
         if self.configbool('smtp', 'tls'):
             s.ehlo()
             s.starttls()
             s.ehlo()
         username = self.config('smtp', 'username')
         password = self.config('smtp', 'password')
         if username and password:
             s.login(username, password)
         return s