From 6c04592e7e76da83361b32175e54287b7d5595d6 Mon Sep 17 00:00:00 2001 From: Prowler Bot Date: Thu, 15 May 2025 16:53:12 +0200 Subject: [PATCH] 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 --- prowler/CHANGELOG.md | 1 + .../providers/aws/services/iam/lib/policy.py | 50 ++++++++ .../sns_topics_not_publicly_accessible.py | 11 ++ .../aws/services/iam/lib/policy_test.py | 35 ++++++ ...sns_topics_not_publicly_accessible_test.py | 107 ++++++++++++++++++ 5 files changed, 204 insertions(+) diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 78b04c0309..e8ecc04af5 100644 --- a/prowler/CHANGELOG.md +++ b/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) --- diff --git a/prowler/providers/aws/services/iam/lib/policy.py b/prowler/providers/aws/services/iam/lib/policy.py index b45dd1bd30..e023b137b5 100644 --- a/prowler/providers/aws/services/iam/lib/policy.py +++ b/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//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. diff --git a/prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.py b/prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.py index 536bf4f800..1dc0776b64 100644 --- a/prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.py +++ b/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." diff --git a/tests/providers/aws/services/iam/lib/policy_test.py b/tests/providers/aws/services/iam/lib/policy_test.py index bac84355ac..02e88739ec 100644 --- a/tests/providers/aws/services/iam/lib/policy_test.py +++ b/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//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": [ diff --git a/tests/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible_test.py b/tests/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible_test.py index 81f2cc8bad..71aa6c4477 100644 --- a/tests/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible_test.py +++ b/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 == []