From 04749c1da1a4dbd4ecd5270e1a4f2aeb770ebb74 Mon Sep 17 00:00:00 2001 From: Andoni Alonso <14891798+andoniaf@users.noreply.github.com> Date: Thu, 24 Jul 2025 12:03:30 +0200 Subject: [PATCH] fix(aws): `sns_topics_not_publicly_accessible` false positive with aws:SourceArn conditions (#8340) Co-authored-by: MrCloudSec --- prowler/CHANGELOG.md | 1 + .../providers/aws/services/iam/lib/policy.py | 252 +++++++++++++---- .../sns_topics_not_publicly_accessible.py | 58 ++-- ...ingress_from_internet_to_all_ports_test.py | 2 +- .../aws/services/iam/lib/policy_test.py | 265 ++++++++++++++++++ ...sns_topics_not_publicly_accessible_test.py | 250 ++++++++++++++++- 6 files changed, 729 insertions(+), 99 deletions(-) diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index e8250cf07f..2bb9dac6e3 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -20,6 +20,7 @@ All notable changes to the **Prowler SDK** are documented in this file. ### Fixed - Add more validations to Azure Storage models when some values are None to avoid serialization issues [(#8325)](https://github.com/prowler-cloud/prowler/pull/8325) +- `sns_topics_not_publicly_accessible` false positive with `aws:SourceArn` conditions [(#8326)](https://github.com/prowler-cloud/prowler/issues/8326) --- diff --git a/prowler/providers/aws/services/iam/lib/policy.py b/prowler/providers/aws/services/iam/lib/policy.py index b560389feb..c3b6eb3dba 100644 --- a/prowler/providers/aws/services/iam/lib/policy.py +++ b/prowler/providers/aws/services/iam/lib/policy.py @@ -223,6 +223,108 @@ def check_full_service_access(service: str, policy: dict) -> bool: return all_target_service_actions.issubset(actions_allowed_on_all_resources) +def has_public_principal(statement: dict) -> bool: + """ + Check if a policy statement has a public principal. + + Args: + statement (dict): IAM policy statement + + Returns: + bool: True if the statement has a public principal, False otherwise + """ + principal = statement.get("Principal", "") + return ( + "*" in principal + or "arn:aws:iam::*:root" in principal + or ( + isinstance(principal, dict) + and ( + "*" in principal.get("AWS", "") + or "arn:aws:iam::*:root" in principal.get("AWS", "") + or ( + isinstance(principal.get("AWS"), list) + and ( + "*" in principal["AWS"] + or "arn:aws:iam::*:root" in principal["AWS"] + ) + ) + or "*" in principal.get("CanonicalUser", "") + or "arn:aws:iam::*:root" in principal.get("CanonicalUser", "") + ) + ) + ) + + +def has_restrictive_source_arn_condition( + statement: dict, source_account: str = "" +) -> bool: + """ + Check if a policy statement has a restrictive aws:SourceArn condition. + + A SourceArn condition is considered restrictive if: + 1. It doesn't contain overly permissive wildcards (like "*" or "arn:aws:s3:::*") + 2. When source_account is provided, the ARN either contains no account field (like S3 buckets) + or contains the source_account + + Args: + statement (dict): IAM policy statement + source_account (str): The account to check restrictions for (optional) + + Returns: + bool: True if the statement has a restrictive aws:SourceArn condition, False otherwise + """ + if "Condition" not in statement: + return False + + for condition_operator in statement["Condition"]: + for condition_key, condition_value in statement["Condition"][ + condition_operator + ].items(): + if condition_key.lower() == "aws:sourcearn": + arn_values = ( + condition_value + if isinstance(condition_value, list) + else [condition_value] + ) + + for arn_value in arn_values: + if ( + arn_value == "*" # Global wildcard + or arn_value.count("*") + >= 3 # Too many wildcards (e.g., arn:aws:*:*:*:*) + or ( + isinstance(arn_value, str) + and ( + arn_value.endswith( + ":::*" + ) # Service-wide wildcard (e.g., arn:aws:s3:::*) + or arn_value.endswith( + ":*" + ) # Resource wildcard (e.g., arn:aws:sns:us-east-1:123456789012:*) + ) + ) + ): + return False + + if source_account: + arn_parts = arn_value.split(":") + if len(arn_parts) > 4 and arn_parts[4] and arn_parts[4] != "*": + if arn_parts[4].isdigit(): + if source_account not in arn_value: + return False + else: + if arn_parts[4] != source_account: + return False + elif len(arn_parts) > 4 and arn_parts[4] == "*": + return False + # else: ARN doesn't contain account field (like S3 bucket), so it's restrictive + + return True + + return False + + def is_condition_restricting_from_private_ip(condition_statement: dict) -> bool: """Check if the policy condition is coming from a private IP address. @@ -303,61 +405,49 @@ def is_policy_public( for statement in policy.get("Statement", []): # Only check allow statements if statement["Effect"] == "Allow": + has_public_access = has_public_principal(statement) + principal = statement.get("Principal", "") - if ( - "*" in principal - or "arn:aws:iam::*:root" in principal - or ( - isinstance(principal, dict) - and ( - "*" in principal.get("AWS", "") - or "arn:aws:iam::*:root" in principal.get("AWS", "") - or ( - isinstance(principal.get("AWS"), str) - and source_account - and not is_cross_account_allowed - and source_account not in principal.get("AWS", "") - ) - or ( - isinstance(principal.get("AWS"), list) - and ( - "*" in principal["AWS"] - or "arn:aws:iam::*:root" in principal["AWS"] - or ( - source_account - and not is_cross_account_allowed - and not any( - source_account in principal_aws - for principal_aws in principal["AWS"] - ) - ) - ) - ) - or "*" in principal.get("CanonicalUser", "") - or "arn:aws:iam::*:root" - in principal.get("CanonicalUser", "") - or check_cross_service_confused_deputy - and ( - # Check if function can be invoked by other AWS services if check_cross_service_confused_deputy is True - ( - ".amazonaws.com" in principal.get("Service", "") - or ".amazon.com" in principal.get("Service", "") - or "*" in principal.get("Service", "") - ) - and ( - "secretsmanager.amazonaws.com" - not in principal.get( - "Service", "" - ) # AWS ensures that resources called by SecretsManager are executed in the same AWS account - or "eks.amazonaws.com" - not in principal.get( - "Service", "" - ) # AWS ensures that resources called by EKS are executed in the same AWS account - ) - ) + if not has_public_access and isinstance(principal, dict): + # Check for cross-account access when not allowed + if ( + isinstance(principal.get("AWS"), str) + and source_account + and not is_cross_account_allowed + and source_account not in principal.get("AWS", "") + ) or ( + isinstance(principal.get("AWS"), list) + and source_account + and not is_cross_account_allowed + and not any( + source_account in principal_aws + for principal_aws in principal["AWS"] ) - ) - ) and ( + ): + has_public_access = True + + # Check for cross-service confused deputy + if check_cross_service_confused_deputy and ( + # Check if function can be invoked by other AWS services if check_cross_service_confused_deputy is True + ( + ".amazonaws.com" in principal.get("Service", "") + or ".amazon.com" in principal.get("Service", "") + or "*" in principal.get("Service", "") + ) + and ( + "secretsmanager.amazonaws.com" + not in principal.get( + "Service", "" + ) # AWS ensures that resources called by SecretsManager are executed in the same AWS account + or "eks.amazonaws.com" + not in principal.get( + "Service", "" + ) # AWS ensures that resources called by EKS are executed in the same AWS account + ) + ): + has_public_access = True + + if has_public_access and ( not not_allowed_actions # If not_allowed_actions is empty, the function will not consider the actions in the policy or ( statement.get( @@ -498,9 +588,29 @@ def is_condition_block_restrictive( "aws:sourcevpc" != value and "aws:sourcevpce" != value ): - if source_account not in item: - is_condition_key_restrictive = False - break + if value == "aws:sourcearn": + # Use the specialized function to properly validate SourceArn restrictions + # Create a minimal statement to test with our function + test_statement = { + "Condition": { + condition_operator: { + value: condition_statement[ + condition_operator + ][value] + } + } + } + is_condition_key_restrictive = ( + has_restrictive_source_arn_condition( + test_statement, source_account + ) + ) + if not is_condition_key_restrictive: + break + else: + if source_account not in item: + is_condition_key_restrictive = False + break if is_condition_key_restrictive: is_condition_valid = True @@ -516,11 +626,31 @@ def is_condition_block_restrictive( if is_cross_account_allowed: is_condition_valid = True else: - if ( - source_account - in condition_statement[condition_operator][value] - ): - is_condition_valid = True + if value == "aws:sourcearn": + # Use the specialized function to properly validate SourceArn restrictions + # Create a minimal statement to test with our function + test_statement = { + "Condition": { + condition_operator: { + value: condition_statement[ + condition_operator + ][value] + } + } + } + is_condition_valid = ( + has_restrictive_source_arn_condition( + test_statement, source_account + ) + ) + else: + if ( + source_account + 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 1dc0776b64..b6c66dc702 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,5 +1,7 @@ from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.iam.lib.policy import ( + has_public_principal, + has_restrictive_source_arn_condition, is_condition_block_restrictive, is_condition_block_restrictive_organization, is_condition_block_restrictive_sns_endpoint, @@ -16,46 +18,26 @@ class sns_topics_not_publicly_accessible(Check): report.status_extended = ( f"SNS topic {topic.name} is not publicly accessible." ) + if topic.policy: for statement in topic.policy["Statement"]: - # Only check allow statements - if statement["Effect"] == "Allow": - if ( - "*" in statement["Principal"] - or ( - "AWS" in statement["Principal"] - and "*" in statement["Principal"]["AWS"] + if statement["Effect"] == "Allow" and has_public_principal( + statement + ): + if has_restrictive_source_arn_condition(statement): + break + elif "Condition" in statement: + condition_account = is_condition_block_restrictive( + statement["Condition"], sns_client.audited_account ) - or ( - "CanonicalUser" in statement["Principal"] - and "*" in statement["Principal"]["CanonicalUser"] + condition_org = is_condition_block_restrictive_organization( + statement["Condition"] ) - ): - condition_account = False - condition_org = False - condition_endpoint = False - if ( - "Condition" in statement - and is_condition_block_restrictive( - statement["Condition"], - sns_client.audited_account, + condition_endpoint = ( + is_condition_block_restrictive_sns_endpoint( + statement["Condition"] ) - ): - condition_account = True - if ( - "Condition" in statement - and is_condition_block_restrictive_organization( - statement["Condition"], - ) - ): - 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." @@ -69,7 +51,11 @@ class sns_topics_not_publicly_accessible(Check): report.status = "FAIL" report.status_extended = f"SNS topic {topic.name} is public because its policy allows public access." break + else: + # Public principal with no conditions = public + report.status = "FAIL" + report.status_extended = f"SNS topic {topic.name} is public because its policy allows public access." + break findings.append(report) - return findings diff --git a/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_all_ports/ec2_securitygroup_allow_ingress_from_internet_to_all_ports_test.py b/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_all_ports/ec2_securitygroup_allow_ingress_from_internet_to_all_ports_test.py index 98bfe0907e..f08d4cfc44 100644 --- a/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_all_ports/ec2_securitygroup_allow_ingress_from_internet_to_all_ports_test.py +++ b/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_all_ports/ec2_securitygroup_allow_ingress_from_internet_to_all_ports_test.py @@ -404,7 +404,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_all_ports: new=EC2(aws_provider), ), mock.patch( - "prowler.providers.aws.services.vpc.vpc_service.VPC", + "prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client", new=VPC(aws_provider), ), mock.patch( diff --git a/tests/providers/aws/services/iam/lib/policy_test.py b/tests/providers/aws/services/iam/lib/policy_test.py index f8022d6f77..4687d38dab 100644 --- a/tests/providers/aws/services/iam/lib/policy_test.py +++ b/tests/providers/aws/services/iam/lib/policy_test.py @@ -6,6 +6,8 @@ from prowler.providers.aws.services.iam.lib.policy import ( check_full_service_access, get_effective_actions, has_codebuild_trusted_principal, + has_public_principal, + has_restrictive_source_arn_condition, is_codebuild_using_allowed_github_org, is_condition_block_restrictive, is_condition_block_restrictive_organization, @@ -2451,3 +2453,266 @@ def test_has_codebuild_trusted_principal_list(): ], } assert has_codebuild_trusted_principal(trust_policy) is True + + +class Test_has_public_principal: + """Tests for the has_public_principal function""" + + def test_has_public_principal_wildcard_string(self): + """Test public principal detection with wildcard string""" + statement = {"Principal": "*"} + assert has_public_principal(statement) is True + + def test_has_public_principal_root_arn_string(self): + """Test public principal detection with root ARN string""" + statement = {"Principal": "arn:aws:iam::*:root"} + assert has_public_principal(statement) is True + + def test_has_public_principal_aws_dict_wildcard(self): + """Test public principal detection with AWS dict containing wildcard""" + statement = {"Principal": {"AWS": "*"}} + assert has_public_principal(statement) is True + + def test_has_public_principal_aws_dict_root_arn(self): + """Test public principal detection with AWS dict containing root ARN""" + statement = {"Principal": {"AWS": "arn:aws:iam::*:root"}} + assert has_public_principal(statement) is True + + def test_has_public_principal_aws_list_wildcard(self): + """Test public principal detection with AWS list containing wildcard""" + statement = {"Principal": {"AWS": ["arn:aws:iam::123456789012:user/test", "*"]}} + assert has_public_principal(statement) is True + + def test_has_public_principal_aws_list_root_arn(self): + """Test public principal detection with AWS list containing root ARN""" + statement = { + "Principal": { + "AWS": ["arn:aws:iam::123456789012:user/test", "arn:aws:iam::*:root"] + } + } + assert has_public_principal(statement) is True + + def test_has_public_principal_canonical_user_wildcard(self): + """Test public principal detection with CanonicalUser wildcard""" + statement = {"Principal": {"CanonicalUser": "*"}} + assert has_public_principal(statement) is True + + def test_has_public_principal_canonical_user_root_arn(self): + """Test public principal detection with CanonicalUser root ARN""" + statement = {"Principal": {"CanonicalUser": "arn:aws:iam::*:root"}} + assert has_public_principal(statement) is True + + def test_has_public_principal_no_principal(self): + """Test with statement that has no Principal field""" + statement = {"Effect": "Allow", "Action": "s3:GetObject"} + assert has_public_principal(statement) is False + + def test_has_public_principal_empty_principal(self): + """Test with empty principal""" + statement = {"Principal": ""} + assert has_public_principal(statement) is False + + def test_has_public_principal_specific_account(self): + """Test with specific account principal (not public)""" + statement = {"Principal": {"AWS": "arn:aws:iam::123456789012:root"}} + assert has_public_principal(statement) is False + + def test_has_public_principal_service_principal(self): + """Test with service principal (not public)""" + statement = {"Principal": {"Service": "lambda.amazonaws.com"}} + assert has_public_principal(statement) is False + + def test_has_public_principal_mixed_principals(self): + """Test with mixed principals including public one""" + statement = { + "Principal": { + "AWS": ["arn:aws:iam::123456789012:user/test"], + "Service": "lambda.amazonaws.com", + "CanonicalUser": "*", + } + } + assert has_public_principal(statement) is True + + +class Test_has_restrictive_source_arn_condition: + """Tests for the has_restrictive_source_arn_condition function""" + + def test_no_condition_block(self): + """Test statement without Condition block""" + statement = {"Effect": "Allow", "Principal": "*", "Action": "s3:GetObject"} + assert has_restrictive_source_arn_condition(statement) is False + + def test_no_source_arn_condition(self): + """Test with condition block but no aws:SourceArn""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "s3:GetObject", + "Condition": {"StringEquals": {"aws:SourceAccount": "123456789012"}}, + } + assert has_restrictive_source_arn_condition(statement) is False + + def test_restrictive_source_arn_s3_bucket(self): + """Test restrictive SourceArn condition with S3 bucket""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": {"ArnLike": {"aws:SourceArn": "arn:aws:s3:::my-bucket"}}, + } + assert has_restrictive_source_arn_condition(statement) is True + + def test_restrictive_source_arn_lambda_function(self): + """Test restrictive SourceArn condition with Lambda function""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": { + "ArnEquals": { + "aws:SourceArn": "arn:aws:lambda:us-east-1:123456789012:function:MyFunction" + } + }, + } + assert has_restrictive_source_arn_condition(statement) is True + + def test_non_restrictive_global_wildcard(self): + """Test non-restrictive SourceArn with global wildcard""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": {"ArnLike": {"aws:SourceArn": "*"}}, + } + assert has_restrictive_source_arn_condition(statement) is False + + def test_non_restrictive_service_wildcard(self): + """Test non-restrictive SourceArn with service wildcard""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": {"ArnLike": {"aws:SourceArn": "arn:aws:s3:::*"}}, + } + assert has_restrictive_source_arn_condition(statement) is False + + def test_non_restrictive_multi_wildcard(self): + """Test non-restrictive SourceArn with multiple wildcards""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": {"ArnLike": {"aws:SourceArn": "arn:aws:*:*:*:*"}}, + } + assert has_restrictive_source_arn_condition(statement) is False + + def test_non_restrictive_resource_wildcard(self): + """Test non-restrictive SourceArn with resource wildcard""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": { + "ArnLike": {"aws:SourceArn": "arn:aws:lambda:us-east-1:123456789012:*"} + }, + } + assert has_restrictive_source_arn_condition(statement) is False + + def test_source_arn_list_with_valid_arn(self): + """Test SourceArn condition with list containing valid ARN""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": { + "ArnLike": { + "aws:SourceArn": ["arn:aws:s3:::bucket1", "arn:aws:s3:::bucket2"] + } + }, + } + assert has_restrictive_source_arn_condition(statement) is True + + def test_source_arn_list_with_wildcard(self): + """Test SourceArn condition with list containing wildcard""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": {"ArnLike": {"aws:SourceArn": ["arn:aws:s3:::bucket1", "*"]}}, + } + assert has_restrictive_source_arn_condition(statement) is False + + def test_source_arn_with_account_validation_match(self): + """Test SourceArn with account validation - matching account""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": { + "ArnLike": { + "aws:SourceArn": "arn:aws:lambda:us-east-1:123456789012:function:MyFunction" + } + }, + } + assert has_restrictive_source_arn_condition(statement, "123456789012") is True + + def test_source_arn_with_account_validation_mismatch(self): + """Test SourceArn with account validation - non-matching account""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": { + "ArnLike": { + "aws:SourceArn": "arn:aws:lambda:us-east-1:123456789012:function:MyFunction" + } + }, + } + assert has_restrictive_source_arn_condition(statement, "987654321098") is False + + def test_source_arn_with_account_wildcard(self): + """Test SourceArn with account wildcard""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": { + "ArnLike": { + "aws:SourceArn": "arn:aws:lambda:us-east-1:*:function:MyFunction" + } + }, + } + assert has_restrictive_source_arn_condition(statement, "123456789012") is False + + def test_source_arn_s3_bucket_no_account_field(self): + """Test SourceArn with S3 bucket (no account field) - should be restrictive""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": {"ArnLike": {"aws:SourceArn": "arn:aws:s3:::my-bucket"}}, + } + assert has_restrictive_source_arn_condition(statement, "123456789012") is True + + def test_source_arn_case_insensitive(self): + """Test SourceArn condition key is case insensitive""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": {"ArnLike": {"AWS:SourceArn": "arn:aws:s3:::my-bucket"}}, + } + assert has_restrictive_source_arn_condition(statement) is True + + def test_source_arn_mixed_operators(self): + """Test SourceArn with multiple condition operators""" + statement = { + "Effect": "Allow", + "Principal": "*", + "Action": "sns:Publish", + "Condition": { + "ArnLike": {"aws:SourceArn": "arn:aws:s3:::my-bucket"}, + "StringEquals": {"aws:SourceAccount": "123456789012"}, + }, + } + assert has_restrictive_source_arn_condition(statement) is True 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 71aa6c4477..a868be5715 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 @@ -2,9 +2,10 @@ from typing import Any, Dict from unittest import mock from uuid import uuid4 +import pytest + 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" @@ -98,6 +99,73 @@ test_policy_restricted_principal_account_organization = { ] } +test_policy_restricted_source_arn = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"AWS": "*"}, + "Action": "SNS:Publish", + "Resource": f"arn:aws:sns:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{topic_name}", + "Condition": { + "ArnLike": {"aws:SourceArn": "arn:aws:s3:::test-bucket-name"} + }, + } + ], +} + +test_policy_invalid_source_arn = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"AWS": "*"}, + "Action": "SNS:Publish", + "Resource": f"arn:aws:sns:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{topic_name}", + "Condition": {"ArnLike": {"aws:SourceArn": "invalid-arn-format"}}, + } + ], +} + +test_policy_unrestricted_source_arn_wildcard = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"AWS": "*"}, + "Action": "SNS:Publish", + "Resource": f"arn:aws:sns:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{topic_name}", + "Condition": {"ArnLike": {"aws:SourceArn": "*"}}, + } + ], +} + +test_policy_unrestricted_source_arn_service_wildcard = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"AWS": "*"}, + "Action": "SNS:Publish", + "Resource": f"arn:aws:sns:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{topic_name}", + "Condition": {"ArnLike": {"aws:SourceArn": "arn:aws:s3:::*"}}, + } + ], +} + +test_policy_unrestricted_source_arn_multi_wildcard = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"AWS": "*"}, + "Action": "SNS:Publish", + "Resource": f"arn:aws:sns:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:{topic_name}", + "Condition": {"ArnLike": {"aws:SourceArn": "arn:aws:*:*:*:*"}}, + } + ], +} + def generate_policy_restricted_on_sns_endpoint(endpoint: str) -> Dict[str, Any]: return { @@ -396,6 +464,78 @@ class Test_sns_topics_not_publicly_accessible: assert result[0].region == AWS_REGION_EU_WEST_1 assert result[0].resource_tags == [] + def test_topic_public_with_source_arn_restriction(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_source_arn, + 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 publicly accessible." + ) + 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_invalid_source_arn(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_invalid_source_arn, + 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 publicly accessible." + ) + 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", [ @@ -443,6 +583,114 @@ class Test_sns_topics_not_publicly_accessible: assert result[0].region == AWS_REGION_EU_WEST_1 assert result[0].resource_tags == [] + def test_topic_public_with_unrestricted_source_arn_wildcard(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_unrestricted_source_arn_wildcard, + 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_unrestricted_source_arn_service_wildcard(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_unrestricted_source_arn_service_wildcard, + 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_unrestricted_source_arn_multi_wildcard(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_unrestricted_source_arn_multi_wildcard, + 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 == [] + @pytest.mark.parametrize( "endpoint", [