diff --git a/docs/docs/services/s3.rst b/docs/docs/services/s3.rst
index 3d0e39d98..d07802e56 100644
--- a/docs/docs/services/s3.rst
+++ b/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
diff --git a/moto/s3/models.py b/moto/s3/models.py
index bf7b48939..2f2dbe8cc 100644
--- a/moto/s3/models.py
+++ b/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)
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 50a753b8c..378d79917 100644
--- a/moto/s3/responses.py
+++ b/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 = """
{{ replication["Role"] }}
"""
+
+S3_BUCKET_GET_OWNERSHIP_RULE = """
+
+
+ {{ownership_rule}}
+
+
+"""
+
+S3_ERROR_BUCKET_ONWERSHIP_NOT_FOUND = """
+
+ OwnershipControlsNotFoundError
+ The bucket ownership controls were not found
+ {{bucket_name}}
+ 294PFVCB9GFVXY2S
+ l/tqqyk7HZbfvFFpdq3+CAzA9JXUiV4ZajKYhwolOIpnmlvZrsI88AKsDLsgQI6EvZ9MuGHhk7M=
+
+"""
diff --git a/tests/test_s3/test_s3_ownership.py b/tests/test_s3/test_s3_ownership.py
new file mode 100644
index 000000000..bc68eb892
--- /dev/null
+++ b/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"
+ )