From 063537b86c7655c11d3ae429cba123fc32f76eb5 Mon Sep 17 00:00:00 2001
From: Bert Blommers <info@bertblommers.nl>
Date: Wed, 11 Sep 2024 19:32:45 +0000
Subject: [PATCH] S3: put_object() now support conditional writes (#8109)

---
 moto/s3/responses.py     |  7 +++++++
 tests/test_s3/test_s3.py | 31 +++++++++++++++++++++++++++++++
 2 files changed, 38 insertions(+)

diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 967382100684..eb79a3228ac4 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -1694,12 +1694,19 @@ def put_object(self) -> TYPE_RESPONSE:
         storage_class = self.headers.get("x-amz-storage-class", "STANDARD")
         encryption = self.headers.get("x-amz-server-side-encryption")
         kms_key_id = self.headers.get("x-amz-server-side-encryption-aws-kms-key-id")
+        if_none_match = self.headers.get("If-None-Match")
         bucket_key_enabled = self.headers.get(
             "x-amz-server-side-encryption-bucket-key-enabled"
         )
         if bucket_key_enabled is not None:
             bucket_key_enabled = str(bucket_key_enabled).lower()
 
+        if (
+            if_none_match == "*"
+            and self.backend.get_object(self.bucket_name, key_name) is not None
+        ):
+            raise PreconditionFailed("If-None-Match")
+
         checksum_algorithm, checksum_value = self._get_checksum(response_headers)
 
         bucket = self.backend.get_bucket(self.bucket_name)
diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py
index 3f057eb72f4c..6b178665f64b 100644
--- a/tests/test_s3/test_s3.py
+++ b/tests/test_s3/test_s3.py
@@ -526,6 +526,37 @@ def test_key_with_special_characters(key):
     assert resp["Body"].read() == b"value"
 
 
+@s3_aws_verified
+@pytest.mark.aws_verified
+@pytest.mark.parametrize("versioned", [True, False], ids=["versioned", "not versioned"])
+def test_conditional_write(versioned, bucket_name=None):
+    s3 = boto3.client("s3", region_name="us-east-1")
+    if versioned:
+        enable_versioning(bucket_name, s3)
+
+    s3.put_object(
+        Key="test_object",
+        Body="test",
+        Bucket=bucket_name,
+        IfNoneMatch="*",
+    )
+
+    with pytest.raises(ClientError) as exc:
+        s3.put_object(
+            Key="test_object",
+            Body="another_test",
+            Bucket=bucket_name,
+            IfNoneMatch="*",
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "PreconditionFailed"
+    assert (
+        err["Message"]
+        == "At least one of the pre-conditions you specified did not hold"
+    )
+    assert err["Condition"] == "If-None-Match"
+
+
 @mock_aws
 def test_bucket_key_listing_order():
     s3_resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)