From f8120b10f73db8a55bcdda831e8d72bc242f1109 Mon Sep 17 00:00:00 2001 From: Denis Kirillov Date: Tue, 8 Oct 2024 12:23:04 +0300 Subject: [PATCH] [#XX] Add test_frostfs.py Signed-off-by: Denis Kirillov --- s3tests_boto3/functional/test_frostfs.py | 177 +++++++++++++++++++++++ 1 file changed, 177 insertions(+) create mode 100644 s3tests_boto3/functional/test_frostfs.py diff --git a/s3tests_boto3/functional/test_frostfs.py b/s3tests_boto3/functional/test_frostfs.py new file mode 100644 index 0000000..6d7bf53 --- /dev/null +++ b/s3tests_boto3/functional/test_frostfs.py @@ -0,0 +1,177 @@ +from botocore.exceptions import ClientError +import json +import time + +from .utils import assert_raises +from .utils import _get_status_and_error_code + +from . import ( + configfile, setup_teardown, # we need this to parse config + get_client, + get_unauthenticated_client, + get_new_bucket, +) + + +def test_bucket_policy_frostfs_deny(): + bucket_name = get_new_bucket() + client = get_client() + key = 'tmp' + + resource1 = "arn:aws:s3:::" + bucket_name + resource2 = "arn:aws:s3:::" + bucket_name + "/*" + policy_document = json.dumps( + { + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Deny", + "Principal": "*", + "Action": "s3:PutObject", + "Resource": [ + "{}".format(resource1), + "{}".format(resource2) + ], + "Condition": { + "StringNotEquals": { + "s3:RequestObjectTag/environment": "production" + } + } + }] + } + ) + client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document) + + # TEST 7 + client.put_object(Bucket=bucket_name, Key=key, Tagging='environment=production') + + # TEST 8 + e = assert_raises(ClientError, client.put_object, Bucket=bucket_name, Key=key, Tagging='environment=development') + status, error_code = _get_status_and_error_code(e.response) + assert status == 403 + assert error_code == 'AccessDenied' + + # TEST 9 + e = assert_raises(ClientError, client.put_object, Bucket=bucket_name, Key=key) + status, error_code = _get_status_and_error_code(e.response) + assert status == 403 + assert error_code == 'AccessDenied' + + + + policy_document = json.dumps( + { + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Deny", + "Principal": "*", + "Action": "s3:PutObject", + "Resource": [ + "{}".format(resource1), + "{}".format(resource2) + ], + "Condition": { + "StringEquals": { + "s3:RequestObjectTag/environment": "production" + } + } + }] + } + ) + client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document) + + # TEST 10 + e = assert_raises(ClientError, client.put_object, Bucket=bucket_name, Key=key, Tagging='environment=production') + status, error_code = _get_status_and_error_code(e.response) + assert status == 403 + assert error_code == 'AccessDenied' + + # TEST 11 + client.put_object(Bucket=bucket_name, Key=key, Tagging='environment=development') + + # TEST 12 + client.put_object(Bucket=bucket_name, Key=key) + +def test_bucket_policy_frostfs_allow(): + bucket_name = get_new_bucket() + client = get_client() + key = 'tmp' + + client.put_object(Bucket=bucket_name, Key=key) + + resource1 = "arn:aws:s3:::" + bucket_name + resource2 = "arn:aws:s3:::" + bucket_name + "/*" + policy_document = json.dumps( + { + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Principal": "*", + "Action": "s3:PutObjectTagging", + "Resource": [ + "{}".format(resource1), + "{}".format(resource2) + ], + "Condition": { + "StringNotEquals": { + "s3:RequestObjectTag/environment": "production" + } + } + }] + } + ) + client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document) + time.sleep(3) + + alt_client = get_unauthenticated_client() + + # TEST 1 + e = assert_raises(ClientError, alt_client.put_object_tagging, Bucket=bucket_name, Key=key, Tagging={'TagSet':[{'Key':'environment','Value':'production'}]}) + status, error_code = _get_status_and_error_code(e.response) + assert status == 403 + assert error_code == 'AccessDenied' + + # TEST 2 + alt_client.put_object_tagging(Bucket=bucket_name, Key=key, Tagging={'TagSet':[{'Key':'environment','Value':'development'}]}) + + # TEST 3 + alt_client.put_object_tagging(Bucket=bucket_name, Key=key, Tagging={'TagSet':[]}) + + + + policy_document2 = json.dumps( + { + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Principal": "*", + "Action": "s3:PutObjectTagging", + "Resource": [ + "{}".format(resource1), + "{}".format(resource2) + ], + "Condition": { + "StringEquals": { + "s3:RequestObjectTag/environment": "production" + } + } + }] + } + ) + client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document2) + time.sleep(15) # probably we can reduce this + + # TEST 4 + alt_client.put_object_tagging(Bucket=bucket_name, Key=key, Tagging={'TagSet':[{'Key':'environment','Value':'production'}]}) + + # TEST 5 + e = assert_raises(ClientError, alt_client.put_object_tagging, Bucket=bucket_name, Key=key, Tagging={'TagSet':[{'Key':'environment','Value':'development'}]}) + status, error_code = _get_status_and_error_code(e.response) + assert status == 403 + assert error_code == 'AccessDenied' + + # TEST 6 + e = assert_raises(ClientError, alt_client.put_object_tagging, Bucket=bucket_name, Key=key, Tagging={'TagSet':[]}) + status, error_code = _get_status_and_error_code(e.response) + assert status == 403 + assert error_code == 'AccessDenied' +