S3: Implement cross-account access

localstack/constants.py

@@ -165,6 +165,7 @@
 # additional credentials used in the test suite (mainly for cross-account access)
 SECONDARY_TEST_AWS_ACCESS_KEY_ID = "000000000002"
 SECONDARY_TEST_AWS_SECRET_ACCESS_KEY = "test2"
+SECONDARY_TEST_AWS_REGION_NAME = "ap-southeast-1"
 
 # credentials being used for internal calls
 INTERNAL_AWS_ACCESS_KEY_ID = "__internal_call__"

tests/integration/s3/test_s3.py

@@ -33,6 +33,9 @@
 from localstack.constants import (
     LOCALHOST_HOSTNAME,
     S3_VIRTUAL_HOSTNAME,
+    SECONDARY_TEST_AWS_ACCESS_KEY_ID,
+    SECONDARY_TEST_AWS_REGION_NAME,
+    SECONDARY_TEST_AWS_SECRET_ACCESS_KEY,
     TEST_AWS_ACCESS_KEY_ID,
     TEST_AWS_SECRET_ACCESS_KEY,
 )
@@ -4725,6 +4728,70 @@ def test_s3_intelligent_tier_config(self, aws_client, s3_create_bucket, snapshot
         snapshot.match("list_bucket_intelligent_tiering_configurations_3", response)
 
 
+class TestS3MultiAccounts:
+    @pytest.fixture
+    def primary_client(self, aws_client):
+        return aws_client.s3
+
+    @pytest.fixture
+    def secondary_client(self, aws_client_factory):
+        """
+        Create a boto client with secondary test credentials and region.
+        """
+        return aws_client_factory.get_client(
+            "s3",
+            aws_access_key_id=SECONDARY_TEST_AWS_ACCESS_KEY_ID,
+            aws_secret_access_key=SECONDARY_TEST_AWS_SECRET_ACCESS_KEY,
+            region_name=SECONDARY_TEST_AWS_REGION_NAME,
+        )
+
+    def test_shared_bucket_namespace(self, primary_client, secondary_client):
+        # Ensure that the bucket name space is shared by all accounts and regions
+        primary_client.create_bucket(Bucket="foo")
+
+        with pytest.raises(ClientError) as exc:
+            secondary_client.create_bucket(
+                Bucket="foo",
+                CreateBucketConfiguration={"LocationConstraint": SECONDARY_TEST_AWS_REGION_NAME},
+            )
+        exc.match("BucketAlreadyExists")
+
+    def test_cross_account_access(self, primary_client, secondary_client):
+        # Ensure that following operations can be performed across accounts
+        # - ListObjects
+        # - PutObject
+        # - GetObject
+
+        bucket_name = "foo"
+        key_name = "lorem/ipsum"
+        body1 = b"zaphod beeblebrox"
+        body2 = b"42"
+
+        # First user creates a bucket and puts an object
+        primary_client.create_bucket(Bucket=bucket_name)
+        response = primary_client.list_buckets()
+        assert bucket_name in [bucket["Name"] for bucket in response["Buckets"]]
+        primary_client.put_object(Bucket=bucket_name, Key=key_name, Body=body1)
+
+        # Second user must not see this bucket in their `ListBuckets` response
+        response = secondary_client.list_buckets()
+        assert bucket_name not in [bucket["Name"] for bucket in response["Buckets"]]
+
+        # Yet they should be able to `ListObjects` in that bucket
+        response = secondary_client.list_objects(Bucket=bucket_name)
+        assert key_name in [key["Key"] for key in response["Contents"]]
+
+        # Along with `GetObject` and `PutObject`
+        # ACL and permission enforcement is currently not implemented
+        response = secondary_client.get_object(Bucket=bucket_name, Key=key_name)
+        assert response["Body"].read() == body1
+        assert secondary_client.put_object(Bucket=bucket_name, Key=key_name, Body=body2)
+
+        # The modified object must be reflected for the first user
+        response = primary_client.get_object(Bucket=bucket_name, Key=key_name)
+        assert response["Body"].read() == body2
+
+
 class TestS3TerraformRawRequests:
     @pytest.mark.only_localstack
     def test_terraform_request_sequence(self, aws_client):