From 2e4149429332e7042a4d8aa5b7a0680397c6996e Mon Sep 17 00:00:00 2001 From: Casey Bodley Date: Tue, 13 Aug 2024 10:23:19 -0400 Subject: [PATCH] s3: reenable tenanted bucket policy test the before-call hook url-encodes the ':' part of tenanted bucket names to resolve SignatureDoesNotMatch errors removed the list-v2 version of the test since it isn't relevant to bucket policy test coverage add a new test case that creates the bucket under the tenanted user, then uses the main client to access it Signed-off-by: Casey Bodley --- s3tests.conf.SAMPLE | 3 ++ s3tests_boto3/functional/__init__.py | 4 ++ s3tests_boto3/functional/test_s3.py | 70 ++++++++++++---------------- 3 files changed, 38 insertions(+), 39 deletions(-) diff --git a/s3tests.conf.SAMPLE b/s3tests.conf.SAMPLE index c0dc89a..3f0992a 100644 --- a/s3tests.conf.SAMPLE +++ b/s3tests.conf.SAMPLE @@ -117,6 +117,9 @@ secret_key = opqrstuvwxyzabcdefghijklmnopqrstuvwxyzab # tenant email set in vstart.sh email = tenanteduser@example.com +# tenant name +tenant = testx + #following section needs to be added for all sts-tests [iam] #used for iam operations in sts-tests diff --git a/s3tests_boto3/functional/__init__.py b/s3tests_boto3/functional/__init__.py index 2f9f7e1..3bdf2ec 100644 --- a/s3tests_boto3/functional/__init__.py +++ b/s3tests_boto3/functional/__init__.py @@ -259,6 +259,7 @@ def configure(): config.tenant_display_name = cfg.get('s3 tenant',"display_name") config.tenant_user_id = cfg.get('s3 tenant',"user_id") config.tenant_email = cfg.get('s3 tenant',"email") + config.tenant_name = cfg.get('s3 tenant',"tenant") config.iam_access_key = cfg.get('iam',"access_key") config.iam_secret_key = cfg.get('iam',"secret_key") @@ -694,6 +695,9 @@ def get_tenant_aws_secret_key(): def get_tenant_display_name(): return config.tenant_display_name +def get_tenant_name(): + return config.tenant_name + def get_tenant_user_id(): return config.tenant_user_id diff --git a/s3tests_boto3/functional/test_s3.py b/s3tests_boto3/functional/test_s3.py index 803279f..d8b81ec 100644 --- a/s3tests_boto3/functional/test_s3.py +++ b/s3tests_boto3/functional/test_s3.py @@ -68,6 +68,7 @@ from . import ( get_alt_client, get_tenant_client, get_tenant_iam_client, + get_tenant_name, get_tenant_user_id, get_buckets_list, get_objects_list, @@ -10573,17 +10574,29 @@ def test_bucketv2_policy_acl(): client.delete_bucket_policy(Bucket=bucket_name) client.put_bucket_acl(Bucket=bucket_name, ACL='public-read') +def tenanted_bucket_name(tenant): + def change_bucket_name(params, **kwargs): + old_name = params['context']['signing']['bucket'] + new_name = "{}:{}".format(tenant, old_name) + params['Bucket'] = new_name + params['context']['signing']['bucket'] = new_name + + # the : needs to be url-encoded for urls + new_name_url = "{}%3A{}".format(tenant, old_name) + params['url'] = params['url'].replace(old_name, new_name_url) + params['url_path'] = params['url_path'].replace(old_name, new_name_url) + + return change_bucket_name + @pytest.mark.bucket_policy -# TODO: remove this fails_on_rgw when I fix it -@pytest.mark.fails_on_rgw def test_bucket_policy_different_tenant(): bucket_name = get_new_bucket() client = get_client() key = 'asdf' client.put_object(Bucket=bucket_name, Key=key, Body='asdf') - resource1 = "arn:aws:s3::*:" + bucket_name - resource2 = "arn:aws:s3::*:" + bucket_name + "/*" + resource1 = "arn:aws:s3:::" + bucket_name + resource2 = "arn:aws:s3:::" + bucket_name + "/*" policy_document = json.dumps( { "Version": "2012-10-17", @@ -10600,35 +10613,22 @@ def test_bucket_policy_different_tenant(): client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document) - # TODO: figure out how to change the bucketname - def change_bucket_name(**kwargs): - kwargs['params']['url'] = "http://localhost:8000/:{bucket_name}?encoding-type=url".format(bucket_name=bucket_name) - kwargs['params']['url_path'] = "/:{bucket_name}".format(bucket_name=bucket_name) - kwargs['params']['context']['signing']['bucket'] = ":{bucket_name}".format(bucket_name=bucket_name) - print(kwargs['request_signer']) - print(kwargs) - - #bucket_name = ":" + bucket_name + # use the tenanted client to list the global tenant's bucket tenant_client = get_tenant_client() - tenant_client.meta.events.register('before-call.s3.ListObjects', change_bucket_name) + tenant_client.meta.events.register('before-call.s3.ListObjects', tenanted_bucket_name('')) response = tenant_client.list_objects(Bucket=bucket_name) - #alt_client = get_alt_client() - #response = alt_client.list_objects(Bucket=bucket_name) assert len(response['Contents']) == 1 @pytest.mark.bucket_policy -# TODO: remove this fails_on_rgw when I fix it -@pytest.mark.fails_on_rgw -@pytest.mark.list_objects_v2 -def test_bucketv2_policy_different_tenant(): - bucket_name = get_new_bucket() - client = get_client() +def test_bucket_policy_tenanted_bucket(): + tenant_client = get_tenant_client() + bucket_name = get_new_bucket(tenant_client) key = 'asdf' - client.put_object(Bucket=bucket_name, Key=key, Body='asdf') + tenant_client.put_object(Bucket=bucket_name, Key=key, Body='asdf') - resource1 = "arn:aws:s3::*:" + bucket_name - resource2 = "arn:aws:s3::*:" + bucket_name + "/*" + resource1 = "arn:aws:s3:::" + bucket_name + resource2 = "arn:aws:s3:::" + bucket_name + "/*" policy_document = json.dumps( { "Version": "2012-10-17", @@ -10643,23 +10643,15 @@ def test_bucketv2_policy_different_tenant(): }] }) - client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document) + tenant_client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document) - # TODO: figure out how to change the bucketname - def change_bucket_name(**kwargs): - kwargs['params']['url'] = "http://localhost:8000/:{bucket_name}?encoding-type=url".format(bucket_name=bucket_name) - kwargs['params']['url_path'] = "/:{bucket_name}".format(bucket_name=bucket_name) - kwargs['params']['context']['signing']['bucket'] = ":{bucket_name}".format(bucket_name=bucket_name) - print(kwargs['request_signer']) - print(kwargs) + tenant = get_tenant_name() - #bucket_name = ":" + bucket_name - tenant_client = get_tenant_client() - tenant_client.meta.events.register('before-call.s3.ListObjects', change_bucket_name) - response = tenant_client.list_objects_v2(Bucket=bucket_name) - #alt_client = get_alt_client() - #response = alt_client.list_objects_v2(Bucket=bucket_name) + # use the global tenant's client to list the tenanted bucket + client = get_client() + client.meta.events.register('before-call.s3.ListObjects', tenanted_bucket_name(tenant)) + response = client.list_objects(Bucket=bucket_name) assert len(response['Contents']) == 1 @pytest.mark.bucket_policy