[3.10] gh-83499: Fix closing file descriptors in tempfile (GH-93874)

Lib/tempfile.py

@@ -203,8 +203,7 @@ def _get_default_tempdir():
                 fd = _os.open(filename, _bin_openflags, 0o600)
                 try:
                     try:
-                        with _io.open(fd, 'wb', closefd=False) as fp:
-                            fp.write(b'blat')
+                        _os.write(fd, b'blat')
                     finally:
                         _os.close(fd)
                 finally:
@@ -244,6 +243,7 @@ def _get_candidate_names():
 def _mkstemp_inner(dir, pre, suf, flags, output_type):
     """Code common to mkstemp, TemporaryFile, and NamedTemporaryFile."""
 
+    dir = _os.path.abspath(dir)
     names = _get_candidate_names()
     if output_type is bytes:
         names = map(_os.fsencode, names)
@@ -264,7 +264,7 @@ def _mkstemp_inner(dir, pre, suf, flags, output_type):
                 continue
             else:
                 raise
-        return (fd, _os.path.abspath(file))
+        return fd, file
 
     raise FileExistsError(_errno.EEXIST,
                           "No usable temporary file name found")
@@ -550,15 +550,26 @@ def NamedTemporaryFile(mode='w+b', buffering=-1, encoding=None,
     if "b" not in mode:
         encoding = _io.text_encoding(encoding)
 
-    (fd, name) = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
+    name = None
+    def opener(*args):
+        nonlocal name
+        fd, name = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
+        return fd
     try:
-        file = _io.open(fd, mode, buffering=buffering,
-                        newline=newline, encoding=encoding, errors=errors)
-
-        return _TemporaryFileWrapper(file, name, delete)
-    except BaseException:
-        _os.unlink(name)
-        _os.close(fd)
+        file = _io.open(dir, mode, buffering=buffering,
+                        newline=newline, encoding=encoding, errors=errors,
+                        opener=opener)
+        try:
+            raw = getattr(file, 'buffer', file)
+            raw = getattr(raw, 'raw', raw)
+            raw.name = name
+            return _TemporaryFileWrapper(file, name, delete)
+        except:
+            file.close()
+            raise
+    except:
+        if name is not None and not (_os.name == 'nt' and delete):
+            _os.unlink(name)
         raise
 
 if _os.name != 'posix' or _sys.platform == 'cygwin':
@@ -597,9 +608,20 @@ def TemporaryFile(mode='w+b', buffering=-1, encoding=None,
 
         flags = _bin_openflags
         if _O_TMPFILE_WORKS:
-            try:
+            fd = None
+            def opener(*args):
+                nonlocal fd
                 flags2 = (flags | _os.O_TMPFILE) & ~_os.O_CREAT
                 fd = _os.open(dir, flags2, 0o600)
+                return fd
+            try:
+                file = _io.open(dir, mode, buffering=buffering,
+                                newline=newline, encoding=encoding,
+                                errors=errors, opener=opener)
+                raw = getattr(file, 'buffer', file)
+                raw = getattr(raw, 'raw', raw)
+                raw.name = fd
+                return file
             except IsADirectoryError:
                 # Linux kernel older than 3.11 ignores the O_TMPFILE flag:
                 # O_TMPFILE is read as O_DIRECTORY. Trying to open a directory
@@ -616,24 +638,25 @@ def TemporaryFile(mode='w+b', buffering=-1, encoding=None,
                 # fails with NotADirectoryError, because O_TMPFILE is read as
                 # O_DIRECTORY.
                 pass
-            else:
-                try:
-                    return _io.open(fd, mode, buffering=buffering,
-                                    newline=newline, encoding=encoding,
-                                    errors=errors)
-                except:
-                    _os.close(fd)
-                    raise
             # Fallback to _mkstemp_inner().
 
-        (fd, name) = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
-        try:
-            _os.unlink(name)
-            return _io.open(fd, mode, buffering=buffering,
-                            newline=newline, encoding=encoding, errors=errors)
-        except:
-            _os.close(fd)
-            raise
+        fd = None
+        def opener(*args):
+            nonlocal fd
+            fd, name = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
+            try:
+                _os.unlink(name)
+            except BaseException as e:
+                _os.close(fd)
+                raise
+            return fd
+        file = _io.open(dir, mode, buffering=buffering,
+                        newline=newline, encoding=encoding, errors=errors,
+                        opener=opener)
+        raw = getattr(file, 'buffer', file)
+        raw = getattr(raw, 'raw', raw)
+        raw.name = fd
+        return file
 
 class SpooledTemporaryFile:
     """Temporary file wrapper, specialized to switch from BytesIO

Lib/test/test_tempfile.py

@@ -290,19 +290,14 @@ def our_candidate_list():
                 def raise_OSError(*args, **kwargs):
                     raise OSError()
 
-                with support.swap_attr(io, "open", raise_OSError):
-                    # test again with failing io.open()
+                with support.swap_attr(os, "open", raise_OSError):
+                    # test again with failing os.open()
                     with self.assertRaises(FileNotFoundError):
                         tempfile._get_default_tempdir()
                     self.assertEqual(os.listdir(our_temp_directory), [])
 
-                def bad_writer(*args, **kwargs):
-                    fp = orig_open(*args, **kwargs)
-                    fp.write = raise_OSError
-                    return fp
-
-                with support.swap_attr(io, "open", bad_writer) as orig_open:
-                    # test again with failing write()
+                with support.swap_attr(os, "write", raise_OSError):
+                    # test again with failing os.write()
                     with self.assertRaises(FileNotFoundError):
                         tempfile._get_default_tempdir()
                     self.assertEqual(os.listdir(our_temp_directory), [])
@@ -977,6 +972,7 @@ def test_del_on_close(self):
         try:
             with tempfile.NamedTemporaryFile(dir=dir) as f:
                 f.write(b'blat')
+            self.assertEqual(os.listdir(dir), [])
             self.assertFalse(os.path.exists(f.name),
                         "NamedTemporaryFile %s exists after close" % f.name)
         finally:
@@ -1016,19 +1012,6 @@ def use_closed():
                 pass
         self.assertRaises(ValueError, use_closed)
 
-    def test_no_leak_fd(self):
-        # Issue #21058: don't leak file descriptor when io.open() fails
-        closed = []
-        os_close = os.close
-        def close(fd):
-            closed.append(fd)
-            os_close(fd)
-
-        with mock.patch('os.close', side_effect=close):
-            with mock.patch('io.open', side_effect=ValueError):
-                self.assertRaises(ValueError, tempfile.NamedTemporaryFile)
-                self.assertEqual(len(closed), 1)
-
     def test_bad_mode(self):
         dir = tempfile.mkdtemp()
         self.addCleanup(os_helper.rmtree, dir)
@@ -1038,6 +1021,24 @@ def test_bad_mode(self):
             tempfile.NamedTemporaryFile(mode=2, dir=dir)
         self.assertEqual(os.listdir(dir), [])
 
+    def test_bad_encoding(self):
+        dir = tempfile.mkdtemp()
+        self.addCleanup(os_helper.rmtree, dir)
+        with self.assertRaises(LookupError):
+            tempfile.NamedTemporaryFile('w', encoding='bad-encoding', dir=dir)
+        self.assertEqual(os.listdir(dir), [])
+
+    def test_unexpected_error(self):
+        dir = tempfile.mkdtemp()
+        self.addCleanup(os_helper.rmtree, dir)
+        with mock.patch('tempfile._TemporaryFileWrapper') as mock_ntf, \
+             mock.patch('io.open', mock.mock_open()) as mock_open:
+            mock_ntf.side_effect = KeyboardInterrupt()
+            with self.assertRaises(KeyboardInterrupt):
+                tempfile.NamedTemporaryFile(dir=dir)
+        mock_open().close.assert_called()
+        self.assertEqual(os.listdir(dir), [])
+
     # How to test the mode and bufsize parameters?
 
 class TestSpooledTemporaryFile(BaseTestCase):
@@ -1068,8 +1069,10 @@ def test_del_on_close(self):
             self.assertTrue(f._rolled)
             filename = f.name
             f.close()
-            self.assertFalse(isinstance(filename, str) and os.path.exists(filename),
-                        "SpooledTemporaryFile %s exists after close" % filename)
+            self.assertEqual(os.listdir(dir), [])
+            if not isinstance(filename, int):
+                self.assertFalse(os.path.exists(filename),
+                    "SpooledTemporaryFile %s exists after close" % filename)
         finally:
             os.rmdir(dir)
 
@@ -1356,19 +1359,34 @@ def roundtrip(input, *args, **kwargs):
             roundtrip("\u039B", "w+", encoding="utf-16")
             roundtrip("foo\r\n", "w+", newline="")
 
-        def test_no_leak_fd(self):
-            # Issue #21058: don't leak file descriptor when io.open() fails
-            closed = []
-            os_close = os.close
-            def close(fd):
-                closed.append(fd)
-                os_close(fd)
-
-            with mock.patch('os.close', side_effect=close):
-                with mock.patch('io.open', side_effect=ValueError):
-                    self.assertRaises(ValueError, tempfile.TemporaryFile)
-                    self.assertEqual(len(closed), 1)
+        def test_bad_mode(self):
+            dir = tempfile.mkdtemp()
+            self.addCleanup(os_helper.rmtree, dir)
+            with self.assertRaises(ValueError):
+                tempfile.TemporaryFile(mode='wr', dir=dir)
+            with self.assertRaises(TypeError):
+                tempfile.TemporaryFile(mode=2, dir=dir)
+            self.assertEqual(os.listdir(dir), [])
+
+        def test_bad_encoding(self):
+            dir = tempfile.mkdtemp()
+            self.addCleanup(os_helper.rmtree, dir)
+            with self.assertRaises(LookupError):
+                tempfile.TemporaryFile('w', encoding='bad-encoding', dir=dir)
+            self.assertEqual(os.listdir(dir), [])
 
+        def test_unexpected_error(self):
+            dir = tempfile.mkdtemp()
+            self.addCleanup(os_helper.rmtree, dir)
+            with mock.patch('tempfile._O_TMPFILE_WORKS', False), \
+                 mock.patch('os.unlink') as mock_unlink, \
+                 mock.patch('os.open') as mock_open, \
+                 mock.patch('os.close') as mock_close:
+                mock_unlink.side_effect = KeyboardInterrupt()
+                with self.assertRaises(KeyboardInterrupt):
+                    tempfile.TemporaryFile(dir=dir)
+            mock_close.assert_called()
+            self.assertEqual(os.listdir(dir), [])
 
 
 # Helper for test_del_on_shutdown

Misc/NEWS.d/next/Library/2022-06-15-21-28-16.gh-issue-83499.u3DQJ-.rst

@@ -0,0 +1 @@
+Fix double closing of file description in :mod:`tempfile`.