gh-160406: Keep the backup and create output exclusively in fileinput

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.