Skip to content

Commit

Permalink
Fix: fix the code about handling getaddrinfo when disable ipv6 in (#…
Browse files Browse the repository at this point in the history
…5906)

CPython and enable ipv6 in system.
Related to #5901; `getaddrinfo` will return an `(int, bytes)` tuple, if
CPython could not handle the address family. It will cause a index out
of range error in aiohttp.
  • Loading branch information
Hanaasagi authored and asvetlov committed Oct 24, 2021
1 parent 2fb3daa commit 2b92ef7
Show file tree
Hide file tree
Showing 3 changed files with 33 additions and 1 deletion.
4 changes: 4 additions & 0 deletions CHANGES/5901.bugfix
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Fix the error in handling the return value of `getaddrinfo`.
`getaddrinfo` will return an `(int, bytes)` tuple, if CPython could not handle the address family.
It will cause a index out of range error in aiohttp. For example, if user compile CPython with
`--disable-ipv6` option but his system enable the ipv6.
4 changes: 3 additions & 1 deletion aiohttp/resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,9 @@ async def resolve(

hosts = []
for family, _, proto, _, address in infos:
if family == socket.AF_INET6 and address[3]: # type: ignore[misc]
if family == socket.AF_INET6:
if not (socket.has_ipv6 and address[3]): # type: ignore[misc]
continue
# This is essential for link-local IPv6 addresses.
# LL IPv6 is a VERY rare case. Strictly speaking, we should use
# getnameinfo() unconditionally, but performance makes sense.
Expand Down
26 changes: 26 additions & 0 deletions tests/test_resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,32 @@ async def test_close_for_threaded_resolver(loop) -> None:
await resolver.close()


async def test_threaded_negative_lookup_with_unknown_result() -> None:
loop = Mock()

# If compile CPython with `--disable-ipv6` option,
# we will get an (int, bytes) tuple, instead of a Exception.
async def unknown_addrinfo(*args, **kwargs):
return [
(
socket.AF_INET6,
socket.SOCK_STREAM,
6,
"",
(10, b"\x01\xbb\x00\x00\x00\x00*\x04NB\x00\x1a\x00\x00"),
)
]

loop.getaddrinfo = unknown_addrinfo
resolver = ThreadedResolver()
resolver._loop = loop
with patch("socket.has_ipv6", False):
res = await resolver.resolve("www.python.org")
assert len(res) == 0

await resolver.close()


@pytest.mark.skipif(aiodns is None, reason="aiodns required")
async def test_close_for_async_resolver(loop) -> None:
resolver = AsyncResolver(loop=loop)
Expand Down

0 comments on commit 2b92ef7

Please sign in to comment.