00435: pythongh-121650: Encode newlines in headers, and verify header…

…s are sound (pythonGH-122233)

Per RFC 2047:

> [...] these encoding schemes allow the
> encoding of arbitrary octet values, mail readers that implement this
> decoding should also ensure that display of the decoded data on the
> recipient's terminal will not cause unwanted side-effects

It seems that the "quoted-word" scheme is a valid way to include
a newline character in a header value, just like we already allow
undecodable bytes or control characters.
They do need to be properly quoted when serialized to text, though.

This should fail for custom fold() implementations that aren't careful
about newlines.

(cherry picked from commit 0976339)

Co-authored-by: Petr Viktorin <encukou@gmail.com>
Co-authored-by: Bas Bloemsaat <bas@bloemsaat.org>
Co-authored-by: Serhiy Storchaka <storchaka@gmail.com>

Doc/library/email.errors.rst

@@ -58,6 +58,13 @@ The following exception classes are defined in the :mod:`email.errors` module:
    :class:`~email.mime.nonmultipart.MIMENonMultipart` (e.g.
    :class:`~email.mime.image.MIMEImage`).
 
+
+.. exception:: HeaderWriteError()
+
+   Raised when an error occurs when the :mod:`~email.generator` outputs
+   headers.
+
+
 .. exception:: MessageDefect()
 
    This is the base class for all defects found when parsing email messages.

Doc/library/email.policy.rst

@@ -228,6 +228,24 @@ added matters.  To illustrate::
 
       .. versionadded:: 3.6
 
+
+   .. attribute:: verify_generated_headers
+
+      If ``True`` (the default), the generator will raise
+      :exc:`~email.errors.HeaderWriteError` instead of writing a header
+      that is improperly folded or delimited, such that it would
+      be parsed as multiple headers or joined with adjacent data.
+      Such headers can be generated by custom header classes or bugs
+      in the ``email`` module.
+
+      As it's a security feature, this defaults to ``True`` even in the
+      :class:`~email.policy.Compat32` policy.
+      For backwards compatible, but unsafe, behavior, it must be set to
+      ``False`` explicitly.
+
+      .. versionadded:: 3.11.10
+
+
    The following :class:`Policy` method is intended to be called by code using
    the email library to create policy instances with custom settings:
 

Doc/whatsnew/3.11.rst

@@ -2728,6 +2728,7 @@ OpenSSL
 
 .. _libb2: https://www.blake2.net/
 
+
 Notable changes in 3.11.10
 ==========================
 
@@ -2736,3 +2737,15 @@ ipaddress
 
 * Fixed ``is_global`` and ``is_private`` behavior in ``IPv4Address``,
   ``IPv6Address``, ``IPv4Network`` and ``IPv6Network``.
+
+email
+-----
+
+* Headers with embedded newlines are now quoted on output.
+
+  The :mod:`~email.generator` will now refuse to serialize (write) headers
+  that are improperly folded or delimited, such that they would be parsed as
+  multiple headers or joined with adjacent data.
+  If you need to turn this safety feature off,
+  set :attr:`~email.policy.Policy.verify_generated_headers`.
+  (Contributed by Bas Bloemsaat and Petr Viktorin in :gh:`121650`.)

Lib/email/_header_value_parser.py

@@ -92,6 +92,8 @@
 ASPECIALS = TSPECIALS | set("*'%")
 ATTRIBUTE_ENDS = ASPECIALS | WSP
 EXTENDED_ATTRIBUTE_ENDS = ATTRIBUTE_ENDS - set('%')
+NLSET = {'\n', '\r'}
+SPECIALSNL = SPECIALS | NLSET
 
 def quote_string(value):
     return '"'+str(value).replace('\\', '\\\\').replace('"', r'\"')+'"'
@@ -2780,9 +2782,13 @@ def _refold_parse_tree(parse_tree, *, policy):
             wrap_as_ew_blocked -= 1
             continue
         tstr = str(part)
-        if part.token_type == 'ptext' and set(tstr) & SPECIALS:
-            # Encode if tstr contains special characters.
-            want_encoding = True
+        if not want_encoding:
+            if part.token_type == 'ptext':
+                # Encode if tstr contains special characters.
+                want_encoding = not SPECIALSNL.isdisjoint(tstr)
+            else:
+                # Encode if tstr contains newlines.
+                want_encoding = not NLSET.isdisjoint(tstr)
         try:
             tstr.encode(encoding)
             charset = encoding

Lib/email/_policybase.py

@@ -157,6 +157,13 @@ class Policy(_PolicyBase, metaclass=abc.ABCMeta):
     message_factory     -- the class to use to create new message objects.
                            If the value is None, the default is Message.
 
+    verify_generated_headers
+                        -- if true, the generator verifies that each header
+                           they are properly folded, so that a parser won't
+                           treat it as multiple headers, start-of-body, or
+                           part of another header.
+                           This is a check against custom Header & fold()
+                           implementations.
     """
 
     raise_on_defect = False
@@ -165,6 +172,7 @@ class Policy(_PolicyBase, metaclass=abc.ABCMeta):
     max_line_length = 78
     mangle_from_ = False
     message_factory = None
+    verify_generated_headers = True
 
     def handle_defect(self, obj, defect):
         """Based on policy, either raise defect or call register_defect.

Lib/email/errors.py

@@ -29,6 +29,10 @@ class CharsetError(MessageError):
     """An illegal charset was given."""
 
 
+class HeaderWriteError(MessageError):
+    """Error while writing headers."""
+
+
 # These are parsing defects which the parser was able to work around.
 class MessageDefect(ValueError):
     """Base class for a message defect."""

Lib/email/generator.py

@@ -14,12 +14,14 @@
 from copy import deepcopy
 from io import StringIO, BytesIO
 from email.utils import _has_surrogates
+from email.errors import HeaderWriteError
 
 UNDERSCORE = '_'
 NL = '\n'  # XXX: no longer used by the code below.
 
 NLCRE = re.compile(r'\r\n|\r|\n')
 fcre = re.compile(r'^From ', re.MULTILINE)
+NEWLINE_WITHOUT_FWSP = re.compile(r'\r\n[^ \t]|\r[^ \n\t]|\n[^ \t]')
 
 
 class Generator:
@@ -222,7 +224,16 @@ def _dispatch(self, msg):
 
     def _write_headers(self, msg):
         for h, v in msg.raw_items():
-            self.write(self.policy.fold(h, v))
+            folded = self.policy.fold(h, v)
+            if self.policy.verify_generated_headers:
+                linesep = self.policy.linesep
+                if not folded.endswith(self.policy.linesep):
+                    raise HeaderWriteError(
+                        f'folded header does not end with {linesep!r}: {folded!r}')
+                if NEWLINE_WITHOUT_FWSP.search(folded.removesuffix(linesep)):
+                    raise HeaderWriteError(
+                        f'folded header contains newline: {folded!r}')
+            self.write(folded)
         # A blank line always separates headers from body
         self.write(self._NL)
 

Lib/test/test_email/test_generator.py

@@ -6,6 +6,7 @@
 from email.generator import Generator, BytesGenerator
 from email.headerregistry import Address
 from email import policy
+import email.errors
 from test.test_email import TestEmailBase, parameterize
 
 
@@ -216,6 +217,44 @@ def test_rfc2231_wrapping_switches_to_default_len_if_too_narrow(self):
         g.flatten(msg)
         self.assertEqual(s.getvalue(), self.typ(expected))
 
+    def test_keep_encoded_newlines(self):
+        msg = self.msgmaker(self.typ(textwrap.dedent("""\
+            To: nobody
+            Subject: Bad subject=?UTF-8?Q?=0A?=Bcc: injection@example.com
+
+            None
+            """)))
+        expected = textwrap.dedent("""\
+            To: nobody
+            Subject: Bad subject=?UTF-8?Q?=0A?=Bcc: injection@example.com
+
+            None
+            """)
+        s = self.ioclass()
+        g = self.genclass(s, policy=self.policy.clone(max_line_length=80))
+        g.flatten(msg)
+        self.assertEqual(s.getvalue(), self.typ(expected))
+
+    def test_keep_long_encoded_newlines(self):
+        msg = self.msgmaker(self.typ(textwrap.dedent("""\
+            To: nobody
+            Subject: Bad subject=?UTF-8?Q?=0A?=Bcc: injection@example.com
+
+            None
+            """)))
+        expected = textwrap.dedent("""\
+            To: nobody
+            Subject: Bad subject
+             =?utf-8?q?=0A?=Bcc:
+             injection@example.com
+
+            None
+            """)
+        s = self.ioclass()
+        g = self.genclass(s, policy=self.policy.clone(max_line_length=30))
+        g.flatten(msg)
+        self.assertEqual(s.getvalue(), self.typ(expected))
+
 
 class TestGenerator(TestGeneratorBase, TestEmailBase):
 
@@ -224,6 +263,29 @@ class TestGenerator(TestGeneratorBase, TestEmailBase):
     ioclass = io.StringIO
     typ = str
 
+    def test_verify_generated_headers(self):
+        """gh-121650: by default the generator prevents header injection"""
+        class LiteralHeader(str):
+            name = 'Header'
+            def fold(self, **kwargs):
+                return self
+
+        for text in (
+            'Value\r\nBad Injection\r\n',
+            'NoNewLine'
+        ):
+            with self.subTest(text=text):
+                message = message_from_string(
+                    "Header: Value\r\n\r\nBody",
+                    policy=self.policy,
+                )
+
+                del message['Header']
+                message['Header'] = LiteralHeader(text)
+
+                with self.assertRaises(email.errors.HeaderWriteError):
+                    message.as_string()
+
 
 class TestBytesGenerator(TestGeneratorBase, TestEmailBase):
 

Lib/test/test_email/test_policy.py

@@ -26,6 +26,7 @@ class PolicyAPITests(unittest.TestCase):
         'raise_on_defect':          False,
         'mangle_from_':             True,
         'message_factory':          None,
+        'verify_generated_headers': True,
         }
     # These default values are the ones set on email.policy.default.
     # If any of these defaults change, the docs must be updated.
@@ -294,6 +295,31 @@ def test_short_maxlen_error(self):
                 with self.assertRaises(email.errors.HeaderParseError):
                     policy.fold("Subject", subject)
 
+    def test_verify_generated_headers(self):
+        """Turning protection off allows header injection"""
+        policy = email.policy.default.clone(verify_generated_headers=False)
+        for text in (
+            'Header: Value\r\nBad: Injection\r\n',
+            'Header: NoNewLine'
+        ):
+            with self.subTest(text=text):
+                message = email.message_from_string(
+                    "Header: Value\r\n\r\nBody",
+                    policy=policy,
+                )
+                class LiteralHeader(str):
+                    name = 'Header'
+                    def fold(self, **kwargs):
+                        return self
+
+                del message['Header']
+                message['Header'] = LiteralHeader(text)
+
+                self.assertEqual(
+                    message.as_string(),
+                    f"{text}\nBody",
+                )
+
     # XXX: Need subclassing tests.
     # For adding subclassed objects, make sure the usual rules apply (subclass
     # wins), but that the order still works (right overrides left).

Misc/NEWS.d/next/Library/2024-07-27-16-10-41.gh-issue-121650.nf6oc9.rst

@@ -0,0 +1,5 @@
+:mod:`email` headers with embedded newlines are now quoted on output. The
+:mod:`~email.generator` will now refuse to serialize (write) headers that
+are unsafely folded or delimited; see
+:attr:`~email.policy.Policy.verify_generated_headers`. (Contributed by Bas
+Bloemsaat and Petr Viktorin in :gh:`121650`.)