[PR aio-libs#12827/ccf218ab backport][3.15] Numeric ipv4 resolver bypass (aio-libs#12850)

Author: bdraco

CHANGES/12827.bugfix.rst

@@ -0,0 +1 @@
+Changed :class:`~aiohttp.TCPConnector` to reject legacy non-canonical numeric IPv4 host forms such as ``2130706433``, ``017700000001`` and ``127.1`` with :exc:`~aiohttp.InvalidUrlClientError`; only canonical dotted-quad IPv4 literals are now treated as IP address literals, while every other host is sent through the configured resolver -- by :user:`bdraco`.

aiohttp/connector.py

@@ -27,6 +27,7 @@
     ClientConnectorSSLError,
     ClientHttpProxyError,
     ClientProxyConnectionError,
+    InvalidUrlClientError,
     ServerFingerprintMismatch,
     UnixClientConnectorError,
     cert_errors,
@@ -37,6 +38,7 @@
 from .helpers import (
     _SENTINEL,
     ceil_timeout,
+    is_canonical_ipv4_address,
     is_ip_address,
     noop,
     sentinel,
@@ -1092,6 +1094,11 @@ async def _resolve_host(
     ) -> list[ResolveResult]:
         """Resolve host and return list of addresses."""
         if is_ip_address(host):
+            # Reject legacy numeric IPv4 forms (e.g. 2130706433, 127.1) that
+            # socket would map onto an address, slipping past a connector-level
+            # policy that only sees the raw host.
+            if ":" not in host and not is_canonical_ipv4_address(host):
+                raise InvalidUrlClientError(host, "is not a canonical IPv4 address")
             return [
                 {
                     "hostname": host,

aiohttp/helpers.py

@@ -511,6 +511,28 @@ def is_ip_address(host: str | None) -> bool:
     return ":" in host or host.replace(".", "").isdigit()
 
 
+def is_canonical_ipv4_address(host: str) -> bool:
+    """Check if host is a canonical dotted-quad IPv4 address.
+
+    Rejects the legacy numeric forms that ``socket`` still accepts and
+    maps onto an address, e.g. ``2130706433``, ``017700000001``, ``127.1``.
+    """
+    parts = host.split(".")
+    if len(parts) != 4:
+        return False
+    for part in parts:
+        # Each octet must be 1-3 ASCII digits; reject unicode digits
+        # (which ``str.isdigit`` accepts but ``int`` may not), octal
+        # leading zeros, and values above 255.
+        if not (1 <= len(part) <= 3) or not part.isascii() or not part.isdigit():
+            return False
+        if part[0] == "0" and len(part) != 1:
+            return False
+        if int(part) > 255:
+            return False
+    return True
+
+
 _cached_current_datetime: int | None = None
 _cached_formatted_datetime = ""
 

tests/test_connector.py

@@ -26,6 +26,7 @@
 from aiohttp import client, connector as connector_module, hdrs, web
 from aiohttp.abc import AbstractResolver
 from aiohttp.client import ClientRequest, ClientTimeout
+from aiohttp.client_exceptions import InvalidUrlClientError
 from aiohttp.client_proto import ResponseHandler
 from aiohttp.client_reqrep import ConnectionKey
 from aiohttp.connector import (
@@ -1253,6 +1254,35 @@ async def test_tcp_connector_resolve_host(loop: asyncio.AbstractEventLoop) -> No
     await conn.close()
 
 
+async def test_tcp_connector_rejects_non_canonical_ipv4_alias() -> None:
+    """Legacy numeric IPv4 aliases must not bypass the configured resolver."""
+    calls: list[str] = []
+
+    class _RecordingResolver(AbstractResolver):
+        async def resolve(
+            self,
+            host: str,
+            port: int = 0,
+            family: socket.AddressFamily = socket.AF_INET,
+        ) -> list[ResolveResult]:
+            assert False
+
+        async def close(self) -> None:
+            """Close the resolver."""
+
+    conn = aiohttp.TCPConnector(resolver=_RecordingResolver())
+    for alias in ("2130706433", "017700000001", "127.1"):
+        with pytest.raises(InvalidUrlClientError, match="canonical IPv4"):
+            await conn._resolve_host(alias, 8080)
+
+    # Resolver is never consulted, and a canonical IP still short-circuits it.
+    assert calls == []
+    res = await conn._resolve_host("127.0.0.1", 8080)
+    assert res[0]["host"] == "127.0.0.1"
+    assert calls == []
+    await conn.close()
+
+
 @pytest.fixture
 def dns_response(loop):
     async def coro():

tests/test_helpers.py

@@ -2,6 +2,8 @@
 import base64
 import datetime
 import gc
+import ipaddress
+import itertools
 import sys
 import warnings
 import weakref
@@ -319,6 +321,93 @@ def test_is_ip_address_invalid_type() -> None:
         helpers.is_ip_address(object())
 
 
+# ------------------------------- is_canonical_ipv4_address() ---------------
+
+
+@pytest.mark.parametrize(
+    "host",
+    [
+        "0.0.0.0",
+        "127.0.0.1",
+        "8.8.8.8",
+        "192.168.0.1",
+        "255.255.255.255",
+    ],
+)
+def test_is_canonical_ipv4_address_accepts_dotted_quad(host: str) -> None:
+    assert helpers.is_canonical_ipv4_address(host)
+
+
+@pytest.mark.parametrize(
+    "host",
+    [
+        "2130706433",  # decimal integer form of 127.0.0.1
+        "017700000001",  # octal form of 127.0.0.1
+        "127.1",  # short-hand form of 127.0.0.1
+        "127.0.1",  # 3-part short-hand
+        "0177.0.0.1",  # octal leading-zero octet
+        "01.2.3.4",  # octal leading-zero octet
+        "256.0.0.1",  # octet out of range
+        "999.0.0.1",  # octet out of range
+        "1.2.3.4.5",  # too many octets
+        "127.0.0.",  # trailing dot / empty octet
+        "12³.0.0.1",  # superscript digit (str.isdigit but not int)
+        "１２７.0.0.1",  # full-width digits
+        "0xa.0.0.0",  # hex octet
+        " 127.0.0.1",  # leading whitespace
+        "127.0.0.1 ",  # trailing whitespace
+        "example.com",  # domain name
+        "",  # empty
+    ],
+)
+def test_is_canonical_ipv4_address_rejects_non_canonical(host: str) -> None:
+    assert not helpers.is_canonical_ipv4_address(host)
+
+
+def _ipaddress_accepts_ipv4(host: str) -> bool:
+    """Oracle: does the stdlib accept ``host`` as a canonical IPv4 address?"""
+    try:
+        ipaddress.IPv4Address(host)
+    except ipaddress.AddressValueError:
+        return False
+    return True
+
+
+def test_is_canonical_ipv4_address_matches_stdlib() -> None:
+    """Prove equivalence with ``ipaddress.IPv4Address`` over a broad corpus.
+
+    The helper is a fast hand-rolled substitute for the stdlib parser; this
+    exhaustively cross-checks the two agree on every combination of a set of
+    octet-like tokens covering the known edge cases (leading zeros, out of
+    range, empty, unicode digits, wrong octet count).
+    """
+    tokens = [
+        "0",
+        "1",
+        "9",
+        "10",
+        "99",
+        "255",
+        "256",
+        "999",
+        "00",
+        "01",
+        "0177",
+        "1234",
+        "",
+        "a",
+        "0x1",
+        "１",  # full-width 1
+        "1²",  # trailing superscript
+    ]
+    for count in range(1, 5):
+        for parts in itertools.product(tokens, repeat=count):
+            host = ".".join(parts)
+            assert helpers.is_canonical_ipv4_address(host) == _ipaddress_accepts_ipv4(
+                host
+            ), host
+
+
 # ----------------------------------- TimeoutHandle -------------------