From c3f25057766a2e96ca66cf3993463c36a511e4a6 Mon Sep 17 00:00:00 2001
From: Seth Larson <seth@python.org>
Date: Fri, 10 Apr 2026 10:21:42 -0500
Subject: [PATCH] gh-146211: Reject CR/LF in HTTP tunnel request headers
 (GH-146212) (cherry picked from commit
 05ed7ce7ae9e17c23a04085b2539fe6d6d3cef69)

Co-authored-by: Seth Larson <seth@python.org>
Co-authored-by: Illia Volochii <illia.volochii@gmail.com>
---
 Lib/http/client.py                            | 11 ++++-
 Lib/test/test_httplib.py                      | 45 +++++++++++++++++++
 ...-03-20-09-29-42.gh-issue-146211.PQVbs7.rst |  2 +
 3 files changed, 57 insertions(+), 1 deletion(-)
 create mode 100644 Misc/NEWS.d/next/Security/2026-03-20-09-29-42.gh-issue-146211.PQVbs7.rst

diff --git a/Lib/http/client.py b/Lib/http/client.py
index dd5f4136e9eab5..c1ff4cef02f087 100644
--- a/Lib/http/client.py
+++ b/Lib/http/client.py
@@ -972,13 +972,22 @@ def _wrap_ipv6(self, ip):
         return ip
 
     def _tunnel(self):
+        if _contains_disallowed_url_pchar_re.search(self._tunnel_host):
+            raise ValueError('Tunnel host can\'t contain control characters %r'
+                             % (self._tunnel_host,))
         connect = b"CONNECT %s:%d %s\r\n" % (
             self._wrap_ipv6(self._tunnel_host.encode("idna")),
             self._tunnel_port,
             self._http_vsn_str.encode("ascii"))
         headers = [connect]
         for header, value in self._tunnel_headers.items():
-            headers.append(f"{header}: {value}\r\n".encode("latin-1"))
+            header_bytes = header.encode("latin-1")
+            value_bytes = value.encode("latin-1")
+            if not _is_legal_header_name(header_bytes):
+                raise ValueError('Invalid header name %r' % (header_bytes,))
+            if _is_illegal_header_value(value_bytes):
+                raise ValueError('Invalid header value %r' % (value_bytes,))
+            headers.append(b"%s: %s\r\n" % (header_bytes, value_bytes))
         headers.append(b"\r\n")
         # Making a single send() call instead of one per line encourages
         # the host OS to use a more optimal packet size instead of
diff --git a/Lib/test/test_httplib.py b/Lib/test/test_httplib.py
index 5267d2fe011766..75b350163ce384 100644
--- a/Lib/test/test_httplib.py
+++ b/Lib/test/test_httplib.py
@@ -370,6 +370,51 @@ def test_invalid_headers(self):
                 with self.assertRaisesRegex(ValueError, 'Invalid header'):
                     conn.putheader(name, value)
 
+    def test_invalid_tunnel_headers(self):
+        cases = (
+            ('Invalid\r\nName', 'ValidValue'),
+            ('Invalid\rName', 'ValidValue'),
+            ('Invalid\nName', 'ValidValue'),
+            ('\r\nInvalidName', 'ValidValue'),
+            ('\rInvalidName', 'ValidValue'),
+            ('\nInvalidName', 'ValidValue'),
+            (' InvalidName', 'ValidValue'),
+            ('\tInvalidName', 'ValidValue'),
+            ('Invalid:Name', 'ValidValue'),
+            (':InvalidName', 'ValidValue'),
+            ('ValidName', 'Invalid\r\nValue'),
+            ('ValidName', 'Invalid\rValue'),
+            ('ValidName', 'Invalid\nValue'),
+            ('ValidName', 'InvalidValue\r\n'),
+            ('ValidName', 'InvalidValue\r'),
+            ('ValidName', 'InvalidValue\n'),
+        )
+        for name, value in cases:
+            with self.subTest((name, value)):
+                conn = client.HTTPConnection('example.com')
+                conn.set_tunnel('tunnel', headers={
+                    name: value
+                })
+                conn.sock = FakeSocket('')
+                with self.assertRaisesRegex(ValueError, 'Invalid header'):
+                    conn._tunnel()  # Called in .connect()
+
+    def test_invalid_tunnel_host(self):
+        cases = (
+            'invalid\r.host',
+            '\ninvalid.host',
+            'invalid.host\r\n',
+            'invalid.host\x00',
+            'invalid host',
+        )
+        for tunnel_host in cases:
+            with self.subTest(tunnel_host):
+                conn = client.HTTPConnection('example.com')
+                conn.set_tunnel(tunnel_host)
+                conn.sock = FakeSocket('')
+                with self.assertRaisesRegex(ValueError, 'Tunnel host can\'t contain control characters'):
+                    conn._tunnel()  # Called in .connect()
+
     def test_headers_debuglevel(self):
         body = (
             b'HTTP/1.1 200 OK\r\n'
diff --git a/Misc/NEWS.d/next/Security/2026-03-20-09-29-42.gh-issue-146211.PQVbs7.rst b/Misc/NEWS.d/next/Security/2026-03-20-09-29-42.gh-issue-146211.PQVbs7.rst
new file mode 100644
index 00000000000000..4993633b8ebebb
--- /dev/null
+++ b/Misc/NEWS.d/next/Security/2026-03-20-09-29-42.gh-issue-146211.PQVbs7.rst
@@ -0,0 +1,2 @@
+Reject CR/LF characters in tunnel request headers for the
+HTTPConnection.set_tunnel() method.