More robust check of socket.type

Author: 1st1

asyncio/base_events.py

@@ -84,20 +84,18 @@ def _set_reuseport(sock):
                              'SO_REUSEPORT defined but not implemented.')
 
 
-# Linux's sock.type is a bitmask that can include extra info about socket.
-_SOCKET_TYPE_MASK = 0
-if hasattr(socket, 'SOCK_NONBLOCK'):
-    _SOCKET_TYPE_MASK |= socket.SOCK_NONBLOCK
-if hasattr(socket, 'SOCK_CLOEXEC'):
-    _SOCKET_TYPE_MASK |= socket.SOCK_CLOEXEC
-
-
 def _is_stream_socket(sock_type):
-    return (sock_type & ~_SOCKET_TYPE_MASK) == socket.SOCK_STREAM
+    # Linux's socket.type is a bitmask that can include extra info
+    # about socket, therefore we can't do simple
+    # `sock_type == socket.SOCK_STREAM`.
+    return (sock_type & socket.SOCK_STREAM) == socket.SOCK_STREAM
 
 
 def _is_dgram_socket(sock_type):
-    return (sock_type & ~_SOCKET_TYPE_MASK) == socket.SOCK_DGRAM
+    # Linux's socket.type is a bitmask that can include extra info
+    # about socket, therefore we can't do simple
+    # `sock_type == socket.SOCK_DGRAM`.
+    return (sock_type & socket.SOCK_DGRAM) == socket.SOCK_DGRAM
 
 
 def _is_ip_socket(sock_family):
@@ -119,8 +117,10 @@ def _ipaddr_info(host, port, family, type, proto):
         return None
 
     if _is_stream_socket(type):
+        type = socket.SOCK_STREAM
         proto = socket.IPPROTO_TCP
     elif _is_dgram_socket(type):
+        type = socket.SOCK_DGRAM
         proto = socket.IPPROTO_UDP
     else:
         return None

asyncio/unix_events.py

@@ -235,7 +235,7 @@ def create_unix_connection(self, protocol_factory, path, *,
             if sock is None:
                 raise ValueError('no path and sock were specified')
             if (sock.family != socket.AF_UNIX or
-                    sock.type != socket.SOCK_STREAM):
+                    not base_events._is_stream_socket(sock.type)):
                 raise ValueError(
                     'A UNIX Domain Stream Socket was expected, got {!r}'
                     .format(sock))
@@ -289,7 +289,7 @@ def create_unix_server(self, protocol_factory, path=None, *,
                     'path was not specified, and no sock specified')
 
             if (sock.family != socket.AF_UNIX or
-                    sock.type != socket.SOCK_STREAM):
+                    not base_events._is_stream_socket(sock.type)):
                 raise ValueError(
                     'A UNIX Domain Stream Socket was expected, got {!r}'
                     .format(sock))

tests/test_base_events.py

@@ -116,6 +116,13 @@ def test_ipaddr_info(self):
         self.assertIsNone(
             base_events._ipaddr_info('::3%lo0', 1, INET6, STREAM, TCP))
 
+        if hasattr(socket, 'SOCK_NONBLOCK'):
+            self.assertEqual(
+                (INET, STREAM, TCP, '', ('1.2.3.4', 1)),
+                base_events._ipaddr_info(
+                    '1.2.3.4', 1, INET, STREAM | socket.SOCK_NONBLOCK, TCP))
+
+
     def test_port_parameter_types(self):
         # Test obscure kinds of arguments for "port".
         INET = socket.AF_INET
@@ -1058,6 +1065,17 @@ def test_create_server_wrong_sock(self):
                                         'A TCP Stream Socket was expected'):
                 self.loop.run_until_complete(coro)
 
+    @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
+                         'no socket.SOCK_NONBLOCK (linux only)')
+    def test_create_server_stream_bittype(self):
+        sock = socket.socket(
+            socket.AF_INET, socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
+        with sock:
+            coro = self.loop.create_server(lambda: None, sock=sock)
+            srv = self.loop.run_until_complete(coro)
+            srv.close()
+            self.loop.run_until_complete(srv.wait_closed())
+
     def test_create_datagram_endpoint_wrong_sock(self):
         sock = socket.socket(socket.AF_INET)
         with sock:

tests/test_unix_events.py

@@ -280,6 +280,33 @@ def test_create_unix_server_path_inetsock(self):
                                         'A UNIX Domain Stream.*was expected'):
                 self.loop.run_until_complete(coro)
 
+    def test_create_unix_server_path_dgram(self):
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+        with sock:
+            coro = self.loop.create_unix_server(lambda: None, path=None,
+                                                sock=sock)
+            with self.assertRaisesRegex(ValueError,
+                                        'A UNIX Domain Stream.*was expected'):
+                self.loop.run_until_complete(coro)
+
+    @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
+                         'no socket.SOCK_NONBLOCK (linux only)')
+    def test_create_unix_server_path_stream_bittype(self):
+        sock = socket.socket(
+            socket.AF_UNIX, socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
+        with tempfile.NamedTemporaryFile() as file:
+            fn = file.name
+        try:
+            with sock:
+                sock.bind(fn)
+                coro = self.loop.create_unix_server(lambda: None, path=None,
+                                                    sock=sock)
+                srv = self.loop.run_until_complete(coro)
+                srv.close()
+                self.loop.run_until_complete(srv.wait_closed())
+        finally:
+            os.unlink(fn)
+
     def test_create_unix_connection_path_inetsock(self):
         sock = socket.socket()
         with sock: