Merge branch '3.14' into backport-77bf4ba-3.14

Author: pablogsal

Lib/email/_header_value_parser.py

@@ -101,6 +101,12 @@ def make_quoted_pairs(value):
     return str(value).replace('\\', '\\\\').replace('"', '\\"')
 
 
+def make_parenthesis_pairs(value):
+    """Escape parenthesis and backslash for use within a comment."""
+    return str(value).replace('\\', '\\\\') \
+        .replace('(', '\\(').replace(')', '\\)')
+
+
 def quote_string(value):
     escaped = make_quoted_pairs(value)
     return f'"{escaped}"'
@@ -939,7 +945,7 @@ def value(self):
         return ' '
 
     def startswith_fws(self):
-        return True
+        return self and self[0] in WSP
 
 
 class ValueTerminal(Terminal):
@@ -2959,6 +2965,13 @@ def _refold_parse_tree(parse_tree, *, policy):
                     [ValueTerminal(make_quoted_pairs(p), 'ptext')
                      for p in newparts] +
                     [ValueTerminal('"', 'ptext')])
+            if part.token_type == 'comment':
+                newparts = (
+                    [ValueTerminal('(', 'ptext')] +
+                    [ValueTerminal(make_parenthesis_pairs(p), 'ptext')
+                     if p.token_type == 'ptext' else p
+                     for p in newparts] +
+                    [ValueTerminal(')', 'ptext')])
             if not part.as_ew_allowed:
                 wrap_as_ew_blocked += 1
                 newparts.append(end_ew_not_allowed)

Lib/email/generator.py

@@ -22,6 +22,7 @@
 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]')
+NEWLINE_WITHOUT_FWSP_BYTES = re.compile(br'\r\n[^ \t]|\r[^ \n\t]|\n[^ \t]')
 
 
 class Generator:
@@ -429,7 +430,16 @@ def _write_headers(self, msg):
         # This is almost the same as the string version, except for handling
         # strings with 8bit bytes.
         for h, v in msg.raw_items():
-            self._fp.write(self.policy.fold_binary(h, v))
+            folded = self.policy.fold_binary(h, v)
+            if self.policy.verify_generated_headers:
+                linesep = self.policy.linesep.encode()
+                if not folded.endswith(linesep):
+                    raise HeaderWriteError(
+                        f'folded header does not end with {linesep!r}: {folded!r}')
+                if NEWLINE_WITHOUT_FWSP_BYTES.search(folded.removesuffix(linesep)):
+                    raise HeaderWriteError(
+                        f'folded header contains newline: {folded!r}')
+            self._fp.write(folded)
         # A blank line always separates headers from body
         self.write(self._NL)
 

Lib/logging/handlers.py

@@ -196,7 +196,11 @@ def shouldRollover(self, record):
         if self.stream is None:                 # delay was set...
             self.stream = self._open()
         if self.maxBytes > 0:                   # are we rolling over?
-            pos = self.stream.tell()
+            try:
+                pos = self.stream.tell()
+            except io.UnsupportedOperation:
+                # gh-143237: Never rollover a named pipe.
+                return False
             if not pos:
                 # gh-116263: Never rollover an empty file
                 return False

Lib/test/test_email/test__header_value_parser.py

@@ -3294,6 +3294,29 @@ def test_address_list_with_specials_in_long_quoted_string(self):
             with self.subTest(to=to):
                 self._test(parser.get_address_list(to)[0], folded, policy=policy)
 
+    def test_address_list_with_long_unwrapable_comment(self):
+        policy = self.policy.clone(max_line_length=40)
+        cases = [
+            # (to, folded)
+            ('(loremipsumdolorsitametconsecteturadipi)<spy@example.org>',
+             '(loremipsumdolorsitametconsecteturadipi)<spy@example.org>\n'),
+            ('<spy@example.org>(loremipsumdolorsitametconsecteturadipi)',
+             '<spy@example.org>(loremipsumdolorsitametconsecteturadipi)\n'),
+            ('(loremipsum dolorsitametconsecteturadipi)<spy@example.org>',
+             '(loremipsum dolorsitametconsecteturadipi)<spy@example.org>\n'),
+             ('<spy@example.org>(loremipsum dolorsitametconsecteturadipi)',
+             '<spy@example.org>(loremipsum\n dolorsitametconsecteturadipi)\n'),
+            ('(Escaped \\( \\) chars \\\\ in comments stay escaped)<spy@example.org>',
+             '(Escaped \\( \\) chars \\\\ in comments stay\n escaped)<spy@example.org>\n'),
+            ('((loremipsum)(loremipsum)(loremipsum)(loremipsum))<spy@example.org>',
+             '((loremipsum)(loremipsum)(loremipsum)(loremipsum))<spy@example.org>\n'),
+            ('((loremipsum)(loremipsum)(loremipsum) (loremipsum))<spy@example.org>',
+             '((loremipsum)(loremipsum)(loremipsum)\n (loremipsum))<spy@example.org>\n'),
+        ]
+        for (to, folded) in cases:
+            with self.subTest(to=to):
+                self._test(parser.get_address_list(to)[0], folded, policy=policy)
+
     # XXX Need tests with comments on various sides of a unicode token,
     # and with unicode tokens in the comments.  Spaces inside the quotes
     # currently don't do the right thing.

Lib/test/test_email/test_generator.py

@@ -313,7 +313,7 @@ def test_flatten_unicode_linesep(self):
         self.assertEqual(s.getvalue(), self.typ(expected))
 
     def test_verify_generated_headers(self):
-        """gh-121650: by default the generator prevents header injection"""
+        # gh-121650: by default the generator prevents header injection
         class LiteralHeader(str):
             name = 'Header'
             def fold(self, **kwargs):
@@ -334,6 +334,8 @@ def fold(self, **kwargs):
 
                 with self.assertRaises(email.errors.HeaderWriteError):
                     message.as_string()
+                with self.assertRaises(email.errors.HeaderWriteError):
+                    message.as_bytes()
 
 
 class TestBytesGenerator(TestGeneratorBase, TestEmailBase):

Lib/test/test_email/test_policy.py

@@ -296,7 +296,7 @@ def test_short_maxlen_error(self):
                     policy.fold("Subject", subject)
 
     def test_verify_generated_headers(self):
-        """Turning protection off allows header injection"""
+        # 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',
@@ -319,6 +319,10 @@ def fold(self, **kwargs):
                     message.as_string(),
                     f"{text}\nBody",
                 )
+                self.assertEqual(
+                    message.as_bytes(),
+                    f"{text}\nBody".encode(),
+                )
 
     # XXX: Need subclassing tests.
     # For adding subclassed objects, make sure the usual rules apply (subclass

Lib/test/test_logging.py

@@ -25,6 +25,7 @@
 
 import codecs
 import configparser
+import contextlib
 import copy
 import datetime
 import pathlib
@@ -6343,6 +6344,32 @@ def test_should_not_rollover_non_file(self):
         self.assertFalse(rh.shouldRollover(self.next_rec()))
         rh.close()
 
+    @unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()')
+    def test_should_not_rollover_named_pipe(self):
+        # gh-143237 - test with non-seekable special file (named pipe)
+        filename = os_helper.TESTFN
+        self.addCleanup(os_helper.unlink, filename)
+        try:
+            os.mkfifo(filename)
+        except PermissionError as e:
+            self.skipTest('os.mkfifo(): %s' % e)
+
+        data = 'not read'
+        def other_side():
+            nonlocal data
+            with open(filename, 'rb') as f:
+                data = f.read()
+
+        thread = threading.Thread(target=other_side)
+        with threading_helper.start_threads([thread]):
+            rh = logging.handlers.RotatingFileHandler(
+                    filename, encoding="utf-8", maxBytes=1)
+            with contextlib.closing(rh):
+                m = self.next_rec()
+                self.assertFalse(rh.shouldRollover(m))
+                rh.emit(m)
+        self.assertEqual(data.decode(), m.msg + os.linesep)
+
     def test_should_rollover(self):
         with open(self.fn, 'wb') as f:
             f.write(b'\n')

Misc/NEWS.d/next/Library/2025-12-28-20-28-05.gh-issue-143237.q1ymuA.rst

@@ -0,0 +1 @@
+Fix support of named pipes in the rotating :mod:`logging` handlers.

Misc/NEWS.d/next/Security/2026-01-16-14-40-31.gh-issue-143935.U2YtKl.rst

@@ -0,0 +1,6 @@
+Fixed a bug in the folding of comments when flattening an email message
+using a modern email policy. Comments consisting of a very long sequence of
+non-foldable characters could trigger a forced line wrap that omitted the
+required leading space on the continuation line, causing the remainder of
+the comment to be interpreted as a new header field. This enabled header
+injection with carefully crafted inputs.

Misc/NEWS.d/next/Security/2026-01-21-12-34-05.gh-issue-144125.TAz5uo.rst

@@ -0,0 +1,4 @@
+:mod:`~email.generator.BytesGenerator` 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`).