gh-157254: Preserve email content on failure and reject unsupported bytes CTEs

Lib/email/contentmanager.py

@@ -34,8 +34,15 @@ def set_content(self, msg, obj, *args, **kw):
             # but we can't add it later, so do it for now.
             raise TypeError("set_content not valid on multipart")
         handler = self._find_set_handler(msg, obj)
-        msg.clear_content()
-        handler(msg, obj, *args, **kw)
+        headers = msg._headers
+        payload = msg._payload
+        try:
+            msg.clear_content()
+            handler(msg, obj, *args, **kw)
+        except BaseException:
+            msg._headers = headers
+            msg._payload = payload
+            raise
 
     def _find_set_handler(self, msg, obj):
         full_path_for_error = None
@@ -234,6 +241,8 @@ def set_bytes_content(msg, data, maintype, subtype, cte='base64',
         data = data.decode('ascii')
     elif cte in ('8bit', 'binary'):
         data = data.decode('ascii', 'surrogateescape')
+    else:
+        raise ValueError("Unknown content transfer encoding {}".format(cte))
     msg.set_payload(data)
     msg['Content-Transfer-Encoding'] = cte
     _finalize_set(msg, disposition, filename, cid, params)

Lib/email/message.py

@@ -1144,13 +1144,22 @@ def set_content(self, *args, content_manager=None, **kw):
             content_manager = self.policy.content_manager
         content_manager.set_content(self, *args, **kw)
 
-    def _make_multipart(self, subtype, disallowed_subtypes, boundary):
+    _multipart_disallowed_subtypes = {
+        'related': ('alternative', 'mixed'),
+        'alternative': ('mixed',),
+        'mixed': (),
+    }
+
+    def _check_multipart_conversion(self, subtype, disallowed_subtypes):
         if self.get_content_maintype() == 'multipart':
             existing_subtype = self.get_content_subtype()
             disallowed_subtypes = disallowed_subtypes + (subtype,)
             if existing_subtype in disallowed_subtypes:
                 raise ValueError("Cannot convert {} to {}".format(
                     existing_subtype, subtype))
+
+    def _make_multipart(self, subtype, disallowed_subtypes, boundary):
+        self._check_multipart_conversion(subtype, disallowed_subtypes)
         keep_headers = []
         part_headers = []
         for name, value in self._headers:
@@ -1172,22 +1181,30 @@ def _make_multipart(self, subtype, disallowed_subtypes, boundary):
             self.set_param('boundary', boundary)
 
     def make_related(self, boundary=None):
-        self._make_multipart('related', ('alternative', 'mixed'), boundary)
+        self._make_multipart(
+            'related', self._multipart_disallowed_subtypes['related'], boundary)
 
     def make_alternative(self, boundary=None):
-        self._make_multipart('alternative', ('mixed',), boundary)
+        self._make_multipart(
+            'alternative', self._multipart_disallowed_subtypes['alternative'],
+            boundary)
 
     def make_mixed(self, boundary=None):
-        self._make_multipart('mixed', (), boundary)
+        self._make_multipart(
+            'mixed', self._multipart_disallowed_subtypes['mixed'], boundary)
 
     def _add_multipart(self, _subtype, *args, _disp=None, **kw):
-        if (self.get_content_maintype() != 'multipart' or
-                self.get_content_subtype() != _subtype):
-            getattr(self, 'make_' + _subtype)()
+        needs_conversion = (self.get_content_maintype() != 'multipart' or
+                            self.get_content_subtype() != _subtype)
+        if needs_conversion:
+            self._check_multipart_conversion(
+                _subtype, self._multipart_disallowed_subtypes[_subtype])
         part = type(self)(policy=self.policy)
         part.set_content(*args, **kw)
         if _disp and 'content-disposition' not in part:
             part['Content-Disposition'] = _disp
+        if needs_conversion:
+            getattr(self, 'make_' + _subtype)()
         self.attach(part)
 
     def add_related(self, *args, **kw):

Lib/test/test_email/test_contentmanager.py

@@ -131,6 +131,20 @@ def test_set_content_calls_clear_content(self):
         self.assertEqual(m['To'], 'test')
         self.assertIsNone(m.get_payload())
 
+    def test_set_content_base_exception_preserves_content(self):
+        def handler(msg, obj):
+            msg['X-New'] = 'new'
+            msg.set_payload(obj)
+            raise KeyboardInterrupt
+        cm = ContentManager()
+        cm.add_set_handler(str, handler)
+        m = self._make_message()
+        m.set_content('original')
+        original = m.as_bytes()
+        with self.assertRaises(KeyboardInterrupt):
+            m.set_content('replacement', content_manager=cm)
+        self.assertEqual(m.as_bytes(), original)
+
 
 @parameterize
 class TestRawDataManager(TestEmailBase):
@@ -145,6 +159,31 @@ class TestRawDataManager(TestEmailBase):
                                   content_manager=raw_data_manager)
     message = EmailMessage
 
+    content_failure_params = {
+        'unknown_charset': ('replacement',
+                            {'charset': 'does-not-exist'}, LookupError),
+        'ascii_charset': ('\N{LATIN SMALL LETTER E WITH ACUTE}',
+                          {'charset': 'ascii'}, UnicodeEncodeError),
+        'text_cte': ('replacement', {'cte': 'unknown'}, ValueError),
+        'bytes_7bit': (b'\xff', {'maintype': 'application',
+                                'subtype': 'octet-stream', 'cte': '7bit'},
+                       UnicodeDecodeError),
+        'header': ('replacement', {'headers': ['Subject: duplicate']},
+                   ValueError),
+        'cid': ('replacement', {'cid': 'bad\nvalue'}, ValueError),
+    }
+
+    def content_failure_as_set_content_preserves_content(self, content, kw,
+                                                        error):
+        m = self._make_message()
+        m['Subject'] = 'original subject'
+        m.set_content('original')
+        original = m.as_bytes()
+        with self.assertRaises(error):
+            m.set_content(content, **kw)
+        self.assertEqual(m.as_bytes(), original)
+        self.assertFalse(m.is_multipart())
+
     def test_get_text_plain(self):
         m = self._str_msg(textwrap.dedent("""\
             Content-Type: text/plain
@@ -725,6 +764,17 @@ def test_set_application_octet_stream_with_8bit_cte(self):
         self.assertEqual(m.get_payload(decode=True), content)
         self.assertEqual(m.get_content(), content)
 
+    def test_set_bytes_unknown_cte_raises(self):
+        m = self._make_message()
+        m.set_content('original')
+        original = bytes(m)
+        with self.assertRaisesRegex(
+                ValueError,
+                'Unknown content transfer encoding not-a-transfer-encoding'):
+            m.set_content(b'abc', 'application', 'octet-stream',
+                          cte='not-a-transfer-encoding')
+        self.assertEqual(bytes(m), original)
+
     def test_set_headers_from_header_objects(self):
         m = self._make_message()
         content = "Simple message.\n"

Lib/test/test_email/test_message.py

@@ -709,6 +709,46 @@ def test_default_content_manager_for_add_comes_from_policy(self):
                 getattr(m, method)('')
             self.assertEqual(str(ar.exception), 'test')
 
+    def test_add_multipart_failure_preserves_content(self):
+        for method, existing_subtypes in (
+                ('add_related', (None, 'related')),
+                ('add_alternative', (None, 'related', 'alternative')),
+                ('add_attachment', (None, 'related', 'mixed'))):
+            for existing in existing_subtypes:
+                for content, kw, error in (
+                        ('replacement', {'charset': 'does-not-exist'},
+                         LookupError),
+                        (b'abc', {'maintype': 'application',
+                                  'subtype': 'octet-stream', 'cte': 'unknown'},
+                         ValueError)):
+                    with self.subTest(method=method, existing=existing, kw=kw):
+                        m = self._make_message()
+                        m.set_content('original')
+                        if existing is not None:
+                            getattr(m, 'make_' + existing)()
+                        original = m.as_bytes()
+                        parts = list(m.iter_parts())
+                        with self.assertRaises(error):
+                            getattr(m, method)(content, **kw)
+                        self.assertEqual(m.as_bytes(), original)
+                        self.assertEqual(m.is_multipart(), existing is not None)
+                        self.assertEqual(list(m.iter_parts()), parts)
+
+    def test_add_multipart_checks_conversion_before_content(self):
+        for existing, target in (('mixed', 'related'),
+                                 ('mixed', 'alternative'),
+                                 ('alternative', 'related')):
+            with self.subTest(existing=existing, target=target):
+                m = self._make_message()
+                m.set_content('original')
+                getattr(m, 'make_' + existing)()
+                original = m.as_bytes()
+                with self.assertRaisesRegex(
+                        ValueError, f'Cannot convert {existing} to {target}'):
+                    getattr(m, 'add_' + target)(
+                        'replacement', charset='does-not-exist')
+                self.assertEqual(m.as_bytes(), original)
+
     def message_as_clear(self, body_parts, attachments, parts, msg):
         m = self._str_msg(msg)
         m.clear()

Misc/NEWS.d/next/Library/2026-09-10-17-18-27.gh-issue-157254.R8VptZ.rst

@@ -0,0 +1,4 @@
+Preserve existing message content when built-in :mod:`email` content
+handlers fail, and avoid changing the parent MIME structure when preparing a
+new part fails. Reject unsupported content transfer encodings for bytes
+content.