diff --git a/moto/s3/models.py b/moto/s3/models.py index 7b6237543..34cf1824d 100644 --- a/moto/s3/models.py +++ b/moto/s3/models.py @@ -57,6 +57,7 @@ from .utils import _VersionedKeyStore, CaseInsensitiveDict from .utils import ARCHIVE_STORAGE_CLASSES, STORAGE_CLASS from ..events.notifications import send_notification as events_send_notification from ..settings import get_s3_default_key_buffer_size, S3_UPLOAD_PART_MIN_SIZE +from ..settings import s3_allow_crossdomain_access MAX_BUCKET_NAME_LENGTH = 63 MIN_BUCKET_NAME_LENGTH = 3 @@ -1518,7 +1519,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider): Note that this only works if the environment variable is set **before** the mock is initialized. - ------------------------------------ + _-_-_-_ When using the MultiPart-API manually, the minimum part size is 5MB, just as with AWS. Use the following environment variable to lower this: @@ -1526,7 +1527,15 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider): S3_UPLOAD_PART_MIN_SIZE=256 - ------------------------------------ + _-_-_-_ + + CrossAccount access is allowed by default. If you want Moto to throw an AccessDenied-error when accessing a bucket in another account, use this environment variable: + + .. sourcecode:: bash + + MOTO_S3_ALLOW_CROSSACCOUNT_ACCESS=false + + _-_-_-_ Install `moto[s3crc32c]` if you use the CRC32C algorithm, and absolutely need the correct value. Alternatively, you can install the `crc32c` dependency manually. @@ -1711,6 +1720,8 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider): return self.buckets[bucket_name] if bucket_name in s3_backends.bucket_accounts: + if not s3_allow_crossdomain_access(): + raise AccessDeniedByLock account_id = s3_backends.bucket_accounts[bucket_name] return s3_backends[account_id]["global"].get_bucket(bucket_name) diff --git a/moto/settings.py b/moto/settings.py index 2c7974879..e332db7f9 100644 --- a/moto/settings.py +++ b/moto/settings.py @@ -65,6 +65,10 @@ def get_s3_default_key_buffer_size() -> int: ) +def s3_allow_crossdomain_access() -> bool: + return os.environ.get("MOTO_S3_ALLOW_CROSSACCOUNT_ACCESS", "true").lower() == "true" + + def ecs_new_arn_format() -> bool: # True by default - only the value 'false' will return false return os.environ.get("MOTO_ECS_NEW_ARN", "true").lower() != "false" diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py index 33cc75797..962ea8602 100644 --- a/tests/test_s3/test_s3.py +++ b/tests/test_s3/test_s3.py @@ -4,7 +4,7 @@ from io import BytesIO import json import os import pickle -from unittest import SkipTest, mock +from unittest import SkipTest from urllib.parse import urlparse, parse_qs import uuid from uuid import uuid4 @@ -3384,46 +3384,3 @@ def test_checksum_response(algorithm): ChecksumAlgorithm=algorithm, ) assert f"Checksum{algorithm}" in response - - -@mock_s3 -def test_cross_account_region_access(): - if settings.TEST_SERVER_MODE: - raise SkipTest("Multi-accounts env config only works serverside") - - client1 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) - client2 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) - - account2 = "222222222222" - bucket_name = "cross-account-bucket" - key = "test-key" - - # Create a bucket in the default account - client1.create_bucket(Bucket=bucket_name) - client1.put_object(Bucket=bucket_name, Key=key, Body=b"data") - - with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}): - # Ensure the bucket can be retrieved from another account - response = client2.list_objects(Bucket=bucket_name) - assert len(response["Contents"]) == 1 - assert response["Contents"][0]["Key"] == key - - assert client2.get_object(Bucket=bucket_name, Key=key) - - assert client2.put_object(Bucket=bucket_name, Key=key, Body=b"kaytranada") - - # Ensure bucket namespace is shared across accounts - with pytest.raises(ClientError) as exc: - client2.create_bucket(Bucket=bucket_name) - assert exc.value.response["Error"]["Code"] == "BucketAlreadyExists" - assert exc.value.response["Error"]["Message"] == ( - "The requested bucket name is not available. The bucket " - "namespace is shared by all users of the system. Please " - "select a different name and try again" - ) - - # Ensure bucket name can be reused if it is deleted - client1.delete_object(Bucket=bucket_name, Key=key) - client1.delete_bucket(Bucket=bucket_name) - with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}): - assert client2.create_bucket(Bucket=bucket_name) diff --git a/tests/test_s3/test_s3_cross_account.py b/tests/test_s3/test_s3_cross_account.py new file mode 100644 index 000000000..52cf63edb --- /dev/null +++ b/tests/test_s3/test_s3_cross_account.py @@ -0,0 +1,60 @@ +import os +from unittest import SkipTest, mock + +import boto3 +from botocore.client import ClientError +import pytest + +from moto import settings, mock_s3 +from moto.s3.responses import DEFAULT_REGION_NAME + + +@mock_s3 +def test_cross_account_region_access(): + if settings.TEST_SERVER_MODE: + raise SkipTest("Multi-accounts env config only works serverside") + + client1 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + client2 = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + + account2 = "222222222222" + bucket_name = "cross-account-bucket" + key = "test-key" + + # Create a bucket in the default account + client1.create_bucket(Bucket=bucket_name) + client1.put_object(Bucket=bucket_name, Key=key, Body=b"data") + + with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}): + # Ensure the bucket can be retrieved from another account + response = client2.list_objects(Bucket=bucket_name) + assert len(response["Contents"]) == 1 + assert response["Contents"][0]["Key"] == key + + assert client2.get_object(Bucket=bucket_name, Key=key) + + assert client2.put_object(Bucket=bucket_name, Key=key, Body=b"kaytranada") + + # Ensure bucket namespace is shared across accounts + with pytest.raises(ClientError) as exc: + client2.create_bucket(Bucket=bucket_name) + assert exc.value.response["Error"]["Code"] == "BucketAlreadyExists" + assert exc.value.response["Error"]["Message"] == ( + "The requested bucket name is not available. The bucket " + "namespace is shared by all users of the system. Please " + "select a different name and try again" + ) + + with mock.patch.dict( + os.environ, {"MOTO_S3_ALLOW_CROSSACCOUNT_ACCESS": "false"} + ): + with pytest.raises(ClientError) as ex: + client2.list_objects(Bucket=bucket_name) + assert ex.value.response["Error"]["Code"] == "AccessDenied" + assert ex.value.response["Error"]["Message"] == "Access Denied" + + # Ensure bucket name can be reused if it is deleted + client1.delete_object(Bucket=bucket_name, Key=key) + client1.delete_bucket(Bucket=bucket_name) + with mock.patch.dict(os.environ, {"MOTO_ACCOUNT_ID": account2}): + assert client2.create_bucket(Bucket=bucket_name)