fix(ses): evaluate all identity authorization policies (#12464)

Co-authored-by: Hugo P.Brito <hugopbrit@gmail.com>

prowler/changelog.d/ses-multiple-authorization-policies.fixed.md

@@ -0,0 +1 @@
+`ses_identity_not_publicly_accessible` now evaluates every SES identity authorization policy and marks mixed public Allow and Deny statements for manual review

prowler/providers/aws/services/ses/ses_identity_not_publicly_accessible/ses_identity_not_publicly_accessible.py

@@ -1,25 +1,58 @@
+from copy import deepcopy
+
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.iam.lib.policy import is_policy_public
 from prowler.providers.aws.services.ses.ses_client import ses_client
 
 
+def _normalize_policy_statements(policy: dict) -> dict:
+    statements = policy.get("Statement", [])
+    if isinstance(statements, dict):
+        return {**policy, "Statement": [statements]}
+    return policy
+
+
+def _has_explicit_deny(policy: dict) -> bool:
+    return any(
+        isinstance(statement, dict) and statement.get("Effect") == "Deny"
+        for statement in _normalize_policy_statements(policy).get("Statement", [])
+    )
+
+
 class ses_identity_not_publicly_accessible(Check):
-    def execute(self):
+    """Ensure SES identities are not publicly accessible through authorization policies."""
+
+    def execute(self) -> list[Check_Report_AWS]:
+        """Evaluate every authorization policy attached to each SES identity.
+
+        Returns:
+            A list of reports containing the public-access result for each identity.
+        """
         findings = []
         for identity in ses_client.email_identities.values():
-            if identity.policy is None:
+            if not identity.policies:
                 continue
             report = Check_Report_AWS(metadata=self.metadata(), resource=identity)
             report.status = "PASS"
             report.status_extended = (
                 f"SES identity {identity.name} is not publicly accessible."
             )
-            if is_policy_public(
-                identity.policy,
-                ses_client.audited_account,
-            ):
-                report.status = "FAIL"
-                report.status_extended = f"SES identity {identity.name} is publicly accessible due to its resource policy."
+            has_public_allow = any(
+                is_policy_public(
+                    _normalize_policy_statements(deepcopy(policy)),
+                    ses_client.audited_account,
+                )
+                for policy in identity.policies.values()
+            )
+            if has_public_allow:
+                if any(
+                    _has_explicit_deny(policy) for policy in identity.policies.values()
+                ):
+                    report.status = "MANUAL"
+                    report.status_extended = f"SES identity {identity.name} has public Allow and explicit Deny statements in its resource policies. Effective public access requires manual review."
+                else:
+                    report.status = "FAIL"
+                    report.status_extended = f"SES identity {identity.name} is publicly accessible due to its resource policies."
 
             findings.append(report)
 

prowler/providers/aws/services/ses/ses_service.py

@@ -1,7 +1,7 @@
 from json import loads
 from typing import Optional
 
-from pydantic.v1 import BaseModel
+from pydantic.v1 import BaseModel, Field
 
 from prowler.lib.logger import logger
 from prowler.lib.scan_filters.scan_filters import is_resource_filtered
@@ -46,8 +46,11 @@ class SES(AWSService):
                 identity_attributes = regional_client.get_email_identity(
                     EmailIdentity=identity.name
                 )
-                for _, content in identity_attributes.get("Policies", {}).items():
-                    identity.policy = loads(content)
+                identity.policies = {
+                    name: loads(content)
+                    for name, content in identity_attributes.get("Policies", {}).items()
+                }
+                identity.policy = next(reversed(identity.policies.values()), None)
                 identity.tags = identity_attributes.get("Tags", [])
                 dkim_attrs = identity_attributes.get("DkimAttributes", {}) or {}
                 identity.dkim_status = dkim_attrs.get("Status")
@@ -72,6 +75,7 @@ class Identity(BaseModel):
     region: str
     type: Optional[str]
     policy: Optional[dict] = None
+    policies: dict[str, dict] = Field(default_factory=dict)
     tags: Optional[list] = []
     dkim_status: Optional[str] = None
     dkim_signing_attributes_origin: Optional[str] = None

tests/providers/aws/services/ses/ses_identity_not_publicly_accessible/ses_identity_not_publicly_accessible_test.py

@@ -1,6 +1,8 @@
+from copy import deepcopy
 from unittest import mock
 
 import botocore
+import pytest
 from boto3 import client
 from moto import mock_aws
 
@@ -54,6 +56,113 @@ def mock_make_api_call_v2(self, operation_name, kwarg):
     return make_api_call(self, operation_name, kwarg)
 
 
+PUBLIC_ALLOW_POLICY = '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"ses:SendEmail","Resource":"*"}]}'
+PRIVATE_ALLOW_POLICY = '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":"arn:aws:iam::123456789012:root"},"Action":"ses:SendEmail","Resource":"*"}]}'
+MATCHING_DENY_POLICY = '{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","Action":"ses:SendEmail","Resource":"*"}]}'
+UNRELATED_DENY_POLICY = '{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","Action":"ses:SendRawEmail","Resource":"*"}]}'
+PUBLIC_ALLOW_AND_DENY_POLICY = '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"ses:SendEmail","Resource":"*"},{"Effect":"Deny","Principal":"*","Action":"ses:SendEmail","Resource":"*"}]}'
+PUBLIC_ALLOW_SINGLE_STATEMENT_POLICY = '{"Version":"2012-10-17","Statement":{"Effect":"Allow","Principal":"*","Action":"ses:SendEmail","Resource":"*"}}'
+PRIVATE_ALLOW_SINGLE_STATEMENT_POLICY = '{"Version":"2012-10-17","Statement":{"Effect":"Allow","Principal":{"AWS":"arn:aws:iam::123456789012:root"},"Action":"ses:SendEmail","Resource":"*"}}'
+MATCHING_DENY_SINGLE_STATEMENT_POLICY = '{"Version":"2012-10-17","Statement":{"Effect":"Deny","Principal":"*","Action":"ses:SendEmail","Resource":"*"}}'
+CONDITIONAL_ALLOW_SINGLE_STATEMENT_POLICY = '{"Version":"2012-10-17","Statement":{"Effect":"Allow","Principal":"*","Action":"ses:SendEmail","Resource":"*","Condition":{"StringEquals":{"AWS:SourceAccount":"123456789012"}}}}'
+
+
+def make_multiple_policies_api_mock(policies):
+    def mock_api_call(self, operation_name, kwarg):
+        if operation_name == "ListEmailIdentities":
+            return {
+                "EmailIdentities": [
+                    {
+                        "IdentityType": "DOMAIN",
+                        "IdentityName": "test-email-identity-multiple-policies",
+                    }
+                ],
+            }
+        elif operation_name == "GetEmailIdentity":
+            return {"Policies": policies, "Tags": {}}
+        return make_api_call(self, operation_name, kwarg)
+
+    return mock_api_call
+
+
+mock_make_api_call_multiple_policies = make_multiple_policies_api_mock(
+    {
+        "public-policy": PUBLIC_ALLOW_POLICY,
+        "private-policy": PRIVATE_ALLOW_POLICY,
+    }
+)
+mock_make_api_call_multiple_policies_reversed = make_multiple_policies_api_mock(
+    {
+        "private-policy": PRIVATE_ALLOW_POLICY,
+        "public-policy": PUBLIC_ALLOW_POLICY,
+    }
+)
+mock_make_api_call_public_allow_and_matching_deny = make_multiple_policies_api_mock(
+    {
+        "public-policy": PUBLIC_ALLOW_POLICY,
+        "deny-policy": MATCHING_DENY_POLICY,
+    }
+)
+mock_make_api_call_matching_deny_and_public_allow = make_multiple_policies_api_mock(
+    {
+        "deny-policy": MATCHING_DENY_POLICY,
+        "public-policy": PUBLIC_ALLOW_POLICY,
+    }
+)
+mock_make_api_call_public_allow_and_unrelated_deny = make_multiple_policies_api_mock(
+    {
+        "public-policy": PUBLIC_ALLOW_POLICY,
+        "deny-policy": UNRELATED_DENY_POLICY,
+    }
+)
+mock_make_api_call_same_policy_allow_and_deny = make_multiple_policies_api_mock(
+    {"combined-policy": PUBLIC_ALLOW_AND_DENY_POLICY}
+)
+mock_make_api_call_multiple_private_policies = make_multiple_policies_api_mock(
+    {
+        "private-policy-1": PRIVATE_ALLOW_POLICY,
+        "private-policy-2": PRIVATE_ALLOW_POLICY,
+    }
+)
+mock_make_api_call_public_single_statement = make_multiple_policies_api_mock(
+    {"public-policy": PUBLIC_ALLOW_SINGLE_STATEMENT_POLICY}
+)
+mock_make_api_call_private_single_statement = make_multiple_policies_api_mock(
+    {"private-policy": PRIVATE_ALLOW_SINGLE_STATEMENT_POLICY}
+)
+mock_make_api_call_public_and_deny_single_statements = make_multiple_policies_api_mock(
+    {
+        "public-policy": PUBLIC_ALLOW_SINGLE_STATEMENT_POLICY,
+        "deny-policy": MATCHING_DENY_SINGLE_STATEMENT_POLICY,
+    }
+)
+mock_make_api_call_conditional_single_statement = make_multiple_policies_api_mock(
+    {"conditional-policy": CONDITIONAL_ALLOW_SINGLE_STATEMENT_POLICY}
+)
+
+
+def execute_check_with_api_mock(api_call_mock):
+    with mock.patch("botocore.client.BaseClient._make_api_call", new=api_call_mock):
+        client("sesv2", region_name=AWS_REGION_EU_WEST_1)
+        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.ses.ses_identity_not_publicly_accessible.ses_identity_not_publicly_accessible.ses_client",
+                new=SES(aws_provider),
+            ),
+        ):
+            from prowler.providers.aws.services.ses.ses_identity_not_publicly_accessible.ses_identity_not_publicly_accessible import (
+                ses_identity_not_publicly_accessible,
+            )
+
+            return ses_identity_not_publicly_accessible().execute()
+
+
 class Test_ses_identities_not_publicly_accessible:
     @mock_aws
     def test_no_identities(self):
@@ -114,6 +223,114 @@ class Test_ses_identities_not_publicly_accessible:
             assert result[0].resource_tags == {"tag1": "value1", "tag2": "value2"}
             assert result[0].region == AWS_REGION_EU_WEST_1
 
+    @mock_aws
+    @pytest.mark.parametrize(
+        "api_call_mock",
+        [
+            mock_make_api_call_multiple_policies,
+            mock_make_api_call_multiple_policies_reversed,
+        ],
+        ids=["public-policy-first", "public-policy-last"],
+    )
+    def test_email_identity_public_when_any_policy_is_public(self, api_call_mock):
+        result = execute_check_with_api_mock(api_call_mock)
+
+        assert len(result) == 1
+        assert result[0].status == "FAIL"
+        assert (
+            result[0].status_extended
+            == "SES identity test-email-identity-multiple-policies is publicly accessible due to its resource policies."
+        )
+
+    @mock_aws
+    @pytest.mark.parametrize(
+        "api_call_mock",
+        [
+            mock_make_api_call_public_allow_and_matching_deny,
+            mock_make_api_call_matching_deny_and_public_allow,
+            mock_make_api_call_public_allow_and_unrelated_deny,
+            mock_make_api_call_same_policy_allow_and_deny,
+        ],
+        ids=[
+            "matching-deny-last",
+            "matching-deny-first",
+            "unrelated-deny",
+            "same-policy-deny",
+        ],
+    )
+    def test_email_identity_public_allow_with_explicit_deny_is_manual(
+        self, api_call_mock
+    ):
+        result = execute_check_with_api_mock(api_call_mock)
+
+        assert len(result) == 1
+        assert result[0].status == "MANUAL"
+        assert (
+            result[0].status_extended
+            == "SES identity test-email-identity-multiple-policies has public Allow and explicit Deny statements in its resource policies. Effective public access requires manual review."
+        )
+
+    @mock_aws
+    def test_email_identity_multiple_private_policies(self):
+        result = execute_check_with_api_mock(
+            mock_make_api_call_multiple_private_policies
+        )
+
+        assert len(result) == 1
+        assert result[0].status == "PASS"
+        assert (
+            result[0].status_extended
+            == "SES identity test-email-identity-multiple-policies is not publicly accessible."
+        )
+
+    @mock_aws
+    @pytest.mark.parametrize(
+        ("api_call_mock", "expected_status"),
+        [
+            (mock_make_api_call_public_single_statement, "FAIL"),
+            (mock_make_api_call_private_single_statement, "PASS"),
+            (mock_make_api_call_public_and_deny_single_statements, "MANUAL"),
+        ],
+        ids=["public", "private", "public-with-deny"],
+    )
+    def test_email_identity_single_statement_policy(
+        self, api_call_mock, expected_status
+    ):
+        result = execute_check_with_api_mock(api_call_mock)
+
+        assert len(result) == 1
+        assert result[0].status == expected_status
+
+    @mock_aws
+    def test_check_preserves_nested_policy_condition_keys(self):
+        with mock.patch(
+            "botocore.client.BaseClient._make_api_call",
+            new=mock_make_api_call_conditional_single_statement,
+        ):
+            client("sesv2", region_name=AWS_REGION_EU_WEST_1)
+            aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+            ses_client = SES(aws_provider)
+            identity = next(iter(ses_client.email_identities.values()))
+            policies_before_check = deepcopy(identity.policies)
+
+            with (
+                mock.patch(
+                    "prowler.providers.common.provider.Provider.get_global_provider",
+                    return_value=aws_provider,
+                ),
+                mock.patch(
+                    "prowler.providers.aws.services.ses.ses_identity_not_publicly_accessible.ses_identity_not_publicly_accessible.ses_client",
+                    new=ses_client,
+                ),
+            ):
+                from prowler.providers.aws.services.ses.ses_identity_not_publicly_accessible.ses_identity_not_publicly_accessible import (
+                    ses_identity_not_publicly_accessible,
+                )
+
+                ses_identity_not_publicly_accessible().execute()
+
+            assert identity.policies == policies_before_check
+
     @mock_aws
     @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call_v2)
     def test_email_identity_public(self):
@@ -140,7 +357,7 @@ class Test_ses_identities_not_publicly_accessible:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == "SES identity test-email-identity-public is publicly accessible due to its resource policy."
+                == "SES identity test-email-identity-public is publicly accessible due to its resource policies."
             )
             assert result[0].resource_id == "test-email-identity-public"
             assert (

tests/providers/aws/services/ses/ses_service_test.py

@@ -27,6 +27,7 @@ def mock_make_api_call(self, operation_name, kwarg):
         return {
             "Policies": {
                 "policy1": '{"policy1": "value1"}',
+                "policy2": '{"policy2": "value2"}',
             },
             "Tags": {"tag1": "value1", "tag2": "value2"},
             "DkimAttributes": {
@@ -81,7 +82,11 @@ class Test_SES_Service:
         assert ses.email_identities[arn].type == "EMAIL_ADDRESS"
         assert ses.email_identities[arn].arn == arn
         assert ses.email_identities[arn].region == AWS_REGION_EU_WEST_1
-        assert ses.email_identities[arn].policy == {"policy1": "value1"}
+        assert ses.email_identities[arn].policy == {"policy2": "value2"}
+        assert ses.email_identities[arn].policies == {
+            "policy1": {"policy1": "value1"},
+            "policy2": {"policy2": "value2"},
+        }
         assert ses.email_identities[arn].tags == {"tag1": "value1", "tag2": "value2"}
         assert ses.email_identities[arn].dkim_status == "SUCCESS"
         assert ses.email_identities[arn].dkim_signing_attributes_origin == "AWS_SES"