# Authenticating avoids using a connection from something else

# able to connect to {host}:{port} instead of us.

# 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

self.cli = None

self.serv = None

def socketpair(self):

# To be overridden by some child classes.

return socket.socketpair()

self.serv, self.cli = self.socketpair()

if self.serv:

self.serv.close()

if self.cli:

self.cli.close()

class PurePythonSocketPairTest(SocketPairTest):

# Explicitly use socketpair AF_INET or AF_INET6 to ensure that is the

# code path we're using regardless platform is the pure python one where

# `_socket.socketpair` does not exist. (AF_INET does not work with

# _socket.socketpair on many platforms).

def socketpair(self):

# called by super().setUp().

try:

return socket.socketpair(socket.AF_INET6)

except OSError:

return socket.socketpair(socket.AF_INET)

# Local imports in this class make for easy security fix backporting.

def setUp(self):

import _socket

self._orig_sp = getattr(_socket, 'socketpair', None)

if self._orig_sp is not None:

# This forces the version using the non-OS provided socketpair

# emulation via an AF_INET socket in Lib/socket.py.

del _socket.socketpair

import importlib

global socket

socket = importlib.reload(socket)

else:

pass # This platform already uses the non-OS provided version.

super().setUp()

def tearDown(self):

super().tearDown()

import _socket

if self._orig_sp is not None:

# Restore the default socket.socketpair definition.

_socket.socketpair = self._orig_sp

import importlib

global socket

socket = importlib.reload(socket)

def test_recv(self):

msg = self.serv.recv(1024)

self.assertEqual(msg, MSG)

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_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

@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):

orig_getsockname = socket.socket.getsockname

inject_sock = None

def inject_getsocketname(self):

nonlocal inject_sock

sockname = orig_getsockname(self)

# Connect to the listening socket ahead of the

# client socket.

if inject_sock is None:

inject_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

inject_sock.setblocking(False)

try:

inject_sock.connect(sockname[:2])

except (BlockingIOError, InterruptedError):

pass

inject_sock.setblocking(True)

return sockname

sock1 = sock2 = None

try:

socket.socket.getsockname = inject_getsocketname

with self.assertRaises(OSError):

sock1, sock2 = socket.socketpair()

finally:

socket.socket.getsockname = orig_getsockname

if inject_sock:

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

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> and Seth Larson <seth@python.org>. Reported by Ellie

<el@horse64.org>