Handle empty request body with chunked encoding

* Identify and handle requests with empty body
  and 'transfer-encoding' request header
  set to value 'chunked'

Change-Id: I54462dad76bdbf7c069ae3031237d98702a77e40
Closes-Bug: #1249501
This commit is contained in:
Rohan Kanade 2014-06-17 22:23:03 +02:00
parent 18c1efa1be
commit 8dad79d2f9
3 changed files with 33 additions and 5 deletions

View File

@ -571,13 +571,18 @@ class Request(webob.Request):
class JSONRequestDeserializer(object):
valid_transfer_encoding = frozenset(['chunked', 'compress', 'deflate',
'gzip', 'identity'])
def has_body(self, request):
"""
Returns whether a Webob.Request object will possess an entity body.
:param request: Webob.Request object
"""
if 'transfer-encoding' in request.headers:
request_encoding = request.headers.get('transfer-encoding', '').lower()
is_valid_encoding = request_encoding in self.valid_transfer_encoding
if is_valid_encoding and request.is_body_readable:
return True
elif request.content_length > 0:
return True

View File

@ -344,7 +344,7 @@ class TestRPCJSONDeserializer(test_utils.BaseTestCase):
request = wsgi.Request.blank('/')
request.method = 'POST'
request.body = 'fake_body'
request.headers['transfer-encoding'] = 0
request.headers['transfer-encoding'] = ''
self.assertIn('transfer-encoding', request.headers)
self.assertTrue(rpc.RPCJSONDeserializer().has_body(request))

View File

@ -447,12 +447,35 @@ class JSONRequestDeserializerTest(test_utils.BaseTestCase):
self.assertEqual(expected, actual)
def test_has_body_has_transfer_encoding(self):
self.assertTrue(self._check_transfer_encoding(
transfer_encoding='chunked'))
def test_has_body_multiple_transfer_encoding(self):
self.assertTrue(self._check_transfer_encoding(
transfer_encoding='chunked, gzip'))
def test_has_body_invalid_transfer_encoding(self):
self.assertFalse(self._check_transfer_encoding(
transfer_encoding='invalid', content_length=0))
def test_has_body_invalid_transfer_encoding_with_content_length(self):
self.assertTrue(self._check_transfer_encoding(
transfer_encoding='invalid', content_length=5))
def test_has_body_valid_transfer_encoding_with_content_length(self):
self.assertTrue(self._check_transfer_encoding(
transfer_encoding='chunked', content_length=0))
def _check_transfer_encoding(self, transfer_encoding=None,
content_length=None):
request = wsgi.Request.blank('/')
request.method = 'POST'
request.body = 'fake_body'
request.headers['transfer-encoding'] = 0
self.assertIn('transfer-encoding', request.headers)
self.assertTrue(wsgi.JSONRequestDeserializer().has_body(request))
request.headers['transfer-encoding'] = transfer_encoding
if content_length is not None:
request.headers['content-length'] = content_length
return wsgi.JSONRequestDeserializer().has_body(request)
def test_get_bind_addr_default_value(self):
expected = ('0.0.0.0', '123456')