sslutil: extracted ssl methods from httpsconnection in url.py

This makes it easier to share ssl cert validation with other http
implementations.

mercurial/sslutil.py

@@ -0,0 +1,126 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.
+import os
+
+from mercurial import util
+from mercurial.i18n import _
+try:
+    # avoid using deprecated/broken FakeSocket in python 2.6
+    import ssl
+    ssl_wrap_socket = ssl.wrap_socket
+    CERT_REQUIRED = ssl.CERT_REQUIRED
+except ImportError:
+    CERT_REQUIRED = 2
+
+    def ssl_wrap_socket(sock, key_file, cert_file,
+                        cert_reqs=CERT_REQUIRED, ca_certs=None):
+        if ca_certs:
+            raise util.Abort(_(
+                'certificate checking requires Python 2.6'))
+
+        ssl = socket.ssl(sock, key_file, cert_file)
+        return httplib.FakeSocket(sock, ssl)
+
+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 _('no certificate received')
+    dnsname = hostname.lower()
+    def matchdnsname(certname):
+        return (certname == dnsname or
+                '.' in dnsname and certname == '*.' + dnsname.split('.', 1)[1])
+
+    san = cert.get('subjectAltName', [])
+    if san:
+        certnames = [value.lower() for key, value in san if key == 'DNS']
+        for name in certnames:
+            if matchdnsname(name):
+                return None
+        return _('certificate is for %s') % ', '.join(certnames)
+
+    # subject is only checked when subjectAltName is empty
+    for s in cert.get('subject', []):
+        key, value = s[0]
+        if key == 'commonName':
+            try:
+                # 'subject' entries are unicode
+                certname = value.lower().encode('ascii')
+            except UnicodeEncodeError:
+                return _('IDN in certificate not supported')
+            if matchdnsname(certname):
+                return None
+            return _('certificate is for %s') % certname
+    return _('no commonName or subjectAltName found in certificate')
+
+
+# CERT_REQUIRED means fetch the cert from the server all the time AND
+# validate it against the CA store provided in web.cacerts.
+#
+# We COMPLETELY ignore CERT_REQUIRED on Python <= 2.5, as it's totally
+# busted on those versions.
+
+def sslkwargs(ui, host):
+    cacerts = ui.config('web', 'cacerts')
+    hostfingerprint = ui.config('hostfingerprints', host)
+    if cacerts and not hostfingerprint:
+        cacerts = util.expandpath(cacerts)
+        if not os.path.exists(cacerts):
+            raise util.Abort(_('could not find web.cacerts: %s') % cacerts)
+        return {'ca_certs': cacerts,
+                'cert_reqs': CERT_REQUIRED,
+                }
+    return {}
+
+class validator(object):
+    def __init__(self, ui, host):
+        self.ui = ui
+        self.host = host
+
+    def __call__(self, sock):
+        host = self.host
+        cacerts = self.ui.config('web', 'cacerts')
+        hostfingerprint = self.ui.config('hostfingerprints', host)
+        if cacerts and not hostfingerprint:
+            msg = _verifycert(sock.getpeercert(), host)
+            if msg:
+                raise util.Abort(_('%s certificate error: %s '
+                                   '(use --insecure to connect '
+                                   'insecurely)') % (host, msg))
+            self.ui.debug('%s certificate successfully verified\n' % host)
+        else:
+            if getattr(sock, 'getpeercert', False):
+                peercert = sock.getpeercert(True)
+                peerfingerprint = util.sha1(peercert).hexdigest()
+                nicefingerprint = ":".join([peerfingerprint[x:x + 2]
+                    for x in xrange(0, len(peerfingerprint), 2)])
+                if hostfingerprint:
+                    if peerfingerprint.lower() != \
+                            hostfingerprint.replace(':', '').lower():
+                        raise util.Abort(_('invalid certificate for %s '
+                                           'with fingerprint %s') %
+                                         (host, nicefingerprint))
+                    self.ui.debug('%s certificate matched fingerprint %s\n' %
+                                  (host, nicefingerprint))
+                else:
+                    self.ui.warn(_('warning: %s certificate '
+                                   'with fingerprint %s not verified '
+                                   '(check hostfingerprints or web.cacerts '
+                                   'config setting)\n') %
+                                 (host, nicefingerprint))
+            else: # python 2.5 ?
+                if hostfingerprint:
+                    raise util.Abort(_('no certificate for %s with '
+                                       'configured hostfingerprint') % host)
+                self.ui.warn(_('warning: %s certificate not verified '
+                               '(check web.cacerts config setting)\n') %
+                             host)

mercurial/url.py

@@ -1,625 +1,530 b''
 # url.py - HTTP 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.
 
 import urllib, urllib2, httplib, os, socket, cStringIO
 import __builtin__
 from i18n import _
-import keepalive, util
+import keepalive, util, sslutil
 
 def readauthforuri(ui, uri):
     # Read configuration
     config = dict()
     for key, val in ui.configitems('auth'):
         if '.' not in key:
             ui.warn(_("ignoring invalid [auth] key '%s'\n") % key)
             continue
         group, setting = key.rsplit('.', 1)
         gdict = config.setdefault(group, dict())
         if setting in ('username', 'cert', 'key'):
             val = util.expandpath(val)
         gdict[setting] = val
 
     # Find the best match
     scheme, hostpath = uri.split('://', 1)
     bestlen = 0
     bestauth = None
     for group, auth in config.iteritems():
         prefix = auth.get('prefix')
         if not prefix:
             continue
         p = prefix.split('://', 1)
         if len(p) > 1:
             schemes, prefix = [p[0]], p[1]
         else:
             schemes = (auth.get('schemes') or 'https').split()
         if (prefix == '*' or hostpath.startswith(prefix)) and \
             len(prefix) > bestlen and scheme in schemes:
             bestlen = len(prefix)
             bestauth = group, auth
     return bestauth
 
 class passwordmgr(urllib2.HTTPPasswordMgrWithDefaultRealm):
     def __init__(self, ui):
         urllib2.HTTPPasswordMgrWithDefaultRealm.__init__(self)
         self.ui = ui
 
     def find_user_password(self, realm, authuri):
         authinfo = urllib2.HTTPPasswordMgrWithDefaultRealm.find_user_password(
             self, realm, authuri)
         user, passwd = authinfo
         if user and passwd:
             self._writedebug(user, passwd)
             return (user, passwd)
 
         if not user:
             res = readauthforuri(self.ui, authuri)
             if res:
                 group, auth = res
                 user, passwd = auth.get('username'), auth.get('password')
                 self.ui.debug("using auth.%s.* for authentication\n" % group)
         if not user or not passwd:
             if not self.ui.interactive():
                 raise util.Abort(_('http authorization required'))
 
             self.ui.write(_("http authorization required\n"))
             self.ui.write(_("realm: %s\n") % realm)
             if user:
                 self.ui.write(_("user: %s\n") % user)
             else:
                 user = self.ui.prompt(_("user:"), default=None)
 
             if not passwd:
                 passwd = self.ui.getpass()
 
         self.add_password(realm, authuri, user, passwd)
         self._writedebug(user, passwd)
         return (user, passwd)
 
     def _writedebug(self, user, passwd):
         msg = _('http auth: user %s, password %s\n')
         self.ui.debug(msg % (user, passwd and '*' * len(passwd) or 'not set'))
 
 class proxyhandler(urllib2.ProxyHandler):
     def __init__(self, ui):
         proxyurl = ui.config("http_proxy", "host") or os.getenv('http_proxy')
         # XXX proxyauthinfo = None
 
         if proxyurl:
             # proxy can be proper url or host[:port]
             if not (proxyurl.startswith('http:') or
                     proxyurl.startswith('https:')):
                 proxyurl = 'http://' + proxyurl + '/'
             proxy = util.url(proxyurl)
             if not proxy.user:
                 proxy.user = ui.config("http_proxy", "user")
                 proxy.passwd = ui.config("http_proxy", "passwd")
 
             # see if we should use a proxy for this url
             no_list = ["localhost", "127.0.0.1"]
             no_list.extend([p.lower() for
                             p in ui.configlist("http_proxy", "no")])
             no_list.extend([p.strip().lower() for
                             p in os.getenv("no_proxy", '').split(',')
                             if p.strip()])
             # "http_proxy.always" config is for running tests on localhost
             if ui.configbool("http_proxy", "always"):
                 self.no_list = []
             else:
                 self.no_list = no_list
 
             proxyurl = str(proxy)
             proxies = {'http': proxyurl, 'https': proxyurl}
             ui.debug('proxying through http://%s:%s\n' %
                       (proxy.host, proxy.port))
         else:
             proxies = {}
 
         # urllib2 takes proxy values from the environment and those
         # will take precedence if found, so drop them
         for env in ["HTTP_PROXY", "http_proxy", "no_proxy"]:
             try:
                 if env in os.environ:
                     del os.environ[env]
             except OSError:
                 pass
 
         urllib2.ProxyHandler.__init__(self, proxies)
         self.ui = ui
 
     def proxy_open(self, req, proxy, type_):
         host = req.get_host().split(':')[0]
         if host in self.no_list:
             return None
 
         # work around a bug in Python < 2.4.2
         # (it leaves a "\n" at the end of Proxy-authorization headers)
         baseclass = req.__class__
         class _request(baseclass):
             def add_header(self, key, val):
                 if key.lower() == 'proxy-authorization':
                     val = val.strip()
                 return baseclass.add_header(self, key, val)
         req.__class__ = _request
 
         return urllib2.ProxyHandler.proxy_open(self, req, proxy, type_)
 
 class httpsendfile(object):
     """This is a wrapper around the objects returned by python's "open".
 
     Its purpose is to send file-like objects via HTTP and, to do so, it
     defines a __len__ attribute to feed the Content-Length header.
     """
 
     def __init__(self, ui, *args, **kwargs):
         # We can't just "self._data = open(*args, **kwargs)" here because there
         # is an "open" function defined in this module that shadows the global
         # one
         self.ui = ui
         self._data = __builtin__.open(*args, **kwargs)
         self.seek = self._data.seek
         self.close = self._data.close
         self.write = self._data.write
         self._len = os.fstat(self._data.fileno()).st_size
         self._pos = 0
         self._total = len(self) / 1024 * 2
 
     def read(self, *args, **kwargs):
         try:
             ret = self._data.read(*args, **kwargs)
         except EOFError:
             self.ui.progress(_('sending'), None)
         self._pos += len(ret)
         # We pass double the max for total because we currently have
         # to send the bundle twice in the case of a server that
         # requires authentication. Since we can't know until we try
         # once whether authentication will be required, just lie to
         # the user and maybe the push succeeds suddenly at 50%.
         self.ui.progress(_('sending'), self._pos / 1024,
                          unit=_('kb'), total=self._total)
         return ret
 
     def __len__(self):
         return self._len
 
 def _gen_sendfile(orgsend):
     def _sendfile(self, data):
         # send a file
         if isinstance(data, httpsendfile):
             # if auth required, some data sent twice, so rewind here
             data.seek(0)
             for chunk in util.filechunkiter(data):
                 orgsend(self, chunk)
         else:
             orgsend(self, data)
     return _sendfile
 
 has_https = hasattr(urllib2, 'HTTPSHandler')
 if has_https:
     try:
-        # avoid using deprecated/broken FakeSocket in python 2.6
-        import ssl
-        _ssl_wrap_socket = ssl.wrap_socket
-        CERT_REQUIRED = ssl.CERT_REQUIRED
-    except ImportError:
-        CERT_REQUIRED = 2
-
-        def _ssl_wrap_socket(sock, key_file, cert_file,
-                             cert_reqs=CERT_REQUIRED, ca_certs=None):
-            if ca_certs:
-                raise util.Abort(_(
-                    'certificate checking requires Python 2.6'))
-
-            ssl = socket.ssl(sock, key_file, cert_file)
-            return httplib.FakeSocket(sock, ssl)
-
-    try:
         _create_connection = socket.create_connection
     except AttributeError:
         _GLOBAL_DEFAULT_TIMEOUT = object()
 
         def _create_connection(address, timeout=_GLOBAL_DEFAULT_TIMEOUT,
                                source_address=None):
             # lifted from Python 2.6
 
             msg = "getaddrinfo returns an empty list"
             host, port = address
             for res in socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM):
                 af, socktype, proto, canonname, sa = res
                 sock = None
                 try:
                     sock = socket.socket(af, socktype, proto)
                     if timeout is not _GLOBAL_DEFAULT_TIMEOUT:
                         sock.settimeout(timeout)
                     if source_address:
                         sock.bind(source_address)
                     sock.connect(sa)
                     return sock
 
                 except socket.error, msg:
                     if sock is not None:
                         sock.close()
 
             raise socket.error, msg
 
 class httpconnection(keepalive.HTTPConnection):
     # must be able to send big bundle as stream.
     send = _gen_sendfile(keepalive.HTTPConnection.send)
 
     def connect(self):
         if has_https and self.realhostport: # use CONNECT proxy
             self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             self.sock.connect((self.host, self.port))
             if _generic_proxytunnel(self):
                 # we do not support client x509 certificates
-                self.sock = _ssl_wrap_socket(self.sock, None, None)
+                self.sock = sslutil.ssl_wrap_socket(self.sock, None, None)
         else:
             keepalive.HTTPConnection.connect(self)
 
     def getresponse(self):
         proxyres = getattr(self, 'proxyres', None)
         if proxyres:
             if proxyres.will_close:
                 self.close()
             self.proxyres = None
             return proxyres
         return keepalive.HTTPConnection.getresponse(self)
 
 # general transaction handler to support different ways to handle
 # HTTPS proxying before and after Python 2.6.3.
 def _generic_start_transaction(handler, h, req):
     if hasattr(req, '_tunnel_host') and req._tunnel_host:
         tunnel_host = req._tunnel_host
         if tunnel_host[:7] not in ['http://', 'https:/']:
             tunnel_host = 'https://' + tunnel_host
         new_tunnel = True
     else:
         tunnel_host = req.get_selector()
         new_tunnel = False
 
     if new_tunnel or tunnel_host == req.get_full_url(): # has proxy
         u = util.url(tunnel_host)
         if new_tunnel or u.scheme == 'https': # only use CONNECT for HTTPS
             h.realhostport = ':'.join([u.host, (u.port or '443')])
             h.headers = req.headers.copy()
             h.headers.update(handler.parent.addheaders)
             return
 
     h.realhostport = None
     h.headers = None
 
 def _generic_proxytunnel(self):
     proxyheaders = dict(
             [(x, self.headers[x]) for x in self.headers
              if x.lower().startswith('proxy-')])
     self._set_hostport(self.host, self.port)
     self.send('CONNECT %s HTTP/1.0\r\n' % self.realhostport)
     for header in proxyheaders.iteritems():
         self.send('%s: %s\r\n' % header)
     self.send('\r\n')
 
     # majority of the following code is duplicated from
     # httplib.HTTPConnection as there are no adequate places to
     # override functions to provide the needed functionality
     res = self.response_class(self.sock,
                               strict=self.strict,
                               method=self._method)
 
     while True:
         version, status, reason = res._read_status()
         if status != httplib.CONTINUE:
             break
         while True:
             skip = res.fp.readline().strip()
             if not skip:
                 break
     res.status = status
     res.reason = reason.strip()
 
     if res.status == 200:
         while True:
             line = res.fp.readline()
             if line == '\r\n':
                 break
         return True
 
     if version == 'HTTP/1.0':
         res.version = 10
     elif version.startswith('HTTP/1.'):
         res.version = 11
     elif version == 'HTTP/0.9':
         res.version = 9
     else:
         raise httplib.UnknownProtocol(version)
 
     if res.version == 9:
         res.length = None
         res.chunked = 0
         res.will_close = 1
         res.msg = httplib.HTTPMessage(cStringIO.StringIO())
         return False
 
     res.msg = httplib.HTTPMessage(res.fp)
     res.msg.fp = None
 
     # are we using the chunked-style of transfer encoding?
     trenc = res.msg.getheader('transfer-encoding')
     if trenc and trenc.lower() == "chunked":
         res.chunked = 1
         res.chunk_left = None
     else:
         res.chunked = 0
 
     # will the connection close at the end of the response?
     res.will_close = res._check_close()
 
     # do we have a Content-Length?
     # NOTE: RFC 2616, S4.4, #3 says we ignore this if tr_enc is "chunked"
     length = res.msg.getheader('content-length')
     if length and not res.chunked:
         try:
             res.length = int(length)
         except ValueError:
             res.length = None
         else:
             if res.length < 0:  # ignore nonsensical negative lengths
                 res.length = None
     else:
         res.length = None
 
     # does the body have a fixed length? (of zero)
     if (status == httplib.NO_CONTENT or status == httplib.NOT_MODIFIED or
         100 <= status < 200 or # 1xx codes
         res._method == 'HEAD'):
         res.length = 0
 
     # if the connection remains open, and we aren't using chunked, and
     # a content-length was not provided, then assume that the connection
     # WILL close.
     if (not res.will_close and
        not res.chunked and
        res.length is None):
         res.will_close = 1
 
     self.proxyres = res
 
     return False
 
 class httphandler(keepalive.HTTPHandler):
     def http_open(self, req):
         return self.do_open(httpconnection, req)
 
     def _start_transaction(self, h, req):
         _generic_start_transaction(self, h, req)
         return keepalive.HTTPHandler._start_transaction(self, h, req)
 
-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 _('no certificate received')
-    dnsname = hostname.lower()
-    def matchdnsname(certname):
-        return (certname == dnsname or
-                '.' in dnsname and certname == '*.' + dnsname.split('.', 1)[1])
-
-    san = cert.get('subjectAltName', [])
-    if san:
-        certnames = [value.lower() for key, value in san if key == 'DNS']
-        for name in certnames:
-            if matchdnsname(name):
-                return None
-        return _('certificate is for %s') % ', '.join(certnames)
-
-    # subject is only checked when subjectAltName is empty
-    for s in cert.get('subject', []):
-        key, value = s[0]
-        if key == 'commonName':
-            try:
-                # 'subject' entries are unicode
-                certname = value.lower().encode('ascii')
-            except UnicodeEncodeError:
-                return _('IDN in certificate not supported')
-            if matchdnsname(certname):
-                return None
-            return _('certificate is for %s') % certname
-    return _('no commonName or subjectAltName found in certificate')
-
 if has_https:
     class httpsconnection(httplib.HTTPSConnection):
         response_class = keepalive.HTTPResponse
         # must be able to send big bundle as stream.
         send = _gen_sendfile(keepalive.safesend)
         getresponse = keepalive.wrapgetresponse(httplib.HTTPSConnection)
 
         def connect(self):
             self.sock = _create_connection((self.host, self.port))
 
             host = self.host
             if self.realhostport: # use CONNECT proxy
                 _generic_proxytunnel(self)
                 host = self.realhostport.rsplit(':', 1)[0]
-
-            cacerts = self.ui.config('web', 'cacerts')
-            hostfingerprint = self.ui.config('hostfingerprints', host)
-
-            if cacerts and not hostfingerprint:
-                cacerts = util.expandpath(cacerts)
-                if not os.path.exists(cacerts):
-                    raise util.Abort(_('could not find '
-                                       'web.cacerts: %s') % cacerts)
-                self.sock = _ssl_wrap_socket(self.sock, self.key_file,
-                    self.cert_file, cert_reqs=CERT_REQUIRED,
-                    ca_certs=cacerts)
-                msg = _verifycert(self.sock.getpeercert(), host)
-                if msg:
-                    raise util.Abort(_('%s certificate error: %s '
-                                       '(use --insecure to connect '
-                                       'insecurely)') % (host, msg))
-                self.ui.debug('%s certificate successfully verified\n' % host)
-            else:
-                self.sock = _ssl_wrap_socket(self.sock, self.key_file,
-                    self.cert_file)
-                if hasattr(self.sock, 'getpeercert'):
-                    peercert = self.sock.getpeercert(True)
-                    peerfingerprint = util.sha1(peercert).hexdigest()
-                    nicefingerprint = ":".join([peerfingerprint[x:x + 2]
-                        for x in xrange(0, len(peerfingerprint), 2)])
-                    if hostfingerprint:
-                        if peerfingerprint.lower() != \
-                                hostfingerprint.replace(':', '').lower():
-                            raise util.Abort(_('invalid certificate for %s '
-                                               'with fingerprint %s') %
-                                             (host, nicefingerprint))
-                        self.ui.debug('%s certificate matched fingerprint %s\n' %
-                                      (host, nicefingerprint))
-                    else:
-                        self.ui.warn(_('warning: %s certificate '
-                                       'with fingerprint %s not verified '
-                                       '(check hostfingerprints or web.cacerts '
-                                       'config setting)\n') %
-                                     (host, nicefingerprint))
-                else: # python 2.5 ?
-                    if hostfingerprint:
-                        raise util.Abort(_('no certificate for %s with '
-                                           'configured hostfingerprint') % host)
-                    self.ui.warn(_('warning: %s certificate not verified '
-                                   '(check web.cacerts config setting)\n') %
-                                 host)
+            self.sock = sslutil.ssl_wrap_socket(
+                self.sock, self.key_file, self.cert_file,
+                **sslutil.sslkwargs(self.ui, host))
+            sslutil.validator(self.ui, host)(self.sock)
 
     class httpshandler(keepalive.KeepAliveHandler, urllib2.HTTPSHandler):
         def __init__(self, ui):
             keepalive.KeepAliveHandler.__init__(self)
             urllib2.HTTPSHandler.__init__(self)
             self.ui = ui
             self.pwmgr = passwordmgr(self.ui)
 
         def _start_transaction(self, h, req):
             _generic_start_transaction(self, h, req)
             return keepalive.KeepAliveHandler._start_transaction(self, h, req)
 
         def https_open(self, req):
             res = readauthforuri(self.ui, req.get_full_url())
             if res:
                 group, auth = res
                 self.auth = auth
                 self.ui.debug("using auth.%s.* for authentication\n" % group)
             else:
                 self.auth = None
             return self.do_open(self._makeconnection, req)
 
         def _makeconnection(self, host, port=None, *args, **kwargs):
             keyfile = None
             certfile = None
 
             if len(args) >= 1: # key_file
                 keyfile = args[0]
             if len(args) >= 2: # cert_file
                 certfile = args[1]
             args = args[2:]
 
             # if the user has specified different key/cert files in
             # hgrc, we prefer these
             if self.auth and 'key' in self.auth and 'cert' in self.auth:
                 keyfile = self.auth['key']
                 certfile = self.auth['cert']
 
             conn = httpsconnection(host, port, keyfile, certfile, *args, **kwargs)
             conn.ui = self.ui
             return conn
 
 class httpdigestauthhandler(urllib2.HTTPDigestAuthHandler):
     def __init__(self, *args, **kwargs):
         urllib2.HTTPDigestAuthHandler.__init__(self, *args, **kwargs)
         self.retried_req = None
 
     def reset_retry_count(self):
         # Python 2.6.5 will call this on 401 or 407 errors and thus loop
         # forever. We disable reset_retry_count completely and reset in
         # http_error_auth_reqed instead.
         pass
 
     def http_error_auth_reqed(self, auth_header, host, req, headers):
         # Reset the retry counter once for each request.
         if req is not self.retried_req:
             self.retried_req = req
             self.retried = 0
         # In python < 2.5 AbstractDigestAuthHandler raises a ValueError if
         # it doesn't know about the auth type requested. This can happen if
         # somebody is using BasicAuth and types a bad password.
         try:
             return urllib2.HTTPDigestAuthHandler.http_error_auth_reqed(
                         self, auth_header, host, req, headers)
         except ValueError, inst:
             arg = inst.args[0]
             if arg.startswith("AbstractDigestAuthHandler doesn't know "):
                 return
             raise
 
 class httpbasicauthhandler(urllib2.HTTPBasicAuthHandler):
     def __init__(self, *args, **kwargs):
         urllib2.HTTPBasicAuthHandler.__init__(self, *args, **kwargs)
         self.retried_req = None
 
     def reset_retry_count(self):
         # Python 2.6.5 will call this on 401 or 407 errors and thus loop
         # forever. We disable reset_retry_count completely and reset in
         # http_error_auth_reqed instead.
         pass
 
     def http_error_auth_reqed(self, auth_header, host, req, headers):
         # Reset the retry counter once for each request.
         if req is not self.retried_req:
             self.retried_req = req
             self.retried = 0
         return urllib2.HTTPBasicAuthHandler.http_error_auth_reqed(
                         self, auth_header, host, req, headers)
 
 handlerfuncs = []
 
 def opener(ui, authinfo=None):
     '''
     construct an opener suitable for urllib2
     authinfo will be added to the password manager
     '''
     handlers = [httphandler()]
     if has_https:
         handlers.append(httpshandler(ui))
 
     handlers.append(proxyhandler(ui))
 
     passmgr = passwordmgr(ui)
     if authinfo is not None:
         passmgr.add_password(*authinfo)
         user, passwd = authinfo[2:4]
         ui.debug('http auth: user %s, password %s\n' %
                  (user, passwd and '*' * len(passwd) or 'not set'))
 
     handlers.extend((httpbasicauthhandler(passmgr),
                      httpdigestauthhandler(passmgr)))
     handlers.extend([h(ui, passmgr) for h in handlerfuncs])
     opener = urllib2.build_opener(*handlers)
 
     # 1.0 here is the _protocol_ version
     opener.addheaders = [('User-agent', 'mercurial/proto-1.0')]
     opener.addheaders.append(('Accept', 'application/mercurial-0.1'))
     return opener
 
 def open(ui, url_, data=None):
     u = util.url(url_)
     if u.scheme:
         u.scheme = u.scheme.lower()
         url_, authinfo = u.authinfo()
     else:
         path = util.normpath(os.path.abspath(url_))
         url_ = 'file://' + urllib.pathname2url(path)
         authinfo = None
     return opener(ui, authinfo).open(url_, data)

tests/test-url.py

@@ -1,205 +1,205 b''
 import sys
 
 def check(a, b):
     if a != b:
         print (a, b)
 
 def cert(cn):
     return dict(subject=((('commonName', cn),),))
 
-from mercurial.url import _verifycert
+from mercurial.sslutil import _verifycert
 
 # Test non-wildcard certificates
 check(_verifycert(cert('example.com'), 'example.com'),
       None)
 check(_verifycert(cert('example.com'), 'www.example.com'),
       'certificate is for example.com')
 check(_verifycert(cert('www.example.com'), 'example.com'),
       'certificate is for www.example.com')
 
 # Test wildcard certificates
 check(_verifycert(cert('*.example.com'), 'www.example.com'),
       None)
 check(_verifycert(cert('*.example.com'), 'example.com'),
       'certificate is for *.example.com')
 check(_verifycert(cert('*.example.com'), 'w.w.example.com'),
       'certificate is for *.example.com')
 
 # Test subjectAltName
 san_cert = {'subject': ((('commonName', 'example.com'),),),
             'subjectAltName': (('DNS', '*.example.net'),
                                ('DNS', 'example.net'))}
 check(_verifycert(san_cert, 'example.net'),
       None)
 check(_verifycert(san_cert, 'foo.example.net'),
       None)
 # subject is only checked when subjectAltName is empty
 check(_verifycert(san_cert, 'example.com'),
       'certificate is for *.example.net, example.net')
 
 # Avoid some pitfalls
 check(_verifycert(cert('*.foo'), 'foo'),
       'certificate is for *.foo')
 check(_verifycert(cert('*o'), 'foo'),
       'certificate is for *o')
 
 check(_verifycert({'subject': ()},
                   'example.com'),
       'no commonName or subjectAltName found in certificate')
 check(_verifycert(None, 'example.com'),
       'no certificate received')
 
 import doctest
 
 def test_url():
     """
     >>> from mercurial.util import url
 
     This tests for edge cases in url.URL's parsing algorithm. Most of
     these aren't useful for documentation purposes, so they aren't
     part of the class's doc tests.
 
     Query strings and fragments:
 
     >>> url('http://host/a?b#c')
     <url scheme: 'http', host: 'host', path: 'a', query: 'b', fragment: 'c'>
     >>> url('http://host/a?')
     <url scheme: 'http', host: 'host', path: 'a'>
     >>> url('http://host/a#b#c')
     <url scheme: 'http', host: 'host', path: 'a', fragment: 'b#c'>
     >>> url('http://host/a#b?c')
     <url scheme: 'http', host: 'host', path: 'a', fragment: 'b?c'>
     >>> url('http://host/?a#b')
     <url scheme: 'http', host: 'host', path: '', query: 'a', fragment: 'b'>
     >>> url('http://host/?a#b', parsequery=False)
     <url scheme: 'http', host: 'host', path: '?a', fragment: 'b'>
     >>> url('http://host/?a#b', parsefragment=False)
     <url scheme: 'http', host: 'host', path: '', query: 'a#b'>
     >>> url('http://host/?a#b', parsequery=False, parsefragment=False)
     <url scheme: 'http', host: 'host', path: '?a#b'>
 
     IPv6 addresses:
 
     >>> url('ldap://[2001:db8::7]/c=GB?objectClass?one')
     <url scheme: 'ldap', host: '[2001:db8::7]', path: 'c=GB',
          query: 'objectClass?one'>
     >>> url('ldap://joe:xxx@[2001:db8::7]:80/c=GB?objectClass?one')
     <url scheme: 'ldap', user: 'joe', passwd: 'xxx', host: '[2001:db8::7]',
          port: '80', path: 'c=GB', query: 'objectClass?one'>
 
     Missing scheme, host, etc.:
 
     >>> url('://192.0.2.16:80/')
     <url path: '://192.0.2.16:80/'>
     >>> url('http://mercurial.selenic.com')
     <url scheme: 'http', host: 'mercurial.selenic.com'>
     >>> url('/foo')
     <url path: '/foo'>
     >>> url('bundle:/foo')
     <url scheme: 'bundle', path: '/foo'>
     >>> url('a?b#c')
     <url path: 'a?b', fragment: 'c'>
     >>> url('http://x.com?arg=/foo')
     <url scheme: 'http', host: 'x.com', query: 'arg=/foo'>
     >>> url('http://joe:xxx@/foo')
     <url scheme: 'http', user: 'joe', passwd: 'xxx', path: 'foo'>
 
     Just a scheme and a path:
 
     >>> url('mailto:[e-mail]')
     <url scheme: 'mailto', path: '[e-mail]'>
     >>> url('a:b:c:d')
     <url path: 'a:b:c:d'>
     >>> url('aa:bb:cc:dd')
     <url scheme: 'aa', path: 'bb:cc:dd'>
 
     SSH examples:
 
     >>> url('ssh://joe@host//home/joe')
     <url scheme: 'ssh', user: 'joe', host: 'host', path: '/home/joe'>
     >>> url('ssh://joe:xxx@host/src')
     <url scheme: 'ssh', user: 'joe', passwd: 'xxx', host: 'host', path: 'src'>
     >>> url('ssh://joe:xxx@host')
     <url scheme: 'ssh', user: 'joe', passwd: 'xxx', host: 'host'>
     >>> url('ssh://joe@host')
     <url scheme: 'ssh', user: 'joe', host: 'host'>
     >>> url('ssh://host')
     <url scheme: 'ssh', host: 'host'>
     >>> url('ssh://')
     <url scheme: 'ssh'>
     >>> url('ssh:')
     <url scheme: 'ssh'>
 
     Non-numeric port:
 
     >>> url('http://example.com:dd')
     <url scheme: 'http', host: 'example.com', port: 'dd'>
     >>> url('ssh://joe:xxx@host:ssh/foo')
     <url scheme: 'ssh', user: 'joe', passwd: 'xxx', host: 'host', port: 'ssh',
          path: 'foo'>
 
     Bad authentication credentials:
 
     >>> url('http://joe@joeville:123@4:@host/a?b#c')
     <url scheme: 'http', user: 'joe@joeville', passwd: '123@4:',
          host: 'host', path: 'a', query: 'b', fragment: 'c'>
     >>> url('http://!*#?/@!*#?/:@host/a?b#c')
     <url scheme: 'http', host: '!*', fragment: '?/@!*#?/:@host/a?b#c'>
     >>> url('http://!*#?@!*#?:@host/a?b#c')
     <url scheme: 'http', host: '!*', fragment: '?@!*#?:@host/a?b#c'>
     >>> url('http://!*@:!*@@host/a?b#c')
     <url scheme: 'http', user: '!*@', passwd: '!*@', host: 'host',
          path: 'a', query: 'b', fragment: 'c'>
 
     File paths:
 
     >>> url('a/b/c/d.g.f')
     <url path: 'a/b/c/d.g.f'>
     >>> url('/x///z/y/')
     <url path: '/x///z/y/'>
     >>> url('/foo:bar')
     <url path: '/foo:bar'>
     >>> url('\\\\foo:bar')
     <url path: '\\\\foo:bar'>
     >>> url('./foo:bar')
     <url path: './foo:bar'>
 
     Non-localhost file URL:
 
     >>> u = url('file://mercurial.selenic.com/foo')
     Traceback (most recent call last):
       File "<stdin>", line 1, in ?
     Abort: file:// URLs can only refer to localhost
 
     Empty URL:
 
     >>> u = url('')
     >>> u
     <url path: ''>
     >>> str(u)
     ''
 
     Empty path with query string:
 
     >>> str(url('http://foo/?bar'))
     'http://foo/?bar'
 
     Invalid path:
 
     >>> u = url('http://foo/bar')
     >>> u.path = 'bar'
     >>> str(u)
     'http://foo/bar'
 
     >>> u = url('file:///foo/bar/baz')
     >>> u
     <url scheme: 'file', path: '/foo/bar/baz'>
     >>> str(u)
     'file:/foo/bar/baz'
     """
 
 doctest.testmod(optionflags=doctest.NORMALIZE_WHITESPACE)
 
 # Unicode (IDN) certname isn't supported
 check(_verifycert(cert(u'\u4f8b.jp'), 'example.jp'),
       'IDN in certificate not supported')