Fix py2.6 for expect 100 continue

botocore/awsrequest.py

@@ -11,20 +11,183 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
-
 import logging
+import select
+import functools
+import inspect
 
 import six
 from botocore.vendored.requests import models
 from botocore.vendored.requests.sessions import REDIRECT_STATI
-
-from botocore.compat import HTTPHeaders, file_type
+from botocore.compat import HTTPHeaders, file_type, HTTPResponse
 from botocore.exceptions import UnseekableStreamError
+from botocore.vendored.requests.packages.urllib3.connection import VerifiedHTTPSConnection
+from botocore.vendored.requests.packages.urllib3.connection import HTTPConnection
+from botocore.vendored.requests.packages.urllib3.connectionpool import HTTPConnectionPool
+from botocore.vendored.requests.packages.urllib3.connectionpool import HTTPSConnectionPool
 
 
 logger = logging.getLogger(__name__)
 
 
+class AWSHTTPResponse(HTTPResponse):
+    # The *args, **kwargs is used because the args are slightly
+    # different in py2.6 than in py2.7/py3.
+    def __init__(self, *args, **kwargs):
+        self._status_tuple = kwargs.pop('status_tuple')
+        HTTPResponse.__init__(self, *args, **kwargs)
+
+    def _read_status(self):
+        if self._status_tuple is not None:
+            status_tuple = self._status_tuple
+            self._status_tuple = None
+            return status_tuple
+        else:
+            return HTTPResponse._read_status(self)
+
+
+class AWSHTTPConnection(HTTPConnection):
+    """HTTPConnection that supports Expect 100-continue.
+
+    This is conceptually a subclass of httplib.HTTPConnection (though
+    technically we subclass from urllib3, which subclasses
+    httplib.HTTPConnection) and we only override this class to support Expect
+    100-continue, which we need for S3.  As far as I can tell, this is
+    general purpose enough to not be specific to S3, but I'm being
+    tentative and keeping it in botocore because I've only tested
+    this against AWS services.
+
+    """
+    def __init__(self, *args, **kwargs):
+        HTTPConnection.__init__(self, *args, **kwargs)
+        self._original_response_cls = self.response_class
+        # We'd ideally hook into httplib's states, but they're all
+        # __mangled_vars so we use our own state var.  This variable is set
+        # when we receive an early response from the server.  If this value is
+        # set to True, any calls to send() are noops.  This value is reset to
+        # false every time _send_request is called.  This is to workaround the
+        # fact that py2.6 (and only py2.6) has a separate send() call for the
+        # body in _send_request, as opposed to endheaders(), which is where the
+        # body is sent in all versions > 2.6.
+        self._response_received = False
+
+    def _send_request(self, method, url, body, headers):
+        self._response_received = False
+        if headers.get('Expect', '') == '100-continue':
+            self._expect_header_set = True
+        else:
+            self._expect_header_set = False
+            self.response_class = self._original_response_cls
+        rval = HTTPConnection._send_request(
+            self, method, url, body, headers)
+        self._expect_header_set = False
+        return rval
+
+    def _send_output(self, message_body=None):
+        self._buffer.extend((b"", b""))
+        msg = b"\r\n".join(self._buffer)
+        del self._buffer[:]
+        # If msg and message_body are sent in a single send() call,
+        # it will avoid performance problems caused by the interaction
+        # between delayed ack and the Nagle algorithm.
+        if isinstance(message_body, bytes):
+            msg += message_body
+            message_body = None
+        self.send(msg)
+        if self._expect_header_set:
+            # This is our custom behavior.  If the Expect header was
+            # set, it will trigger this custom behavior.
+            logger.debug("Waiting for 100 Continue response.")
+            # Wait for 1 second for the server to send a response.
+            read, write, exc = select.select([self.sock], [], [self.sock], 1)
+            if read:
+                self._handle_expect_response(message_body)
+                return
+            else:
+                # From the RFC:
+                # Because of the presence of older implementations, the
+                # protocol allows ambiguous situations in which a client may
+                # send "Expect: 100-continue" without receiving either a 417
+                # (Expectation Failed) status or a 100 (Continue) status.
+                # Therefore, when a client sends this header field to an origin
+                # server (possibly via a proxy) from which it has never seen a
+                # 100 (Continue) status, the client SHOULD NOT wait for an
+                # indefinite period before sending the request body.
+                logger.debug("No response seen from server, continuing to "
+                             "send the response body.")
+        if message_body is not None:
+            # message_body was not a string (i.e. it is a file), and
+            # we must run the risk of Nagle.
+            self.send(message_body)
+
+    def _handle_expect_response(self, message_body):
+        # This is called when we sent the request headers containing
+        # an Expect: 100-continue header and received a response.
+        # We now need to figure out what to do.
+        fp = self.sock.makefile('rb', 0)
+        try:
+            maybe_status_line = fp.readline()
+            parts = maybe_status_line.split(None, 2)
+            if self._is_100_continue_status(maybe_status_line):
+                # Read an empty line as per the RFC.
+                fp.readline()
+                logger.debug("100 Continue response seen, now sending request body.")
+                self._send_message_body(message_body)
+            elif len(parts) == 3 and parts[0].startswith(b'HTTP/'):
+                # From the RFC:
+                # Requirements for HTTP/1.1 origin servers:
+                #
+                # - Upon receiving a request which includes an Expect
+                #   request-header field with the "100-continue"
+                #   expectation, an origin server MUST either respond with
+                #   100 (Continue) status and continue to read from the
+                #   input stream, or respond with a final status code.
+                #
+                # So if we don't get a 100 Continue response, then
+                # whatever the server has sent back is the final response
+                # and don't send the message_body.
+                logger.debug("Received a non 100 Continue response "
+                            "from the server, NOT sending request body.")
+                status_tuple = (parts[0].decode('ascii'),
+                                int(parts[1]), parts[2].decode('ascii'))
+                response_class = functools.partial(
+                    AWSHTTPResponse, status_tuple=status_tuple)
+                self.response_class = response_class
+                self._response_received = True
+        finally:
+            fp.close()
+
+    def _send_message_body(self, message_body):
+        if message_body is not None:
+            self.send(message_body)
+
+    def send(self, str):
+        if self._response_received:
+            logger.debug("send() called, but reseponse already received. "
+                         "Not sending data.")
+            return
+        return HTTPConnection.send(self, str)
+
+    def _is_100_continue_status(self, maybe_status_line):
+        parts = maybe_status_line.split(None, 2)
+        # Check for HTTP/<version> 100 Continue\r\n
+        return (
+            len(parts) == 3 and parts[0].startswith(b'HTTP/') and
+            parts[1] == b'100' and parts[2].startswith(b'Continue'))
+
+
+class AWSHTTPSConnection(VerifiedHTTPSConnection):
+    pass
+
+
+# Now we need to set the methods we overrode from AWSHTTPConnection
+# onto AWSHTTPSConnection.  This is just a shortcut to avoid
+# copy/pasting the same code into AWSHTTPSConnection.
+for name, function in AWSHTTPConnection.__dict__.items():
+    if inspect.isfunction(function):
+        setattr(AWSHTTPSConnection, name, function)
+
+
 class AWSRequest(models.RequestEncodingMixin, models.Request):
     def __init__(self, *args, **kwargs):
         self.auth_path = None
@@ -104,3 +267,7 @@ def reset_stream(self):
         except Exception as e:
             logger.debug("Unable to rewind stream: %s", e)
             raise UnseekableStreamError(stream_object=self.body)
+
+
+HTTPSConnectionPool.ConnectionCls = AWSHTTPSConnection
+HTTPConnectionPool.ConnectionCls = AWSHTTPConnection

botocore/compat.py

@@ -31,6 +31,7 @@ class HTTPHeaders(http_client.HTTPMessage):
     from urllib.parse import parse_qsl
     from urllib.parse import parse_qs
     from urllib.parse import urlencode
+    from http.client import HTTPResponse
     from io import IOBase as _IOBase
     file_type = _IOBase
     zip = zip
@@ -67,6 +68,7 @@ def accepts_kwargs(func):
     from email.message import Message
     file_type = file
     from itertools import izip as zip
+    from httplib import HTTPResponse
 
     class HTTPHeaders(Message):
 

botocore/endpoint.py

@@ -129,6 +129,8 @@ def _get_response(self, request, operation, attempts):
                 stream=operation.is_streaming(),
                 proxies=self.proxies)
         except Exception as e:
+            logger.debug("Exception received when sending HTTP request.",
+                         exc_info=True)
             return (None, e)
         # This returns the http_response and the parsed_data.
         return (botocore.response.get_response(self.session, operation,

botocore/handlers.py

@@ -25,6 +25,7 @@
 
 from botocore.compat import urlsplit, urlunsplit, unquote, json, quote
 from botocore import retryhandler
+from botocore.payload import Payload
 import botocore.auth
 
 
@@ -239,6 +240,18 @@ def signature_overrides(service_data, service_name, session, **kwargs):
         service_data['signature_version'] = signature_version_override
 
 
+def add_expect_header(operation, params, **kwargs):
+    if operation.http.get('method', '') not in ['PUT', 'POST']:
+        return
+    if params['payload'].__class__ == Payload:
+        payload = params['payload'].getvalue()
+        if hasattr(payload, 'read'):
+            # Any file like object will use an expect 100-continue
+            # header regardless of size.
+            logger.debug("Adding expect 100 continue header to request.")
+            params['headers']['Expect'] = '100-continue'
+
+
 def quote_source_header(params, **kwargs):
     if params['headers'] and 'x-amz-copy-source' in params['headers']:
         value = params['headers']['x-amz-copy-source']
@@ -285,6 +298,7 @@ def copy_snapshot_encrypted(operation, params, **kwargs):
     ('before-call.s3.DeleteObjects', calculate_md5),
     ('before-call.s3.UploadPartCopy', quote_source_header),
     ('before-call.s3.CopyObject', quote_source_header),
+    ('before-call.s3', add_expect_header),
     ('before-call.ec2.CopySnapshot', copy_snapshot_encrypted),
     ('before-auth.s3', fix_s3_host),
     ('needs-retry.s3.UploadPartCopy', check_for_200_error),

tests/integration/test_s3.py

@@ -14,7 +14,7 @@
 import os
 import time
 import random
-from tests import unittest
+from tests import unittest, temporary_file
 from collections import defaultdict
 import tempfile
 import shutil
@@ -465,5 +465,56 @@ def test_presign_does_not_change_host(self):
             "got: %s" % presigned_url)
 
 
+class TestCreateBucketInOtherRegion(BaseS3Test):
+    def setUp(self):
+        super(TestCreateBucketInOtherRegion, self).setUp()
+        self.bucket_name = 'botocoretest%s-%s' % (
+            int(time.time()), random.randint(1, 1000))
+        self.bucket_location = 'us-west-2'
+
+        operation = self.service.get_operation('CreateBucket')
+        response = operation.call(self.endpoint, bucket=self.bucket_name,
+            create_bucket_configuration={'LocationConstraint': self.bucket_location})
+        self.assertEqual(response[0].status_code, 200)
+        self.keys = []
+
+    def tearDown(self):
+        for key in self.keys:
+            op = self.service.get_operation('DeleteObject')
+            response = op.call(self.endpoint, bucket=self.bucket_name, key=key)
+            self.assertEqual(response[0].status_code, 204)
+        self.delete_bucket(self.bucket_name)
+
+    def test_bucket_in_other_region(self):
+        # This verifies expect 100-continue behavior.  We previously
+        # had a bug where we did not support this behavior and trying to
+        # create a bucket and immediately PutObject with a file like object
+        # would actually cause errors.
+        with temporary_file('w') as f:
+            f.write('foobarbaz' * 1024 * 1024)
+            f.flush()
+            op = self.service.get_operation('PutObject')
+            response = op.call(self.endpoint,
+                               bucket=self.bucket_name,
+                               key='foo.txt',
+                               body=open(f.name, 'rb'))
+            self.assertEqual(response[0].status_code, 200)
+            self.keys.append('foo.txt')
+
+    def test_bucket_in_other_region_using_http(self):
+        http_endpoint = self.service.get_endpoint(
+            endpoint_url='http://s3.amazonaws.com/')
+        with temporary_file('w') as f:
+            f.write('foobarbaz' * 1024 * 1024)
+            f.flush()
+            op = self.service.get_operation('PutObject')
+            response = op.call(http_endpoint,
+                               bucket=self.bucket_name,
+                               key='foo.txt',
+                               body=open(f.name, 'rb'))
+            self.assertEqual(response[0].status_code, 200)
+            self.keys.append('foo.txt')
+
+
 if __name__ == '__main__':
     unittest.main()