Add event handler to quote the x-amz-copy-source header value as expected by S3.

botocore/handlers.py

@@ -31,7 +31,7 @@
 
 import six
 
-from botocore.compat import urlsplit, urlunsplit, unquote, json
+from botocore.compat import urlsplit, urlunsplit, unquote, json, quote
 from botocore import retryhandler
 import botocore.auth
 
@@ -77,6 +77,13 @@ def calculate_md5(event_name, params, **kwargs):
         params['headers']['Content-MD5'] = value
 
 
+def quote_source_header(event_name, params, **kwargs):
+    if params['headers'] and 'x-amz-copy-source' in params['headers']:
+        value = params['headers']['x-amz-copy-source']
+        value = quote(value.encode('utf-8'), safe='/~')
+        params['headers']['x-amz-copy-source'] = value
+
+
 def check_dns_name(bucket_name):
     """
     Check to see if the ``bucket_name`` complies with the
@@ -200,6 +207,8 @@ def maybe_switch_to_sigv4(service, region_name, **kwargs):
     ('before-call.s3.PutBucketLifecycle', calculate_md5),
     ('before-call.s3.PutBucketCors', calculate_md5),
     ('before-call.s3.DeleteObjects', calculate_md5),
+    ('before-call.s3.UploadPartCopy', quote_source_header),
+    ('before-call.s3.CopyObject', quote_source_header),
     ('before-auth.s3', fix_s3_host),
     ('service-created', register_retries_for_service),
     ('creating-endpoint.s3', maybe_switch_to_s3sigv4),

tests/unit/test_handlers.py

@@ -64,6 +64,14 @@ def test_decode_jsondoc(self):
         converted_value = first_non_none_response(rv)
         self.assertEqual(converted_value, {'foo':'bar'})
 
+    def test_quote_source_header(self):
+        for op in ('UploadPartCopy', 'CopyObject'):
+            event = self.session.create_event(
+                'before-call', 's3', op)
+            params = {'headers': {'x-amz-copy-source': 'foo++bar.txt'}}
+            self.session.emit(event, params=params)
+            self.assertEqual(
+                params['headers']['x-amz-copy-source'], 'foo%2B%2Bbar.txt')
 
 
 if __name__ == '__main__':