feat(eventbridge): add EventBridge checks (#4020)

tests/providers/aws/services/eventbridge/eventbridge_bus_cross_account_access/eventbridge_bus_cross_account_access_test.py

@@ -0,0 +1,170 @@
+from unittest import mock
+from uuid import uuid4
+
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    set_mocked_aws_provider,
+)
+
+test_bus_name = str(uuid4())
+test_bus_arn = f"arn:aws:eventbridge:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/{test_bus_name}"
+
+
+class Test_eventbridge_bus_cross_account_access:
+    @mock_aws
+    def test_default_bus(self):
+        from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+            EventBridge,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.eventbridge.eventbridge_bus_cross_account_access.eventbridge_bus_cross_account_access.eventbridge_client",
+            new=EventBridge(aws_provider),
+        ):
+            from prowler.providers.aws.services.eventbridge.eventbridge_bus_cross_account_access.eventbridge_bus_cross_account_access import (
+                eventbridge_bus_cross_account_access,
+            )
+
+            check = eventbridge_bus_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                "EventBridge event bus default does not allow cross-account access."
+            )
+            assert result[0].resource_id == "default"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:events:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/default"
+            )
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    @mock_aws
+    def test_buses_self_account(self):
+        from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+            EventBridge,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        events_client = client("events", region_name=AWS_REGION_EU_WEST_1)
+        events_client.put_permission(
+            EventBusName="default",
+            Action="events:PutEvents",
+            Principal=AWS_ACCOUNT_NUMBER,
+            StatementId="test-statement",
+        )
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.eventbridge.eventbridge_bus_cross_account_access.eventbridge_bus_cross_account_access.eventbridge_client",
+            new=EventBridge(aws_provider),
+        ):
+            from prowler.providers.aws.services.eventbridge.eventbridge_bus_cross_account_access.eventbridge_bus_cross_account_access import (
+                eventbridge_bus_cross_account_access,
+            )
+
+            check = eventbridge_bus_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                "EventBridge event bus default does not allow cross-account access."
+            )
+            assert result[0].resource_id == "default"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:events:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/default"
+            )
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    @mock_aws
+    def test_buses_other_account(self):
+        from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+            EventBridge,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        events_client = client("events", region_name=AWS_REGION_EU_WEST_1)
+        events_client.put_permission(
+            EventBusName="default",
+            Action="events:PutEvents",
+            Principal="111122223333",
+            StatementId="test-statement",
+        )
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.eventbridge.eventbridge_bus_cross_account_access.eventbridge_bus_cross_account_access.eventbridge_client",
+            new=EventBridge(aws_provider),
+        ):
+            from prowler.providers.aws.services.eventbridge.eventbridge_bus_cross_account_access.eventbridge_bus_cross_account_access import (
+                eventbridge_bus_cross_account_access,
+            )
+
+            check = eventbridge_bus_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == (
+                "EventBridge event bus default allows cross-account access."
+            )
+            assert result[0].resource_id == "default"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:events:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/default"
+            )
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    @mock_aws
+    def test_buses_asterisk_principal(self):
+        from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+            EventBridge,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        events_client = client("events", region_name=AWS_REGION_EU_WEST_1)
+        events_client.put_permission(
+            EventBusName="default",
+            Action="events:PutEvents",
+            Principal="*",
+            StatementId="test-statement",
+        )
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.eventbridge.eventbridge_bus_cross_account_access.eventbridge_bus_cross_account_access.eventbridge_client",
+            new=EventBridge(aws_provider),
+        ):
+            from prowler.providers.aws.services.eventbridge.eventbridge_bus_cross_account_access.eventbridge_bus_cross_account_access import (
+                eventbridge_bus_cross_account_access,
+            )
+
+            check = eventbridge_bus_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == (
+                "EventBridge event bus default allows cross-account access."
+            )
+            assert result[0].resource_id == "default"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:events:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/default"
+            )
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1

tests/providers/aws/services/eventbridge/eventbridge_bus_exposed/eventbridge_bus_exposed_test.py

@@ -0,0 +1,170 @@
+from unittest import mock
+from uuid import uuid4
+
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    set_mocked_aws_provider,
+)
+
+test_bus_name = str(uuid4())
+test_bus_arn = f"arn:aws:eventbridge:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/{test_bus_name}"
+
+
+class Test_eventbridge_bus_exposed:
+    @mock_aws
+    def test_default_bus(self):
+        from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+            EventBridge,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.eventbridge.eventbridge_bus_exposed.eventbridge_bus_exposed.eventbridge_client",
+            new=EventBridge(aws_provider),
+        ):
+            from prowler.providers.aws.services.eventbridge.eventbridge_bus_exposed.eventbridge_bus_exposed import (
+                eventbridge_bus_exposed,
+            )
+
+            check = eventbridge_bus_exposed()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                "EventBridge event bus default is not exposed to everyone."
+            )
+            assert result[0].resource_id == "default"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:events:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/default"
+            )
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    @mock_aws
+    def test_buses_self_account(self):
+        from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+            EventBridge,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        events_client = client("events", region_name=AWS_REGION_EU_WEST_1)
+        events_client.put_permission(
+            EventBusName="default",
+            Action="events:PutEvents",
+            Principal=AWS_ACCOUNT_NUMBER,
+            StatementId="test-statement",
+        )
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.eventbridge.eventbridge_bus_exposed.eventbridge_bus_exposed.eventbridge_client",
+            new=EventBridge(aws_provider),
+        ):
+            from prowler.providers.aws.services.eventbridge.eventbridge_bus_exposed.eventbridge_bus_exposed import (
+                eventbridge_bus_exposed,
+            )
+
+            check = eventbridge_bus_exposed()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                "EventBridge event bus default is not exposed to everyone."
+            )
+            assert result[0].resource_id == "default"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:events:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/default"
+            )
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    @mock_aws
+    def test_buses_other_account(self):
+        from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+            EventBridge,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        events_client = client("events", region_name=AWS_REGION_EU_WEST_1)
+        events_client.put_permission(
+            EventBusName="default",
+            Action="events:PutEvents",
+            Principal="111122223333",
+            StatementId="test-statement",
+        )
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.eventbridge.eventbridge_bus_exposed.eventbridge_bus_exposed.eventbridge_client",
+            new=EventBridge(aws_provider),
+        ):
+            from prowler.providers.aws.services.eventbridge.eventbridge_bus_exposed.eventbridge_bus_exposed import (
+                eventbridge_bus_exposed,
+            )
+
+            check = eventbridge_bus_exposed()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                "EventBridge event bus default is not exposed to everyone."
+            )
+            assert result[0].resource_id == "default"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:events:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/default"
+            )
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    @mock_aws
+    def test_buses_asterisk_principal(self):
+        from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+            EventBridge,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        events_client = client("events", region_name=AWS_REGION_EU_WEST_1)
+        events_client.put_permission(
+            EventBusName="default",
+            Action="events:PutEvents",
+            Principal="*",
+            StatementId="test-statement",
+        )
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.eventbridge.eventbridge_bus_exposed.eventbridge_bus_exposed.eventbridge_client",
+            new=EventBridge(aws_provider),
+        ):
+            from prowler.providers.aws.services.eventbridge.eventbridge_bus_exposed.eventbridge_bus_exposed import (
+                eventbridge_bus_exposed,
+            )
+
+            check = eventbridge_bus_exposed()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == (
+                "EventBridge event bus default is exposed to everyone."
+            )
+            assert result[0].resource_id == "default"
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:events:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/default"
+            )
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1

tests/providers/aws/services/eventbridge/eventbridge_service_test.py

@@ -0,0 +1,171 @@
+from unittest.mock import patch
+
+import botocore
+from boto3 import client
+from moto import mock_aws
+
+from prowler.providers.aws.services.eventbridge.eventbridge_service import (
+    EventBridge,
+    Schema,
+)
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+# Mocking Calls
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwargs):
+    """We have to mock every AWS API call using Boto3"""
+    if operation_name == "ListRegistries":
+        return {
+            "Registries": [
+                {
+                    "RegistryArn": "arn:aws:schemas:us-east-1:123456789012:registry/test",
+                    "RegistryName": "test",
+                    "Tags": {"key1": "value1"},
+                },
+            ]
+        }
+    if operation_name == "GetResourcePolicy":
+        return {
+            "Policy": '{"Version":"2012-10-17","Statement":[{"Sid":"AllowReadWrite","Effect":"Allow","Principal":{"AWS":"arn:aws:iam::123456789012:root"},"Action":"schemas:*","Resource":"arn:aws:schemas:eu-west-1:123456789012:registry/test"}]}',
+            "RevisionId": "1",
+        }
+
+    return make_api_call(self, operation_name, kwargs)
+
+
+@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+class Test_EventBridge_Service:
+    # Test EventBridge Service
+    @mock_aws
+    def test_service(self):
+        # EventBridge client for this test class
+        aws_provider = set_mocked_aws_provider()
+        eventbridge = EventBridge(aws_provider)
+        assert eventbridge.service == "events"
+
+    # Test EventBridge Client
+    @mock_aws
+    def test_client(self):
+        # EventBridge client for this test class
+        aws_provider = set_mocked_aws_provider()
+        eventbridge = EventBridge(aws_provider)
+        for client_ in eventbridge.regional_clients.values():
+            assert client_.__class__.__name__ == "EventBridge"
+
+    # Test EventBridge Session
+    @mock_aws
+    def test__get_session__(self):
+        # EventBridge client for this test class
+        aws_provider = set_mocked_aws_provider()
+        eventbridge = EventBridge(aws_provider)
+        assert eventbridge.session.__class__.__name__ == "Session"
+
+    # Test EventBridge Session
+    @mock_aws
+    def test_audited_account(self):
+        # EventBridge client for this test class
+        aws_provider = set_mocked_aws_provider()
+        eventbridge = EventBridge(aws_provider)
+        assert eventbridge.audited_account == AWS_ACCOUNT_NUMBER
+
+    # Test Schema Service
+    @mock_aws
+    def test_schema_service(self):
+        # Schema client for this test class
+        aws_provider = set_mocked_aws_provider()
+        schema = Schema(aws_provider)
+        assert schema.service == "schemas"
+
+    # Test Schema Client
+    @mock_aws
+    def test_schema_client(self):
+        # Schema client for this test class
+        aws_provider = set_mocked_aws_provider()
+        schema = Schema(aws_provider)
+        for client_ in schema.regional_clients.values():
+            assert client_.__class__.__name__ == "Schemas"
+
+    # Test Schema Session
+    @mock_aws
+    def test__schema_get_session__(self):
+        # Schema client for this test class
+        aws_provider = set_mocked_aws_provider()
+        schema = Schema(aws_provider)
+        assert schema.session.__class__.__name__ == "Session"
+
+    # Test Schema Session
+    @mock_aws
+    def test_schema_audited_account(self):
+        # Schema client for this test class
+        aws_provider = set_mocked_aws_provider()
+        schema = Schema(aws_provider)
+        assert schema.audited_account == AWS_ACCOUNT_NUMBER
+
+    # Test EventBridge Buses
+    @mock_aws
+    def test__list_event_buses__(self):
+        # EventBridge client for this test class
+        events_client = client("events", region_name=AWS_REGION_US_EAST_1)
+        events_client.create_event_bus(Name="test")
+        events_client.tag_resource(
+            ResourceARN=f"arn:aws:events:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/test",
+            Tags=[{"Key": "key-1", "Value": "value-1"}],
+        )
+        events_client.put_permission(
+            EventBusName="test",
+            Action="events:PutEvents",
+            Principal="123456789012",
+            StatementId="test-statement",
+        )
+        aws_provider = set_mocked_aws_provider()
+        eventbridge = EventBridge(aws_provider)
+        assert len(eventbridge.buses) == 30  # 1 per region
+        for bus in eventbridge.buses.values():
+            if bus.name == "test":
+                assert (
+                    bus.arn
+                    == f"arn:aws:events:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:event-bus/test"
+                )
+                assert bus.name == "test"
+                assert bus.region == AWS_REGION_US_EAST_1
+                assert bus.policy == {
+                    "Statement": [
+                        {
+                            "Action": "events:PutEvents",
+                            "Effect": "Allow",
+                            "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
+                            "Resource": "arn:aws:events:us-east-1:123456789012:event-bus/test",
+                            "Sid": "test-statement",
+                        }
+                    ],
+                    "Version": "2012-10-17",
+                }
+                assert bus.tags == [{"Key": "key-1", "Value": "value-1"}]
+
+    # Test Schema Registries
+    def test__list_policies__(self):
+        aws_provider = set_mocked_aws_provider()
+        schema = Schema(aws_provider)
+        assert len(schema.registries) == 1
+        schema_arn = "arn:aws:schemas:us-east-1:123456789012:registry/test"
+        assert schema.registries[schema_arn].arn == schema_arn
+        assert schema.registries[schema_arn].name == "test"
+        assert schema.registries[schema_arn].tags == [{"key1": "value1"}]
+        assert schema.registries[schema_arn].policy == {
+            "Statement": [
+                {
+                    "Action": "schemas:*",
+                    "Effect": "Allow",
+                    "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
+                    "Resource": "arn:aws:schemas:eu-west-1:123456789012:registry/test",
+                    "Sid": "AllowReadWrite",
+                }
+            ],
+            "Version": "2012-10-17",
+        }

tests/providers/aws/services/iam/lib/policy_test.py

@@ -0,0 +1,91 @@
+from prowler.providers.aws.services.iam.lib.policy import (
+    is_policy_cross_account,
+    is_policy_public,
+)
+from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER
+
+
+class Test_Policy:
+    def test_is_policy_cross_account(self):
+        policy1 = {
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {"AWS": ["arn:aws:iam::123456789012:root", "*"]},
+                    "Action": "s3:*",
+                    "Resource": "arn:aws:s3:::example_bucket/*",
+                }
+            ]
+        }
+        policy2 = {
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {"AWS": ["arn:aws:iam::123456789012:root"]},
+                    "Action": "s3:*",
+                    "Resource": "arn:aws:s3:::example_bucket/*",
+                }
+            ]
+        }
+        policy3 = {
+            "Statement": [
+                {
+                    "Effect": "Deny",
+                    "Principal": {"AWS": ["arn:aws:iam::123456789012:root"]},
+                    "Action": "s3:*",
+                    "Resource": "arn:aws:s3:::example_bucket/*",
+                }
+            ]
+        }
+
+        assert is_policy_cross_account(policy1, AWS_ACCOUNT_NUMBER)
+        assert not is_policy_cross_account(policy2, AWS_ACCOUNT_NUMBER)
+        assert not is_policy_cross_account(policy3, AWS_ACCOUNT_NUMBER)
+
+    def test_is_policy_public(self):
+        policy1 = {
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": "*",
+                    "Action": "s3:*",
+                    "Resource": "arn:aws:s3:::example_bucket/*",
+                }
+            ]
+        }
+        policy2 = {
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {"AWS": "*"},
+                    "Action": "s3:*",
+                    "Resource": "arn:aws:s3:::example_bucket/*",
+                }
+            ]
+        }
+        policy3 = {
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
+                    "Action": "s3:*",
+                    "Resource": "arn:aws:s3:::example_bucket/*",
+                }
+            ]
+        }
+        policy4 = {
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {"AWS": "*"},
+                    "Action": "s3:*",
+                    "Resource": "arn:aws:s3:::example_bucket/*",
+                    "Condition": {"IpAddress": {"aws:SourceIp": "192.0.2.0/24"}},
+                }
+            ]
+        }
+
+        assert is_policy_public(policy1)
+        assert is_policy_public(policy2)
+        assert not is_policy_public(policy3)
+        assert not is_policy_public(policy4)