Fix passing of metadata on S3 presigned URL put
diff --git a/localstack/services/s3/s3_listener.py b/localstack/services/s3/s3_listener.py
index 3176b59f03198..7e5b8a0951c51 100644
--- a/localstack/services/s3/s3_listener.py
+++ b/localstack/services/s3/s3_listener.py
@@ -42,6 +42,9 @@
 # list of destination types for bucket notifications
 NOTIFICATION_DESTINATION_TYPES = ('Queue', 'Topic', 'CloudFunction', 'LambdaFunction')
 
+# prefix for object metadata keys in headers and query params
+OBJECT_METADATA_KEY_PREFIX = 'x-amz-meta-'
+
 # response header overrides the client may request
 ALLOWED_HEADER_OVERRIDES = {
     'response-content-type': 'Content-Type',
@@ -306,6 +309,13 @@ def append_list_objects_marker(method, path, data, response):
             response.headers['Content-Length'] = str(len(response._content))
 
 
+def append_metadata_headers(method, query_map, headers):
+    for key, value in query_map.items():
+        if key.lower().startswith(OBJECT_METADATA_KEY_PREFIX):
+            if headers.get(key) is None:
+                headers[key] = value[0]
+
+
 def get_lifecycle(bucket_name):
     lifecycle = BUCKET_LIFECYCLE.get(bucket_name)
     if not lifecycle:
@@ -561,6 +571,10 @@ def forward_request(self, method, path, data, headers):
         path = parsed.path
         bucket = path.split('/')[1]
         query_map = urlparse.parse_qs(query, keep_blank_values=True)
+
+        # remap metadata query params (not supported in moto) to request headers
+        append_metadata_headers(method, query_map, headers)
+
         if query == 'notification' or 'notification' in query_map:
             # handle and return response for ?notification request
             response = handle_notification_request(bucket, method, data)
@@ -662,7 +676,8 @@ def return_response(self, method, path, data, headers, response):
 
             # Remove body from PUT response on presigned URL
             # https://github.com/localstack/localstack/issues/1317
-            if method == 'PUT' and ('X-Amz-Security-Token=' in path or 'AWSAccessKeyId=' in path):
+            if method == 'PUT' and ('X-Amz-Security-Token=' in path or
+                    'X-Amz-Credential=' in path or 'AWSAccessKeyId=' in path):
                 response._content = ''
                 reset_content_length = True
 
diff --git a/tests/integration/test_s3.py b/tests/integration/test_s3.py
index 2097ee3365d3a..846ea6345bf64 100644
--- a/tests/integration/test_s3.py
+++ b/tests/integration/test_s3.py
@@ -176,8 +176,7 @@ def test_s3_get_response_default_content_type(self):
         object_key = 'key-by-hostname'
         self.s3_client.put_object(Bucket=bucket_name, Key=object_key, Body='something')
         url = self.s3_client.generate_presigned_url(
-            'get_object', Params={'Bucket': bucket_name, 'Key': object_key}
-        )
+            'get_object', Params={'Bucket': bucket_name, 'Key': object_key})
 
         # get object and assert headers
         response = requests.get(url, verify=False)
@@ -185,6 +184,32 @@ def test_s3_get_response_default_content_type(self):
         # clean up
         self._delete_bucket(bucket_name, [object_key])
 
+    def test_s3_put_presigned_url_metadata(self):
+        # Object metadata should be passed as query params via presigned URL
+        # https://github.com/localstack/localstack/issues/544
+
+        bucket_name = 'test-bucket-%s' % short_uid()
+        self.s3_client.create_bucket(Bucket=bucket_name)
+
+        # put object
+        object_key = 'key-by-hostname'
+        url = self.s3_client.generate_presigned_url(
+            'put_object', Params={'Bucket': bucket_name, 'Key': object_key})
+        # append metadata manually to URL (this is not easily possible with boto3, as "Metadata" cannot
+        # be passed to generate_presigned_url, and generate_presigned_post works differently)
+        url += '&x-amz-meta-foo=bar'
+
+        # get object and assert metadata is present
+        response = requests.put(url, data='content 123', verify=False)
+        self.assertLess(response.status_code, 400)
+        # response body should be empty, see https://github.com/localstack/localstack/issues/1317
+        self.assertEqual('', to_str(response.content))
+        response = self.s3_client.head_object(Bucket=bucket_name, Key=object_key)
+        self.assertEquals('bar', response.get('Metadata', {}).get('foo'))
+
+        # clean up
+        self._delete_bucket(bucket_name, [object_key])
+
     def test_s3_get_response_content_type_same_as_upload(self):
         bucket_name = 'test-bucket-%s' % short_uid()
         self.s3_client.create_bucket(Bucket=bucket_name)

Fix passing of metadata on S3 presigned URL put - addresses #544