diff --git a/Lib/mailbox.py b/Lib/mailbox.py index 99426220154360b..510330fcdff54b6 100644 --- a/Lib/mailbox.py +++ b/Lib/mailbox.py @@ -18,6 +18,8 @@ import email.generator import io import contextlib +import shutil +import tempfile from types import GenericAlias try: import fcntl @@ -1073,8 +1075,11 @@ def __setitem__(self, key, message): if self._locked: _lock_file(f) try: - os.close(os.open(path, os.O_WRONLY | os.O_TRUNC)) - self._dump_message(message, f) + with tempfile.TemporaryFile(mode='w+b') as new_file: + self._dump_message(message, new_file) + new_file.seek(0) + os.close(os.open(path, os.O_WRONLY | os.O_TRUNC)) + shutil.copyfileobj(new_file, f) if isinstance(message, MHMessage): self._dump_sequences(message, key) finally: diff --git a/Lib/test/test_mailbox.py b/Lib/test/test_mailbox.py index 019c699bff55c42..c864a08d695540a 100644 --- a/Lib/test/test_mailbox.py +++ b/Lib/test/test_mailbox.py @@ -1,4 +1,5 @@ import os +import stat import sys import time import socket @@ -1338,6 +1339,68 @@ class TestMH(TestMailbox, unittest.TestCase): def assertMailboxEmpty(self): self.assertEqual(os.listdir(self._path), ['.mh_sequences']) + def test_set_item_nonascii_string_raises_without_modifying_message(self): + key = self._box.add(self._template % 'original') + original = self._box.get_bytes(key) + with self.assertRaisesRegex(ValueError, "ASCII-only"): + self._box[key] = self._nonascii_msg + self.assertEqual(self._box.get_bytes(key), original) + self._box.close() + self._box = self._factory(self._path) + self.assertEqual(self._box.get_bytes(key), original) + + def test_set_item_read_error_does_not_modify_message(self): + class CustomError(Exception): + pass + + class FaultyMessage: + def __init__(self): + self.first_read = True + + def read(self): + raise AssertionError + + def readline(self): + if self.first_read: + self.first_read = False + return b'Subject: replacement\n' + raise CustomError + + key = self._box.add(self._template % 'original') + original = self._box.get_bytes(key) + original_files = set(os.listdir(self._path)) + with self.assertRaises(CustomError): + self._box[key] = FaultyMessage() + self.assertEqual(self._box.get_bytes(key), original) + self.assertEqual(set(os.listdir(self._path)), original_files) + self._box.close() + self._box = self._factory(self._path) + self.assertEqual(self._box.get_bytes(key), original) + + @unittest.skipUnless(os.name == 'posix', 'requires POSIX permissions') + def test_set_item_preserves_mode(self): + key = self._box.add(self._template % 'original') + path = os.path.join(self._path, str(key)) + mode = 0o640 + os.chmod(path, mode) + if stat.S_IMODE(os.stat(path).st_mode) != mode: + self.skipTest('filesystem does not support POSIX permissions') + + self._box[key] = self._template % 'replacement' + + self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), mode) + + def test_set_item_with_open_file(self): + key = self._box.add(self._template % 'original') + replacement = self._template % 'replacement' + self._box.lock() + try: + with self._box.get_file(key): + self._box[key] = replacement + self.assertEqual(self._box.get_bytes(key), replacement.encode('ascii')) + finally: + self._box.unlock() + def test_list_folders(self): # List folders self._box.add_folder('one') diff --git a/Misc/NEWS.d/next/Library/2026-08-24-19-37-56.gh-issue-156312.tF9p9m.rst b/Misc/NEWS.d/next/Library/2026-08-24-19-37-56.gh-issue-156312.tF9p9m.rst new file mode 100644 index 000000000000000..9fcfc2c6a0a8d6f --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-08-24-19-37-56.gh-issue-156312.tF9p9m.rst @@ -0,0 +1,3 @@ +Prevent :class:`mailbox.MH` message replacement from truncating or partially +overwriting the original message when serializing the replacement raises an +exception.