bpo-11671: add header validation from http.client to wsgiref.headers.Headers

Lib/test/test_wsgiref.py

@@ -525,6 +525,40 @@ def testExtras(self):
  '\r\n'
  )
 
+ validation_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'),
+ (b'InvalidName', 'ValidValue', AssertionError),
+ ('ValidName', b'InvalidValue', AssertionError)
+ )
+
+ def test_add_header_validation(self):
+ h = Headers([])
+ for name, value, *exception in self.validation_cases:
+ with self.subTest((name, value, exception)):
+ with self.assertRaises(exception[0]) if len(exception) else self.assertRaisesRegex(ValueError, 'Invalid header'):
+ h.add_header(name, value)
+
+ def test_initialize_validation(self):
+ for name, value, *exception in self.validation_cases:
+ with self.subTest((name, value, exception)):
+ with self.assertRaises(exception[0]) if len(exception) else self.assertRaisesRegex(ValueError, 'Invalid header'):
+ Headers([(name, value)])
+
 class ErrorHandler(BaseCGIHandler):
  """Simple handler subclass for testing BaseHandler"""
 

Lib/wsgiref/headers.py

@@ -10,6 +10,8 @@
 import re
 tspecials = re.compile(r'[ \(\)<>@,;:\\"/\[\]\?=]')
 
+from http.client import _is_legal_header_name, _is_illegal_header_value
+
 def _formatparam(param, value=None, quote=1):
  """Convenience function to format and return a key=value pair.
 
@@ -32,7 +34,9 @@ def __init__(self, headers=None):
  headers = headers if headers is not None else []
  if type(headers) is not list:
  raise TypeError("Headers must be a list of name/value tuples")
- self._headers = headers
+ self._headers = []
+ for header, value in headers:
+ self.add_header(header, value)
  if __debug__:
  for k, v in headers:
  self._convert_string_type(k)
@@ -52,8 +56,7 @@ def __len__(self):
  def __setitem__(self, name, val):
  """Set the value of a header."""
  del self[name]
- self._headers.append(
- (self._convert_string_type(name), self._convert_string_type(val)))
+ self.add_header(name, val)
 
  def __delitem__(self,name):
  """Delete all occurrences of a header, if present.
@@ -148,8 +151,7 @@ def setdefault(self,name,value):
  and value 'value'."""
  result = self.get(name)
  if result is None:
- self._headers.append((self._convert_string_type(name),
- self._convert_string_type(value)))
+ self.add_header(name, value)
  return value
  else:
  return result
@@ -181,4 +183,10 @@ def add_header(self, _name, _value, **_params):
  else:
  v = self._convert_string_type(v)
  parts.append(_formatparam(k.replace('_', '-'), v))
- self._headers.append((self._convert_string_type(_name), "; ".join(parts)))
+ header = self._convert_string_type(_name)
+ value = "; ".join(parts)
+ if not _is_legal_header_name(header.encode('ascii')):
+ raise ValueError('Invalid header name %r' % (header,))
+ if _is_illegal_header_value(value.encode('ascii')):
+ raise ValueError('Invalid header value %r' % (value,))
+ self._headers.append((header, value))

Misc/NEWS.d/next/Security/2019-08-27-01-08-25.bpo-11671.wfJY-D.rst

@@ -0,0 +1 @@
+Add validation in wsgiref.headers.Headers to disallow invalid characters in header names and values. Patch by Ashwin Ramaswami, initial patch by Devin Cook.