Address review comments

Author: sethmlarson

Lib/socket.py

@@ -653,11 +653,20 @@ def socketpair(family=AF_INET, type=SOCK_STREAM, proto=0):
 
         # Authenticating avoids using a connection from something else
         # able to connect to {host}:{port} instead of us.
-        if (
-            ssock.getsockname()[:2] != csock.getpeername()[:2]
-            or csock.getsockname()[:2] != ssock.getpeername()[:2]
-        ):
-            raise ConnectionError("Unexpected peer connection")
+        # We expect only AF_INET and AF_INET6 families.
+        try:
+            if (
+                ssock.getsockname() != csock.getpeername()
+                or csock.getsockname() != ssock.getpeername()
+            ):
+                raise ConnectionError("Unexpected peer connection")
+        except:
+            # getsockname() and getpeername() can fail
+            # if either socket isn't connected.
+            ssock.close()
+            csock.close()
+            raise
+
         return (ssock, csock)
     __all__.append("socketpair")
 

Lib/test/test_socket.py

@@ -595,12 +595,12 @@ def __init__(self, methodName='runTest'):
         self.cli = None
         self.serv = None
 
-    def call_socketpair(self):
+    def socketpair(self):
         # To be overridden by some child classes.
         return socket.socketpair()
 
     def setUp(self):
-        self.serv, self.cli = self.call_socketpair()
+        self.serv, self.cli = self.socketpair()
 
     def tearDown(self):
         if self.serv:
@@ -4907,17 +4907,30 @@ def _test_recv(self):
         self.cli.send(MSG)
 
     def test_send(self):
-        self.serv.send(MSG)
-
-    def _test_send(self):
         msg = self.cli.recv(1024)
         self.assertEqual(msg, MSG)
 
-    def _test_injected_authentication_failure(self):
-        # No-op.  Exists for base class threading infrastructure to call.
-        # We could refactor this test into its own lesser class along with the
-        # setUp and tearDown code to construct an ideal; it is simpler to keep
-        # it here and live with extra overhead one this _one_ failure test.
+    def _test_send(self):
+        self.serv.send(MSG)
+
+    def test_ipv4(self):
+        cli, srv = socket.socketpair(socket.AF_INET)
+        cli.close()
+        srv.close()
+
+    def _test_ipv4(self):
+        pass
+
+    @unittest.skipIf(not hasattr(_socket, 'IPPROTO_IPV6') or
+                     not hasattr(_socket, 'IPV6_V6ONLY'),
+                     "IPV6_V6ONLY option not supported")
+    @unittest.skipUnless(socket_helper.IPV6_ENABLED, 'IPv6 required for this test')
+    def test_ipv6(self):
+        cli, srv = socket.socketpair(socket.AF_INET6)
+        cli.close()
+        srv.close()
+
+    def _test_ipv6(self):
         pass
 
     def test_injected_authentication_failure(self):
@@ -4950,8 +4963,16 @@ def inject_getsocketname(self):
                 inject_sock.close()
             if sock1:  # This cleanup isn't needed on a successful test.
                 sock1.close()
+            if sock2:
                 sock2.close()
 
+    def _test_injected_authentication_failure(self):
+        # No-op.  Exists for base class threading infrastructure to call.
+        # We could refactor this test into its own lesser class along with the
+        # setUp and tearDown code to construct an ideal; it is simpler to keep
+        # it here and live with extra overhead one this _one_ failure test.
+        pass
+
 
 class NonBlockingTCPTests(ThreadedTCPSocketTest):
 

Misc/NEWS.d/next/Security/2024-07-22-13-11-28.gh-issue-122133.0mPeta.rst

@@ -1,5 +1,5 @@
 Authenticate the socket connection for the ``socket.socketpair()`` fallback
 on platforms where ``AF_UNIX`` is not available like Windows.
 
-Patch by Gregory P. Smith <greg@krypto.org>. Reported by Ellie
+Patch by Gregory P. Smith <greg@krypto.org> and Seth Larson <seth@python.org>. Reported by Ellie
 <el@horse64.org>