proxy: Don't trust Content-Length for chunked transfers

Previously we'd
- complain that a client disconnected even though they finished their
  chunked transfer just fine, and
- on EC, send a X-Backend-Obj-Content-Length for pre-allocation even
  though Content-Length doesn't determine request body size.

Change-Id: Ia80e595f713695cbb41dab575963f2cb9bebfa09
Related-Bug: 1840507

3 files changed, 159 insertions, 18 deletions

diff --git a/swift/proxy/controllers/obj.py b/swift/proxy/controllers/obj.py
index b79aafeef..8508e2a0b 100644
--- a/swift/proxy/controllers/obj.py
+++ b/swift/proxy/controllers/obj.py
@@ -927,8 +927,8 @@ class ReplicatedObjectController(BaseObjectController):
 
                     send_chunk(chunk)
 
-                if req.content_length and (
-                        bytes_transferred < req.content_length):
+                ml = req.message_length()
+                if ml and bytes_transferred < ml:
                     req.client_disconnect = True
                     self.app.logger.warning(
                         _('Client disconnected without sending enough data'))
@@ -2638,8 +2638,8 @@ class ECObjectController(BaseObjectController):
 
                     send_chunk(chunk)
 
-                if req.content_length and (
-                        bytes_transferred < req.content_length):
+                ml = req.message_length()
+                if ml and bytes_transferred < ml:
                     req.client_disconnect = True
                     self.app.logger.warning(
                         _('Client disconnected without sending enough data'))
@@ -2787,7 +2787,8 @@ class ECObjectController(BaseObjectController):
         policy = POLICIES.get_by_index(policy_index)
 
         expected_frag_size = None
-        if req.content_length:
+        ml = req.message_length()
+        if ml:
             # TODO: PyECLib <= 1.2.0 looks to return the segment info
             # different from the input for aligned data efficiency but
             # Swift never does. So calculate the fragment length Swift
@@ -2797,12 +2798,12 @@ class ECObjectController(BaseObjectController):
             # and the next call is to get info for the last segment
 
             # get number of fragments except the tail - use truncation //
-            num_fragments = req.content_length // policy.ec_segment_size
+            num_fragments = ml // policy.ec_segment_size
             expected_frag_size = policy.fragment_size * num_fragments
 
             # calculate the tail fragment_size by hand and add it to
             # expected_frag_size
-            last_segment_size = req.content_length % policy.ec_segment_size
+            last_segment_size = ml % policy.ec_segment_size
             if last_segment_size:
                 last_info = policy.pyeclib_driver.get_segment_info(
                     last_segment_size, policy.ec_segment_size)
diff --git a/test/unit/proxy/controllers/test_obj.py b/test/unit/proxy/controllers/test_obj.py
index 3a73b7b98..dadfd96fc 100644
--- a/test/unit/proxy/controllers/test_obj.py
+++ b/test/unit/proxy/controllers/test_obj.py
@@ -4677,6 +4677,15 @@ class TestECObjControllerMimePutter(BaseObjectControllerMixin,
         self.assertEqual(resp.status_int, 201)
 
     def test_PUT_with_body(self):
+        self._test_PUT_with_body()
+
+    def test_PUT_with_chunked_body(self):
+        self._test_PUT_with_body(chunked=True, content_length=False)
+
+    def test_PUT_with_both_body(self):
+        self._test_PUT_with_body(chunked=True, content_length=True)
+
+    def _test_PUT_with_body(self, chunked=False, content_length=True):
         segment_size = self.policy.ec_segment_size
         test_body = (b'asdf' * segment_size)[:-10]
         # make the footers callback not include Etag footer so that we can
@@ -4689,6 +4698,10 @@ class TestECObjControllerMimePutter(BaseObjectControllerMixin,
         etag = md5(test_body).hexdigest()
         size = len(test_body)
         req.body = test_body
+        if chunked:
+            req.headers['Transfer-Encoding'] = 'chunked'
+        if not content_length:
+            del req.headers['Content-Length']
         codes = [201] * self.replicas()
         resp_headers = {
             'Some-Other-Header': 'Four',
@@ -4705,8 +4718,8 @@ class TestECObjControllerMimePutter(BaseObjectControllerMixin,
             conn_id = kwargs['connection_id']
             put_requests[conn_id]['boundary'] = headers[
                 'X-Backend-Obj-Multipart-Mime-Boundary']
-            put_requests[conn_id]['backend-content-length'] = headers[
-                'X-Backend-Obj-Content-Length']
+            put_requests[conn_id]['backend-content-length'] = headers.get(
+                'X-Backend-Obj-Content-Length')
             put_requests[conn_id]['x-timestamp'] = headers[
                 'X-Timestamp']
 
@@ -4734,9 +4747,6 @@ class TestECObjControllerMimePutter(BaseObjectControllerMixin,
             self.assertIsNotNone(info['boundary'],
                                  "didn't get boundary for conn %r" % (
                                      connection_id,))
-            self.assertTrue(size > int(info['backend-content-length']) > 0,
-                            "invalid backend-content-length for conn %r" % (
-                                connection_id,))
 
             # email.parser.FeedParser doesn't know how to take a multipart
             # message and boundary together and parse it; it only knows how
@@ -4759,12 +4769,19 @@ class TestECObjControllerMimePutter(BaseObjectControllerMixin,
             obj_payload = obj_part.get_payload(decode=True)
             frag_archives.append(obj_payload)
 
-            # assert length was correct for this connection
-            self.assertEqual(int(info['backend-content-length']),
-                             len(frag_archives[-1]))
-            # assert length was the same for all connections
-            self.assertEqual(int(info['backend-content-length']),
-                             len(frag_archives[0]))
+            if chunked:
+                self.assertIsNone(info['backend-content-length'])
+            else:
+                self.assertTrue(
+                    size > int(info['backend-content-length']) > 0,
+                    "invalid backend-content-length for conn %r" % (
+                        connection_id,))
+                # assert length was correct for this connection
+                self.assertEqual(int(info['backend-content-length']),
+                                 len(frag_archives[-1]))
+                # assert length was the same for all connections
+                self.assertEqual(int(info['backend-content-length']),
+                                 len(frag_archives[0]))
 
             # validate some footer metadata
             self.assertEqual(footer_part['X-Document'], 'object metadata')
diff --git a/test/unit/proxy/test_server.py b/test/unit/proxy/test_server.py
index bcef252b7..cfa24344f 100644
--- a/test/unit/proxy/test_server.py
+++ b/test/unit/proxy/test_server.py
@@ -4317,6 +4317,31 @@ class TestReplicatedObjectController(
             resp = req.get_response(self.app)
             self.assertEqual(resp.status_int, 499)
 
+            # chunked transfers basically go "until I stop sending bytes"
+            req = Request.blank('/v1/a/c/o',
+                                environ={'REQUEST_METHOD': 'PUT',
+                                         'wsgi.input': DisconnectedBody()},
+                                headers={'Transfer-Encoding': 'chunked',
+                                         'Content-Type': 'text/plain'})
+            self.app.update_request(req)
+            set_http_connect(200, 200, 201, 201, 201)
+            #                acct cont obj  obj  obj
+            resp = req.get_response(self.app)
+            self.assertEqual(resp.status_int, 201)  # ... so, no disconnect
+
+            # chunked transfer trumps content-length
+            req = Request.blank('/v1/a/c/o',
+                                environ={'REQUEST_METHOD': 'PUT',
+                                         'wsgi.input': DisconnectedBody()},
+                                headers={'Content-Length': '4',
+                                         'Transfer-Encoding': 'chunked',
+                                         'Content-Type': 'text/plain'})
+            self.app.update_request(req)
+            set_http_connect(200, 200, 201, 201, 201)
+            #                acct cont obj  obj  obj
+            resp = req.get_response(self.app)
+            self.assertEqual(resp.status_int, 201)
+
     def test_node_read_timeout(self):
         with save_globals():
             self.app.account_ring.get_nodes('account')
@@ -7223,6 +7248,104 @@ class BaseTestECObjectController(BaseTestObjectController):
         errors = _test_servers[0].logger.get_lines_for_level('error')
         self.assertEqual([], errors)
 
+        # try it chunked
+        _test_servers[0].logger.clear()
+        chunk = 'a' * 64 * 2 ** 10
+        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
+        fd = sock.makefile('rwb')
+        fd.write(('PUT /v1/a/%s-discon/test HTTP/1.1\r\n'
+                  'Host: localhost\r\n'
+                  'Transfer-Encoding: chunked\r\n'
+                  'X-Storage-Token: t\r\n'
+                  'Content-Type: donuts\r\n'
+                  '\r\n' % (self.ec_policy.name,)).encode('ascii'))
+        fd.write(('%x\r\n%s\r\n' % (len(chunk), chunk)).encode('ascii'))
+        # no zero-byte end chunk
+        fd.flush()
+        fd.close()
+        sock.close()
+        # sleep to trampoline enough
+        condition = \
+            lambda: _test_servers[0].logger.get_lines_for_level('warning')
+        self._sleep_enough(condition)
+        expected = ['Client disconnected without sending last chunk']
+        warns = _test_servers[0].logger.get_lines_for_level('warning')
+        self.assertEqual(expected, warns)
+        errors = _test_servers[0].logger.get_lines_for_level('error')
+        self.assertEqual([], errors)
+
+        _test_servers[0].logger.clear()
+        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
+        fd = sock.makefile('rwb')
+        fd.write(('PUT /v1/a/%s-discon/test HTTP/1.1\r\n'
+                  'Host: localhost\r\n'
+                  'Transfer-Encoding: chunked\r\n'
+                  'X-Storage-Token: t\r\n'
+                  'Content-Type: donuts\r\n'
+                  '\r\n' % (self.ec_policy.name,)).encode('ascii'))
+        fd.write(('%x\r\n%s\r\n' % (len(chunk), chunk)).encode('ascii')[:-10])
+        fd.flush()
+        fd.close()
+        sock.close()
+        # sleep to trampoline enough
+        condition = \
+            lambda: _test_servers[0].logger.get_lines_for_level('warning')
+        self._sleep_enough(condition)
+        expected = ['Client disconnected without sending last chunk']
+        warns = _test_servers[0].logger.get_lines_for_level('warning')
+        self.assertEqual(expected, warns)
+        errors = _test_servers[0].logger.get_lines_for_level('error')
+        self.assertEqual([], errors)
+
+        _test_servers[0].logger.clear()
+        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
+        fd = sock.makefile('rwb')
+        fd.write(('PUT /v1/a/%s-discon/test HTTP/1.1\r\n'
+                  'Host: localhost\r\n'
+                  'Transfer-Encoding: chunked\r\n'
+                  'X-Storage-Token: t\r\n'
+                  'Content-Type: donuts\r\n'
+                  '\r\n' % (self.ec_policy.name,)).encode('ascii'))
+        fd.write(('%x\r\n' % len(chunk)).encode('ascii'))
+        fd.flush()
+        fd.close()
+        sock.close()
+        # sleep to trampoline enough
+        condition = \
+            lambda: _test_servers[0].logger.get_lines_for_level('warning')
+        self._sleep_enough(condition)
+        expected = ['Client disconnected without sending last chunk']
+        warns = _test_servers[0].logger.get_lines_for_level('warning')
+        self.assertEqual(expected, warns)
+        errors = _test_servers[0].logger.get_lines_for_level('error')
+        self.assertEqual([], errors)
+
+        # Do a valid guy with conflicting headers
+        _test_servers[0].logger.clear()
+        chunk = 'a' * 64 * 2 ** 10
+        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
+        fd = sock.makefile('rwb')
+        fd.write(('PUT /v1/a/%s-discon/test HTTP/1.1\r\n'
+                  'Host: localhost\r\n'
+                  'Transfer-Encoding: chunked\r\n'
+                  'Content-Length: 999999999999999999999999\r\n'
+                  'X-Storage-Token: t\r\n'
+                  'Content-Type: donuts\r\n'
+                  '\r\n' % (self.ec_policy.name,)).encode('ascii'))
+        fd.write(('%x\r\n%s\r\n0\r\n\r\n' % (
+            len(chunk), chunk)).encode('ascii'))
+        # no zero-byte end chunk
+        fd.flush()
+        headers = readuntil2crlfs(fd)
+        exp = b'HTTP/1.1 201'
+        self.assertEqual(headers[:len(exp)], exp)
+        fd.close()
+        sock.close()
+        warns = _test_servers[0].logger.get_lines_for_level('warning')
+        self.assertEqual([], warns)
+        errors = _test_servers[0].logger.get_lines_for_level('error')
+        self.assertEqual([], errors)
+
 
 class TestECObjectController(BaseTestECObjectController, unittest.TestCase):
     def setUp(self):