feat(aws): add s3_bucket_shadow_resource_vulnerability check (#8398)

Co-authored-by: MrCloudSec <hello@mistercloudsec.com>

tests/providers/aws/services/s3/s3_bucket_shadow_resource_vulnerability/s3_bucket_shadow_resource_vulnerability_test.py

@@ -0,0 +1,210 @@
+from unittest import mock
+
+from moto import mock_aws
+
+from prowler.providers.aws.services.s3.s3_service import Bucket
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+class Test_s3_bucket_shadow_resource_vulnerability:
+    @mock_aws
+    def test_no_buckets(self):
+        s3_client = mock.MagicMock
+        s3_client.buckets = {}
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        aws_provider.identity.identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
+        s3_client = mock.MagicMock
+        s3_client.provider = aws_provider
+        s3_client._head_bucket = mock.MagicMock(return_value=False)
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability.s3_client",
+                new=s3_client,
+            ),
+        ):
+            from prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability import (
+                s3_bucket_shadow_resource_vulnerability,
+            )
+
+            check = s3_bucket_shadow_resource_vulnerability()
+            result = check.execute()
+            assert len(result) == 0
+
+    @mock_aws
+    def test_bucket_owned_by_account(self):
+        s3_client = mock.MagicMock
+        bucket_name = f"sagemaker-{AWS_REGION_US_EAST_1}-{AWS_ACCOUNT_NUMBER}"
+        s3_client.audited_account_id = AWS_ACCOUNT_NUMBER
+        s3_client.audited_identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
+        s3_client.audited_canonical_id = AWS_ACCOUNT_NUMBER
+        s3_client.buckets = {
+            bucket_name: Bucket(
+                name=bucket_name,
+                arn=f"arn:aws:s3:::{bucket_name}",
+                region=AWS_REGION_US_EAST_1,
+                owner_id=AWS_ACCOUNT_NUMBER,
+            )
+        }
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        aws_provider.identity.identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
+        s3_client = mock.MagicMock
+        s3_client.provider = aws_provider
+        s3_client._head_bucket = mock.MagicMock(return_value=False)
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability.s3_client",
+                new=s3_client,
+            ),
+        ):
+            from prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability import (
+                s3_bucket_shadow_resource_vulnerability,
+            )
+
+            check = s3_bucket_shadow_resource_vulnerability()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                "is correctly owned by the audited account" in result[0].status_extended
+            )
+
+    @mock_aws
+    def test_bucket_not_predictable(self):
+        s3_client = mock.MagicMock
+        bucket_name = "my-non-predictable-bucket"
+        s3_client.audited_account_id = AWS_ACCOUNT_NUMBER
+        s3_client.audited_identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
+        s3_client.audited_canonical_id = AWS_ACCOUNT_NUMBER
+        s3_client.buckets = {
+            bucket_name: Bucket(
+                name=bucket_name,
+                arn=f"arn:aws:s3:::{bucket_name}",
+                region=AWS_REGION_US_EAST_1,
+                owner_id=AWS_ACCOUNT_NUMBER,
+            )
+        }
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        aws_provider.identity.identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
+        s3_client = mock.MagicMock
+        s3_client.provider = aws_provider
+        s3_client._head_bucket = mock.MagicMock(return_value=False)
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability.s3_client",
+                new=s3_client,
+            ),
+        ):
+            from prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability import (
+                s3_bucket_shadow_resource_vulnerability,
+            )
+
+            check = s3_bucket_shadow_resource_vulnerability()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert "is not a known shadow resource" in result[0].status_extended
+
+    @mock_aws
+    def test_shadow_resource_in_other_account(self):
+        # Mock S3 client with no buckets in current account
+        s3_client = mock.MagicMock()
+        s3_client.buckets = {}
+        s3_client.audited_account_id = AWS_ACCOUNT_NUMBER
+        s3_client.audited_identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
+        s3_client.audited_canonical_id = AWS_ACCOUNT_NUMBER
+        s3_client.audited_partition = "aws"
+
+        # Mock regional clients - this is what the check uses to determine regions to test
+        s3_client.regional_clients = {
+            "us-east-1": mock.MagicMock(),
+            "us-west-2": mock.MagicMock(),
+            "eu-west-1": mock.MagicMock(),
+        }
+
+        # Define the shadow resources we want to simulate
+        shadow_resources = [
+            f"aws-glue-assets-{AWS_ACCOUNT_NUMBER}-us-west-2",
+            f"sagemaker-us-east-1-{AWS_ACCOUNT_NUMBER}",
+            f"aws-emr-studio-{AWS_ACCOUNT_NUMBER}-eu-west-1",
+        ]
+
+        # Mock the _head_bucket method to simulate finding shadow resources
+        def mock_head_bucket(bucket_name):
+            return bucket_name in shadow_resources
+
+        s3_client._head_bucket = mock_head_bucket
+
+        # Mock provider with multiple regions to test
+        aws_provider = set_mocked_aws_provider(["us-east-1", "us-west-2", "eu-west-1"])
+        aws_provider.identity.identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
+        aws_provider.identity.account = AWS_ACCOUNT_NUMBER
+        s3_client.provider = aws_provider
+
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability.s3_client",
+                new=s3_client,
+            ),
+        ):
+            from prowler.providers.aws.services.s3.s3_bucket_shadow_resource_vulnerability.s3_bucket_shadow_resource_vulnerability import (
+                s3_bucket_shadow_resource_vulnerability,
+            )
+
+            check = s3_bucket_shadow_resource_vulnerability()
+            result = check.execute()
+
+            # Should find shadow resources
+            assert len(result) >= 3
+
+            # Check if we found all expected shadow resources
+            found_services = set()
+            for finding in result:
+                if (
+                    finding.status == "FAIL"
+                    and "shadow resource" in finding.status_extended
+                ):
+                    if (
+                        "aws-glue-assets" in finding.status_extended
+                        and "Glue" in finding.status_extended
+                    ):
+                        found_services.add("Glue")
+                        assert "us-west-2" in finding.status_extended
+                    elif (
+                        "sagemaker" in finding.status_extended
+                        and "SageMaker" in finding.status_extended
+                    ):
+                        found_services.add("SageMaker")
+                        assert "us-east-1" in finding.status_extended
+                    elif (
+                        "aws-emr-studio" in finding.status_extended
+                        and "EMR" in finding.status_extended
+                    ):
+                        found_services.add("EMR")
+                        assert "eu-west-1" in finding.status_extended
+
+                    # Verify common attributes
+                    assert "owned by another account" in finding.status_extended
+
+            # Verify we found all expected services
+            expected_services = {"Glue", "SageMaker", "EMR"}
+            assert found_services == expected_services