bpo-27377: Add socket.fdtype()

Doc/library/socket.rst

@@ -874,6 +874,24 @@ The :mod:`socket` module also offers various network-related services:
    .. versionadded:: 3.3
 
 
+.. function:: fdtype(fd)
+
+   Get socket information from a file descriptor.  The return value
+   is a 3-tuple with the following structure:
+
+   ``(family, type, proto)``
+
+   The values of *family*, *type*, *proto* are all integers and are
+   meant to be passed to the :func:`.socket` function.  If the file
+   descriptor is not a socket, :exc:`OSError` is raised.  On some
+   platforms, determining the protocol is not possible and in those
+   cases it is returned as zero.
+
+   Availability: Unix
+
+   .. versionadded:: 3.7
+
+
 .. _socket-objects:
 
 Socket Objects

Lib/socket.py

@@ -24,6 +24,7 @@
 inet_ntoa() -- convert 32-bit packed format IP to string (123.45.67.89)
 socket.getdefaulttimeout() -- get the default timeout value
 socket.setdefaulttimeout() -- set the default timeout value
+fdtype() -- get (family, type, protocol) from a socket file descriptor [*]
 create_connection() -- connects to an address, with an optional timeout and
                        optional source address.
 
@@ -460,6 +461,19 @@ def fromfd(fd, family, type, proto=0):
     nfd = dup(fd)
     return socket(family, type, proto, nfd)
 
+if hasattr(_socket, 'fdtype'):
+    def fdtype(fd):
+        """fdtype(fd) -> (family, type, proto)
+
+        Return (family, type, proto) for a socket given a file descriptor.
+        Raises OSError if the file descriptor is not a socket.
+        """
+        family, type, proto = _socket.fdtype(fd)
+        return (_intenum_converter(family, AddressFamily),
+                _intenum_converter(type, SocketKind),
+                proto)
+    __all__.append('fdtype')
+
 if hasattr(_socket.socket, "share"):
     def fromshare(info):
         """ fromshare(info) -> socket object

Lib/test/test_socket.py

@@ -5047,6 +5047,43 @@ def test_SOCK_NONBLOCK(self):
         socket.setdefaulttimeout(t)
 
 
+@unittest.skipUnless(hasattr(socket, "fdtype"),
+                     "socket.fdtype() not defined")
+class FdTypeTests(unittest.TestCase):
+    TYPES = [
+        (socket.AF_INET, socket.SOCK_STREAM, 0),
+        (socket.AF_INET, socket.SOCK_DGRAM, 0),
+        ]
+    _add = TYPES.append
+    if hasattr(socket, 'IPPROTO_TCP'):
+        _add((socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP))
+    if hasattr(socket, 'AF_UNIX'):
+        _add((socket.AF_UNIX, socket.SOCK_STREAM, 0))
+        _add((socket.AF_UNIX, socket.SOCK_DGRAM, 0))
+    if HAVE_SOCKET_CAN:
+        _add((socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW))
+        # This fails, not sure if Linux bug?
+        #if hasattr(socket, 'CAN_BCM'):
+        #    _add((socket.PF_CAN, socket.SOCK_DGRAM, socket.CAN_BCM))
+    if hasattr(socket, 'IPPROTO_SCTP'):
+        _add((socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_SCTP))
+    if HAVE_SOCKET_RDS:
+        _add((socket.PF_RDS, socket.SOCK_SEQPACKET, 0))
+
+    def test_fdtype(self):
+        for family, kind, proto in self.TYPES:
+            s = socket.socket(family, kind, proto)
+            with s:
+                family2, kind2, proto2 = socket.fdtype(s.fileno())
+                self.assertEqual(family, family2)
+                self.assertEqual(kind, kind2)
+                # depending on platform, we may not be able to find proto,
+                # the returned value could be zero or the actual protocol
+                # used for the socket
+                if proto != 0:
+                    self.assertIn(proto2, {proto, 0})
+
+
 @unittest.skipUnless(os.name == "nt", "Windows specific")
 @unittest.skipUnless(multiprocessing, "need multiprocessing")
 class TestSocketSharing(SocketTCPTest):
@@ -5613,7 +5650,8 @@ def test_main():
         NetworkConnectionBehaviourTest,
         ContextManagersTest,
         InheritanceTest,
-        NonblockConstantTest
+        NonblockConstantTest,
+        FdTypeTests,
     ])
     tests.append(BasicSocketPairTest)
     tests.append(TestUnixDomain)

Modules/socketmodule.c

@@ -5470,6 +5470,61 @@ AF_UNIX if defined on the platform; otherwise, the default is AF_INET.");
 #endif /* HAVE_SOCKETPAIR */
 
 
+/* socket.fdtype() function */
+
+#if defined(SO_TYPE) && !defined(MS_WINDOWS)
+/* set if we can implement fdtype().  On Windows, getsockname() fails with
+   error 10022.  There may be other platforms that have SO_TYPE but also
+   don't provide the necessary functionality. In those cases the #if above
+   may need tweaking. */
+#define HAVE_FDTYPE
+#endif
+
+#ifdef HAVE_FDTYPE
+static PyObject *
+socket_fdtype(PyObject *self, PyObject *fdobj)
+{
+    SOCKET_T fd;
+    int sock_type;
+    struct sockaddr sa;
+    socklen_t l;
+    int protocol;
+
+    fd = PyLong_AsSocket_t(fdobj);
+    if (fd == (SOCKET_T)(-1) && PyErr_Occurred())
+        return NULL;
+
+    l = sizeof(sock_type);
+    if (getsockopt(fd, SOL_SOCKET, SO_TYPE, &sock_type, &l) < 0) {
+        return set_error();
+    }
+
+    l = sizeof(sa);
+    if (getsockname(fd, &sa, &l) < 0) {
+        return set_error();
+    }
+#ifdef SO_PROTOCOL
+    l = sizeof(protocol);
+    if (getsockopt(fd, SOL_SOCKET, SO_PROTOCOL, &protocol, &l) < 0) {
+        return set_error();
+    }
+#else
+    protocol = 0;
+#endif
+    return Py_BuildValue("iii",
+                         sa.sa_family,
+                         sock_type,
+                         protocol);
+}
+
+PyDoc_STRVAR(fdtype_doc,
+"fdtype(integer) -> (family, type, protocol)\n\
+\n\
+Return the family, type and protocol for socket given a file descriptor.\
+");
+#endif /* HAVE_FDTYPE */
+
+
 static PyObject *
 socket_ntohs(PyObject *self, PyObject *args)
 {
@@ -6381,6 +6436,10 @@ static PyMethodDef socket_methods[] = {
 #ifdef HAVE_SOCKETPAIR
     {"socketpair",              socket_socketpair,
      METH_VARARGS, socketpair_doc},
+#endif
+#ifdef HAVE_FDTYPE
+    {"fdtype",                 socket_fdtype,
+     METH_O, fdtype_doc},
 #endif
     {"ntohs",                   socket_ntohs,
      METH_VARARGS, ntohs_doc},