diff --git a/botocore/handlers.py b/botocore/handlers.py
index 584130b68e..817a2e379d 100644
--- a/botocore/handlers.py
+++ b/botocore/handlers.py
@@ -23,7 +23,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
 
@@ -191,6 +191,13 @@ def signature_overrides(service_data, service_name, session, **kwargs):
         service_data['signature_version'] = signature_version_override
 
 
+def quote_source_header(params, **kwargs):
+    if params['headers'] and 'x-amz-copy-source' in params['headers']:
+        value = params['headers']['x-amz-copy-source']
+        params['headers']['x-amz-copy-source'] = quote(
+            value.encode('utf-8'), '/~')
+
+
 # This is a list of (event_name, handler).
 # When a Session is created, everything in this list will be
 # automatically registered with that Session.
@@ -205,6 +212,8 @@ def signature_overrides(service_data, service_name, session, **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),
diff --git a/tests/integration/test_s3.py b/tests/integration/test_s3.py
index 9ee7685d94..94a7220343 100644
--- a/tests/integration/test_s3.py
+++ b/tests/integration/test_s3.py
@@ -361,5 +361,38 @@ def test_reset_stream_on_redirects(self):
         self.assertEqual(data['Body'].read(), b'foo' * 1024)
 
 
+class TestS3Copy(BaseS3Test):
+    def setUp(self):
+        super(TestS3Copy, 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')
+        operation.call(self.endpoint, bucket=self.bucket_name,
+            create_bucket_configuration={'LocationConstraint': self.bucket_location})
+
+    def tearDown(self):
+        operation = self.service.get_operation('DeleteBucket')
+        operation.call(self.endpoint, bucket=self.bucket_name)
+
+    def test_copy_with_quoted_char(self):
+        key_name = 'a+b/foo'
+        self.create_object(key_name=key_name)
+
+        operation = self.service.get_operation('CopyObject')
+        http, parsed = operation.call(
+            self.endpoint, bucket=self.bucket_name, key=key_name + 'bar',
+            copy_source='%s/%s' % (self.bucket_name, key_name))
+        self.assertEqual(http.status_code, 200)
+
+        # Now verify we can retrieve the copied object.
+        operation = self.service.get_operation('GetObject')
+        response = operation.call(self.endpoint, bucket=self.bucket_name,
+                                  key=key_name + 'bar')
+        data = response[1]
+        self.assertEqual(data['Body'].read().decode('utf-8'), 'foo')
+
+
 if __name__ == '__main__':
     unittest.main()