rgw test-case to support x-expected-bucket-owner

If the bucket is owned by a different account, the request fails with the HTTP status code 403 Forbidden (access denied).
PR: ceph/ceph#61215
Fixes: https://tracker.ceph.com/issues/64526
Signed-off-by: Raja Sharma <raja@ibm.com>

diff --git a/s3tests_boto3/functional/test_s3.py b/s3tests_boto3/functional/test_s3.py
index ec0dfc66047c3548820bd27a69d0d6d2d7def008..cffe480240d3f3a59295827dc0fc9f426019ab86 100644 (file)
--- a/s3tests_boto3/functional/test_s3.py
+++ b/s3tests_boto3/functional/test_s3.py
@@ -1694,6 +1694,27 @@ def test_multi_object_delete():
     response = client.list_objects(Bucket=bucket_name)
     assert 'Contents' not in response
 
+@pytest.mark.list_objects_v2
+def test_expected_bucket_owner():
+  bucket_name = get_new_bucket()
+  client = get_client()
+  client.put_bucket_acl(Bucket=bucket_name, ACL='public-read-write')
+  client.list_objects(Bucket=bucket_name)
+  client.put_object(Bucket=bucket_name, Key='foo', Body='bar')
+
+  unauthenticated_client = get_unauthenticated_client()
+  incorrect_expected_owner = get_main_user_id() + 'foo'
+
+  e = assert_raises(ClientError, unauthenticated_client.list_objects, Bucket=bucket_name, ExpectedBucketOwner=incorrect_expected_owner)
+  status, error_code = _get_status_and_error_code(e.response)
+  assert status == 403
+  assert error_code == 'AccessDenied'
+
+  e = assert_raises(ClientError, unauthenticated_client.put_object, Bucket=bucket_name, Key='bar', Body='coffee', ExpectedBucketOwner=incorrect_expected_owner)
+  status, error_code = _get_status_and_error_code(e.response)
+  assert status == 403
+  assert error_code == 'AccessDenied'
+
 @pytest.mark.list_objects_v2
 def test_multi_objectv2_delete():
     key_names = ['key0', 'key1', 'key2']