From fb3138825137e8152fad9f883c9e4eb7262e745d Mon Sep 17 00:00:00 2001 From: N3XT3R1337 Date: Wed, 18 Mar 2026 11:43:28 +0300 Subject: [PATCH] fix: improve type safety and error handling --- CHANGES/12253.bugfix.rst | 5 +++++ aiohttp/resolver.py | 3 ++- tests/test_resolver.py | 45 ++++++++++++++++++++++++++++++++++++---- 3 files changed, 48 insertions(+), 5 deletions(-) create mode 100644 CHANGES/12253.bugfix.rst diff --git a/CHANGES/12253.bugfix.rst b/CHANGES/12253.bugfix.rst new file mode 100644 index 00000000000..40fdbff365e --- /dev/null +++ b/CHANGES/12253.bugfix.rst @@ -0,0 +1,5 @@ +Fixed an off-by-one error in :class:`~aiohttp.resolver.AsyncResolver` that caused +an :exc:`IndexError` when handling :exc:`aiodns.error.DNSError` exceptions with +a single argument. Also fixed the resolver not extracting the port from +``getnameinfo()`` results for link-local IPv6 addresses, which could lead to +incorrect port values in resolved addresses. diff --git a/aiohttp/resolver.py b/aiohttp/resolver.py index d29ffff1d95..162160a0361 100644 --- a/aiohttp/resolver.py +++ b/aiohttp/resolver.py @@ -113,7 +113,7 @@ async def resolve( flags=_AI_ADDRCONFIG, ) except aiodns.error.DNSError as exc: - msg = exc.args[1] if len(exc.args) >= 1 else "DNS lookup failed" + msg = exc.args[1] if len(exc.args) >= 2 else "DNS lookup failed" raise OSError(None, msg) from exc hosts: list[ResolveResult] = [] for node in resp.nodes: @@ -128,6 +128,7 @@ async def resolve( _NAME_SOCKET_FLAGS, ) resolved_host = result.node + port = int(result.service) else: resolved_host = address[0].decode("ascii") port = address[1] diff --git a/tests/test_resolver.py b/tests/test_resolver.py index e19df43e7e2..c0b4540f9f9 100644 --- a/tests/test_resolver.py +++ b/tests/test_resolver.py @@ -105,9 +105,9 @@ def __init__(self, hosts: Collection[str]) -> None: class FakeAIODNSNameInfoIPv6Result: - def __init__(self, host: str) -> None: + def __init__(self, host: str, service: str = "0") -> None: self.node = host - self.service = None + self.service = service async def fake_aiodns_getaddrinfo_ipv4_result( @@ -123,9 +123,9 @@ async def fake_aiodns_getaddrinfo_ipv6_result( async def fake_aiodns_getnameinfo_ipv6_result( - host: str, + host: str, service: str = "0" ) -> FakeAIODNSNameInfoIPv6Result: - return FakeAIODNSNameInfoIPv6Result(host) + return FakeAIODNSNameInfoIPv6Result(host, service=service) def fake_addrinfo(hosts: Collection[str]) -> Callable[..., Awaitable[_AddrInfo4]]: @@ -209,6 +209,27 @@ async def test_async_resolver_positive_link_local_ipv6_lookup( type=socket.SOCK_STREAM, ) mock().getnameinfo.assert_called_with(("fe80::1", 0, 0, 3), _NAME_SOCKET_FLAGS) + assert real[0]["port"] == 0 + await resolver.close() + + +@pytest.mark.skipif(not getaddrinfo, reason="aiodns >=3.2.0 required") +@pytest.mark.usefixtures("check_no_lingering_resolvers") +async def test_async_resolver_link_local_ipv6_port_from_getnameinfo( + loop: asyncio.AbstractEventLoop, +) -> None: + """Ensure the port is correctly extracted from getnameinfo for link-local IPv6.""" + with patch("aiodns.DNSResolver") as mock: + mock().getaddrinfo.return_value = fake_aiodns_getaddrinfo_ipv6_result( + ["fe80::1"] + ) + mock().getnameinfo.return_value = fake_aiodns_getnameinfo_ipv6_result( + "fe80::1%eth0", service="8080" + ) + resolver = AsyncResolver() + real = await resolver.resolve("www.python.org") + assert real[0]["port"] == 8080 + assert real[0]["host"] == "fe80::1%eth0" await resolver.close() @@ -400,6 +421,22 @@ async def test_async_resolver_error_messages_passed( await resolver.close() +@pytest.mark.skipif(not getaddrinfo, reason="aiodns >=3.2.0 required") +@pytest.mark.usefixtures("check_no_lingering_resolvers") +async def test_async_resolver_error_single_arg_dns_error( + loop: asyncio.AbstractEventLoop, +) -> None: + """Ensure DNSError with a single arg does not cause IndexError.""" + with patch("aiodns.DNSResolver", autospec=True, spec_set=True) as mock: + mock().getaddrinfo.side_effect = aiodns.error.DNSError(1) + resolver = AsyncResolver() + with pytest.raises(OSError, match="DNS lookup failed") as excinfo: + await resolver.resolve("x.org") + + assert excinfo.value.strerror == "DNS lookup failed" + await resolver.close() + + @pytest.mark.skipif(not getaddrinfo, reason="aiodns >=3.2.0 required") @pytest.mark.usefixtures("check_no_lingering_resolvers") async def test_async_resolver_error_messages_passed_no_hosts(