comparison mercurial/url.py @ 12592:f2937d6492c5 stable

url: verify correctness of https server certificates (issue2407)

Pythons SSL module verifies that certificates received for HTTPS are valid
according to the specified cacerts, but it doesn't verify that the certificate
is for the host we connect to.

We now explicitly verify that the commonName in the received certificate
matches the requested hostname and is valid for the time being.

This is a minimal patch where we try to fail to the safe side, but we do still
rely on Python's SSL functionality and do not try to implement the standards
fully and correctly. CRLs and subjectAltName are not handled and proxies
haven't been considered.

This change might break connections to some sites if cacerts is specified and
the certificates (by our definition) isn't correct. The workaround is to
disable cacerts which in most cases isn't much worse than it was before with
cacerts.

12576:1c9bb7e00f71

# 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, urlparse, httplib, os, re, socket, cStringIO
from i18n import _
import keepalive, util

def _urlunparse(scheme, netloc, path, params, query, fragment, url):
    '''Handle cases where urlunparse(urlparse(x://)) doesn't preserve the "//"'''

    def _start_transaction(self, h, req):
        _generic_start_transaction(self, h, req)
        return keepalive.HTTPHandler._start_transaction(self, h, req)

if has_https:
    class BetterHTTPS(httplib.HTTPSConnection):
        send = keepalive.safesend

        def connect(self):

            if cacerts:
                sock = _create_connection((self.host, self.port))
                self.sock = _ssl_wrap_socket(sock, self.key_file,
                        self.cert_file, cert_reqs=CERT_REQUIRED,
                        ca_certs=cacerts)
                self.ui.debug(_('server identity verification succeeded\n'))
            else:
                httplib.HTTPSConnection.connect(self)

    class httpsconnection(BetterHTTPS):
        response_class = keepalive.HTTPResponse

12592:f2937d6492c5

# 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, urlparse, httplib, os, re, socket, cStringIO, time
from i18n import _
import keepalive, util

def _urlunparse(scheme, netloc, path, params, query, fragment, url):
    '''Handle cases where urlunparse(urlparse(x://)) doesn't preserve the "//"'''

    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 and is 
    valid at this time. CRLs and subjectAltName are not handled.
    
    Returns error message if any problems are found and None on success.
    '''
    if not cert:
        return _('no certificate received')
    notafter = cert.get('notAfter')
    if notafter and time.time() > ssl.cert_time_to_seconds(notafter):
        return _('certificate expired %s') % notafter
    notbefore = cert.get('notBefore')
    if notbefore and time.time() < ssl.cert_time_to_seconds(notbefore):
        return _('certificate not valid before %s') % notbefore
    dnsname = hostname.lower()
    for s in cert.get('subject', []):
        key, value = s[0]
        if key == 'commonName':
            certname = value.lower()
            if (certname == dnsname or
                '.' in dnsname and certname == '*.' + dnsname.split('.', 1)[1]):
                return None
            return _('certificate is for %s') % certname
    return _('no commonName found in certificate')

if has_https:
    class BetterHTTPS(httplib.HTTPSConnection):
        send = keepalive.safesend

        def connect(self):

            if cacerts:
                sock = _create_connection((self.host, self.port))
                self.sock = _ssl_wrap_socket(sock, self.key_file,
                        self.cert_file, cert_reqs=CERT_REQUIRED,
                        ca_certs=cacerts)
                msg = _verifycert(self.sock.getpeercert(), self.host)
                if msg:
                    raise util.Abort('%s certificate error: %s' % (self.host, msg))
                self.ui.debug(_('%s certificate successfully verified\n') % 
                    self.host)
            else:
                httplib.HTTPSConnection.connect(self)

    class httpsconnection(BetterHTTPS):
        response_class = keepalive.HTTPResponse