if fcntl is None:

return None

try:

with open("/dev/vsock", "rb") as f:

r = fcntl.ioctl(f, socket.IOCTL_VM_SOCKETS_GET_LOCAL_CID, " ")

except OSError:

return None

else:

"""Check whether CAN sockets are supported on this host."""

try:

s = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)

except (AttributeError, OSError):

return False

else:

s.close()

"""Check whether CAN ISOTP sockets are supported on this host."""

try:

s = socket.socket(socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_ISOTP)

except (AttributeError, OSError):

return False

else:

s.close()

"""Check whether RDS sockets are supported on this host."""

try:

s = socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0)

except (AttributeError, OSError):

return False

else:

s.close()

"""Check whether AF_ALG sockets are supported on this host."""

try:

s = socket.socket(socket.AF_ALG, socket.SOCK_SEQPACKET, 0)

except (AttributeError, OSError):

return False

else:

s.close()

def setUp(self):

self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

self.port = support.bind_port(self.serv)

self.serv.listen()

def tearDown(self):

self.serv.close()

def setUp(self):

self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

self.port = support.bind_port(self.serv)

def tearDown(self):

self.serv.close()

"""Subclass of unittest.TestCase with thread-safe cleanup methods.

def __init__(self, *args, **kwargs):

super().__init__(*args, **kwargs)

self._cleanup_lock = threading.RLock()

def addCleanup(self, *args, **kwargs):

with self._cleanup_lock:

return super().addCleanup(*args, **kwargs)

def doCleanups(self, *args, **kwargs):

with self._cleanup_lock:

"""To be able to run this test, a `vcan0` CAN interface can be created with

interface = 'vcan0'

bufsize = 128

"""The CAN frame structure is defined in <linux/can.h>:

can_frame_fmt = "=IB3x8s"

can_frame_size = struct.calcsize(can_frame_fmt)

"""The Broadcast Management Command frame structure is defined

bcm_cmd_msg_fmt = "@3I4l2I"

bcm_cmd_msg_fmt += "x" * (struct.calcsize(bcm_cmd_msg_fmt) % 8)

def setUp(self):

self.s = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)

self.addCleanup(self.s.close)

try:

self.s.bind((self.interface,))

except OSError:

self.skipTest('network interface `%s` does not exist' %

"""To be able to run this test, the `rds` kernel module must be loaded:

bufsize = 8192

def setUp(self):

self.serv = socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0)

self.addCleanup(self.serv.close)

try:

self.port = support.bind_port(self.serv)

except OSError:

"""Threadable Test class

def __init__(self):

# Swap the true setup function

self.__setUp = self.setUp

self.__tearDown = self.tearDown

self.setUp = self._setUp

self.tearDown = self._tearDown

def serverExplicitReady(self):

"""This method allows the server to explicitly indicate that

self.server_ready.set()

def _setUp(self):

self.wait_threads = support.wait_threads_exit()

self.wait_threads.__enter__()

self.server_ready = threading.Event()

self.client_ready = threading.Event()

self.done = threading.Event()

self.queue = queue.Queue(1)

self.server_crashed = False

# Do some munging to start the client test.

methodname = self.id()

i = methodname.rfind('.')

methodname = methodname[i+1:]

test_method = getattr(self, '_' + methodname)

self.client_thread = thread.start_new_thread(

self.clientRun, (test_method,))

try:

self.__setUp()

except:

self.server_crashed = True

raise

finally:

self.server_ready.set()

self.client_ready.wait()

def _tearDown(self):

self.__tearDown()

self.done.wait()

self.wait_threads.__exit__(None, None, None)

if self.queue.qsize():

exc = self.queue.get()

raise exc

def clientRun(self, test_func):

self.server_ready.wait()

try:

self.clientSetUp()

except BaseException as e:

self.queue.put(e)

self.clientTearDown()

return

finally:

self.client_ready.set()

if self.server_crashed:

self.clientTearDown()

return

if not hasattr(test_func, '__call__'):

raise TypeError("test_func must be a callable function")

try:

test_func()

except BaseException as e:

self.queue.put(e)

finally:

self.clientTearDown()

def clientSetUp(self):

raise NotImplementedError("clientSetUp must be implemented.")

def clientTearDown(self):

self.done.set()

def __init__(self, methodName='runTest'):

SocketTCPTest.__init__(self, methodName=methodName)

ThreadableTest.__init__(self)

def clientSetUp(self):

self.cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

def clientTearDown(self):

self.cli.close()

self.cli = None

def __init__(self, methodName='runTest'):

SocketUDPTest.__init__(self, methodName=methodName)

ThreadableTest.__init__(self)

def clientSetUp(self):

self.cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

def clientTearDown(self):

self.cli.close()

self.cli = None

def __init__(self, methodName='runTest'):

SocketCANTest.__init__(self, methodName=methodName)

ThreadableTest.__init__(self)

def clientSetUp(self):

self.cli = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)

try:

self.cli.bind((self.interface,))

except OSError:

# skipTest should not be called here, and will be called in the

# server instead

pass

def clientTearDown(self):

self.cli.close()

self.cli = None

def __init__(self, methodName='runTest'):

SocketRDSTest.__init__(self, methodName=methodName)

ThreadableTest.__init__(self)

def clientSetUp(self):

self.cli = socket.socket(socket.PF_RDS, socket.SOCK_SEQPACKET, 0)

try:

# RDS sockets must be bound explicitly to send or receive data

self.cli.bind((HOST, 0))

self.cli_addr = self.cli.getsockname()

except OSError:

# skipTest should not be called here, and will be called in the

# server instead

pass

def clientTearDown(self):

self.cli.close()

self.cli = None

def __init__(self, methodName='runTest'):

unittest.TestCase.__init__(self, methodName=methodName)

ThreadableTest.__init__(self)

def setUp(self):

self.serv = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)

self.addCleanup(self.serv.close)

self.serv.bind((socket.VMADDR_CID_ANY, VSOCKPORT))

self.serv.listen()

self.serverExplicitReady()

self.conn, self.connaddr = self.serv.accept()

self.addCleanup(self.conn.close)

def clientSetUp(self):

time.sleep(0.1)

self.cli = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)

self.addCleanup(self.cli.close)

cid = get_cid()

self.cli.connect((cid, VSOCKPORT))

def testStream(self):

msg = self.conn.recv(1024)

self.assertEqual(msg, MSG)

def _testStream(self):

self.cli.send(MSG)

"""Socket tests for client-server connection.

def __init__(self, methodName='runTest'):

ThreadedTCPSocketTest.__init__(self, methodName=methodName)

def setUp(self):

ThreadedTCPSocketTest.setUp(self)

# Indicate explicitly we're ready for the client thread to

# proceed and then perform the blocking call to accept

self.serverExplicitReady()

conn, addr = self.serv.accept()

self.cli_conn = conn

def tearDown(self):

self.cli_conn.close()

self.cli_conn = None

ThreadedTCPSocketTest.tearDown(self)

def clientSetUp(self):

ThreadedTCPSocketTest.clientSetUp(self)

self.cli.connect((HOST, self.port))

self.serv_conn = self.cli

def clientTearDown(self):

self.serv_conn.close()

self.serv_conn = None

def __init__(self, methodName='runTest'):

unittest.TestCase.__init__(self, methodName=methodName)

ThreadableTest.__init__(self)

def setUp(self):

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

def tearDown(self):

self.serv.close()

self.serv = None

def clientSetUp(self):

pass

def clientTearDown(self):

self.cli.close()

self.cli = None

"""A base class for socket tests.

def setUp(self):

self.serv = self.newSocket()

self.bindServer()

def bindServer(self):

"""Bind server socket and set self.serv_addr to its address."""

self.bindSock(self.serv)

self.serv_addr = self.serv.getsockname()

def tearDown(self):

self.serv.close()

"""Mixin to listen on the server socket."""

def setUp(self):

super().setUp()

ThreadableTest):

"""Mixin to add client socket and allow client/server tests.

def __init__(self, *args, **kwargs):

super().__init__(*args, **kwargs)

ThreadableTest.__init__(self)

def clientSetUp(self):

self.cli = self.newClientSocket()

self.bindClient()

def newClientSocket(self):

"""Return a new socket for use as client."""

return self.newSocket()

def bindClient(self):

"""Bind client socket and set self.cli_addr to its address."""

self.bindSock(self.cli)

self.cli_addr = self.cli.getsockname()

def clientTearDown(self):

self.cli.close()

self.cli = None

ThreadedSocketTestMixin):

"""Mixin to allow client/server stream tests with connected client.

def setUp(self):

super().setUp()

# Indicate explicitly we're ready for the client thread to

# proceed and then perform the blocking call to accept

self.serverExplicitReady()

conn, addr = self.serv.accept()

self.cli_conn = conn

def tearDown(self):

self.cli_conn.close()

self.cli_conn = None

super().tearDown()

def clientSetUp(self):

super().clientSetUp()

self.cli.connect(self.serv_addr)

self.serv_conn = self.cli

def clientTearDown(self):

try:

self.serv_conn.close()

self.serv_conn = None

except AttributeError:

pass

"""Base class for Unix-domain socket tests."""

# This class is used for file descriptor passing tests, so we

# create the sockets in a private directory so that other users

# can't send anything that might be problematic for a privileged

# user running the tests.

def setUp(self):

self.dir_path = tempfile.mkdtemp()

self.addCleanup(os.rmdir, self.dir_path)

super().setUp()

def bindSock(self, sock):

path = tempfile.mktemp(dir=self.dir_path)

support.bind_unix_socket(sock, path)

"""Base class for Unix-domain SOCK_STREAM tests."""

def newSocket(self):

"""Base class for IPv4 socket tests."""

host = HOST

def setUp(self):

super().setUp()

self.port = self.serv_addr[1]

def bindSock(self, sock):

"""Base class for TCP-over-IPv4 tests."""

def newSocket(self):

"""Base class for UDP-over-IPv4 tests."""

def newSocket(self):

"""Base class for SCTP tests in one-to-one (SOCK_STREAM) mode."""

def newSocket(self):

return socket.socket(socket.AF_INET, socket.SOCK_STREAM,

"""Base class for UDP-over-IPv6 tests."""

def newSocket(self):

"""Skip decorated test if condition is true, add client_skip decorator.

def client_pass(*args, **kwargs):

pass

def skipdec(obj):

retval = unittest.skip(reason)(obj)

if not isinstance(obj, type):

retval.client_skip = lambda f: client_pass

return retval

def noskipdec(obj):

if not (isinstance(obj, type) or hasattr(obj, "client_skip")):

obj.client_skip = lambda f: f

return obj

"""Skip decorated test if obj is missing any of the given attributes.

missing = [name for name in attributes if not hasattr(obj, name)]

return skipWithClientIf(

"""Skip decorated test if a socket cannot be created with given arguments.

err = None

missing = [obj for obj in args if

isinstance(obj, str) and not hasattr(socket, obj)]

if missing:

err = "don't have " + ", ".join(name for name in missing)

else:

callargs = [getattr(socket, obj) if isinstance(obj, str) else obj

for obj in args]

try:

s = socket.socket(*callargs)

except OSError as e:

# XXX: check errno?

err = str(e)

else:

s.close()

return skipWithClientIf(

err is not None,

"can't create socket({0}): {1}".format(

def test_SocketType_is_socketobject(self):

import _socket

self.assertTrue(socket.SocketType is _socket.socket)

s = socket.socket()

self.assertIsInstance(s, socket.SocketType)

s.close()

def test_repr(self):

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

with s:

self.assertIn('fd=%i' % s.fileno(), repr(s))

self.assertIn('family=%s' % socket.AF_INET, repr(s))

self.assertIn('type=%s' % socket.SOCK_STREAM, repr(s))

self.assertIn('proto=0', repr(s))

self.assertNotIn('raddr', repr(s))

s.bind(('127.0.0.1', 0))

self.assertIn('laddr', repr(s))

self.assertIn(str(s.getsockname()), repr(s))

self.assertIn('[closed]', repr(s))

self.assertNotIn('laddr', repr(s))

@unittest.skipUnless(_socket is not None, 'need _socket module')

def test_csocket_repr(self):

s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)

try:

expected = ('<socket object, fd=%s, family=%s, type=%s, proto=%s>'

% (s.fileno(), s.family, s.type, s.proto))

self.assertEqual(repr(s), expected)

finally:

s.close()

expected = ('<socket object, fd=-1, family=%s, type=%s, proto=%s>'

% (s.family, s.type, s.proto))

self.assertEqual(repr(s), expected)

def test_weakref(self):

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

p = proxy(s)

self.assertEqual(p.fileno(), s.fileno())

s.close()

s = None

try:

p.fileno()

except ReferenceError:

pass

else:

self.fail('Socket proxy still exists')

def testSocketError(self):

# Testing socket module exceptions

msg = "Error raising socket exception (%s)."

with self.assertRaises(OSError, msg=msg % 'OSError'):

raise OSError

with self.assertRaises(OSError, msg=msg % 'socket.herror'):

raise socket.herror

with self.assertRaises(OSError, msg=msg % 'socket.gaierror'):

raise socket.gaierror

def testSendtoErrors(self):

# Testing that sendto doesn't mask failures. See #10169.

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

self.addCleanup(s.close)

s.bind(('', 0))

sockname = s.getsockname()

# 2 args

with self.assertRaises(TypeError) as cm:

s.sendto('\u2620', sockname)

self.assertEqual(str(cm.exception),

"a bytes-like object is required, not 'str'")

with self.assertRaises(TypeError) as cm:

s.sendto(5j, sockname)

self.assertEqual(str(cm.exception),

"a bytes-like object is required, not 'complex'")

with self.assertRaises(TypeError) as cm:

s.sendto(b'foo', None)

self.assertIn('not NoneType',str(cm.exception))

# 3 args

with self.assertRaises(TypeError) as cm:

s.sendto('\u2620', 0, sockname)

self.assertEqual(str(cm.exception),

"a bytes-like object is required, not 'str'")

with self.assertRaises(TypeError) as cm:

s.sendto(5j, 0, sockname)

self.assertEqual(str(cm.exception),

"a bytes-like object is required, not 'complex'")

with self.assertRaises(TypeError) as cm:

s.sendto(b'foo', 0, None)

self.assertIn('not NoneType', str(cm.exception))

with self.assertRaises(TypeError) as cm:

s.sendto(b'foo', 'bar', sockname)

self.assertIn('an integer is required', str(cm.exception))

with self.assertRaises(TypeError) as cm:

s.sendto(b'foo', None, None)

self.assertIn('an integer is required', str(cm.exception))

# wrong number of args

with self.assertRaises(TypeError) as cm:

s.sendto(b'foo')

self.assertIn('(1 given)', str(cm.exception))

with self.assertRaises(TypeError) as cm:

s.sendto(b'foo', 0, sockname, 4)

self.assertIn('(4 given)', str(cm.exception))

def testCrucialConstants(self):

# Testing for mission critical constants

socket.AF_INET

socket.SOCK_STREAM

socket.SOCK_DGRAM

socket.SOCK_RAW

socket.SOCK_RDM

socket.SOCK_SEQPACKET

socket.SOL_SOCKET

socket.SO_REUSEADDR

def testHostnameRes(self):

# Testing hostname resolution mechanisms

hostname = socket.gethostname()

try:

ip = socket.gethostbyname(hostname)

except OSError:

# Probably name lookup wasn't set up right; skip this test

self.skipTest('name lookup failure')

self.assertTrue(ip.find('.') >= 0, "Error resolving host to ip.")

try:

hname, aliases, ipaddrs = socket.gethostbyaddr(ip)

except OSError:

# Probably a similar problem as above; skip this test

self.skipTest('name lookup failure')

all_host_names = [hostname, hname] + aliases

fqhn = socket.getfqdn(ip)

if not fqhn in all_host_names:

self.fail("Error testing host resolution mechanisms. (fqdn: %s, all: %s)" % (fqhn, repr(all_host_names)))

def test_host_resolution(self):

for addr in [support.HOSTv4, '10.0.0.1', '255.255.255.255']:

self.assertEqual(socket.gethostbyname(addr), addr)

# we don't test support.HOSTv6 because there's a chance it doesn't have

# a matching name entry (e.g. 'ip6-localhost')

for host in [support.HOSTv4]:

self.assertIn(host, socket.gethostbyaddr(host)[2])

def test_host_resolution_bad_address(self):

# These are all malformed IP addresses and expected not to resolve to

# any result. But some ISPs, e.g. AWS, may successfully resolve these

# IPs.

explanation = (

"resolving an invalid IP address did not raise OSError; "

"can be caused by a broken DNS server"

)

for addr in ['0.1.1.~1', '1+.1.1.1', '::1q', '::1::2',

'1:1:1:1:1:1:1:1:1']:

with self.assertRaises(OSError):

socket.gethostbyname(addr)

with self.assertRaises(OSError, msg=explanation):

socket.gethostbyaddr(addr)

@unittest.skipUnless(hasattr(socket, 'sethostname'), "test needs socket.sethostname()")

@unittest.skipUnless(hasattr(socket, 'gethostname'), "test needs socket.gethostname()")

def test_sethostname(self):

oldhn = socket.gethostname()

try:

socket.sethostname('new')

except OSError as e:

if e.errno == errno.EPERM:

self.skipTest("test should be run as root")

else:

raise

try:

# running test as root!

self.assertEqual(socket.gethostname(), 'new')

# Should work with bytes objects too

socket.sethostname(b'bar')

self.assertEqual(socket.gethostname(), 'bar')

finally:

socket.sethostname(oldhn)

@unittest.skipUnless(hasattr(socket, 'if_nameindex'),

def testInterfaceNameIndex(self):

interfaces = socket.if_nameindex()

for index, name in interfaces:

self.assertIsInstance(index, int)

self.assertIsInstance(name, str)

# interface indices are non-zero integers

self.assertGreater(index, 0)

_index = socket.if_nametoindex(name)

self.assertIsInstance(_index, int)

self.assertEqual(index, _index)

_name = socket.if_indextoname(index)

self.assertIsInstance(_name, str)

self.assertEqual(name, _name)

@unittest.skipUnless(hasattr(socket, 'if_nameindex'),

def testInvalidInterfaceNameIndex(self):

# test nonexistent interface index/name

self.assertRaises(OSError, socket.if_indextoname, 0)

self.assertRaises(OSError, socket.if_nametoindex, '_DEADBEEF')

# test with invalid values

self.assertRaises(TypeError, socket.if_nametoindex, 0)

self.assertRaises(TypeError, socket.if_indextoname, '_DEADBEEF')

@unittest.skipUnless(hasattr(sys, 'getrefcount'),

def testRefCountGetNameInfo(self):

# Testing reference count for getnameinfo

try:

# On some versions, this loses a reference

orig = sys.getrefcount(__name__)

socket.getnameinfo(__name__,0)

except TypeError:

if sys.getrefcount(__name__) != orig:

self.fail("socket.getnameinfo loses a reference")

def testInterpreterCrash(self):

# Making sure getnameinfo doesn't crash the interpreter

try:

# On some versions, this crashes the interpreter.

socket.getnameinfo(('x', 0, 0, 0), 0)

except OSError:

pass

def testNtoH(self):

# This just checks that htons etc. are their own inverse,

# when looking at the lower 16 or 32 bits.

sizes = {socket.htonl: 32, socket.ntohl: 32,

socket.htons: 16, socket.ntohs: 16}

for func, size in sizes.items():

mask = (1<<size) - 1

for i in (0, 1, 0xffff, ~0xffff, 2, 0x01234567, 0x76543210):

self.assertEqual(i & mask, func(func(i&mask)) & mask)

swapped = func(mask)

self.assertEqual(swapped & mask, mask)

self.assertRaises(OverflowError, func, 1<<34)

@support.cpython_only

def testNtoHErrors(self):

import _testcapi

s_good_values = [0, 1, 2, 0xffff]

l_good_values = s_good_values + [0xffffffff]

l_bad_values = [-1, -2, 1<<32, 1<<1000]

s_bad_values = l_bad_values + [_testcapi.INT_MIN - 1,

_testcapi.INT_MAX + 1]

s_deprecated_values = [1<<16, _testcapi.INT_MAX]

for k in s_good_values:

socket.ntohs(k)

socket.htons(k)

for k in l_good_values:

socket.ntohl(k)

socket.htonl(k)

for k in s_bad_values:

self.assertRaises(OverflowError, socket.ntohs, k)

self.assertRaises(OverflowError, socket.htons, k)

for k in l_bad_values:

self.assertRaises(OverflowError, socket.ntohl, k)