• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import os
2import socket
3import atexit
4import re
5import functools
6
7from setuptools.extern.six.moves import urllib, http_client, map, filter
8
9from pkg_resources import ResolutionError, ExtractionError
10
11try:
12    import ssl
13except ImportError:
14    ssl = None
15
16__all__ = [
17    'VerifyingHTTPSHandler', 'find_ca_bundle', 'is_available', 'cert_paths',
18    'opener_for'
19]
20
21cert_paths = """
22/etc/pki/tls/certs/ca-bundle.crt
23/etc/ssl/certs/ca-certificates.crt
24/usr/share/ssl/certs/ca-bundle.crt
25/usr/local/share/certs/ca-root.crt
26/etc/ssl/cert.pem
27/System/Library/OpenSSL/certs/cert.pem
28/usr/local/share/certs/ca-root-nss.crt
29/etc/ssl/ca-bundle.pem
30""".strip().split()
31
32try:
33    HTTPSHandler = urllib.request.HTTPSHandler
34    HTTPSConnection = http_client.HTTPSConnection
35except AttributeError:
36    HTTPSHandler = HTTPSConnection = object
37
38is_available = ssl is not None and object not in (HTTPSHandler, HTTPSConnection)
39
40
41try:
42    from ssl import CertificateError, match_hostname
43except ImportError:
44    try:
45        from backports.ssl_match_hostname import CertificateError
46        from backports.ssl_match_hostname import match_hostname
47    except ImportError:
48        CertificateError = None
49        match_hostname = None
50
51if not CertificateError:
52
53    class CertificateError(ValueError):
54        pass
55
56
57if not match_hostname:
58
59    def _dnsname_match(dn, hostname, max_wildcards=1):
60        """Matching according to RFC 6125, section 6.4.3
61
62        http://tools.ietf.org/html/rfc6125#section-6.4.3
63        """
64        pats = []
65        if not dn:
66            return False
67
68        # Ported from python3-syntax:
69        # leftmost, *remainder = dn.split(r'.')
70        parts = dn.split(r'.')
71        leftmost = parts[0]
72        remainder = parts[1:]
73
74        wildcards = leftmost.count('*')
75        if wildcards > max_wildcards:
76            # Issue #17980: avoid denials of service by refusing more
77            # than one wildcard per fragment.  A survey of established
78            # policy among SSL implementations showed it to be a
79            # reasonable choice.
80            raise CertificateError(
81                "too many wildcards in certificate DNS name: " + repr(dn))
82
83        # speed up common case w/o wildcards
84        if not wildcards:
85            return dn.lower() == hostname.lower()
86
87        # RFC 6125, section 6.4.3, subitem 1.
88        # The client SHOULD NOT attempt to match a presented identifier in which
89        # the wildcard character comprises a label other than the left-most label.
90        if leftmost == '*':
91            # When '*' is a fragment by itself, it matches a non-empty dotless
92            # fragment.
93            pats.append('[^.]+')
94        elif leftmost.startswith('xn--') or hostname.startswith('xn--'):
95            # RFC 6125, section 6.4.3, subitem 3.
96            # The client SHOULD NOT attempt to match a presented identifier
97            # where the wildcard character is embedded within an A-label or
98            # U-label of an internationalized domain name.
99            pats.append(re.escape(leftmost))
100        else:
101            # Otherwise, '*' matches any dotless string, e.g. www*
102            pats.append(re.escape(leftmost).replace(r'\*', '[^.]*'))
103
104        # add the remaining fragments, ignore any wildcards
105        for frag in remainder:
106            pats.append(re.escape(frag))
107
108        pat = re.compile(r'\A' + r'\.'.join(pats) + r'\Z', re.IGNORECASE)
109        return pat.match(hostname)
110
111    def match_hostname(cert, hostname):
112        """Verify that *cert* (in decoded format as returned by
113        SSLSocket.getpeercert()) matches the *hostname*.  RFC 2818 and RFC 6125
114        rules are followed, but IP addresses are not accepted for *hostname*.
115
116        CertificateError is raised on failure. On success, the function
117        returns nothing.
118        """
119        if not cert:
120            raise ValueError("empty or no certificate")
121        dnsnames = []
122        san = cert.get('subjectAltName', ())
123        for key, value in san:
124            if key == 'DNS':
125                if _dnsname_match(value, hostname):
126                    return
127                dnsnames.append(value)
128        if not dnsnames:
129            # The subject is only checked when there is no dNSName entry
130            # in subjectAltName
131            for sub in cert.get('subject', ()):
132                for key, value in sub:
133                    # XXX according to RFC 2818, the most specific Common Name
134                    # must be used.
135                    if key == 'commonName':
136                        if _dnsname_match(value, hostname):
137                            return
138                        dnsnames.append(value)
139        if len(dnsnames) > 1:
140            raise CertificateError("hostname %r "
141                "doesn't match either of %s"
142                % (hostname, ', '.join(map(repr, dnsnames))))
143        elif len(dnsnames) == 1:
144            raise CertificateError("hostname %r "
145                "doesn't match %r"
146                % (hostname, dnsnames[0]))
147        else:
148            raise CertificateError("no appropriate commonName or "
149                "subjectAltName fields were found")
150
151
152class VerifyingHTTPSHandler(HTTPSHandler):
153    """Simple verifying handler: no auth, subclasses, timeouts, etc."""
154
155    def __init__(self, ca_bundle):
156        self.ca_bundle = ca_bundle
157        HTTPSHandler.__init__(self)
158
159    def https_open(self, req):
160        return self.do_open(
161            lambda host, **kw: VerifyingHTTPSConn(host, self.ca_bundle, **kw), req
162        )
163
164
165class VerifyingHTTPSConn(HTTPSConnection):
166    """Simple verifying connection: no auth, subclasses, timeouts, etc."""
167
168    def __init__(self, host, ca_bundle, **kw):
169        HTTPSConnection.__init__(self, host, **kw)
170        self.ca_bundle = ca_bundle
171
172    def connect(self):
173        sock = socket.create_connection(
174            (self.host, self.port), getattr(self, 'source_address', None)
175        )
176
177        # Handle the socket if a (proxy) tunnel is present
178        if hasattr(self, '_tunnel') and getattr(self, '_tunnel_host', None):
179            self.sock = sock
180            self._tunnel()
181            # http://bugs.python.org/issue7776: Python>=3.4.1 and >=2.7.7
182            # change self.host to mean the proxy server host when tunneling is
183            # being used. Adapt, since we are interested in the destination
184            # host for the match_hostname() comparison.
185            actual_host = self._tunnel_host
186        else:
187            actual_host = self.host
188
189        if hasattr(ssl, 'create_default_context'):
190            ctx = ssl.create_default_context(cafile=self.ca_bundle)
191            self.sock = ctx.wrap_socket(sock, server_hostname=actual_host)
192        else:
193            # This is for python < 2.7.9 and < 3.4?
194            self.sock = ssl.wrap_socket(
195                sock, cert_reqs=ssl.CERT_REQUIRED, ca_certs=self.ca_bundle
196            )
197        try:
198            match_hostname(self.sock.getpeercert(), actual_host)
199        except CertificateError:
200            self.sock.shutdown(socket.SHUT_RDWR)
201            self.sock.close()
202            raise
203
204
205def opener_for(ca_bundle=None):
206    """Get a urlopen() replacement that uses ca_bundle for verification"""
207    return urllib.request.build_opener(
208        VerifyingHTTPSHandler(ca_bundle or find_ca_bundle())
209    ).open
210
211
212# from jaraco.functools
213def once(func):
214    @functools.wraps(func)
215    def wrapper(*args, **kwargs):
216        if not hasattr(func, 'always_returns'):
217            func.always_returns = func(*args, **kwargs)
218        return func.always_returns
219    return wrapper
220
221
222@once
223def get_win_certfile():
224    try:
225        import wincertstore
226    except ImportError:
227        return None
228
229    class CertFile(wincertstore.CertFile):
230        def __init__(self):
231            super(CertFile, self).__init__()
232            atexit.register(self.close)
233
234        def close(self):
235            try:
236                super(CertFile, self).close()
237            except OSError:
238                pass
239
240    _wincerts = CertFile()
241    _wincerts.addstore('CA')
242    _wincerts.addstore('ROOT')
243    return _wincerts.name
244
245
246def find_ca_bundle():
247    """Return an existing CA bundle path, or None"""
248    extant_cert_paths = filter(os.path.isfile, cert_paths)
249    return (
250        get_win_certfile()
251        or next(extant_cert_paths, None)
252        or _certifi_where()
253    )
254
255
256def _certifi_where():
257    try:
258        return __import__('certifi').where()
259    except (ImportError, ResolutionError, ExtractionError):
260        pass
261