gh-160406: Keep the backup and create output exclusively in fileinput (GH-160407)

In-place filtering deleted the backup file, the only copy of the data,
when the output file could not be created.  An existing backup file is
now only replaced once there is a file to back up, so trying again does
not delete it either.  The output file is now created with O_EXCL and
its mode is set through the file descriptor.

ANT-2026-3TZS4H87

Author: gpshead

Doc/library/fileinput.rst

@@ -189,6 +189,16 @@ and the backup file remains around; by default, the extension is ``'.bak'`` and
 it is deleted when the output file is closed.  In-place filtering is disabled
 when standard input is read.
 
+The output file is created exclusively: if something else creates a file or
+symbolic link with the name of the input file after it has been moved to the
+backup file, :exc:`FileExistsError` is raised.  If the output file cannot be
+created, the backup file is kept.  In-place filtering of files in a directory
+that other, untrusted users can modify is not supported.
+
+.. versionchanged:: next
+   The output file is now created exclusively, and the backup file is no
+   longer deleted if the output file cannot be created.
+
 
 The two following opening hooks are provided by this module:
 

Lib/fileinput.py

@@ -281,7 +281,9 @@ def nextfile(self):
             finally:
                 backupfilename = self._backupfilename
                 self._backupfilename = None
-                if backupfilename and not self._backup:
+                # If the output file could not be created, the backup file
+                # is the only copy of the data: keep it.
+                if backupfilename and not self._backup and output:
                     try: os.unlink(backupfilename)
                     except OSError: pass
 
@@ -329,29 +331,34 @@ def _readline(self):
             if self._inplace:
                 self._backupfilename = (
                     os.fspath(self._filename) + (self._backup or ".bak"))
-                try:
-                    os.unlink(self._backupfilename)
-                except OSError:
-                    pass
-                # The next few lines may raise OSError
-                os.rename(self._filename, self._backupfilename)
+                # The next few lines may raise OSError.  An existing backup
+                # file is only replaced once there is a file to back up: it
+                # can be the only copy of the data after an earlier failure.
+                os.replace(self._filename, self._backupfilename)
                 self._file = open(self._backupfilename, self._mode,
                                   encoding=encoding, errors=self._errors)
+                # The file has just been moved away, so nothing should exist
+                # under its name: create the output file exclusively rather
+                # than write through a file or symlink that has appeared there.
                 try:
                     perm = os.fstat(self._file.fileno()).st_mode
                 except OSError:
-                    self._output = open(self._filename, self._write_mode,
+                    self._output = open(self._filename,
+                                        self._write_mode.replace('w', 'x'),
                                         encoding=encoding, errors=self._errors)
                 else:
-                    mode = os.O_CREAT | os.O_WRONLY | os.O_TRUNC
+                    mode = os.O_CREAT | os.O_WRONLY | os.O_EXCL
                     if hasattr(os, 'O_BINARY'):
                         mode |= os.O_BINARY
 
                     fd = os.open(self._filename, mode, perm)
                     self._output = os.fdopen(fd, self._write_mode,
                                              encoding=encoding, errors=self._errors)
                     try:
-                        os.chmod(self._filename, perm)
+                        if os.chmod in os.supports_fd:
+                            os.chmod(fd, perm)
+                        else:
+                            os.chmod(self._filename, perm)
                     except OSError:
                         pass
                 self._savestdout = sys.stdout

Lib/test/test_fileinput.py

@@ -335,6 +335,101 @@ def test_inplace_encoding_errors(self):
         with open(temp_file, 'rb') as f:
             self.assertEqual(f.read().rstrip(b'\r\n'), b'New line ?')
 
+    def check_inplace_name_taken(self, take_name):
+        # Something else gets the name of the file after it has been moved
+        # to the backup file and before the output file is created.
+        temp_file = self.writeTmp('Initial text.')
+        self.addCleanup(os_helper.unlink, temp_file + '.bak')
+        orig_replace = os.replace
+        def replace(src, dst):
+            orig_replace(src, dst)
+            take_name(temp_file)
+        with mock.patch('os.replace', replace):
+            with self.assertRaises(FileExistsError):
+                with FileInput(temp_file, inplace=True, encoding="utf-8") as fi:
+                    fi.readline()
+        # The backup file is the only copy of the data, it must be kept.
+        with open(temp_file + '.bak', encoding="utf-8") as f:
+            self.assertEqual(f.read(), 'Initial text.')
+
+    @os_helper.skip_unless_symlink
+    def test_inplace_does_not_follow_symlink(self):
+        target = self.writeTmp('Other text.')
+        os.chmod(target, 0o600)
+        mode = os.stat(target).st_mode
+        def take_name(name):
+            os.chmod(name + '.bak', 0o644)
+            os.symlink(target, name)
+        self.check_inplace_name_taken(take_name)
+        with support.swap_attr(os, 'fstat', UnconditionallyRaise(OSError)):
+            self.check_inplace_name_taken(take_name)
+        with open(target, encoding="utf-8") as f:
+            self.assertEqual(f.read(), 'Other text.')
+        self.assertEqual(os.stat(target).st_mode, mode)
+
+    def test_inplace_does_not_reuse_existing_file(self):
+        names = []
+        def take_name(name):
+            with open(name, 'w', encoding="utf-8") as f:
+                f.write('Other text.')
+            names.append(name)
+        self.check_inplace_name_taken(take_name)
+        with support.swap_attr(os, 'fstat', UnconditionallyRaise(OSError)):
+            self.check_inplace_name_taken(take_name)
+        for name in names:
+            with open(name, encoding="utf-8") as f:
+                self.assertEqual(f.read(), 'Other text.')
+
+    def check_inplace_backup_kept(self, temp_file, exc_type):
+        with self.assertRaises(exc_type):
+            with FileInput(temp_file, inplace=True, encoding="utf-8") as fi:
+                fi.readline()
+        self.assertFalse(os.path.exists(temp_file))
+        with open(temp_file + '.bak', encoding="utf-8") as f:
+            self.assertEqual(f.read(), 'Initial text.')
+
+    def test_inplace_backup_kept_if_open_fails(self):
+        # The file can be moved to the backup file, but not opened.
+        temp_file = self.writeTmp('Initial text.')
+        self.addCleanup(os_helper.unlink, temp_file + '.bak')
+        with mock.patch('fileinput.open', create=True,
+                        side_effect=PermissionError):
+            self.check_inplace_backup_kept(temp_file, PermissionError)
+            # Trying again must not delete the backup file either.
+            self.check_inplace_backup_kept(temp_file, FileNotFoundError)
+        self.check_inplace_backup_kept(temp_file, FileNotFoundError)
+
+    def test_inplace_backup_kept_if_output_not_created(self):
+        temp_file = self.writeTmp('Initial text.')
+        self.addCleanup(os_helper.unlink, temp_file + '.bak')
+        with mock.patch('os.open', side_effect=PermissionError):
+            self.check_inplace_backup_kept(temp_file, PermissionError)
+            self.check_inplace_backup_kept(temp_file, FileNotFoundError)
+        self.check_inplace_backup_kept(temp_file, FileNotFoundError)
+
+    @os_helper.skip_unless_symlink
+    @unittest.skipUnless(os.chmod in os.supports_fd,
+                         "os.chmod() does not support file descriptors")
+    @unittest.skipIf(sys.platform == 'win32',
+                     "cannot unlink an open file on Windows")
+    def test_inplace_chmod_does_not_follow_symlink(self):
+        # The output file is replaced by a symlink after it has been created
+        # and before its permissions are set.
+        target = self.writeTmp('Other text.')
+        os.chmod(target, 0o600)
+        mode = os.stat(target).st_mode
+        temp_file = self.writeTmp('Initial text.')
+        os.chmod(temp_file, 0o644)
+        orig_fdopen = os.fdopen
+        def fdopen(fd, *args, **kwargs):
+            os.unlink(temp_file)
+            os.symlink(target, temp_file)
+            return orig_fdopen(fd, *args, **kwargs)
+        with mock.patch('os.fdopen', fdopen):
+            with FileInput(temp_file, inplace=True, encoding="utf-8") as fi:
+                fi.readline()
+        self.assertEqual(os.stat(target).st_mode, mode)
+
     def test_file_hook_backward_compatibility(self):
         def old_hook(filename, mode):
             return io.StringIO("I used to receive only filename and mode")

Misc/NEWS.d/next/Library/2026-10-22-08-09-29.gh-issue-160406.q7Rk2d.rst

@@ -0,0 +1,4 @@
+Fix in-place filtering in :mod:`fileinput` deleting the file when it cannot
+be opened or the output file cannot be created: the backup file is now kept.
+The output file is now created exclusively, so :exc:`FileExistsError` is
+raised if a file or symbolic link appears under the name of the input file.