fix: improve type safety and error handling

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.

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]

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(