CHANGES.rst

@@ -1,6 +1,14 @@
 Changes
 =======
 
+1.26.7 (2021-09-22)
+-------------------
+* Fixed a bug with HTTPS hostname verification involving IP addresses and lack
+  of SNI. (Issue #2400)
+* Fixed a bug where IPv6 braces weren't stripped during certificate hostname
+  matching. (Issue #2240)
+
+
 1.26.6 (2021-06-25)
 -------------------
 

src/urllib3/_version.py

@@ -1,2 +1,2 @@
 # This file is protected via CODEOWNERS
-__version__ = "1.26.6"
+__version__ = "1.26.7"

src/urllib3/connection.py

@@ -56,6 +56,7 @@ class BrokenPipeError(Exception):
 from .util.ssl_ import (
     assert_fingerprint,
     create_urllib3_context,
+    is_ipaddress,
     resolve_cert_reqs,
     resolve_ssl_version,
     ssl_wrap_socket,
@@ -107,6 +108,10 @@ class HTTPConnection(_HTTPConnection, object):
     #: Whether this connection verifies the host's certificate.
     is_verified = False
 
+    #: Whether this proxy connection (if used) verifies the proxy host's
+    #: certificate.
+    proxy_is_verified = None
+
     def __init__(self, *args, **kw):
         if not six.PY2:
             kw.pop("strict", None)
@@ -490,14 +495,10 @@ def _connect_tls_proxy(self, hostname, conn):
             self.ca_cert_dir,
             self.ca_cert_data,
         )
-        # By default urllib3's SSLContext disables `check_hostname` and uses
-        # a custom check. For proxies we're good with relying on the default
-        # verification.
-        ssl_context.check_hostname = True
 
         # If no cert was provided, use only the default options for server
         # certificate validation
-        return ssl_wrap_socket(
+        socket = ssl_wrap_socket(
             sock=conn,
             ca_certs=self.ca_certs,
             ca_cert_dir=self.ca_cert_dir,
@@ -506,8 +507,37 @@ def _connect_tls_proxy(self, hostname, conn):
             ssl_context=ssl_context,
         )
 
+        if ssl_context.verify_mode != ssl.CERT_NONE and not getattr(
+            ssl_context, "check_hostname", False
+        ):
+            # While urllib3 attempts to always turn off hostname matching from
+            # the TLS library, this cannot always be done. So we check whether
+            # the TLS Library still thinks it's matching hostnames.
+            cert = socket.getpeercert()
+            if not cert.get("subjectAltName", ()):
+                warnings.warn(
+                    (
+                        "Certificate for {0} has no `subjectAltName`, falling back to check for a "
+                        "`commonName` for now. This feature is being removed by major browsers and "
+                        "deprecated by RFC 2818. (See https://github.com/urllib3/urllib3/issues/497 "
+                        "for details.)".format(hostname)
+                    ),
+                    SubjectAltNameWarning,
+                )
+            _match_hostname(cert, hostname)
+
+        self.proxy_is_verified = ssl_context.verify_mode == ssl.CERT_REQUIRED
+        return socket
+
 
 def _match_hostname(cert, asserted_hostname):
+    # Our upstream implementation of ssl.match_hostname()
+    # only applies this normalization to IP addresses so it doesn't
+    # match DNS SANs so we do the same thing!
+    stripped_hostname = asserted_hostname.strip("u[]")
+    if is_ipaddress(stripped_hostname):
+        asserted_hostname = stripped_hostname
+
     try:
         match_hostname(cert, asserted_hostname)
     except CertificateError as e:

src/urllib3/connectionpool.py

@@ -1020,6 +1020,17 @@ def _validate_conn(self, conn):
                 InsecureRequestWarning,
             )
 
+        if getattr(conn, "proxy_is_verified", None) is False:
+            warnings.warn(
+                (
+                    "Unverified HTTPS connection done to an HTTPS proxy. "
+                    "Adding certificate verification is strongly advised. See: "
+                    "https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html"
+                    "#ssl-warnings"
+                ),
+                InsecureRequestWarning,
+            )
+
 
 def connection_from_url(url, **kw):
     """

src/urllib3/contrib/_securetransport/low_level.py

@@ -188,6 +188,7 @@ def _cert_array_from_pem(pem_bundle):
         # We only want to do that if an error occurs: otherwise, the caller
         # should free.
         CoreFoundation.CFRelease(cert_array)
+        raise
 
     return cert_array
 

src/urllib3/util/proxy.py

@@ -45,6 +45,7 @@ def create_proxy_ssl_context(
         ssl_version=resolve_ssl_version(ssl_version),
         cert_reqs=resolve_cert_reqs(cert_reqs),
     )
+
     if (
         not ca_certs
         and not ca_cert_dir

test/conftest.py

@@ -11,6 +11,7 @@
 from tornado import ioloop, web
 
 from dummyserver.handlers import TestingApp
+from dummyserver.proxy import ProxyHandler
 from dummyserver.server import HAS_IPV6, run_tornado_app
 from dummyserver.testcase import HTTPSDummyServerTestCase
 from urllib3.util import ssl_
@@ -30,16 +31,21 @@ def configure_windows_event_loop():
 ServerConfig = collections.namedtuple("ServerConfig", ["host", "port", "ca_certs"])
 
 
+def _write_cert_to_dir(cert, tmpdir, file_prefix="server"):
+    cert_path = str(tmpdir / ("%s.pem" % file_prefix))
+    key_path = str(tmpdir / ("%s.key" % file_prefix))
+    cert.private_key_pem.write_to_path(key_path)
+    cert.cert_chain_pems[0].write_to_path(cert_path)
+    certs = {"keyfile": key_path, "certfile": cert_path}
+    return certs
+
+
 @contextlib.contextmanager
 def run_server_in_thread(scheme, host, tmpdir, ca, server_cert):
     ca_cert_path = str(tmpdir / "ca.pem")
-    server_cert_path = str(tmpdir / "server.pem")
-    server_key_path = str(tmpdir / "server.key")
     ca.cert_pem.write_to_path(ca_cert_path)
-    server_cert.private_key_pem.write_to_path(server_key_path)
-    server_cert.cert_chain_pems[0].write_to_path(server_cert_path)
-    server_certs = {"keyfile": server_key_path, "certfile": server_cert_path}
 
+    server_certs = _write_cert_to_dir(server_cert, tmpdir)
     io_loop = ioloop.IOLoop.current()
     app = web.Application([(r".*", TestingApp)])
     server, port = run_tornado_app(app, io_loop, server_certs, scheme, host)
@@ -53,6 +59,39 @@ def run_server_in_thread(scheme, host, tmpdir, ca, server_cert):
     server_thread.join()
 
 
+@contextlib.contextmanager
+def run_server_and_proxy_in_thread(
+    proxy_scheme, proxy_host, tmpdir, ca, proxy_cert, server_cert
+):
+    ca_cert_path = str(tmpdir / "ca.pem")
+    ca.cert_pem.write_to_path(ca_cert_path)
+
+    server_certs = _write_cert_to_dir(server_cert, tmpdir)
+    proxy_certs = _write_cert_to_dir(proxy_cert, tmpdir, "proxy")
+
+    io_loop = ioloop.IOLoop.current()
+    server = web.Application([(r".*", TestingApp)])
+    server, port = run_tornado_app(server, io_loop, server_certs, "https", "localhost")
+    server_config = ServerConfig("localhost", port, ca_cert_path)
+
+    proxy = web.Application([(r".*", ProxyHandler)])
+    proxy_app, proxy_port = run_tornado_app(
+        proxy, io_loop, proxy_certs, proxy_scheme, proxy_host
+    )
+    proxy_config = ServerConfig(proxy_host, proxy_port, ca_cert_path)
+
+    server_thread = threading.Thread(target=io_loop.start)
+    server_thread.start()
+
+    yield (proxy_config, server_config)
+
+    io_loop.add_callback(server.stop)
+    io_loop.add_callback(proxy_app.stop)
+    io_loop.add_callback(io_loop.stop)
+
+    server_thread.join()
+
+
 @pytest.fixture
 def no_san_server(tmp_path_factory):
     tmpdir = tmp_path_factory.mktemp("certs")
@@ -64,6 +103,20 @@ def no_san_server(tmp_path_factory):
         yield cfg
 
 
+@pytest.fixture
+def no_san_proxy(tmp_path_factory):
+    tmpdir = tmp_path_factory.mktemp("certs")
+    ca = trustme.CA()
+    # only common name, no subject alternative names
+    proxy_cert = ca.issue_cert(common_name=u"localhost")
+    server_cert = ca.issue_cert(u"localhost")
+
+    with run_server_and_proxy_in_thread(
+        "https", "localhost", tmpdir, ca, proxy_cert, server_cert
+    ) as cfg:
+        yield cfg
+
+
 @pytest.fixture
 def no_localhost_san_server(tmp_path_factory):
     tmpdir = tmp_path_factory.mktemp("certs")
@@ -76,7 +129,37 @@ def no_localhost_san_server(tmp_path_factory):
 
 
 @pytest.fixture
-def ip_san_server(tmp_path_factory):
+def ipv4_san_proxy(tmp_path_factory):
+    tmpdir = tmp_path_factory.mktemp("certs")
+    ca = trustme.CA()
+    # IP address in Subject Alternative Name
+    proxy_cert = ca.issue_cert(u"127.0.0.1")
+
+    server_cert = ca.issue_cert(u"localhost")
+
+    with run_server_and_proxy_in_thread(
+        "https", "127.0.0.1", tmpdir, ca, proxy_cert, server_cert
+    ) as cfg:
+        yield cfg
+
+
+@pytest.fixture
+def ipv6_san_proxy(tmp_path_factory):
+    tmpdir = tmp_path_factory.mktemp("certs")
+    ca = trustme.CA()
+    # IP addresses in Subject Alternative Name
+    proxy_cert = ca.issue_cert(u"::1")
+
+    server_cert = ca.issue_cert(u"localhost")
+
+    with run_server_and_proxy_in_thread(
+        "https", "::1", tmpdir, ca, proxy_cert, server_cert
+    ) as cfg:
+        yield cfg
+
+
+@pytest.fixture
+def ipv4_san_server(tmp_path_factory):
     tmpdir = tmp_path_factory.mktemp("certs")
     ca = trustme.CA()
     # IP address in Subject Alternative Name

test/contrib/test_pyopenssl.py

@@ -34,7 +34,7 @@ def teardown_module():
 from ..test_util import TestUtilSSL  # noqa: E402, F401
 from ..with_dummyserver.test_https import (  # noqa: E402, F401
     TestHTTPS,
-    TestHTTPS_IPSAN,
+    TestHTTPS_IPV4SAN,
     TestHTTPS_IPv6Addr,
     TestHTTPS_IPV6SAN,
     TestHTTPS_NoSAN,

test/contrib/test_securetransport.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+import base64
 import contextlib
 import socket
 import ssl
@@ -52,3 +53,15 @@ def test_no_crash_with_empty_trust_bundle():
         ws = WrappedSocket(s)
         with pytest.raises(ssl.SSLError):
             ws._custom_validate(True, b"")
+
+
+def test_no_crash_with_invalid_trust_bundle():
+    invalid_cert = base64.b64encode(b"invalid-cert")
+    cert_bundle = (
+        b"-----BEGIN CERTIFICATE-----\n" + invalid_cert + b"\n-----END CERTIFICATE-----"
+    )
+
+    with contextlib.closing(socket.socket()) as s:
+        ws = WrappedSocket(s)
+        with pytest.raises(ssl.SSLError):
+            ws._custom_validate(True, cert_bundle)

test/test_connection.py

@@ -43,6 +43,42 @@ def test_match_hostname_mismatch(self):
             )
             assert e._peer_cert == cert
 
+    def test_match_hostname_ip_address_ipv6(self):
+        cert = {"subjectAltName": (("IP Address", "1:2::2:1"),)}
+        asserted_hostname = "1:2::2:2"
+        try:
+            with mock.patch("urllib3.connection.log.warning") as mock_log:
+                _match_hostname(cert, asserted_hostname)
+        except CertificateError as e:
+            assert "hostname '1:2::2:2' doesn't match '1:2::2:1'" in str(e)
+            mock_log.assert_called_once_with(
+                "Certificate did not match expected hostname: %s. Certificate: %s",
+                "1:2::2:2",
+                {"subjectAltName": (("IP Address", "1:2::2:1"),)},
+            )
+            assert e._peer_cert == cert
+
+    def test_match_hostname_dns_with_brackets_doesnt_match(self):
+        cert = {
+            "subjectAltName": (
+                ("DNS", "localhost"),
+                ("IP Address", "localhost"),
+            )
+        }
+        asserted_hostname = "[localhost]"
+        with pytest.raises(CertificateError) as e:
+            _match_hostname(cert, asserted_hostname)
+        assert (
+            "hostname '[localhost]' doesn't match either of 'localhost', 'localhost'"
+            in str(e.value)
+        )
+
+    def test_match_hostname_ip_address_ipv6_brackets(self):
+        cert = {"subjectAltName": (("IP Address", "1:2::2:1"),)}
+        asserted_hostname = "[1:2::2:1]"
+        # Assert no error is raised
+        _match_hostname(cert, asserted_hostname)
+
     def test_recent_date(self):
         # This test is to make sure that the RECENT_DATE value
         # doesn't get too far behind what the current date is.

test/with_dummyserver/test_https.py

@@ -859,29 +859,25 @@ def test_warning_for_certs_without_a_san(self, no_san_server):
                 assert warn.called
 
 
-class TestHTTPS_IPSAN:
-    def test_can_validate_ip_san(self, ip_san_server):
+class TestHTTPS_IPV4SAN:
+    def test_can_validate_ip_san(self, ipv4_san_server):
         """Ensure that urllib3 can validate SANs with IP addresses in them."""
-        try:
-            import ipaddress  # noqa: F401
-        except ImportError:
-            pytest.skip("Only runs on systems with an ipaddress module")
-
         with HTTPSConnectionPool(
-            ip_san_server.host,
-            ip_san_server.port,
+            ipv4_san_server.host,
+            ipv4_san_server.port,
             cert_reqs="CERT_REQUIRED",
-            ca_certs=ip_san_server.ca_certs,
+            ca_certs=ipv4_san_server.ca_certs,
         ) as https_pool:
             r = https_pool.request("GET", "/")
             assert r.status == 200
 
 
 class TestHTTPS_IPv6Addr:
-    def test_strip_square_brackets_before_validating(self, ipv6_addr_server):
+    @pytest.mark.parametrize("host", ["::1", "[::1]"])
+    def test_strip_square_brackets_before_validating(self, ipv6_addr_server, host):
         """Test that the fix for #760 works."""
         with HTTPSConnectionPool(
-            "[::1]",
+            host,
             ipv6_addr_server.port,
             cert_reqs="CERT_REQUIRED",
             ca_certs=ipv6_addr_server.ca_certs,
@@ -891,15 +887,11 @@ def test_strip_square_brackets_before_validating(self, ipv6_addr_server):
 
 
 class TestHTTPS_IPV6SAN:
-    def test_can_validate_ipv6_san(self, ipv6_san_server):
+    @pytest.mark.parametrize("host", ["::1", "[::1]"])
+    def test_can_validate_ipv6_san(self, ipv6_san_server, host):
         """Ensure that urllib3 can validate SANs with IPv6 addresses in them."""
-        try:
-            import ipaddress  # noqa: F401
-        except ImportError:
-            pytest.skip("Only runs on systems with an ipaddress module")
-
         with HTTPSConnectionPool(
-            "[::1]",
+            host,
             ipv6_san_server.port,
             cert_reqs="CERT_REQUIRED",
             ca_certs=ipv6_san_server.ca_certs,