gh-77589: Add unix domain socket for Windows

Lib/http/server.py

@@ -117,6 +117,14 @@ class HTTPServer(socketserver.TCPServer):
     allow_reuse_address = True    # Seems to make sense in testing environment
     allow_reuse_port = False
 
+    def __init__(self, *args, **kwargs):
+        if sys.platform == 'win32' and hasattr(socket, 'AF_UNIX') and\
+                self.address_family == socket.AF_UNIX:
+            # reuse address with AF_UNIX is not supported on Windows
+            self.allow_reuse_address = False
+
+        super().__init__(*args, **kwargs)
+
     def server_bind(self):
         """Override server_bind to store the server name."""
         socketserver.TCPServer.server_bind(self)

Lib/test/_test_multiprocessing.py

@@ -5536,11 +5536,6 @@ def test_invalid_family(self):
         with self.assertRaises(ValueError):
             multiprocessing.connection.Listener(r'\\.\test')
 
-    @unittest.skipUnless(WIN32, "skipped on non-Windows platforms")
-    def test_invalid_family_win32(self):
-        with self.assertRaises(ValueError):
-            multiprocessing.connection.Listener('/var/test.pipe')
-
 #
 # Issue 12098: check sys.flags of child matches that for parent
 #

Lib/test/test_asyncio/test_base_events.py

@@ -1939,6 +1939,8 @@ def test_create_datagram_endpoint_sock(self):
         self.assertEqual('CLOSED', protocol.state)
 
     @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
+    @unittest.skipIf(sys.platform == 'win32', 'AF_UNIX support for asyncio is '
+                     'not implemented on Windows for now')
     def test_create_datagram_endpoint_sock_unix(self):
         fut = self.loop.create_datagram_endpoint(
             lambda: MyDatagramProto(create_future=True, loop=self.loop),

Lib/test/test_asyncio/test_events.py

@@ -1035,6 +1035,9 @@ def test_create_server_reuse_port(self):
         server.close()
 
     def _make_unix_server(self, factory, **kwargs):
+        if sys.platform == 'win32':
+            raise unittest.SkipTest('AF_UNIX support for asyncio is not '
+                                    'implemented on Windows for now')
         path = test_utils.gen_unix_socket_path()
         self.addCleanup(lambda: os.path.exists(path) and os.unlink(path))
 
@@ -1072,6 +1075,8 @@ def test_create_unix_server(self):
         server.close()
 
     @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
+    @unittest.skipIf(sys.platform == 'win32', 'AF_UNIX support for asyncio is '
+                     'not implemented on Windows for now')
     def test_create_unix_server_path_socket_error(self):
         proto = MyProto(loop=self.loop)
         sock = socket.socket()

Lib/test/test_asyncio/utils.py

@@ -243,6 +243,9 @@ def gen_unix_socket_path():
 
     @contextlib.contextmanager
     def unix_socket_path():
+        if sys.platform == 'win32':
+            raise unittest.SkipTest('AF_UNIX support for asyncio is not '
+                                    'implemented on Windows for now')
         path = gen_unix_socket_path()
         try:
             yield path
@@ -255,6 +258,9 @@ def unix_socket_path():
 
     @contextlib.contextmanager
     def run_test_unix_server(*, use_ssl=False):
+        if sys.platform == 'win32':
+            raise unittest.SkipTest('AF_UNIX support for asyncio is not '
+                                    'implemented on Windows for now')
         with unix_socket_path() as path:
             yield from _run_test_server(address=path, use_ssl=use_ssl,
                                         server_cls=SilentUnixWSGIServer,

Lib/test/test_logging.py

@@ -2089,6 +2089,8 @@ def test_output(self):
         self.assertEqual(self.log_output, b'<11>h\xc3\xa4m-sp\xc3\xa4m')
 
     def test_udp_reconnection(self):
+        if self.server_exception:
+            self.skipTest(self.server_exception)
         logger = logging.getLogger("slh")
         self.sl_hdlr.close()
         self.handled.clear()

Lib/test/test_pathlib/test_pathlib.py

@@ -2747,6 +2747,8 @@ def test_is_socket_false(self):
     @unittest.skipIf(
         is_wasi, "Cannot create socket on WASI."
     )
+    @unittest.skipIf(sys.platform=='win32',
+                     "detecting if file is socket is not supported by Windows")
     def test_is_socket_true(self):
         P = self.cls(self.base, 'mysock')
         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -2763,6 +2765,25 @@ def test_is_socket_true(self):
         self.assertIs(self.cls(self.base, 'mysock\udfff').is_socket(), False)
         self.assertIs(self.cls(self.base, 'mysock\x00').is_socket(), False)
 
+    @unittest.skipUnless(hasattr(socket, "AF_UNIX"), "Unix sockets required")
+    @unittest.skipUnless(sys.platform=='win32',
+                         "socket file on Windows is a normal file")
+    def test_is_socket_on_windows(self):
+        P = self.cls(self.base, 'mysock')
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.addCleanup(sock.close)
+        try:
+            sock.bind(str(P))
+        except OSError as e:
+            if (isinstance(e, PermissionError) or
+                    "AF_UNIX path too long" in str(e)):
+                self.skipTest("cannot bind Unix socket: " + str(e))
+        self.assertFalse(P.is_socket())
+        self.assertFalse(P.is_fifo())
+        self.assertTrue(P.is_file())
+        self.assertIs(self.cls(self.base, 'mysock\udfff').is_socket(), False)
+        self.assertIs(self.cls(self.base, 'mysock\x00').is_socket(), False)
+
     def test_is_block_device_false(self):
         P = self.cls(self.base)
         self.assertFalse((P / 'fileA').is_block_device())

Lib/test/test_socket.py

@@ -5131,7 +5131,7 @@ def __init__(self, methodName='runTest'):
 
     def _check_defaults(self, sock):
         self.assertIsInstance(sock, socket.socket)
-        if hasattr(socket, 'AF_UNIX'):
+        if sys.platform != 'win32' and hasattr(socket, 'AF_UNIX'):
             self.assertEqual(sock.family, socket.AF_UNIX)
         else:
             self.assertEqual(sock.family, socket.AF_INET)
@@ -6188,10 +6188,17 @@ def bind(self, sock, path):
             else:
                 raise
 
+    @unittest.skipIf(sys.platform == 'win32',
+                     'Windows will raise Error if is not bound')
     def testUnbound(self):
         # Issue #30205 (note getsockname() can return None on OS X)
         self.assertIn(self.sock.getsockname(), ('', None))
 
+    @unittest.skipUnless(sys.platform == 'win32',
+                         'Windows-specific behavior')
+    def test_unbound_on_windows(self):
+        self.assertRaisesRegex(OSError, 'WinError 10022', self.sock.getsockname)
+
     def testStrAddr(self):
         # Test binding to and retrieving a normal string pathname.
         path = os.path.abspath(os_helper.TESTFN)
@@ -6206,6 +6213,8 @@ def testBytesAddr(self):
         self.addCleanup(os_helper.unlink, path)
         self.assertEqual(self.sock.getsockname(), path)
 
+    @unittest.skipIf(sys.platform == 'win32',
+                     'surrogateescape file path is not supported on Windows')
     def testSurrogateescapeBind(self):
         # Test binding to a valid non-ASCII pathname, with the
         # non-ASCII bytes supplied using surrogateescape encoding.
@@ -6215,6 +6224,9 @@ def testSurrogateescapeBind(self):
         self.addCleanup(os_helper.unlink, path)
         self.assertEqual(self.sock.getsockname(), path)
 
+    @unittest.skipIf(sys.platform == 'win32',
+                     'Windows have a bug which can\'t unlink sock file with '
+                     'TESTFN_UNENCODABLE in it\'s name')
     def testUnencodableAddr(self):
         # Test binding to a pathname that cannot be encoded in the
         # file system encoding.
@@ -6227,10 +6239,18 @@ def testUnencodableAddr(self):
 
     @unittest.skipIf(sys.platform in ('linux', 'android'),
                      'Linux behavior is tested by TestLinuxAbstractNamespace')
+    @unittest.skipIf(sys.platform == 'win32',
+                     'Windows allow bind on empty path')
     def testEmptyAddress(self):
         # Test that binding empty address fails.
         self.assertRaises(OSError, self.sock.bind, "")
 
+    @unittest.skipUnless(sys.platform == 'win32',
+                         'Windows-specified behavior')
+    def test_empty_address_on_windows(self):
+        self.sock.bind('')
+        self.assertEqual(self.sock.getsockname(), '')
+
 
 class BufferIOTest(SocketConnectedTest):
     """

Lib/test/test_socketserver.py

@@ -8,6 +8,7 @@
 import select
 import signal
 import socket
+import sys
 import threading
 import unittest
 import socketserver
@@ -218,18 +219,24 @@ def test_ForkingUDPServer(self):
                             self.dgram_examine)
 
     @requires_unix_sockets
+    @unittest.skipIf(sys.platform=="win32",
+                     "Unix with Dadagram is not supported on Windows")
     def test_UnixDatagramServer(self):
         self.run_server(socketserver.UnixDatagramServer,
                         socketserver.DatagramRequestHandler,
                         self.dgram_examine)
 
     @requires_unix_sockets
+    @unittest.skipIf(sys.platform=="win32",
+                     "Unix with Dadagram is not supported on Windows")
     def test_ThreadingUnixDatagramServer(self):
         self.run_server(socketserver.ThreadingUnixDatagramServer,
                         socketserver.DatagramRequestHandler,
                         self.dgram_examine)
 
     @requires_unix_sockets
+    @unittest.skipIf(sys.platform=="win32",
+                     "Unix with Dadagram is not supported on Windows")
     @requires_forking
     def test_ForkingUnixDatagramServer(self):
         self.run_server(socketserver.ForkingUnixDatagramServer,

Lib/test/test_stat.py

@@ -215,13 +215,23 @@ def test_devices(self):
                 break
 
     @socket_helper.skip_unless_bind_unix_socket
+    @unittest.skipIf(sys.platform=='win32', "didn't work on Windows")
     def test_socket(self):
         with socket.socket(socket.AF_UNIX) as s:
             s.bind(TESTFN)
             st_mode, modestr = self.get_mode()
             self.assertEqual(modestr[0], 's')
             self.assertS_IS("SOCK", st_mode)
 
+    @socket_helper.skip_unless_bind_unix_socket
+    @unittest.skipUnless(sys.platform=='win32', "didn't work on Windows")
+    def test_socket_on_windows(self):
+        with socket.socket(socket.AF_UNIX) as s:
+            s.bind(TESTFN)
+            st_mode, modestr = self.get_mode()
+            self.assertNotEqual(modestr[0], 's')
+            self.assertS_IS("REG", st_mode)
+
     def test_module_attributes(self):
         for key, value in self.stat_struct.items():
             modvalue = getattr(self.statmod, key)

Misc/NEWS.d/next/Windows/2025-08-05-23-59-59.gh-issue-77589.soytRy.rst

@@ -0,0 +1 @@
+Add Unix domain socket on Windows. Patched by AN Long.

Modules/socketmodule.c

@@ -334,6 +334,8 @@ typedef struct {
 
 /* IMPORTANT: make sure the list ordered by descending build_number */
 static FlagRuntimeInfo win_runtime_flags[] = {
+    /* available starting with Windows 10 1803 */
+    {17134, "AF_UNIX"},
     /* available starting with Windows 10 1709 */
     {16299, "TCP_KEEPIDLE"},
     {16299, "TCP_KEEPINTVL"},
@@ -1900,7 +1902,7 @@ getsockaddrarg(PySocketSockObject *s, PyObject *args,
             addr->sun_path[path.len] = 0;
 
             /* including the tailing NUL */
-            *len_ret = path.len + offsetof(struct sockaddr_un, sun_path) + 1;
+            *len_ret = (int)path.len + offsetof(struct sockaddr_un, sun_path) + 1;
         }
         addr->sun_family = s->sock_family;
         memcpy(addr->sun_path, path.buf, path.len);

Modules/socketmodule.h

@@ -96,6 +96,8 @@ typedef int socklen_t;
 
 #ifdef HAVE_SYS_UN_H
 # include <sys/un.h>
+#elif HAVE_AFUNIX_H
+# include <afunix.h>
 #else
 # undef AF_UNIX
 #endif

PC/pyconfig.h

@@ -671,6 +671,13 @@ Py_NO_ENABLE_SHARED to find out.  Also support MS_NO_COREDLL for b/w compat */
 /* Define if you have the <sys/un.h> header file.  */
 /* #define HAVE_SYS_UN_H 1 */
 
+/* Define if you have the <afunix.h> header file.  */
+#if defined(__has_include) && __has_include(<afunix.h>)
+    #define HAVE_AFUNIX_H 1
+#else
+    #define HAVE_AFUNIX_H 0
+#endif
+
 /* Define if you have the <sys/utime.h> header file.  */
 /* #define HAVE_SYS_UTIME_H 1 */