gh-127794: Validate header name according to rfc-5322

Lib/email/_policybase.py

@@ -4,6 +4,7 @@
 """
 
 import abc
+import re
 from email import header
 from email import charset as _charset
 from email.utils import _has_surrogates
@@ -14,6 +15,12 @@
     'compat32',
     ]
 
+valid_header_name_re = re.compile("[\041-\071\073-\176]+$")
+
+def validate_header_name(name):
+    # Validate header name according to RFC 5322
+    if not valid_header_name_re.match(name):
+        raise ValueError(f"Header field name contains invalid characters: {name!r}")
 
 class _PolicyBase:
 
@@ -314,6 +321,7 @@ def header_store_parse(self, name, value):
         """+
         The name and value are returned unmodified.
         """
+        validate_header_name(name)
         return (name, value)
 
     def header_fetch_parse(self, name, value):

Lib/email/policy.py

@@ -4,7 +4,13 @@
 
 import re
 import sys
-from email._policybase import Policy, Compat32, compat32, _extend_docstrings
+from email._policybase import (
+    _extend_docstrings,
+    Compat32,
+    compat32,
+    Policy,
+    validate_header_name
+)
 from email.utils import _has_surrogates
 from email.headerregistry import HeaderRegistry as HeaderRegistry
 from email.contentmanager import raw_data_manager
@@ -140,6 +146,7 @@ def header_store_parse(self, name, value):
         """
         if hasattr(value, 'name') and value.name.lower() == name.lower():
             return (name, value)
+        validate_header_name(name)
         if isinstance(value, str) and len(value.splitlines())>1:
             # XXX this error message isn't quite right when we use splitlines
             # (see issue 22233), but I'm not sure what should happen here.

Lib/test/test_email/test_email.py

@@ -728,6 +728,33 @@ def test_nonascii_add_header_with_tspecial(self):
             "attachment; filename*=utf-8''Fu%C3%9Fballer%20%5Bfilename%5D.ppt",
             msg['Content-Disposition'])
 
+    def test_invalid_header_names(self):
+        invalid_headers = [
+            ('Invalid Header', 'contains space'),
+            ('Tab\tHeader', 'contains tab'),
+            ('Colon:Header', 'contains colon'),
+            ('', 'Empty name'),
+            (' LeadingSpace', 'starts with space'),
+            ('TrailingSpace ', 'ends with space'),
+            ('Header\x7F', 'Non-ASCII character'),
+            ('Header\x80', 'Extended ASCII'),
+        ]
+        for thispolicy in (email.policy.default, email.policy.compat32):
+            for method in ('__setitem__', 'add_header'):
+                for name, value in invalid_headers:
+                    with self.subTest(
+                            name=name,
+                            description=value,
+                            policy=thispolicy.__class__.__name__,
+                            method=method,
+                            ):
+                        with self.assertRaises(ValueError) as cm:
+                            getattr(Message(policy=thispolicy), method)(name, value)
+                        msg = str(cm.exception)
+                        self.assertRegex(msg,'(?i)(?=.*invalid)(?=.*header)(?=.*name)')
+                        self.assertIn(f"{name!r}", msg)
+
+
     def test_binary_quopri_payload(self):
         for charset in ('latin-1', 'ascii'):
             msg = Message()

Lib/test/test_email/test_message.py

@@ -1004,6 +1004,32 @@ def test_folding_with_long_nospace_http_policy_1(self):
         parsed_msg = message_from_bytes(m.as_bytes(), policy=policy.default)
         self.assertEqual(parsed_msg['Message-ID'], m['Message-ID'])
 
+    def test_invalid_header_names(self):
+        invalid_headers = [
+            ('Invalid Header', 'contains space'),
+            ('Tab\tHeader', 'contains tab'),
+            ('Colon:Header', 'contains colon'),
+            ('', 'Empty name'),
+            (' LeadingSpace', 'starts with space'),
+            ('TrailingSpace ', 'ends with space'),
+            ('Header\x7F', 'Non-ASCII character'),
+            ('Header\x80', 'Extended ASCII'),
+        ]
+        for thispolicy in (policy.default, policy.compat32):
+            for method in ('__setitem__', 'add_header'):
+                for name, value in invalid_headers:
+                    with self.subTest(
+                            name=name,
+                            description=value,
+                            policy=thispolicy.__class__.__name__,
+                            method=method,
+                            ):
+                        with self.assertRaises(ValueError) as cm:
+                            getattr(EmailMessage(policy=thispolicy), method)(name, value)
+                        msg = str(cm.exception)
+                        self.assertRegex( msg, '(?i)(?=.*invalid)(?=.*header)(?=.*name)')
+                        self.assertIn(f"{name!r}", msg)
+
     def test_get_body_malformed(self):
         """test for bpo-42892"""
         msg = textwrap.dedent("""\

Misc/NEWS.d/next/Library/2024-12-11-17-44-36.gh-issue-127794.VwmRsp.rst

@@ -0,0 +1,3 @@
+The :meth:`email.message.Message.add_header` method now validates header
+field names according to :rfc:`5322` when headers are added via add_header
+or __setitem__ and raises a ValueError if they contain invalid characters.