99from debugpy .common import log
1010from debugpy .common .util import hide_thread_from_debugger
1111
12+ def can_bind_ipv4_localhost ():
13+ """Check if we can bind to IPv4 localhost."""
14+ try :
15+ sock = socket .socket (socket .AF_INET , socket .SOCK_STREAM )
16+ sock .setsockopt (socket .SOL_SOCKET , socket .SO_REUSEADDR , 1 )
17+ # Try to bind to IPv4 localhost on port 0 (any available port)
18+ sock .bind (("127.0.0.1" , 0 ))
19+ sock .close ()
20+ return True
21+ except (socket .error , OSError , AttributeError ):
22+ return False
23+
24+ def can_bind_ipv6_localhost ():
25+ """Check if we can bind to IPv6 localhost."""
26+ try :
27+ sock = socket .socket (socket .AF_INET6 , socket .SOCK_STREAM )
28+ sock .setsockopt (socket .SOL_SOCKET , socket .SO_REUSEADDR , 1 )
29+ # Try to bind to IPv6 localhost on port 0 (any available port)
30+ sock .bind (("::1" , 0 ))
31+ sock .close ()
32+ return True
33+ except (socket .error , OSError , AttributeError ):
34+ return False
35+
36+ def get_default_localhost ():
37+ """Get the default localhost address.
38+ Defaults to IPv4 '127.0.0.1', but falls back to IPv6 '::1' if IPv4 is unavailable.
39+ """
40+ # First try IPv4 (preferred default)
41+ if can_bind_ipv4_localhost ():
42+ return "127.0.0.1"
43+
44+ # Fall back to IPv6 if IPv4 is not available
45+ if can_bind_ipv6_localhost ():
46+ return "::1"
47+
48+ # If neither works, still return IPv4 as a last resort
49+ # (this is a very unusual situation)
50+ return "127.0.0.1"
51+
52+ def get_address (sock ):
53+ """Gets the socket address host and port."""
54+ try :
55+ host , port = sock .getsockname ()[:2 ]
56+ except Exception as exc :
57+ log .swallow_exception ("Failed to get socket address:" )
58+ raise RuntimeError (f"Failed to get socket address: { exc } ) from exc
59+
60+ return host , port
1261
1362def create_server (host , port = 0 , backlog = socket .SOMAXCONN , timeout = None ):
1463 """Return a local server socket listening on the given port."""
1564
1665 assert backlog > 0
1766 if host is None :
18- host = "127.0.0.1"
67+ host = get_default_localhost ()
1968 if port is None :
2069 port = 0
70+ ipv6 = host .count (":" ) > 1
2171
2272 try :
23- server = _new_sock ()
73+ server = _new_sock (ipv6 )
2474 if port != 0 :
2575 # If binding to a specific port, make sure that the user doesn't have
2676 # to wait until the OS times out the socket to be able to use that port
@@ -42,13 +92,14 @@ def create_server(host, port=0, backlog=socket.SOMAXCONN, timeout=None):
4292 return server
4393
4494
45- def create_client ():
95+ def create_client (ipv6 = False ):
4696 """Return a client socket that may be connected to a remote address."""
47- return _new_sock ()
97+ return _new_sock (ipv6 )
4898
4999
50- def _new_sock ():
51- sock = socket .socket (socket .AF_INET , socket .SOCK_STREAM , socket .IPPROTO_TCP )
100+ def _new_sock (ipv6 = False ):
101+ address_family = socket .AF_INET6 if ipv6 else socket .AF_INET
102+ sock = socket .socket (address_family , socket .SOCK_STREAM , socket .IPPROTO_TCP )
52103
53104 # Set TCP keepalive on an open socket.
54105 # It activates after 1 second (TCP_KEEPIDLE,) of idleness,
@@ -102,13 +153,14 @@ def serve(name, handler, host, port=0, backlog=socket.SOMAXCONN, timeout=None):
102153 log .reraise_exception (
103154 "Error listening for incoming {0} connections on {1}:{2}:" , name , host , port
104155 )
105- host , port = listener . getsockname ( )
156+ host , port = get_address ( listener )
106157 log .info ("Listening for incoming {0} connections on {1}:{2}..." , name , host , port )
107158
108159 def accept_worker ():
109160 while True :
110161 try :
111- sock , (other_host , other_port ) = listener .accept ()
162+ sock , address = listener .accept ()
163+ other_host , other_port = address [:2 ]
112164 except (OSError , socket .error ):
113165 # Listener socket has been closed.
114166 break
0 commit comments