S3: Fix bucket policy enforcement for s3:PutObject when creating new …

…objects (#7840)

moto/s3/responses.py

@@ -1299,7 +1299,9 @@ def _key_response(
             bucket = self.backend.get_bucket(bucket_name)
         except S3ClientError:
             key = bucket = None
-        if key:
+
+        # Enforces policy when creating new objects, `if key` does not https://github.com/getmoto/moto/issues/7837
+        if bucket:
             resource = f"arn:{bucket.partition}:s3:::{bucket_name}/{key_name}"  # type: ignore[union-attr]
 
             # Authorization Workflow
@@ -1313,18 +1315,18 @@ def _key_response(
             if bucket_permissions == PermissionResult.DENIED:
                 return 403, {}, ""
 
-            # If the request is not authorized, and not signed,
-            # that means that the action should be allowed for anonymous users
-            if not authorized_request and not signed_url:
-                # We already know that the bucket permissions do not explicitly deny this
-                # So bucket permissions are either not set, or do not explicitly allow
-                # Next check is to see if the ACL of the individual key allows this action
-                if bucket_permissions != PermissionResult.PERMITTED and (
-                    key.acl and not key.acl.public_read
-                ):
-                    return 403, {}, ""
-
-        elif signed_url and not authorized_request:
+            if key:
+                # If the request is not authorized, and not signed,
+                # that means that the action should be allowed for anonymous users
+                if not authorized_request and not signed_url:
+                    # We already know that the bucket permissions do not explicitly deny this
+                    # So bucket permissions are either not set, or do not explicitly allow
+                    # Next check is to see if the ACL of the individual key allows this action
+                    if bucket_permissions != PermissionResult.PERMITTED and (
+                        key.acl and not key.acl.public_read
+                    ):
+                        return 403, {}, ""
+        if not key and signed_url and not authorized_request:
             # coming in from requests.get(s3.generate_presigned_url())
             if self._invalid_headers(request.url, dict(request.headers)):
                 return 403, {}, S3_INVALID_PRESIGNED_PARAMETERS

tests/test_s3/test_s3_bucket_policy.py

@@ -70,7 +70,10 @@ def test_block_or_allow_get_object(self, kwargs, boto3_status, unauthorized_stat
 
         assert requests.get(self.key_name).status_code == unauthorized_status
 
-    def test_block_put_object(self):
+    @pytest.mark.parametrize(
+        "key", ["test_txt", "new_txt"], ids=["update_object", "create_object"]
+    )
+    def test_block_put_object(self, key):
         # Block Put-access
         self._put_policy(**{"effect": "Deny", "actions": ["s3:PutObject"]})
 
@@ -79,7 +82,7 @@ def test_block_put_object(self):
 
         # But Put (via boto3 or requests) is not allowed
         with pytest.raises(ClientError) as exc:
-            self.client.put_object(Bucket="mybucket", Key="test_txt", Body="new data")
+            self.client.put_object(Bucket="mybucket", Key=key, Body="new data")
         err = exc.value.response["Error"]
         assert err["Message"] == "Forbidden"