S3: Support for ownership rule configuration (#5409)

docs/docs/services/s3.rst

@@ -40,7 +40,7 @@ s3
 - [ ] delete_bucket_inventory_configuration
 - [X] delete_bucket_lifecycle
 - [ ] delete_bucket_metrics_configuration
-- [ ] delete_bucket_ownership_controls
+- [X] delete_bucket_ownership_controls
 - [X] delete_bucket_policy
 - [X] delete_bucket_replication
 - [X] delete_bucket_tagging
@@ -63,7 +63,7 @@ s3
 - [ ] get_bucket_metrics_configuration
 - [ ] get_bucket_notification
 - [X] get_bucket_notification_configuration
-- [ ] get_bucket_ownership_controls
+- [X] get_bucket_ownership_controls
 - [X] get_bucket_policy
 - [ ] get_bucket_policy_status
 - [X] get_bucket_replication
@@ -105,7 +105,7 @@ s3
 - [ ] put_bucket_metrics_configuration
 - [ ] put_bucket_notification
 - [X] put_bucket_notification_configuration
-  
+
         The configuration can be persisted, but at the moment we only send notifications to the following targets:
 
          - AWSLambda
@@ -117,7 +117,7 @@ s3
          - 's3:ObjectCreated:Put'
         
 
-- [ ] put_bucket_ownership_controls
+- [X] put_bucket_ownership_controls
 - [X] put_bucket_policy
 - [X] put_bucket_replication
 - [ ] put_bucket_request_payment

moto/s3/models.py

@@ -893,6 +893,7 @@ class FakeBucket(CloudFormationModel):
         self.default_lock_mode = ""
         self.default_lock_days = 0
         self.default_lock_years = 0
+        self.ownership_rule = None
 
     @property
     def location(self):
@@ -1617,6 +1618,15 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
     def delete_bucket_encryption(self, bucket_name):
         self.get_bucket(bucket_name).encryption = None
 
+    def get_bucket_ownership_rule(self, bucket_name):
+        return self.get_bucket(bucket_name).ownership_rule
+
+    def put_bucket_ownership_rule(self, bucket_name, ownership):
+        self.get_bucket(bucket_name).ownership_rule = ownership
+
+    def delete_bucket_ownership_rule(self, bucket_name):
+        self.get_bucket(bucket_name).ownership_rule = None
+
     def get_bucket_replication(self, bucket_name):
         bucket = self.get_bucket(bucket_name)
         return getattr(bucket, "replication", None)

moto/s3/responses.py

@@ -563,6 +563,13 @@ class S3Response(BaseResponse):
                 return 404, {}, template.render(bucket_name=bucket_name)
             template = self.response_template(S3_REPLICATION_CONFIG)
             return 200, {}, template.render(replication=replication)
+        elif "ownershipControls" in querystring:
+            ownership_rule = self.backend.get_bucket_ownership_rule(bucket_name)
+            if not ownership_rule:
+                template = self.response_template(S3_ERROR_BUCKET_ONWERSHIP_NOT_FOUND)
+                return 404, {}, template.render(bucket_name=bucket_name)
+            template = self.response_template(S3_BUCKET_GET_OWNERSHIP_RULE)
+            return 200, {}, template.render(ownership_rule=ownership_rule)
 
         bucket = self.backend.get_bucket(bucket_name)
         prefix = querystring.get("prefix", [None])[0]
@@ -837,6 +844,13 @@ class S3Response(BaseResponse):
             replication_config = self._replication_config_from_xml(self.body)
             self.backend.put_bucket_replication(bucket_name, replication_config)
             return ""
+        elif "ownershipControls" in querystring:
+            ownership_rule = self._ownership_rule_from_body()
+            self.backend.put_bucket_ownership_rule(
+                bucket_name, ownership=ownership_rule
+            )
+            return ""
+
         else:
             # us-east-1, the default AWS region behaves a bit differently
             # - you should not use it as a location constraint --> it fails
@@ -893,6 +907,10 @@ class S3Response(BaseResponse):
                 new_bucket.object_lock_enabled = True
                 new_bucket.versioning_status = "Enabled"
 
+            ownership_rule = request.headers.get("x-amz-object-ownership")
+            if ownership_rule:
+                new_bucket.ownership_rule = ownership_rule
+
             template = self.response_template(S3_BUCKET_CREATE_RESPONSE)
             return 200, {}, template.render(bucket=new_bucket)
 
@@ -924,6 +942,9 @@ class S3Response(BaseResponse):
         elif "replication" in querystring:
             self.backend.delete_bucket_replication(bucket_name)
             return 204, {}, ""
+        elif "ownershipControls" in querystring:
+            self.backend.delete_bucket_ownership_rule(bucket_name)
+            return 204, {}, ""
 
         removed_bucket = self.backend.delete_bucket(bucket_name)
 
@@ -1761,6 +1782,14 @@ class S3Response(BaseResponse):
 
         return parsed_xml["ServerSideEncryptionConfiguration"]
 
+    def _ownership_rule_from_body(self):
+        parsed_xml = xmltodict.parse(self.body)
+
+        if not parsed_xml["OwnershipControls"]["Rule"].get("ObjectOwnership"):
+            raise MalformedXML()
+
+        return parsed_xml["OwnershipControls"]["Rule"]["ObjectOwnership"]
+
     def _logging_from_body(self):
         parsed_xml = xmltodict.parse(self.body)
 
@@ -2784,3 +2813,21 @@ S3_REPLICATION_CONFIG = """<?xml version="1.0" encoding="UTF-8"?>
 <Role>{{ replication["Role"] }}</Role>
 </ReplicationConfiguration>
 """
+
+S3_BUCKET_GET_OWNERSHIP_RULE = """<?xml version="1.0" encoding="UTF-8"?>
+<OwnershipControls xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+    <Rule>
+        <ObjectOwnership>{{ownership_rule}}</ObjectOwnership>
+    </Rule>
+</OwnershipControls>
+"""
+
+S3_ERROR_BUCKET_ONWERSHIP_NOT_FOUND = """
+<Error>
+    <Code>OwnershipControlsNotFoundError</Code>
+    <Message>The bucket ownership controls were not found</Message>
+    <BucketName>{{bucket_name}}</BucketName>
+    <RequestId>294PFVCB9GFVXY2S</RequestId>
+    <HostId>l/tqqyk7HZbfvFFpdq3+CAzA9JXUiV4ZajKYhwolOIpnmlvZrsI88AKsDLsgQI6EvZ9MuGHhk7M=</HostId>
+</Error>
+"""

tests/test_s3/test_s3_ownership.py

@@ -0,0 +1,50 @@
+import boto3
+from botocore.client import ClientError
+
+import pytest
+import sure  # noqa # pylint: disable=unused-import
+from moto import mock_s3
+
+
+@mock_s3
+def test_create_bucket_with_ownership():
+    bucket = "bucket-with-owner"
+    ownership = "BucketOwnerPreferred"
+    client = boto3.client("s3")
+    client.create_bucket(Bucket=bucket, ObjectOwnership=ownership)
+
+    response = client.get_bucket_ownership_controls(Bucket=bucket)
+    response["OwnershipControls"]["Rules"][0]["ObjectOwnership"].should.equal(ownership)
+
+
+@mock_s3
+def test_put_ownership_to_bucket():
+    bucket = "bucket-updated-with-owner"
+    ownership = "ObjectWriter"
+    client = boto3.client("s3")
+    client.create_bucket(Bucket=bucket)
+
+    client.put_bucket_ownership_controls(
+        Bucket=bucket, OwnershipControls={"Rules": [{"ObjectOwnership": ownership}]}
+    )
+
+    response = client.get_bucket_ownership_controls(Bucket=bucket)
+    response["OwnershipControls"]["Rules"][0]["ObjectOwnership"].should.equal(ownership)
+
+
+@mock_s3
+def test_delete_ownership_from_bucket():
+    bucket = "bucket-with-owner-removed"
+    ownership = "BucketOwnerEnforced"
+    client = boto3.client("s3")
+    client.create_bucket(Bucket=bucket, ObjectOwnership=ownership)
+
+    client.delete_bucket_ownership_controls(Bucket=bucket)
+
+    with pytest.raises(ClientError) as ex:
+        client.get_bucket_ownership_controls(Bucket=bucket)
+
+    ex.value.response["Error"]["Code"].should.equal("OwnershipControlsNotFoundError")
+    ex.value.response["Error"]["Message"].should.equal(
+        "The bucket ownership controls were not found"
+    )