fix(check): Add support for condition with restriction on SNS endpoint (#7757)

Co-authored-by: Ogonna Iwunze <1915636+wunzeco@users.noreply.github.com>
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>

prowler/CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 
 ### Fixed
 - Update CIS 4.0 for M365 provider. [(#7699)](https://github.com/prowler-cloud/prowler/pull/7699)
+- Cover policies with conditions with SNS endpoint in `sns_topics_not_publicly_accessible`. [(#7750)](https://github.com/prowler-cloud/prowler/pull/7750)
 
 ---
 

prowler/providers/aws/services/iam/lib/policy.py

@@ -1,4 +1,5 @@
 from ipaddress import ip_address, ip_network
+import re
 
 from prowler.lib.logger import logger
 from prowler.providers.aws.aws_provider import read_aws_regions_file
@@ -415,6 +416,55 @@ def is_condition_block_restrictive_organization(
     return is_condition_valid
 
 
+def is_condition_block_restrictive_sns_endpoint(
+    condition_statement: dict,
+):
+    """
+    is_condition_block_restrictive_sns_endpoint parses the IAM Condition policy block and returns True if the condition_statement is restrictive for an endpoint, False if not.
+
+    @param condition_statement: dict with an IAM Condition block, e.g.:
+        {
+            "StringLike": {
+                "SNS:Endpoint": "https://events.pagerduty.com/integration/<api-key>/enqueue"
+            }
+        }
+
+    """
+    is_condition_valid = False
+
+    # The conditions must be defined in lowercase since the context key names are not case-sensitive.
+    # For example, including the aws:PrincipalOrgID context key is equivalent to testing for AWS:PrincipalOrgID
+    # https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition.html
+    valid_condition_options = {
+        "StringEquals": [
+            "sns:endpoint",
+        ],
+        "StringLike": [
+            "sns:endpoint",
+        ],
+    }
+
+    for condition_operator, condition_operator_key in valid_condition_options.items():
+        if condition_operator in condition_statement:
+            # https://docs.aws.amazon.com/sns/latest/dg/sns-using-identity-based-policies.html#sns-policy-keys
+            # sns:endpoint - The URL, email address, or ARN from a Subscribe request or a previously confirmed subscription.
+            pattern = re.compile(r".+@[^*]+|^https:\/\/[^*]+|^arn:aws:sns:[^*]+")
+            for value in condition_operator_key:
+                # We need to transform the condition_statement into lowercase
+                condition_statement[condition_operator] = {
+                    k.lower(): v
+                    for k, v in condition_statement[condition_operator].items()
+                }
+
+                if value in condition_statement[condition_operator]:
+                    if pattern.fullmatch(
+                        condition_statement[condition_operator][value]
+                    ):
+                        is_condition_valid = True
+
+    return is_condition_valid
+
+
 def process_actions(effect, actions, target_set):
     """
     process_actions processes the actions in the policy.

prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.py

@@ -2,6 +2,7 @@ from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.iam.lib.policy import (
     is_condition_block_restrictive,
     is_condition_block_restrictive_organization,
+    is_condition_block_restrictive_sns_endpoint,
 )
 from prowler.providers.aws.services.sns.sns_client import sns_client
 
@@ -32,6 +33,7 @@ class sns_topics_not_publicly_accessible(Check):
                         ):
                             condition_account = False
                             condition_org = False
+                            condition_endpoint = False
                             if (
                                 "Condition" in statement
                                 and is_condition_block_restrictive(
@@ -47,6 +49,13 @@ class sns_topics_not_publicly_accessible(Check):
                                 )
                             ):
                                 condition_org = True
+                            if (
+                                "Condition" in statement
+                                and is_condition_block_restrictive_sns_endpoint(
+                                    statement["Condition"],
+                                )
+                            ):
+                                condition_endpoint = True
 
                             if condition_account and condition_org:
                                 report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from the account {sns_client.audited_account} and an organization."
@@ -54,6 +63,8 @@ class sns_topics_not_publicly_accessible(Check):
                                 report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from the account {sns_client.audited_account}."
                             elif condition_org:
                                 report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from an organization."
+                            elif condition_endpoint:
+                                report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from an endpoint."
                             else:
                                 report.status = "FAIL"
                                 report.status_extended = f"SNS topic {topic.name} is public because its policy allows public access."

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

@@ -3,9 +3,11 @@ from prowler.providers.aws.services.iam.lib.policy import (
     check_full_service_access,
     is_condition_block_restrictive,
     is_condition_block_restrictive_organization,
+    is_condition_block_restrictive_sns_endpoint,
     is_condition_restricting_from_private_ip,
     is_policy_public,
 )
+import pytest
 
 TRUSTED_AWS_ACCOUNT_NUMBER = "123456789012"
 NON_TRUSTED_AWS_ACCOUNT_NUMBER = "111222333444"
@@ -1442,6 +1444,39 @@ class Test_Policy:
         condition_statement = {"StringEquals": {"aws:PrincipalOrgID": ALL_ORGS}}
         assert not is_condition_block_restrictive_organization(condition_statement)
 
+    @pytest.mark.parametrize(
+        "condition_value,expected",
+        [
+            ("*@example.com", True),
+            ("https://events.pagerduty.com/integration/<api-key>/enqueue", True),
+            (
+                "arn:aws:sns:eu-west-2:123456789012:example-topic:995be20c-a7e3-44ca-8c18-77cb263d15e7",
+                True,
+            ),
+            ("*@*.com", False),
+            ("*@*", False),
+            ("*@example.*", False),
+            ("https://events.pagerduty.com/integration/*/enqueue", False),
+            ("arn:aws:sns:eu-west-2:123456789012:example-topic:*", False),
+            (
+                "arn:aws:sns:eu-west-2:*:example-topic:995be20c-a7e3-44ca-8c18-77cb263d15e7",
+                False,
+            ),
+        ],
+    )
+    def test_condition_parser_string_equals_sns_endpoint_str(
+        self, condition_value: str, expected: bool
+    ):
+        condition_statement = {"StringEquals": {"SNS:Endpoint": condition_value}}
+        assert (
+            is_condition_block_restrictive_sns_endpoint(condition_statement) == expected
+        )
+
+        condition_statement = {"StringLike": {"SNS:Endpoint": condition_value}}
+        assert (
+            is_condition_block_restrictive_sns_endpoint(condition_statement) == expected
+        )
+
     def test_policy_allows_cross_account_access_with_root_and_wildcard_principal(self):
         policy_allow_root_and_wildcard_principal = {
             "Statement": [

tests/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible_test.py

@@ -1,8 +1,10 @@
+from typing import Any, Dict
 from unittest import mock
 from uuid import uuid4
 
 from prowler.providers.aws.services.sns.sns_service import Topic
 from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+import pytest
 
 kms_key_id = str(uuid4())
 topic_name = "test-topic"
@@ -97,6 +99,20 @@ test_policy_restricted_principal_account_organization = {
 }
 
 
+def generate_policy_restricted_on_sns_endpoint(endpoint: str) -> Dict[str, Any]:
+    return {
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Principal": {"AWS": "*"},
+                "Action": ["sns:Publish"],
+                "Resource": f"arn:aws:sns:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{topic_name}",
+                "Condition": {"StringEquals": {"SNS:Endpoint": endpoint}},
+            }
+        ]
+    }
+
+
 class Test_sns_topics_not_publicly_accessible:
     def test_no_topics(self):
         sns_client = mock.MagicMock
@@ -379,3 +395,94 @@ class Test_sns_topics_not_publicly_accessible:
             assert result[0].resource_arn == topic_arn
             assert result[0].region == AWS_REGION_EU_WEST_1
             assert result[0].resource_tags == []
+
+    @pytest.mark.parametrize(
+        "endpoint",
+        [
+            ("*@example.com"),
+            ("user@example.com"),
+            ("https://events.pagerduty.com/integration/987654321/enqueue"),
+            (
+                "arn:aws:sns:eu-west-2:123456789012:example-topic:995be20c-a7e3-44ca-8c18-77cb263d15e7"
+            ),
+        ],
+    )
+    def test_topic_public_with_sns_endpoint(self, endpoint: str):
+        sns_client = mock.MagicMock
+        sns_client.audited_account = AWS_ACCOUNT_NUMBER
+        sns_client.topics = []
+        sns_client.topics.append(
+            Topic(
+                arn=topic_arn,
+                name=topic_name,
+                policy=generate_policy_restricted_on_sns_endpoint(endpoint=endpoint),
+                region=AWS_REGION_EU_WEST_1,
+            )
+        )
+        sns_client.provider = mock.MagicMock()
+        sns_client.provider.organizations_metadata = mock.MagicMock()
+        sns_client.provider.organizations_metadata.organization_id = org_id
+        with mock.patch(
+            "prowler.providers.aws.services.sns.sns_service.SNS",
+            sns_client,
+        ):
+            from prowler.providers.aws.services.sns.sns_topics_not_publicly_accessible.sns_topics_not_publicly_accessible import (
+                sns_topics_not_publicly_accessible,
+            )
+
+            check = sns_topics_not_publicly_accessible()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"SNS topic {topic_name} is not public because its policy only allows access from an endpoint."
+            )
+            assert result[0].resource_id == topic_name
+            assert result[0].resource_arn == topic_arn
+            assert result[0].region == AWS_REGION_EU_WEST_1
+            assert result[0].resource_tags == []
+
+    @pytest.mark.parametrize(
+        "endpoint",
+        [
+            ("*@*"),
+            ("https://events.pagerduty.com/integration/*/enqueue"),
+            ("arn:aws:sns:eu-west-2:*:example-topic:*"),
+        ],
+    )
+    def test_topic_public_with_unrestricted_sns_endpoint(self, endpoint: str):
+        sns_client = mock.MagicMock
+        sns_client.audited_account = AWS_ACCOUNT_NUMBER
+        sns_client.topics = []
+        sns_client.topics.append(
+            Topic(
+                arn=topic_arn,
+                name=topic_name,
+                policy=generate_policy_restricted_on_sns_endpoint(endpoint=endpoint),
+                region=AWS_REGION_EU_WEST_1,
+            )
+        )
+        sns_client.provider = mock.MagicMock()
+        sns_client.provider.organizations_metadata = mock.MagicMock()
+        sns_client.provider.organizations_metadata.organization_id = org_id
+        with mock.patch(
+            "prowler.providers.aws.services.sns.sns_service.SNS",
+            sns_client,
+        ):
+            from prowler.providers.aws.services.sns.sns_topics_not_publicly_accessible.sns_topics_not_publicly_accessible import (
+                sns_topics_not_publicly_accessible,
+            )
+
+            check = sns_topics_not_publicly_accessible()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"SNS topic {topic_name} is public because its policy allows public access."
+            )
+            assert result[0].resource_id == topic_name
+            assert result[0].resource_arn == topic_arn
+            assert result[0].region == AWS_REGION_EU_WEST_1
+            assert result[0].resource_tags == []