Fix fileno parameter handling for socket class (#1638)

* Fix fileno handling and add tests

Tests are backported from py3.11

* Expect failure for non socket fd on posix

Author: Grub4K

Src/IronPython.Modules/_socket.cs

@@ -143,11 +143,19 @@ public void __init__(CodeContext/*!*/ context, int family = DefaultAddressFamily
                     // we now own the lifetime of the socket
                     GC.SuppressFinalize(sock);
                 } else if (fileno != null) {
-                    if (Converter.TryConvertToInt64(fileno, out long l)) {
-                        socket = HandleToSocket(l);
+                    if (!PythonOps.TryToIndex(fileno, out object? handleObj)) {
+                        throw PythonOps.TypeErrorForUnIndexableObject(fileno);
                     }
+                    long handle = Converter.ConvertToInt64(handleObj);
+                    // Windows reserves only INVALID_SOCKET (~0) as an invalid handle
+                    if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows) ? handle == -1 : handle < 0) {
+                        throw PythonOps.ValueError("negative file descriptor");
+                    }
+                    socket = HandleToSocket(handle);
                     if (socket is null) {
-                        throw PythonOps.OSError("Bad file descriptor");
+                        throw PythonOps.OSError(RuntimeInformation.IsOSPlatform(OSPlatform.Windows)
+                            ? PythonErrorNumber.WSAENOTSOCK : PythonErrorNumber.EBADF,
+                            "Bad file descriptor");
                     }
                 } else {
                     try {

Src/StdLib/Lib/test/test_socket.py

@@ -30,6 +30,45 @@
 except ImportError:
     fcntl = None
 
+# ironpython: backported shims for older python
+try:
+    from test.support import os_helper
+except ImportError:
+    class os_helper:
+        make_bad_fd = support.make_bad_fd
+        unlink = support.unlink
+
+try:
+    from test.support import socket_helper
+except ImportError:
+    class socket_helper:
+        HOST = "localhost"
+        HOSTv6 = "::1"
+
+        @staticmethod
+        def create_unix_domain_name():
+            """
+            Create a UNIX domain name: socket.bind() argument of a AF_UNIX socket.
+            Return a path relative to the current directory to get a short path
+            (around 27 ASCII characters).
+            """
+            return tempfile.mktemp(prefix="test_python_", suffix='.sock',
+                                   dir=os.path.curdir)
+
+        IPV6_ENABLED = False
+
+    if socket.has_ipv6:
+        sock = None
+        try:
+            sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+            sock.bind((socket_helper.HOSTv6, 0))
+            socket_helper.IPV6_ENABLED = True
+        except OSError:
+            pass
+        finally:
+            if sock:
+                sock.close()
+
 HOST = support.HOST
 MSG = 'Michael Gilfix was here\u1234\r\n'.encode('utf-8') ## test unicode string and carriage return
 MAIN_TIMEOUT = 60.0
@@ -1531,6 +1570,94 @@ def fileno(self):
             with self.assertRaises(TypeError):
                 sock._sendfile_use_sendfile(File(None))
 
+    # ironpython: test_socket_fileno tests are backported from py3.11
+    def _test_socket_fileno(self, s, family, stype):
+        self.assertEqual(s.family, family)
+        self.assertEqual(s.type, stype)
+
+        fd = s.fileno()
+        s2 = socket.socket(fileno=fd)
+        self.addCleanup(s2.close)
+        # detach old fd to avoid double close
+        s.detach()
+        self.assertEqual(s2.family, family)
+        self.assertEqual(s2.type, stype)
+        self.assertEqual(s2.fileno(), fd)
+
+    def test_socket_fileno(self):
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.addCleanup(s.close)
+        s.bind((socket_helper.HOST, 0))
+        self._test_socket_fileno(s, socket.AF_INET, socket.SOCK_STREAM)
+
+        if hasattr(socket, "SOCK_DGRAM"):
+            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            self.addCleanup(s.close)
+            s.bind((socket_helper.HOST, 0))
+            self._test_socket_fileno(s, socket.AF_INET, socket.SOCK_DGRAM)
+
+        if socket_helper.IPV6_ENABLED:
+            s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+            self.addCleanup(s.close)
+            s.bind((socket_helper.HOSTv6, 0, 0, 0))
+            self._test_socket_fileno(s, socket.AF_INET6, socket.SOCK_STREAM)
+
+        if hasattr(socket, "AF_UNIX"):
+            unix_name = socket_helper.create_unix_domain_name()
+            self.addCleanup(os_helper.unlink, unix_name)
+
+            s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            with s:
+                try:
+                    s.bind(unix_name)
+                except PermissionError:
+                    pass
+                else:
+                    self._test_socket_fileno(s, socket.AF_UNIX,
+                                             socket.SOCK_STREAM)
+
+    def test_socket_fileno_rejects_float(self):
+        with self.assertRaises(TypeError):
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno=42.5)
+
+    def test_socket_fileno_rejects_other_types(self):
+        with self.assertRaises(TypeError):
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno="foo")
+
+    def test_socket_fileno_rejects_invalid_socket(self):
+        with self.assertRaisesRegex(ValueError, "negative file descriptor"):
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno=-1)
+
+    @unittest.skipIf(os.name == "nt", "Windows disallows -1 only")
+    def test_socket_fileno_rejects_negative(self):
+        with self.assertRaisesRegex(ValueError, "negative file descriptor"):
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno=-42)
+
+    def test_socket_fileno_requires_valid_fd(self):
+        WSAENOTSOCK = 10038
+        with self.assertRaises(OSError) as cm:
+            socket.socket(fileno=os_helper.make_bad_fd())
+        self.assertIn(cm.exception.errno, (errno.EBADF, WSAENOTSOCK))
+
+        with self.assertRaises(OSError) as cm:
+            socket.socket(
+                socket.AF_INET,
+                socket.SOCK_STREAM,
+                fileno=os_helper.make_bad_fd())
+        self.assertIn(cm.exception.errno, (errno.EBADF, WSAENOTSOCK))
+
+    def test_socket_fileno_requires_socket_fd(self):
+        with tempfile.NamedTemporaryFile() as afile:
+            with self.assertRaises(OSError):
+                socket.socket(fileno=afile.fileno())
+
+            with self.assertRaises(OSError) as cm:
+                socket.socket(
+                    socket.AF_INET,
+                    socket.SOCK_STREAM,
+                    fileno=afile.fileno())
+            self.assertEqual(cm.exception.errno, errno.ENOTSOCK)
+
 
 @unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.')
 class BasicCANTest(unittest.TestCase):

Tests/test_socket_stdlib.py

@@ -49,6 +49,10 @@ def load_tests(loader, standard_tests, pattern):
             failing_tests += [
                 test.test_socket.GeneralModuleTests('test_idna'), # TODO: figure out
             ]
+        if is_posix: # https://github.com/IronLanguages/ironpython3/pull/1638#discussion_r1059535867
+            failing_tests += [
+                test.test_socket.GeneralModuleTests('test_socket_fileno_requires_socket_fd'),
+            ]
 
         skip_tests = [
             test.test_socket.UnbufferedFileObjectClassTestCase('testWriteNonBlocking') # fails intermittently during CI