1
1
import asyncio
2
2
import socket
3
+ import sys
3
4
from contextlib import asynccontextmanager
4
- from typing import AsyncGenerator , Callable , Optional , cast
5
+ from typing import AsyncGenerator , Callable , Optional , Tuple , Union , cast
5
6
6
7
from ..quic .configuration import QuicConfiguration
7
8
from ..quic .connection import QuicConnection , QuicTokenHandler
@@ -59,11 +60,24 @@ async def connect(
59
60
token_handler = token_handler ,
60
61
)
61
62
63
+ # OpenBSD well known to not support dual-stack
64
+ dual_stack = not sys .platform .startswith ("openbsd" )
65
+
66
+ # Use AI_ADDRCONFIG on platforms which doesn't support dual-stack
67
+ flags = 0
68
+ if not dual_stack :
69
+ flags = socket .AI_ADDRCONFIG
70
+
62
71
# lookup remote address
63
- infos = await loop .getaddrinfo (host , port , type = socket .SOCK_DGRAM , flags = socket . AI_ADDRCONFIG )
72
+ infos = await loop .getaddrinfo (host , port , type = socket .SOCK_DGRAM , flags = flags )
64
73
74
+ local_tuple : Union [Tuple [str , int ], Tuple [str , int , int , int ]]
65
75
addr = infos [0 ][4 ]
66
76
# addr is 2-tuple for AF_INET and 4-tuple for AF_INET6
77
+ if dual_stack and len (addr ) == 2 :
78
+ addr = ("::ffff:" + addr [0 ], addr [1 ], 0 , 0 )
79
+ local_tuple = ("::" , local_port , 0 , 0 )
80
+ sock = socket .socket (socket .AF_INET6 , socket .SOCK_DGRAM )
67
81
if len (addr ) == 2 :
68
82
local_tuple = ("0.0.0.0" , local_port )
69
83
sock = socket .socket (socket .AF_INET , socket .SOCK_DGRAM )
@@ -75,6 +89,8 @@ async def connect(
75
89
76
90
completed = False
77
91
try :
92
+ if dual_stack :
93
+ sock .setsockopt (socket .IPPROTO_IPV6 , socket .IPV6_V6ONLY , 0 )
78
94
sock .bind (local_tuple )
79
95
completed = True
80
96
finally :
0 commit comments