[1.26] Improve testing for IPv6 scoped addresses

src/urllib3/util/url.py

@@ -279,6 +279,9 @@ def _normalize_host(host, scheme):
         if scheme in NORMALIZABLE_SCHEMES:
             is_ipv6 = IPV6_ADDRZ_RE.match(host)
             if is_ipv6:
+                # IPv6 hosts of the form 'a::b%zone' are encoded in a URL as
+                # such per RFC 6874: 'a::b%25zone'. Unquote the ZoneID
+                # separator as necessary to return a valid RFC 4007 scoped IP.
                 match = ZONE_ID_RE.search(host)
                 if match:
                     start, end = match.span(1)
@@ -331,7 +334,7 @@ def parse_url(url):
     """
     Given a url, return a parsed :class:`.Url` namedtuple. Best-effort is
     performed to parse incomplete urls. Fields not provided will be None.
-    This parser is RFC 3986 compliant.
+    This parser is RFC 3986 and RFC 6874 compliant.
 
     The parser logic and helper functions are based heavily on
     work done in the ``rfc3986`` module.

test/test_poolmanager.py

@@ -2,6 +2,7 @@
 from test import resolvesLocalhostFQDN
 
 import pytest
+from mock import patch
 
 from urllib3 import connection_from_url
 from urllib3.exceptions import ClosedPoolError, LocationValueError
@@ -372,3 +373,27 @@ def test_pool_manager_no_url_absolute_form(self):
         p = PoolManager(strict=True)
         assert p._proxy_requires_url_absolute_form("http://example.com") is False
         assert p._proxy_requires_url_absolute_form("https://example.com") is False
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "[a::b%zone]",
+            "[a::b%25zone]",
+            "http://[a::b%zone]",
+            "http://[a::b%25zone]",
+        ],
+    )
+    @patch("urllib3.util.connection.create_connection")
+    def test_e2e_connect_to_ipv6_scoped(self, create_connection, url):
+        """Checks that IPv6 scoped addresses are properly handled end-to-end.
+
+        This is not strictly speaking a pool manager unit test - this test
+        lives here in absence of a better code location for e2e/integration
+        tests.
+        """
+        p = PoolManager()
+        conn_pool = p.connection_from_url(url)
+        conn = conn_pool._get_conn()
+        conn.connect()
+
+        assert create_connection.call_args[0][0] == ("a::b%zone", 80)

test/test_util.py

@@ -106,6 +106,9 @@ class TestUtil(object):
             "http://[2010:836b:4179::836b:4179]",
             ("http", "[2010:836b:4179::836b:4179]", None),
         ),
+        # Scoped IPv6 (with ZoneID), both RFC 6874 compliant and not.
+        ("http://[a::b%25zone]", ("http", "[a::b%zone]", None)),
+        ("http://[a::b%zone]", ("http", "[a::b%zone]", None)),
         # Hosts
         ("HTTP://GOOGLE.COM/mail/", ("http", "google.com", None)),
         ("GOogle.COM/mail", ("http", "google.com", None)),
@@ -181,6 +184,10 @@ def test_invalid_url(self, url):
             ),
             ("HTTPS://Example.Com/?Key=Value", "https://example.com/?Key=Value"),
             ("Https://Example.Com/#Fragment", "https://example.com/#Fragment"),
+            # IPv6 addresses with zone IDs. Both RFC 6874 (%25) as well as
+            # non-standard (unquoted %) variants.
+            ("[::1%zone]", "[::1%zone]"),
+            ("[::1%25zone]", "[::1%zone]"),
             ("[::1%25]", "[::1%25]"),
             ("[::Ff%etH0%Ff]/%ab%Af", "[::ff%etH0%FF]/%AB%AF"),
             (
@@ -824,6 +831,28 @@ def test_create_connection_with_valid_idna_labels(self, socket, getaddrinfo, hos
         socket.return_value = Mock()
         create_connection((host, 80))
 
+    @patch("socket.getaddrinfo")
+    @patch("socket.socket")
+    def test_create_connection_with_scoped_ipv6(self, socket, getaddrinfo):
+        # Check that providing create_connection with a scoped IPv6 address
+        # properly propagates the scope to getaddrinfo, and that the returned
+        # scoped ID makes it to the socket creation call.
+        fake_scoped_sa6 = ("a::b", 80, 0, 42)
+        getaddrinfo.return_value = [
+            (
+                socket.AF_INET6,
+                socket.SOCK_STREAM,
+                socket.IPPROTO_TCP,
+                "",
+                fake_scoped_sa6,
+            )
+        ]
+        socket.return_value = fake_sock = Mock()
+
+        create_connection(("a::b%iface", 80))
+        assert getaddrinfo.call_args[0][0] == "a::b%iface"
+        fake_sock.connect.assert_called_once_with(fake_scoped_sa6)
+
 
 class TestUtilSSL(object):
     """Test utils that use an SSL backend."""