CVE-2016-9964

@@ -1413,21 +1413,21 @@ def _hval(value):
 
 
 class HeaderProperty(object):
-    def __init__(self, name, reader=None, writer=str, default=''):
+    def __init__(self, name, reader=None, writer=None, default=''):
         self.name, self.default = name, default
         self.reader, self.writer = reader, writer
         self.__doc__ = 'Current value of the %r header.' % name.title()
 
     def __get__(self, obj, cls):
         if obj is None: return self
-        value = obj.headers.get(self.name, self.default)
+        value = obj.get_header(self.name, self.default)
         return self.reader(value) if self.reader else value
 
     def __set__(self, obj, value):
-        obj.headers[self.name] = self.writer(value)
+        obj[self.name] = self.writer(value) if self.writer else value
 
     def __delete__(self, obj):
-        del obj.headers[self.name]
+        del obj[self.name]
 
 
 class BaseResponse(object):
@@ -1534,7 +1534,7 @@ def headers(self):
     def __contains__(self, name): return _hkey(name) in self._headers
     def __delitem__(self, name):  del self._headers[_hkey(name)]
     def __getitem__(self, name):  return self._headers[_hkey(name)][-1]
-    def __setitem__(self, name, value): self._headers[_hkey(name)] = [str(value)]
+    def __setitem__(self, name, value): self._headers[_hkey(name)] = [_hval(value)]
 
     def get_header(self, name, default=None):
         ''' Return the value of a previously defined header. If there is no
@@ -1544,11 +1544,11 @@ def get_header(self, name, default=None):
     def set_header(self, name, value):
         ''' Create a new response header, replacing any previously defined
             headers with the same name. '''
-        self._headers[_hkey(name)] = [str(value)]
+        self._headers[_hkey(name)] = [_hval(value)]
 
     def add_header(self, name, value):
         ''' Add an additional response header, not removing duplicates. '''
-        self._headers.setdefault(_hkey(name), []).append(str(value))
+        self._headers.setdefault(_hkey(name), []).append(_hval(value))
 
     def iter_headers(self):
         ''' Yield (header, value) tuples, skipping headers that are not

@@ -669,16 +669,27 @@ def test_non_string_header(self):
         self.assertEqual('None', response['x-test'])
 
     def test_prevent_control_characters_in_headers(self):
-        apis = 'append', 'replace', '__setitem__', 'setdefault'
         masks = '{}test', 'test{}', 'te{}st'
         tests = '\n', '\r', '\n\r', '\0'
+
+        # Test HeaderDict
+        apis = 'append', 'replace', '__setitem__', 'setdefault'
         for api, mask, test in product(apis, masks, tests):
             hd = bottle.HeaderDict()
             func = getattr(hd, api)
             value = mask.replace("{}", test)
             self.assertRaises(ValueError, func, value, "test-value")
             self.assertRaises(ValueError, func, "test-name", value)
 
+        # Test functions on BaseResponse
+        apis = 'add_header', 'set_header', '__setitem__'
+        for api, mask, test in product(apis, masks, tests):
+            rs = bottle.BaseResponse()
+            func = getattr(rs, api)
+            value = mask.replace("{}", test)
+            self.assertRaises(ValueError, func, value, "test-value")
+            self.assertRaises(ValueError, func, "test-name", value)
+
     def test_expires_header(self):
         import datetime
         response = BaseResponse()

@@ -1573,9 +1573,16 @@ def __delattr__(self, name, value):
             raise AttributeError("Attribute not defined: %s" % name)
 
 
-def _hkey(s):
-    return s.title().replace('_', '-')
-
+def _hkey(key):
+    if '\n' in key or '\r' in key or '\0' in key:
+        raise ValueError("Header names must not contain control characters: %r" % key)
+    return key.title().replace('_', '-')
+
+def _hval(value):
+    value = value if isinstance(value, unicode) else str(value)
+    if '\n' in value or '\r' in value or '\0' in value:
+        raise ValueError("Header value must not contain control characters: %r" % value)
+    return value
 
 class HeaderProperty(object):
     def __init__(self, name, reader=None, writer=str, default=''):
@@ -2170,7 +2177,6 @@ def __getattr__(self, name, default=unicode()):
             return super(FormsDict, self).__getattr__(name)
         return self.getunicode(name, default=default)
 
-
 class HeaderDict(MultiDict):
     """ A case-insensitive version of :class:`MultiDict` that defaults to
         replace the old value instead of appending it. """
@@ -2189,16 +2195,13 @@ def __getitem__(self, key):
         return self.dict[_hkey(key)][-1]
 
     def __setitem__(self, key, value):
-        self.dict[_hkey(key)] = [value if isinstance(value, unicode) else
-                                 str(value)]
+        self.dict[_hkey(key)] = [_hval(value)]
 
     def append(self, key, value):
-        self.dict.setdefault(_hkey(key), []).append(
-            value if isinstance(value, unicode) else str(value))
+        self.dict.setdefault(_hkey(key), []).append(_hval(value))
 
     def replace(self, key, value):
-        self.dict[_hkey(key)] = [value if isinstance(value, unicode) else
-                                 str(value)]
+        self.dict[_hkey(key)] = [_hval(value)]
 
     def getall(self, key):
         return self.dict.get(_hkey(key)) or []
@@ -2207,7 +2210,7 @@ def get(self, key, default=None, index=-1):
         return MultiDict.get(self, _hkey(key), default, index)
 
     def filter(self, names):
-        for name in [_hkey(n) for n in names]:
+        for name in (_hkey(n) for n in names):
             if name in self.dict:
                 del self.dict[name]
 

@@ -3,6 +3,9 @@
 
 import unittest
 import sys
+
+import itertools
+
 import bottle
 from bottle import request, tob, touni, tonat, json_dumps, HTTPError, parse_date
 from test import tools
@@ -695,6 +698,17 @@ def test_non_string_header(self):
         response['x-test'] = None
         self.assertEqual('None', response['x-test'])
 
+    def test_prevent_control_characters_in_headers(self):
+        apis = 'append', 'replace', '__setitem__', 'setdefault'
+        masks = '{}test', 'test{}', 'te{}st'
+        tests = '\n', '\r', '\n\r', '\0'
+        for api, mask, test in itertools.product(apis, masks, tests):
+            hd = bottle.HeaderDict()
+            func = getattr(hd, api)
+            value = mask.replace("{}", test)
+            self.assertRaises(ValueError, func, value, "test-value")
+            self.assertRaises(ValueError, func, "test-name", value)
+
     def test_expires_header(self):
         import datetime
         response = BaseResponse()