From 3056e6d0394987d20ba6c363f238c4922e5cbcab Mon Sep 17 00:00:00 2001 From: Seena Fallah Date: Mon, 19 Feb 2024 22:40:12 +0100 Subject: [PATCH 1/2] PublicAccessBlock: test 404 on no block configuration Make sure NoSuchPublicAccessBlockConfiguration is returned when no public block is configured on bucket: Refs: https://github.com/ceph/ceph/pull/55652 Signed-off-by: Seena Fallah --- s3tests_boto3/functional/test_s3.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/s3tests_boto3/functional/test_s3.py b/s3tests_boto3/functional/test_s3.py index 22ed5ab..c8a0933 100644 --- a/s3tests_boto3/functional/test_s3.py +++ b/s3tests_boto3/functional/test_s3.py @@ -12590,16 +12590,22 @@ def test_get_nonpublicpolicy_deny_bucket_policy_status(): resp = client.get_bucket_policy_status(Bucket=bucket_name) assert resp['PolicyStatus']['IsPublic'] == True -def test_get_default_public_block(): - #client = get_svc_client(svc='s3control', client_config=Config(s3={'addressing_style': 'path'})) +def test_get_undefined_public_block(): bucket_name = get_new_bucket() client = get_client() - resp = client.get_public_access_block(Bucket=bucket_name) - assert resp['PublicAccessBlockConfiguration']['BlockPublicAcls'] == False - assert resp['PublicAccessBlockConfiguration']['BlockPublicPolicy'] == False - assert resp['PublicAccessBlockConfiguration']['IgnorePublicAcls'] == False - assert resp['PublicAccessBlockConfiguration']['RestrictPublicBuckets'] == False + # delete the existing public access block configuration + # as AWS creates a default public access block configuration + resp = client.delete_public_access_block(Bucket=bucket_name) + assert resp['ResponseMetadata']['HTTPStatusCode'] == 204 + + response_code = "" + try: + resp = client.get_public_access_block(Bucket=bucket_name) + except ClientError as e: + response_code = e.response['Error']['Code'] + + assert response_code == 'NoSuchPublicAccessBlockConfiguration' def test_put_public_block(): #client = get_svc_client(svc='s3control', client_config=Config(s3={'addressing_style': 'path'})) From 3af42312bf4870e507e883748314946dd0b01859 Mon Sep 17 00:00:00 2001 From: Seena Fallah Date: Mon, 19 Feb 2024 22:43:43 +0100 Subject: [PATCH 2/2] PublicAccessBlock: test access deny via bucket policy Make sure 403 is returned when access is denied via s3:GetBucketPublicAccessBlock action on GetBucketPublicAccessBlock Refs: https://github.com/ceph/ceph/pull/55652 Signed-off-by: Seena Fallah --- s3tests_boto3/functional/policy.py | 4 ++-- s3tests_boto3/functional/test_s3.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/s3tests_boto3/functional/policy.py b/s3tests_boto3/functional/policy.py index aae5454..123496a 100644 --- a/s3tests_boto3/functional/policy.py +++ b/s3tests_boto3/functional/policy.py @@ -37,10 +37,10 @@ class Policy(object): return json.dumps(policy_dict) -def make_json_policy(action, resource, principal={"AWS": "*"}, conditions=None): +def make_json_policy(action, resource, principal={"AWS": "*"}, effect="Allow", conditions=None): """ Helper function to make single statement policies """ - s = Statement(action, resource, principal, condition=conditions) + s = Statement(action, resource, principal, effect=effect, condition=conditions) p = Policy() return p.add_statement(s).to_json() diff --git a/s3tests_boto3/functional/test_s3.py b/s3tests_boto3/functional/test_s3.py index c8a0933..83f130a 100644 --- a/s3tests_boto3/functional/test_s3.py +++ b/s3tests_boto3/functional/test_s3.py @@ -12607,6 +12607,34 @@ def test_get_undefined_public_block(): assert response_code == 'NoSuchPublicAccessBlockConfiguration' +def test_get_public_block_deny_bucket_policy(): + bucket_name = get_new_bucket() + client = get_client() + + access_conf = {'BlockPublicAcls': True, + 'IgnorePublicAcls': True, + 'BlockPublicPolicy': True, + 'RestrictPublicBuckets': False} + client.put_public_access_block(Bucket=bucket_name, PublicAccessBlockConfiguration=access_conf) + + # make sure we can get the public access block + resp = client.get_public_access_block(Bucket=bucket_name) + assert resp['PublicAccessBlockConfiguration']['BlockPublicAcls'] == access_conf['BlockPublicAcls'] + assert resp['PublicAccessBlockConfiguration']['BlockPublicPolicy'] == access_conf['BlockPublicPolicy'] + assert resp['PublicAccessBlockConfiguration']['IgnorePublicAcls'] == access_conf['IgnorePublicAcls'] + assert resp['PublicAccessBlockConfiguration']['RestrictPublicBuckets'] == access_conf['RestrictPublicBuckets'] + + # make bucket policy to deny access + resource = _make_arn_resource(bucket_name) + policy_document = make_json_policy("s3:GetBucketPublicAccessBlock", + resource, effect="Deny") + client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document) + + # check if the access is denied + e = assert_raises(ClientError, client.get_public_access_block, Bucket=bucket_name) + status, error_code = _get_status_and_error_code(e.response) + assert status == 403 + def test_put_public_block(): #client = get_svc_client(svc='s3control', client_config=Config(s3={'addressing_style': 'path'})) bucket_name = get_new_bucket()