From 52df393b5ae688eba021b2fd579c46d3e113dd7b Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Tue, 12 Oct 2021 17:50:36 +0000 Subject: [PATCH] S3 - Move ACL tests to dedicated file (#4397) --- tests/test_s3/test_s3.py | 288 +----------------------------- tests/test_s3/test_s3_acl.py | 328 +++++++++++++++++++++++++++++++++++ 2 files changed, 329 insertions(+), 287 deletions(-) create mode 100644 tests/test_s3/test_s3_acl.py diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py index d275bfe6b..f637dd1ef 100644 --- a/tests/test_s3/test_s3.py +++ b/tests/test_s3/test_s3.py @@ -2,7 +2,6 @@ from __future__ import unicode_literals import datetime -import sys import os from boto3 import Session from urllib.request import urlopen @@ -21,7 +20,6 @@ import boto3 from botocore.client import ClientError import botocore.exceptions from boto.exception import S3CreateError, S3ResponseError -from botocore.handlers import disable_signing from boto.s3.connection import S3Connection from boto.s3.key import Key from freezegun import freeze_time @@ -37,7 +35,6 @@ import sure # noqa from moto import settings, mock_s3, mock_s3_deprecated, mock_config import moto.s3.models as s3model from moto.core.exceptions import InvalidNextTokenException -from moto.core.utils import py2_strip_unicode_keys from moto.settings import get_s3_default_key_buffer_size, S3_UPLOAD_PART_MIN_SIZE if settings.TEST_SERVER_MODE: @@ -1887,31 +1884,6 @@ def test_acl_setting(): ), grants -@mock_s3 -def test_acl_setting_boto3(): - s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) - client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) - bucket = s3.Bucket("foobar") - bucket.create() - - content = b"imafile" - keyname = "test.txt" - bucket.put_object( - Key=keyname, Body=content, ContentType="text/plain", ACL="public-read" - ) - - grants = client.get_object_acl(Bucket="foobar", Key=keyname)["Grants"] - grants.should.contain( - { - "Grantee": { - "Type": "Group", - "URI": "http://acs.amazonaws.com/groups/global/AllUsers", - }, - "Permission": "READ", - } - ) - - # Has boto3 equivalent @mock_s3_deprecated def test_acl_setting_via_headers(): @@ -1941,30 +1913,6 @@ def test_acl_setting_via_headers(): ), grants -@mock_s3 -def test_acl_setting_via_headers_boto3(): - s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) - client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) - bucket = s3.Bucket("foobar") - bucket.create() - - keyname = "test.txt" - - bucket.put_object(Key=keyname, Body=b"imafile") - client.put_object_acl(ACL="public-read", Bucket="foobar", Key=keyname) - - grants = client.get_object_acl(Bucket="foobar", Key=keyname)["Grants"] - grants.should.contain( - { - "Grantee": { - "Type": "Group", - "URI": "http://acs.amazonaws.com/groups/global/AllUsers", - }, - "Permission": "READ", - } - ) - - # Has boto3 equivalent @mock_s3_deprecated def test_acl_switching(): @@ -1986,40 +1934,6 @@ def test_acl_switching(): ), grants -@mock_s3 -def test_acl_switching_boto3(): - s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) - client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) - bucket = s3.Bucket("foobar") - bucket.create() - keyname = "test.txt" - - bucket.put_object(Key=keyname, Body=b"asdf", ACL="public-read") - client.put_object_acl(ACL="private", Bucket="foobar", Key=keyname) - - grants = client.get_object_acl(Bucket="foobar", Key=keyname)["Grants"] - grants.shouldnt.contain( - { - "Grantee": { - "Type": "Group", - "URI": "http://acs.amazonaws.com/groups/global/AllUsers", - }, - "Permission": "READ", - } - ) - - -@mock_s3 -def test_acl_switching_nonexistent_key(): - s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) - s3.create_bucket(Bucket="mybucket") - - with pytest.raises(ClientError) as e: - s3.put_object_acl(Bucket="mybucket", Key="nonexistent", ACL="private") - - e.value.response["Error"]["Code"].should.equal("NoSuchKey") - - @mock_s3_deprecated def test_bucket_acl_setting(): conn = boto.connect_s3() @@ -2051,50 +1965,6 @@ def test_bucket_acl_switching(): ), grants -@mock_s3 -def test_s3_object_in_public_bucket(): - s3 = boto3.resource("s3") - bucket = s3.Bucket("test-bucket") - bucket.create( - ACL="public-read", CreateBucketConfiguration={"LocationConstraint": "us-west-1"} - ) - bucket.put_object(Body=b"ABCD", Key="file.txt") - - s3_anonymous = boto3.resource("s3") - s3_anonymous.meta.client.meta.events.register("choose-signer.s3.*", disable_signing) - - contents = ( - s3_anonymous.Object(key="file.txt", bucket_name="test-bucket") - .get()["Body"] - .read() - ) - contents.should.equal(b"ABCD") - - bucket.put_object(ACL="private", Body=b"ABCD", Key="file.txt") - - with pytest.raises(ClientError) as exc: - s3_anonymous.Object(key="file.txt", bucket_name="test-bucket").get() - exc.value.response["Error"]["Code"].should.equal("403") - - -@mock_s3 -def test_s3_object_in_public_bucket_using_multiple_presigned_urls(): - s3 = boto3.resource("s3") - bucket = s3.Bucket("test-bucket") - bucket.create( - ACL="public-read", CreateBucketConfiguration={"LocationConstraint": "us-west-1"} - ) - bucket.put_object(Body=b"ABCD", Key="file.txt") - - params = {"Bucket": "test-bucket", "Key": "file.txt"} - presigned_url = boto3.client("s3").generate_presigned_url( - "get_object", params, ExpiresIn=900 - ) - for i in range(1, 10): - response = requests.get(presigned_url) - assert response.status_code == 200, "Failed on req number {}".format(i) - - @mock_s3 def test_streaming_upload_from_file_to_presigned_url(): s3 = boto3.resource("s3", region_name="us-east-1") @@ -2200,31 +2070,6 @@ def test_default_key_buffer_size(): os.environ["MOTO_S3_DEFAULT_KEY_BUFFER_SIZE"] = original_default_key_buffer_size -@mock_s3 -def test_s3_object_in_private_bucket(): - s3 = boto3.resource("s3") - bucket = s3.Bucket("test-bucket") - bucket.create( - ACL="private", CreateBucketConfiguration={"LocationConstraint": "us-west-1"} - ) - bucket.put_object(ACL="private", Body=b"ABCD", Key="file.txt") - - s3_anonymous = boto3.resource("s3") - s3_anonymous.meta.client.meta.events.register("choose-signer.s3.*", disable_signing) - - with pytest.raises(ClientError) as exc: - s3_anonymous.Object(key="file.txt", bucket_name="test-bucket").get() - exc.value.response["Error"]["Code"].should.equal("403") - - bucket.put_object(ACL="public-read", Body=b"ABCD", Key="file.txt") - contents = ( - s3_anonymous.Object(key="file.txt", bucket_name="test-bucket") - .get()["Body"] - .read() - ) - contents.should.equal(b"ABCD") - - # Has boto3 equivalent @mock_s3_deprecated def test_unicode_key(): @@ -3392,7 +3237,7 @@ def test_boto3_s3_content_type(): content_type = "text/python-x" s3.Object(my_bucket.name, s3_path).put( - ContentType=content_type, Body=b"some python code", ACL="public-read" + ContentType=content_type, Body=b"some python code" ) s3.Object(my_bucket.name, s3_path).content_type.should.equal(content_type) @@ -4339,137 +4184,6 @@ def test_boto3_delete_bucket_cors(): e.response["Error"]["Message"].should.equal("The CORS configuration does not exist") -@mock_s3 -def test_put_bucket_acl_body(): - s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) - s3.create_bucket(Bucket="bucket") - bucket_owner = s3.get_bucket_acl(Bucket="bucket")["Owner"] - s3.put_bucket_acl( - Bucket="bucket", - AccessControlPolicy={ - "Grants": [ - { - "Grantee": { - "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", - "Type": "Group", - }, - "Permission": "WRITE", - }, - { - "Grantee": { - "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", - "Type": "Group", - }, - "Permission": "READ_ACP", - }, - ], - "Owner": bucket_owner, - }, - ) - - result = s3.get_bucket_acl(Bucket="bucket") - assert len(result["Grants"]) == 2 - for g in result["Grants"]: - assert g["Grantee"]["URI"] == "http://acs.amazonaws.com/groups/s3/LogDelivery" - assert g["Grantee"]["Type"] == "Group" - assert g["Permission"] in ["WRITE", "READ_ACP"] - - # With one: - s3.put_bucket_acl( - Bucket="bucket", - AccessControlPolicy={ - "Grants": [ - { - "Grantee": { - "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", - "Type": "Group", - }, - "Permission": "WRITE", - } - ], - "Owner": bucket_owner, - }, - ) - result = s3.get_bucket_acl(Bucket="bucket") - assert len(result["Grants"]) == 1 - - # With no owner: - with pytest.raises(ClientError) as err: - s3.put_bucket_acl( - Bucket="bucket", - AccessControlPolicy={ - "Grants": [ - { - "Grantee": { - "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", - "Type": "Group", - }, - "Permission": "WRITE", - } - ] - }, - ) - assert err.value.response["Error"]["Code"] == "MalformedACLError" - - # With incorrect permission: - with pytest.raises(ClientError) as err: - s3.put_bucket_acl( - Bucket="bucket", - AccessControlPolicy={ - "Grants": [ - { - "Grantee": { - "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", - "Type": "Group", - }, - "Permission": "lskjflkasdjflkdsjfalisdjflkdsjf", - } - ], - "Owner": bucket_owner, - }, - ) - assert err.value.response["Error"]["Code"] == "MalformedACLError" - - # Clear the ACLs: - result = s3.put_bucket_acl( - Bucket="bucket", AccessControlPolicy={"Grants": [], "Owner": bucket_owner} - ) - assert not result.get("Grants") - - -@mock_s3 -def test_object_acl_with_presigned_post(): - s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) - - bucket_name = "imageS3Bucket" - object_name = "text.txt" - fields = {"acl": "public-read"} - file = open("text.txt", "w") - file.write("test") - file.close() - - s3.create_bucket(Bucket=bucket_name) - response = s3.generate_presigned_post( - bucket_name, object_name, Fields=fields, ExpiresIn=60000 - ) - - with open(object_name, "rb") as f: - files = {"file": (object_name, f)} - requests.post(response["url"], data=response["fields"], files=files) - - response = s3.get_object_acl(Bucket=bucket_name, Key=object_name) - - assert "Grants" in response - assert len(response["Grants"]) == 2 - assert response["Grants"][1]["Permission"] == "READ" - - response = s3.get_object(Bucket=bucket_name, Key=object_name) - - assert "ETag" in response - assert "Body" in response - os.remove("text.txt") - - @mock_s3 def test_put_bucket_notification(): s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) diff --git a/tests/test_s3/test_s3_acl.py b/tests/test_s3/test_s3_acl.py new file mode 100644 index 000000000..78291fe75 --- /dev/null +++ b/tests/test_s3/test_s3_acl.py @@ -0,0 +1,328 @@ +import boto3 +import os +import pytest +import requests +import sure # noqa + +from botocore.exceptions import ClientError +from botocore.handlers import disable_signing +from moto import mock_s3 +from uuid import uuid4 + +DEFAULT_REGION_NAME = "us-east-1" + + +@pytest.mark.parametrize( + "type_key", [("CanonicalUser", "id", "ID"), ("Group", "url", "URI")] +) +@pytest.mark.parametrize( + "value", ["AllUsers", "http://acs.amazonaws.com/groups/global/AllUsers"] +) +@pytest.mark.parametrize("has_quotes", [True, False]) +@pytest.mark.parametrize("readwrite", ["Read", "Write"]) +@mock_s3 +def test_put_object_acl_using_grant(readwrite, type_key, value, has_quotes): + s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) + client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + bucket_name = str(uuid4()) + bucket = s3.Bucket(bucket_name) + bucket.create() + keyname = "test.txt" + + bucket.put_object(Key=keyname, Body=b"asdf") + _type, key, response_key = type_key + header_value = f'"{value}"' if has_quotes else value + args = { + "Bucket": bucket_name, + "Key": keyname, + f"Grant{readwrite}": f"{key}={header_value}", + } + client.put_object_acl(**args) + + grants = client.get_object_acl(Bucket=bucket_name, Key=keyname)["Grants"] + grants.should.have.length_of(1) + grants[0].should.have.key("Grantee").equal({"Type": _type, response_key: value}) + grants[0].should.have.key("Permission").equal(readwrite.upper()) + + +@mock_s3 +def test_acl_switching_boto3(): + s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) + client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + bucket = s3.Bucket("foobar") + bucket.create() + keyname = "test.txt" + + bucket.put_object(Key=keyname, Body=b"asdf", ACL="public-read") + client.put_object_acl(ACL="private", Bucket="foobar", Key=keyname) + + grants = client.get_object_acl(Bucket="foobar", Key=keyname)["Grants"] + grants.shouldnt.contain( + { + "Grantee": { + "Type": "Group", + "URI": "http://acs.amazonaws.com/groups/global/AllUsers", + }, + "Permission": "READ", + } + ) + + +@mock_s3 +def test_acl_switching_nonexistent_key(): + s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + s3.create_bucket(Bucket="mybucket") + + with pytest.raises(ClientError) as e: + s3.put_object_acl(Bucket="mybucket", Key="nonexistent", ACL="private") + + e.value.response["Error"]["Code"].should.equal("NoSuchKey") + + +@mock_s3 +def test_s3_object_in_public_bucket(): + s3 = boto3.resource("s3") + bucket = s3.Bucket("test-bucket") + bucket.create( + ACL="public-read", CreateBucketConfiguration={"LocationConstraint": "us-west-1"} + ) + bucket.put_object(Body=b"ABCD", Key="file.txt") + + s3_anonymous = boto3.resource("s3") + s3_anonymous.meta.client.meta.events.register("choose-signer.s3.*", disable_signing) + + contents = ( + s3_anonymous.Object(key="file.txt", bucket_name="test-bucket") + .get()["Body"] + .read() + ) + contents.should.equal(b"ABCD") + + bucket.put_object(ACL="private", Body=b"ABCD", Key="file.txt") + + with pytest.raises(ClientError) as exc: + s3_anonymous.Object(key="file.txt", bucket_name="test-bucket").get() + exc.value.response["Error"]["Code"].should.equal("403") + + +@mock_s3 +def test_s3_object_in_public_bucket_using_multiple_presigned_urls(): + s3 = boto3.resource("s3") + bucket = s3.Bucket("test-bucket") + bucket.create( + ACL="public-read", CreateBucketConfiguration={"LocationConstraint": "us-west-1"} + ) + bucket.put_object(Body=b"ABCD", Key="file.txt") + + params = {"Bucket": "test-bucket", "Key": "file.txt"} + presigned_url = boto3.client("s3").generate_presigned_url( + "get_object", params, ExpiresIn=900 + ) + for i in range(1, 10): + response = requests.get(presigned_url) + assert response.status_code == 200, "Failed on req number {}".format(i) + + +@mock_s3 +def test_s3_object_in_private_bucket(): + s3 = boto3.resource("s3") + bucket = s3.Bucket("test-bucket") + bucket.create( + ACL="private", CreateBucketConfiguration={"LocationConstraint": "us-west-1"} + ) + bucket.put_object(ACL="private", Body=b"ABCD", Key="file.txt") + + s3_anonymous = boto3.resource("s3") + s3_anonymous.meta.client.meta.events.register("choose-signer.s3.*", disable_signing) + + with pytest.raises(ClientError) as exc: + s3_anonymous.Object(key="file.txt", bucket_name="test-bucket").get() + exc.value.response["Error"]["Code"].should.equal("403") + + bucket.put_object(ACL="public-read", Body=b"ABCD", Key="file.txt") + contents = ( + s3_anonymous.Object(key="file.txt", bucket_name="test-bucket") + .get()["Body"] + .read() + ) + contents.should.equal(b"ABCD") + + +@mock_s3 +def test_put_bucket_acl_body(): + s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + s3.create_bucket(Bucket="bucket") + bucket_owner = s3.get_bucket_acl(Bucket="bucket")["Owner"] + s3.put_bucket_acl( + Bucket="bucket", + AccessControlPolicy={ + "Grants": [ + { + "Grantee": { + "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", + "Type": "Group", + }, + "Permission": "WRITE", + }, + { + "Grantee": { + "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", + "Type": "Group", + }, + "Permission": "READ_ACP", + }, + ], + "Owner": bucket_owner, + }, + ) + + result = s3.get_bucket_acl(Bucket="bucket") + assert len(result["Grants"]) == 2 + for g in result["Grants"]: + assert g["Grantee"]["URI"] == "http://acs.amazonaws.com/groups/s3/LogDelivery" + assert g["Grantee"]["Type"] == "Group" + assert g["Permission"] in ["WRITE", "READ_ACP"] + + # With one: + s3.put_bucket_acl( + Bucket="bucket", + AccessControlPolicy={ + "Grants": [ + { + "Grantee": { + "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", + "Type": "Group", + }, + "Permission": "WRITE", + } + ], + "Owner": bucket_owner, + }, + ) + result = s3.get_bucket_acl(Bucket="bucket") + assert len(result["Grants"]) == 1 + + # With no owner: + with pytest.raises(ClientError) as err: + s3.put_bucket_acl( + Bucket="bucket", + AccessControlPolicy={ + "Grants": [ + { + "Grantee": { + "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", + "Type": "Group", + }, + "Permission": "WRITE", + } + ] + }, + ) + assert err.value.response["Error"]["Code"] == "MalformedACLError" + + # With incorrect permission: + with pytest.raises(ClientError) as err: + s3.put_bucket_acl( + Bucket="bucket", + AccessControlPolicy={ + "Grants": [ + { + "Grantee": { + "URI": "http://acs.amazonaws.com/groups/s3/LogDelivery", + "Type": "Group", + }, + "Permission": "lskjflkasdjflkdsjfalisdjflkdsjf", + } + ], + "Owner": bucket_owner, + }, + ) + assert err.value.response["Error"]["Code"] == "MalformedACLError" + + # Clear the ACLs: + result = s3.put_bucket_acl( + Bucket="bucket", AccessControlPolicy={"Grants": [], "Owner": bucket_owner} + ) + assert not result.get("Grants") + + +@mock_s3 +def test_object_acl_with_presigned_post(): + s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + + bucket_name = "imageS3Bucket" + object_name = "text.txt" + fields = {"acl": "public-read"} + file = open("text.txt", "w") + file.write("test") + file.close() + + s3.create_bucket(Bucket=bucket_name) + response = s3.generate_presigned_post( + bucket_name, object_name, Fields=fields, ExpiresIn=60000 + ) + + with open(object_name, "rb") as f: + files = {"file": (object_name, f)} + requests.post(response["url"], data=response["fields"], files=files) + + response = s3.get_object_acl(Bucket=bucket_name, Key=object_name) + + assert "Grants" in response + assert len(response["Grants"]) == 2 + assert response["Grants"][1]["Permission"] == "READ" + + response = s3.get_object(Bucket=bucket_name, Key=object_name) + + assert "ETag" in response + assert "Body" in response + os.remove("text.txt") + + +@mock_s3 +def test_acl_setting_boto3(): + s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) + client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + bucket = s3.Bucket("foobar") + bucket.create() + + content = b"imafile" + keyname = "test.txt" + bucket.put_object( + Key=keyname, Body=content, ContentType="text/plain", ACL="public-read" + ) + + grants = client.get_object_acl(Bucket="foobar", Key=keyname)["Grants"] + grants.should.contain( + { + "Grantee": { + "Type": "Group", + "URI": "http://acs.amazonaws.com/groups/global/AllUsers", + }, + "Permission": "READ", + } + ) + + +@mock_s3 +def test_acl_setting_via_headers_boto3(): + s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) + client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + bucket = s3.Bucket("foobar") + bucket.create() + + keyname = "test.txt" + + bucket.put_object(Key=keyname, Body=b"imafile") + client.put_object_acl(ACL="public-read", Bucket="foobar", Key=keyname) + + grants = client.get_object_acl(Bucket="foobar", Key=keyname)["Grants"] + grants.should.contain( + { + "Grantee": { + "Type": "Group", + "URI": "http://acs.amazonaws.com/groups/global/AllUsers", + }, + "Permission": "READ", + } + )