From bbc51114b0bbd4e3a26003b4f545fde3f5bf57c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Thu, 1 Aug 2024 17:54:36 +0200 Subject: [PATCH] fix(sns): add condition to sns topics (#4498) Co-authored-by: Pepe Fagoaga Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com> --- .../policy_condition_parser.py | 67 +++++++- .../sns_topics_not_publicly_accessible.py | 22 ++- .../policy_condition_parser_test.py | 48 ++++++ ...sns_topics_not_publicly_accessible_test.py | 157 +++++++++++++++++- 4 files changed, 289 insertions(+), 5 deletions(-) diff --git a/prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py b/prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py index 7b86d7ec89..3d313a1f71 100644 --- a/prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py +++ b/prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py @@ -1,5 +1,7 @@ def is_condition_block_restrictive( - condition_statement: dict, source_account: str, is_cross_account_allowed=False + condition_statement: dict, + source_account: str, + is_cross_account_allowed=False, ): """ is_condition_block_restrictive parses the IAM Condition policy block and, by default, returns True if the source_account passed as argument is within, False if not. @@ -15,6 +17,9 @@ def is_condition_block_restrictive( } @param source_account: str with a 12-digit AWS Account number, e.g.: 111122223333 + + @param is_cross_account_allowed: bool to allow cross-account access, e.g.: True + """ is_condition_valid = False @@ -90,3 +95,63 @@ def is_condition_block_restrictive( is_condition_valid = True return is_condition_valid + + +def is_condition_block_restrictive_organization( + condition_statement: dict, +): + """ + is_condition_block_restrictive_organization parses the IAM Condition policy block and returns True if the condition_statement is restrictive for the organization, False if not. + + @param condition_statement: dict with an IAM Condition block, e.g.: + { + "StringLike": { + "AWS:PrincipalOrgID": "o-111122223333" + } + } + + """ + 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": [ + "aws:principalorgid", + ], + "StringLike": [ + "aws:principalorgid", + ], + } + + for condition_operator, condition_operator_key in valid_condition_options.items(): + if condition_operator in condition_statement: + 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]: + # values are a list + if isinstance( + condition_statement[condition_operator][value], + list, + ): + is_condition_valid = True + for item in condition_statement[condition_operator][value]: + if item == "*": + is_condition_valid = False + break + + # value is a string + elif isinstance( + condition_statement[condition_operator][value], + str, + ): + if "*" not in condition_statement[condition_operator][value]: + is_condition_valid = True + + return is_condition_valid 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 6844206218..fc93cbbcc4 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 @@ -1,6 +1,7 @@ from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import ( is_condition_block_restrictive, + is_condition_block_restrictive_organization, ) from prowler.providers.aws.services.sns.sns_client import sns_client @@ -33,13 +34,30 @@ class sns_topics_not_publicly_accessible(Check): and "*" in statement["Principal"]["CanonicalUser"] ) ): + condition_account = False + condition_org = False if ( "Condition" in statement and is_condition_block_restrictive( - statement["Condition"], sns_client.audited_account + statement["Condition"], + sns_client.audited_account, ) ): - report.status_extended = f"SNS topic {topic.name} is not public because its policy only allows access from the same account." + condition_account = True + if ( + "Condition" in statement + and is_condition_block_restrictive_organization( + statement["Condition"], + ) + ): + condition_org = 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." + elif condition_account: + 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." 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/lib/policy_condition_parser/policy_condition_parser_test.py b/tests/providers/aws/lib/policy_condition_parser/policy_condition_parser_test.py index 120b8be4e9..0162064bd2 100644 --- a/tests/providers/aws/lib/policy_condition_parser/policy_condition_parser_test.py +++ b/tests/providers/aws/lib/policy_condition_parser/policy_condition_parser_test.py @@ -1,10 +1,16 @@ from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import ( is_condition_block_restrictive, + is_condition_block_restrictive_organization, ) TRUSTED_AWS_ACCOUNT_NUMBER = "123456789012" NON_TRUSTED_AWS_ACCOUNT_NUMBER = "111222333444" +TRUSTED_ORGANIZATION_ID = "o-123456789012" +NON_TRUSTED_ORGANIZATION_ID = "o-111222333444" + +ALL_ORGS = "*" + class Test_policy_condition_parser: # Test lowercase context key name --> aws @@ -1389,3 +1395,45 @@ class Test_policy_condition_parser: assert is_condition_block_restrictive( condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True ) + + def test_condition_parser_string_equals_aws_PrincipalOrgID_list(self): + condition_statement = { + "StringEquals": {"aws:PrincipalOrgID": [TRUSTED_ORGANIZATION_ID]} + } + assert is_condition_block_restrictive_organization(condition_statement) + + def test_condition_parser_string_equals_aws_PrincipalOrgID_list_multiple_items( + self, + ): + condition_statement = { + "StringEquals": { + "aws:PrincipalOrgID": [ + TRUSTED_ORGANIZATION_ID, + NON_TRUSTED_ORGANIZATION_ID, + ] + } + } + assert is_condition_block_restrictive_organization(condition_statement) + + def test_condition_parser_string_equals_aws_PrincipalOrgID_str(self): + condition_statement = { + "StringEquals": {"aws:PrincipalOrgID": TRUSTED_ORGANIZATION_ID} + } + assert is_condition_block_restrictive_organization(condition_statement) + + def test_condition_parser_string_equals_aws_All_Orgs_list_multiple_items( + self, + ): + condition_statement = { + "StringEquals": { + "aws:PrincipalOrgID": [ + TRUSTED_ORGANIZATION_ID, + ALL_ORGS, + ] + } + } + assert not is_condition_block_restrictive_organization(condition_statement) + + def test_condition_parser_string_equals_aws_All_Orgs_str(self): + condition_statement = {"StringEquals": {"aws:PrincipalOrgID": ALL_ORGS}} + assert not is_condition_block_restrictive_organization(condition_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 2be9ac514d..30fc8dee7d 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 @@ -6,6 +6,7 @@ from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1 kms_key_id = str(uuid4()) topic_name = "test-topic" +org_id = "o-123456" topic_arn = f"arn:aws:sns:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{topic_name}" test_policy_restricted = { "Statement": [ @@ -53,6 +54,48 @@ test_policy_not_restricted = { ] } +test_policy_restricted_principal_org_id = { + "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": {"aws:PrincipalOrgID": org_id}}, + } + ] +} + +test_policy_restricted_all_org = { + "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": {"aws:PrincipalOrgID": "*"}}, + } + ] +} + + +test_policy_restricted_principal_account_organization = { + "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": { + "aws:PrincipalOrgID": org_id, + "aws:SourceAccount": AWS_ACCOUNT_NUMBER, + } + }, + } + ] +} + class Test_sns_topics_not_publicly_accessible: def test_no_topics(self): @@ -81,6 +124,7 @@ class Test_sns_topics_not_publicly_accessible: region=AWS_REGION_EU_WEST_1, ) ) + with mock.patch( "prowler.providers.aws.services.sns.sns_service.SNS", sns_client, @@ -108,6 +152,7 @@ class Test_sns_topics_not_publicly_accessible: sns_client.topics.append( Topic(arn=topic_arn, name=topic_name, region=AWS_REGION_EU_WEST_1) ) + with mock.patch( "prowler.providers.aws.services.sns.sns_service.SNS", sns_client, @@ -155,7 +200,7 @@ class Test_sns_topics_not_publicly_accessible: 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 the same account." + == f"SNS topic {topic_name} is not public because its policy only allows access from the account {AWS_ACCOUNT_NUMBER}." ) assert result[0].resource_id == topic_name assert result[0].resource_arn == topic_arn @@ -188,7 +233,7 @@ class Test_sns_topics_not_publicly_accessible: 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 the same account." + == f"SNS topic {topic_name} is not public because its policy only allows access from the account {AWS_ACCOUNT_NUMBER}." ) assert result[0].resource_id == topic_name assert result[0].resource_arn == topic_arn @@ -226,3 +271,111 @@ 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 == [] + + def test_topic_public_with_principal_organization(self): + 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=test_policy_restricted_principal_org_id, + 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 organization." + ) + 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 == [] + + def test_topic_public_not_with_principal_organization(self): + 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=test_policy_restricted_all_org, + 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 == [] + + def test_topic_public_with_principal_account_and_organization(self): + 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=test_policy_restricted_principal_account_organization, + 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 the account {AWS_ACCOUNT_NUMBER} and an organization." + ) + 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 == []