diff --git a/Lib/http/client.py b/Lib/http/client.py
index 77f8d2629..6fb7d254e 100644
--- a/Lib/http/client.py
+++ b/Lib/http/client.py
@@ -972,13 +972,22 @@ class HTTPConnection:
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/http/cookies.py b/Lib/http/cookies.py
index d5b8ba939..5c5b14788 100644
--- a/Lib/http/cookies.py
+++ b/Lib/http/cookies.py
@@ -391,17 +391,21 @@ class Morsel(dict):
return '<%s: %s>' % (self.__class__.__name__, self.OutputString())
def js_output(self, attrs=None):
+ import base64
# Print javascript
output_string = self.OutputString(attrs)
if _has_control_character(output_string):
raise CookieError("Control characters are not allowed in cookies")
+ # Base64-encode value to avoid template
+ # injection in cookie values.
+ output_encoded = base64.b64encode(output_string.encode('utf-8')).decode("ascii")
return """
- """ % (output_string.replace('"', r'\"'))
+ """ % (output_encoded,)
def OutputString(self, attrs=None):
# Build up our result
diff --git a/Lib/test/test_http_cookies.py b/Lib/test/test_http_cookies.py
index b4685f6e7..ab01d68d1 100644
--- a/Lib/test/test_http_cookies.py
+++ b/Lib/test/test_http_cookies.py
@@ -1,5 +1,5 @@
# Simple test suite for http/cookies.py
-
+import base64
import copy
import unittest
import doctest
@@ -153,17 +153,19 @@ class CookieTests(unittest.TestCase):
self.assertEqual(C.output(['path']),
'Set-Cookie: Customer="WILE_E_COYOTE"; Path=/acme')
- self.assertEqual(C.js_output(), r"""
+ cookie_encoded = base64.b64encode(b'Customer="WILE_E_COYOTE"; Path=/acme; Version=1').decode('ascii')
+ self.assertEqual(C.js_output(), fr"""
""")
- self.assertEqual(C.js_output(['path']), r"""
+ cookie_encoded = base64.b64encode(b'Customer="WILE_E_COYOTE"; Path=/acme').decode('ascii')
+ self.assertEqual(C.js_output(['path']), fr"""
""")
@@ -268,17 +270,19 @@ class CookieTests(unittest.TestCase):
self.assertEqual(C.output(['path']),
'Set-Cookie: Customer="WILE_E_COYOTE"; Path=/acme')
- self.assertEqual(C.js_output(), r"""
+ expected_encoded_cookie = base64.b64encode(b'Customer=\"WILE_E_COYOTE\"; Path=/acme; Version=1').decode('ascii')
+ self.assertEqual(C.js_output(), fr"""
""")
- self.assertEqual(C.js_output(['path']), r"""
+ expected_encoded_cookie = base64.b64encode(b'Customer=\"WILE_E_COYOTE\"; Path=/acme').decode('ascii')
+ self.assertEqual(C.js_output(['path']), fr"""
""")
@@ -369,13 +373,16 @@ class MorselTests(unittest.TestCase):
self.assertEqual(
M.output(),
"Set-Cookie: %s=%s; Path=/foo" % (i, "%s_coded_val" % i))
+ expected_encoded_cookie = base64.b64encode(
+ ("%s=%s; Path=/foo" % (i, "%s_coded_val" % i)).encode("ascii")
+ ).decode('ascii')
expected_js_output = """
- """ % (i, "%s_coded_val" % i)
+ """ % (expected_encoded_cookie,)
self.assertEqual(M.js_output(), expected_js_output)
for i in ["foo bar", "foo@bar"]:
# Try some illegal characters
diff --git a/Lib/test/test_httplib.py b/Lib/test/test_httplib.py
index 8ce5f853b..f7ab3e576 100644
--- a/Lib/test/test_httplib.py
+++ b/Lib/test/test_httplib.py
@@ -369,6 +369,51 @@ class HeaderTests(TestCase):
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'