bpo-17561: add socket.create_server_sock() and socket.has_dual_stack()

Doc/library/socket.rst

@@ -595,6 +595,26 @@ The following functions all create :ref:`socket objects <socket-objects>`.
    .. versionchanged:: 3.2
       *source_address* was added.
 
+.. function:: create_server_sock(address, *, dual_stack=True, queue_size=5, reuse_addr=True)
+
+   Convenience function which creates a TCP server bound to *address* (a
+   2-tuple ``(host, port)``) and return a socket object.  If *host* is an empty
+   string or ``None`` all network interfaces are used.
+   If dual-stack is supported by kernel and *dual_stack* parameter
+   is ``True`` the socket will be able to serve both :data:`AF_INET` and
+   :data:`AF_INET6` (IPv4 / IPv6) connections. If not the right address family
+   will be chosen based on the host specified in *address*.
+   *queue_size* is the argument passed to :meth:`socket.listen`.
+   If *reuse_addr* is ``True`` :data:`socket.SO_REUSEADDR` socket flag is set.
+
+   .. versionadded:: 3.8
+
+.. function:: has_dual_stack()
+
+   Return ``True`` if the kernel supports creating a socket which can handle
+   both :data:`AF_INET` and :data:`AF_INET6` (IPv4 / IPv6) connections.
+
+   .. versionadded:: 3.8
 
 .. function:: fromfd(fd, family, type, proto=0)
 

Doc/whatsnew/3.8.rst

@@ -196,6 +196,16 @@ pathlib
 contain characters unrepresentable at the OS level.
 (Contributed by Serhiy Storchaka in :issue:`33721`.)
 
+socket
+------
+
+Added :meth:`~socket.create_server_sock()`, a convenience function which
+creates a TCP server socket able to serve both IPv4 and IPv6 connections.
+(Contributed by Giampaolo Rodola in :issue:`17561`.)
+
+Added :meth:`~socket.has_dual_stack()` function.
+(Contributed by Giampaolo Rodola in :issue:`17561`.)
+
 ssl
 ---
 

Lib/socket.py

@@ -728,6 +728,76 @@ def create_connection(address, timeout=_GLOBAL_DEFAULT_TIMEOUT,
     else:
         raise error("getaddrinfo returns an empty list")
 
+def has_dual_stack():
+    """Return True if the kernel supports creating a socket which
+    can handle both AF_INET and AF_INET6 (IPv4 / IPv6) connections.
+    """
+    if not has_ipv6 \
+            or not hasattr(_socket, 'AF_INET6') \
+            or not hasattr(_socket, 'IPV6_V6ONLY'):
+        return False
+    try:
+        with socket(AF_INET6, SOCK_STREAM) as sock:
+            if not sock.getsockopt(IPPROTO_IPV6, IPV6_V6ONLY):
+                return True
+            else:
+                sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, False)
+                return True
+    except error:
+        return False
+
+def create_server_sock(address, *, dual_stack=has_dual_stack(), queue_size=5,
+                       reuse_addr=os.name == 'posix' and sys.platform != 'cygwin'):
+    """Convenience function which creates a TCP server bound to
+    *address* (a 2-tuple (host, port)) and return a socket object.
+
+    If *host* is an empty string or None all network interfaces are assumed.
+    If dual-stack is supported by kernel and *dual_stack* parameter
+    is True the socket will be able to serve both AF_INET and AF_INET6
+    connections. If not the right address family will be chosen
+    based on the host specified in *address*.
+    *queue_size* is the argument passed to socket.listen() method,
+    *reuse_addr* dictates whether to set SO_REUSEADDR flag.
+
+    >>> with create_server_sock((None, 8000)) as server:
+    ...     while True:
+    ...         sock, addr = server.accept()
+    ...         # handle new sock connection
+    """
+    AF_INET6_ = getattr(_socket, "AF_INET6", 0)
+    host, port = address
+    if host == "":
+        # http://mail.python.org/pipermail/python-ideas/2013-March/019937.html
+        host = None
+    # If dual stack is supported and no host was specified '::' will
+    # accept AF_INET and AF_INET6 connections and all interfaces.
+    if host is None and dual_stack:
+        host = "::"
+    err = None
+    info = getaddrinfo(host, port, AF_UNSPEC, SOCK_STREAM, 0, AI_PASSIVE)
+    for res in info:
+        af, socktype, proto, canonname, sa = res
+        sock = None
+        try:
+            sock = socket(af, socktype, proto)
+            if reuse_addr:
+                sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
+            if af == AF_INET6_ and dual_stack:
+                if sock.getsockopt(IPPROTO_IPV6, IPV6_V6ONLY):
+                    sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, False)
+            sock.bind(sa)
+            sock.listen(queue_size)
+            return sock
+        except error as _:
+            err = _
+            if sock is not None:
+                sock.close()
+
+    if err is not None:
+        raise err
+    else:
+        raise error("getaddrinfo() returned an empty list")
+
 def getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
     """Resolve host and port into list of address info entries.
 

Lib/test/test_socket.py

@@ -6048,9 +6048,67 @@ def test_new_tcp_flags(self):
         self.assertEqual([], unknown,
             "New TCP flags were discovered. See bpo-32394 for more information")
 
+class CreateServerSockTest(unittest.TestCase):
+
+    def echo_server(self, sock, port=None):
+        def run():
+            with sock:
+                conn, _ = sock.accept()
+                with conn:
+                    msg = conn.recv(1024)
+                    if not msg:
+                        return
+                    conn.sendall(msg)
+
+        sock.settimeout(2)
+        t = threading.Thread(target=run)
+        t.start()
+        time.sleep(.1)
+
+    def test_base(self):
+        port = support.find_unused_port()
+        with socket.create_server_sock((None, port)) as sock:
+            self.assertEqual(sock.getsockname()[1], port)
+            self.assertEqual(sock.type, socket.SOCK_STREAM)
+            if socket.has_dual_stack():
+                self.assertEqual(sock.family, socket.AF_INET6)
+            else:
+                self.assertEqual(sock.family, socket.AF_INET)
+            self.echo_server(sock)
+            with socket.create_connection(('localhost', port), timeout=2) as cl:
+                cl.sendall(b'foo')
+                self.assertEqual(cl.recv(1024), b'foo')
+
+    def test_dual_stack(self):
+        with socket.create_server_sock((None, 0)) as sock:
+            self.echo_server(sock)
+            port = sock.getsockname()[1]
+            with socket.create_connection(("127.0.0.1", port), timeout=2) as cl:
+                cl.sendall(b'foo')
+                self.assertEqual(cl.recv(1024), b'foo')
+
+        with socket.create_server_sock((None, 0)) as sock:
+            self.echo_server(sock)
+            port = sock.getsockname()[1]
+            if socket.has_dual_stack():
+                with socket.create_connection(("::1", port), timeout=2) as cl:
+                    cl.sendall(b'foo')
+                    self.assertEqual(cl.recv(1024), b'foo')
+            else:
+                self.assertRaises(ConnectionRefusedError,
+                                  socket.create_connection, ("::1", port))
+                # just stop server
+                with socket.create_connection(("127.0.0.1", port), timeout=2) \
+                        as cl:
+                    cl.sendall(b'foo')
+                    cl.recv(1024)
+                unittest.skip('dual stack cannot be tested as not supported')
+
+
 def test_main():
     tests = [GeneralModuleTests, BasicTCPTest, TCPCloserTest, TCPTimeoutTest,
-             TestExceptions, BufferIOTest, BasicTCPTest2, BasicUDPTest, UDPTimeoutTest ]
+             TestExceptions, BufferIOTest, BasicTCPTest2, BasicUDPTest,
+             UDPTimeoutTest, CreateServerSockTest]
 
     tests.extend([
         NonBlockingTCPTests,

Misc/NEWS.d/next/Library/2018-12-18-20-06-38.bpo-17561.RbIk_v.rst

@@ -0,0 +1,4 @@
+Add socket.create_server_sock() function returning a TCP socket able to serve
+both IPv4 and IPv6 connections. Also add socket.has_dual_stack() function,
+which tells whether the kernel supports creating a socket which can handle
+both IPv4 a IPv6 connections.  (patch by Giampaolo Rodola)