diff --git a/Doc/library/fileinput.rst b/Doc/library/fileinput.rst index 5be16797be908c..24132ee39f130c 100644 --- a/Doc/library/fileinput.rst +++ b/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: diff --git a/Lib/fileinput.py b/Lib/fileinput.py index 3dba3d2fbfa967..34cb7943c9f508 100644 --- a/Lib/fileinput.py +++ b/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,21 +331,23 @@ 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 @@ -351,7 +355,10 @@ def _readline(self): 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 diff --git a/Lib/test/test_fileinput.py b/Lib/test/test_fileinput.py index 6524baabe7f96f..a11e43714dad58 100644 --- a/Lib/test/test_fileinput.py +++ b/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") diff --git a/Misc/NEWS.d/next/Library/2026-10-22-08-09-29.gh-issue-160406.q7Rk2d.rst b/Misc/NEWS.d/next/Library/2026-10-22-08-09-29.gh-issue-160406.q7Rk2d.rst new file mode 100644 index 00000000000000..42f1774b829eee --- /dev/null +++ b/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.