fix pre-signed POST when file is passed as regular form field (#10314)

localstack/services/s3/v3/provider.py

@@ -4,6 +4,7 @@
 import json
 import logging
 from collections import defaultdict
+from io import BytesIO
 from operator import itemgetter
 from secrets import token_urlsafe
 from typing import IO, Optional, Union
@@ -280,7 +281,7 @@
 from localstack.services.s3.website_hosting import register_website_hosting_routes
 from localstack.state import AssetDirectory, StateVisitor
 from localstack.utils.aws.arns import s3_bucket_name
-from localstack.utils.strings import short_uid, to_str
+from localstack.utils.strings import short_uid, to_bytes, to_str
 
 LOG = logging.getLogger(__name__)
 
@@ -3614,6 +3615,11 @@ def get_object_torrent(
     def post_object(
         self, context: RequestContext, bucket: BucketName, body: IO[Body] = None, **kwargs
     ) -> PostResponse:
+        if "multipart/form-data" not in context.request.headers.get("Content-Type", ""):
+            raise PreconditionFailed(
+                "At least one of the pre-conditions you specified did not hold",
+                Condition="Bucket POST must be of the enclosure-type multipart/form-data",
+            )
         # see https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectPOST.html
         # TODO: signature validation is not implemented for pre-signed POST
         # policy validation is not implemented either, except expiration and mandatory fields
@@ -3623,11 +3629,21 @@ def post_object(
 
         form = context.request.form
         validate_post_policy(form)
-
-        fileobj = context.request.files["file"]
         object_key = context.request.form.get("key")
-        if "${filename}" in object_key:
-            object_key = object_key.replace("${filename}", fileobj.filename)
+
+        if "file" in form:
+            # in AWS, you can pass the file content as a string in the form field and not as a file object
+            stream = BytesIO(to_bytes(form["file"]))
+        else:
+            # this is the default behaviour
+            fileobj = context.request.files["file"]
+            stream = fileobj.stream
+            if "${filename}" in object_key:
+                # TODO: ${filename} is actually usable in all form fields
+                # See https://docs.aws.amazon.com/sdk-for-ruby/v3/api/Aws/S3/PresignedPost.html
+                # > The string ${filename} is automatically replaced with the name of the file provided by the user and
+                # is recognized by all form fields.
+                object_key = object_key.replace("${filename}", fileobj.filename)
 
         if canned_acl := form.get("acl"):
             validate_canned_acl(canned_acl)
@@ -3656,7 +3672,7 @@ def post_object(
         }
 
         if tagging := form.get("tagging"):
-            # this is weird, as it's direct XML in the form, we need to parse it direcly
+            # this is weird, as it's direct XML in the form, we need to parse it directly
             tagging = parse_post_object_tagging_xml(tagging)
 
         if (storage_class := form.get("x-amz-storage-class")) is not None and (
@@ -3706,7 +3722,7 @@ def post_object(
         )
 
         s3_stored_object = self._storage_backend.open(bucket, s3_object)
-        s3_stored_object.write(fileobj.stream)
+        s3_stored_object.write(stream)
 
         if checksum_algorithm and s3_object.checksum_value != s3_stored_object.checksum:
             self._storage_backend.remove(bucket, s3_object)

tests/aws/services/s3/test_s3.py

@@ -10333,6 +10333,114 @@ def test_post_object_with_storage_class(self, s3_bucket, aws_client, snapshot):
         assert response.status_code == 400
         snapshot.match("invalid-storage-error", xmltodict.parse(response.content))
 
+    @markers.aws.validated
+    @pytest.mark.skipif(
+        condition=LEGACY_V2_S3_PROVIDER,
+        reason="not implemented in moto",
+    )
+    @markers.snapshot.skip_snapshot_verify(
+        paths=["$..HostId"],  # FIXME: in CI, it fails sporadically and the form is empty
+    )
+    def test_post_object_with_wrong_content_type(self, s3_bucket, aws_client, snapshot):
+        snapshot.add_transformers_list(
+            [
+                snapshot.transform.key_value("HostId"),
+                snapshot.transform.key_value("RequestId"),
+            ]
+        )
+        object_key = "test-presigned-post-key-wrong-content-type"
+        presigned_request = aws_client.s3.generate_presigned_post(
+            Bucket=s3_bucket,
+            Key=object_key,
+            ExpiresIn=60,
+            Conditions=[
+                {"bucket": s3_bucket},
+            ],
+        )
+        # PostObject
+        response = requests.post(
+            presigned_request["url"],
+            data=presigned_request["fields"],
+            files={"file": "test-body-wrong-content-type"},
+            headers={"Content-Type": "text/html"},
+            verify=False,
+        )
+
+        assert response.status_code == 412
+        snapshot.match("invalid-content-type-error", xmltodict.parse(response.content))
+
+    @markers.aws.validated
+    @pytest.mark.skipif(
+        condition=LEGACY_V2_S3_PROVIDER,
+        reason="not implemented in moto",
+    )
+    @markers.snapshot.skip_snapshot_verify(
+        paths=[
+            "$..ContentLength",
+            "$..ETag",
+            "$..HostId",
+        ],  # FIXME: in CI, it fails sporadically and the form is empty
+    )
+    def test_post_object_with_file_as_string(self, s3_bucket, aws_client, snapshot):
+        # this is a test for https://github.com/localstack/localstack/issues/10309
+        # You can send requests with node.js with a different format than what we can with Python
+        # (the actual file would just be a regular `file` key of the form with content)
+        snapshot.add_transformers_list(
+            [
+                snapshot.transform.key_value("HostId"),
+                snapshot.transform.key_value("RequestId"),
+                snapshot.transform.key_value("Name"),
+            ]
+        )
+        object_key = "test-presigned-post-file-as-field"
+        presigned_request = aws_client.s3.generate_presigned_post(
+            Bucket=s3_bucket,
+            Key=object_key,
+            ExpiresIn=60,
+            Conditions=[
+                {"bucket": s3_bucket},
+            ],
+        )
+
+        # we need to define a proper format for `files` so that we don't add the filename= field to the form
+        # see https://github.com/psf/requests/issues/1081
+
+        # PostObject
+        response = requests.post(
+            presigned_request["url"],
+            data=presigned_request["fields"],
+            files={
+                "file": (None, "test-body-file-as-field"),
+            },
+            verify=False,
+        )
+        assert response.status_code == 204
+
+        head_object = aws_client.s3.head_object(Bucket=s3_bucket, Key=object_key)
+        snapshot.match("head-object", head_object)
+
+        object_key = "file-as-field-${filename}"
+        presigned_request = aws_client.s3.generate_presigned_post(
+            Bucket=s3_bucket,
+            Key=object_key,
+            ExpiresIn=60,
+            Conditions=[
+                {"bucket": s3_bucket},
+            ],
+        )
+        response = requests.post(
+            presigned_request["url"],
+            data=presigned_request["fields"],
+            files={
+                "file": (None, "test-body-file-as-field-filename-replacement"),
+            },
+            verify=False,
+        )
+        assert response.status_code == 204
+
+        response = aws_client.s3.list_objects_v2(Bucket=s3_bucket)
+        snapshot.match("list-objects", response)
+
 
 def _s3_client_pre_signed_client(conf: Config, endpoint_url: str = None):
     if is_aws_cloud():

tests/aws/services/s3/test_s3.snapshot.json

@@ -11150,5 +11150,65 @@
         }
       }
     }
+  },
+  "tests/aws/services/s3/test_s3.py::TestS3PresignedPost::test_post_object_with_wrong_content_type": {
+    "recorded-date": "23-02-2024, 23:47:44",
+    "recorded-content": {
+      "invalid-content-type-error": {
+        "Error": {
+          "Code": "PreconditionFailed",
+          "Condition": "Bucket POST must be of the enclosure-type multipart/form-data",
+          "HostId": "<host-id:1>",
+          "Message": "At least one of the pre-conditions you specified did not hold",
+          "RequestId": "<request-id:1>"
+        }
+      }
+    }
+  },
+  "tests/aws/services/s3/test_s3.py::TestS3PresignedPost::test_post_object_with_file_as_string": {
+    "recorded-date": "24-02-2024, 01:01:59",
+    "recorded-content": {
+      "head-object": {
+        "AcceptRanges": "bytes",
+        "ContentLength": 23,
+        "ContentType": "binary/octet-stream",
+        "ETag": "\"518feee9a33977e5047cda470999729a\"",
+        "LastModified": "datetime",
+        "Metadata": {},
+        "ServerSideEncryption": "AES256",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      },
+      "list-objects": {
+        "Contents": [
+          {
+            "ETag": "\"31ac3a102af06a3d79f0172d01158b49\"",
+            "Key": "file-as-field-${filename}",
+            "LastModified": "datetime",
+            "Size": 44,
+            "StorageClass": "STANDARD"
+          },
+          {
+            "ETag": "\"518feee9a33977e5047cda470999729a\"",
+            "Key": "test-presigned-post-file-as-field",
+            "LastModified": "datetime",
+            "Size": 23,
+            "StorageClass": "STANDARD"
+          }
+        ],
+        "EncodingType": "url",
+        "IsTruncated": false,
+        "KeyCount": 2,
+        "MaxKeys": 1000,
+        "Name": "<name:1>",
+        "Prefix": "",
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
   }
 }

tests/aws/services/s3/test_s3.validation.json

@@ -512,6 +512,9 @@
   "tests/aws/services/s3/test_s3.py::TestS3ObjectLockRetention::test_s3_object_retention_exc": {
     "last_validated_date": "2023-08-09T15:58:37+00:00"
   },
+  "tests/aws/services/s3/test_s3.py::TestS3PresignedPost::test_post_object_with_file_as_string": {
+    "last_validated_date": "2024-02-24T01:01:59+00:00"
+  },
   "tests/aws/services/s3/test_s3.py::TestS3PresignedPost::test_post_object_with_metadata": {
     "last_validated_date": "2023-08-14T17:54:15+00:00"
   },
@@ -530,6 +533,9 @@
   "tests/aws/services/s3/test_s3.py::TestS3PresignedPost::test_post_object_with_tags[single]": {
     "last_validated_date": "2023-08-14T17:32:11+00:00"
   },
+  "tests/aws/services/s3/test_s3.py::TestS3PresignedPost::test_post_object_with_wrong_content_type": {
+    "last_validated_date": "2024-02-23T23:47:44+00:00"
+  },
   "tests/aws/services/s3/test_s3.py::TestS3PresignedPost::test_post_request_expires": {
     "last_validated_date": "2023-08-04T21:58:47+00:00"
   },