diff --git a/prowler/changelog.d/ses-multiple-authorization-policies.fixed.md b/prowler/changelog.d/ses-multiple-authorization-policies.fixed.md new file mode 100644 index 0000000000..0fe2961905 --- /dev/null +++ b/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 diff --git a/prowler/providers/aws/services/ses/ses_identity_not_publicly_accessible/ses_identity_not_publicly_accessible.py b/prowler/providers/aws/services/ses/ses_identity_not_publicly_accessible/ses_identity_not_publicly_accessible.py index f668b05d06..6f4440a679 100644 --- a/prowler/providers/aws/services/ses/ses_identity_not_publicly_accessible/ses_identity_not_publicly_accessible.py +++ b/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) diff --git a/prowler/providers/aws/services/ses/ses_service.py b/prowler/providers/aws/services/ses/ses_service.py index 7d0feb8f54..b865b0128e 100644 --- a/prowler/providers/aws/services/ses/ses_service.py +++ b/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 diff --git a/tests/providers/aws/services/ses/ses_identity_not_publicly_accessible/ses_identity_not_publicly_accessible_test.py b/tests/providers/aws/services/ses/ses_identity_not_publicly_accessible/ses_identity_not_publicly_accessible_test.py index 15b8a9dd32..dbd12d189d 100644 --- a/tests/providers/aws/services/ses/ses_identity_not_publicly_accessible/ses_identity_not_publicly_accessible_test.py +++ b/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 ( diff --git a/tests/providers/aws/services/ses/ses_service_test.py b/tests/providers/aws/services/ses/ses_service_test.py index 223ed94365..92146f6a01 100644 --- a/tests/providers/aws/services/ses/ses_service_test.py +++ b/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"