ghndrx/prowler
1
0
Fork 0
mirror of https://github.com/ghndrx/prowler.git synced 2026-02-10 14:55:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sns): handle topic policy conditions (#2660)

Browse Source
This commit is contained in:
Sergio Garcia
2023-08-02 11:45:27 +02:00
committed by GitHub
parent 49c9d2b077
commit aced44f051
2 changed files with 15 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.py
View File

@@ -1,4 +1,7 @@
from prowler.lib.check.models import Check, Check_Report_AWS from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
is_account_only_allowed_in_condition,
)
from prowler.providers.aws.services.sns.sns_client import sns_client from prowler.providers.aws.services.sns.sns_client import sns_client
@@ -28,14 +31,16 @@ class sns_topics_not_publicly_accessible(Check):
and "*" in statement["Principal"]["CanonicalUser"] and "*" in statement["Principal"]["CanonicalUser"]
) )
): ):
if "Condition" not in statement: if (
report.status = "FAIL" "Condition" in statement
report.status_extended = ( and is_account_only_allowed_in_condition(
f"SNS topic {topic.name} is publicly accesible" 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"
else: else:
report.status = "PASS" report.status = "FAIL"
report.status_extended = f"SNS topic {topic.name} is publicly accesible but has a Condition that could filter it" report.status_extended = f"SNS topic {topic.name} is public because its policy allows public access"
findings.append(report) findings.append(report)

7
tests/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible_test.py
View File

@@ -27,7 +27,7 @@ test_policy_restricted_condition = {
"Principal": {"AWS": "*"}, "Principal": {"AWS": "*"},
"Action": ["sns:Publish"], "Action": ["sns:Publish"],
"Resource": f"arn:aws:sns:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:{topic_name}", "Resource": f"arn:aws:sns:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:{topic_name}",
"Condition": {"StringEquals": {"sns:Protocol": "https"}}, "Condition": {"StringEquals": {"aws:SourceAccount": AWS_ACCOUNT_NUMBER}},
} }
] ]
} }
@@ -121,6 +121,7 @@ class Test_sns_topics_not_publicly_accessible:
def test_topic_public_with_condition(self): def test_topic_public_with_condition(self):
sns_client = mock.MagicMock sns_client = mock.MagicMock
sns_client.audited_account = AWS_ACCOUNT_NUMBER
sns_client.topics = [] sns_client.topics = []
sns_client.topics.append( sns_client.topics.append(
Topic( Topic(
@@ -144,7 +145,7 @@ class Test_sns_topics_not_publicly_accessible:
assert result[0].status == "PASS" assert result[0].status == "PASS"
assert ( assert (
result[0].status_extended result[0].status_extended
== f"SNS topic {topic_name} is publicly accesible but has a Condition that could filter it" == f"SNS topic {topic_name} is not public because its policy only allows access from the same account"
) )
assert result[0].resource_id == topic_name assert result[0].resource_id == topic_name
assert result[0].resource_arn == topic_arn assert result[0].resource_arn == topic_arn
@@ -176,7 +177,7 @@ class Test_sns_topics_not_publicly_accessible:
assert result[0].status == "FAIL" assert result[0].status == "FAIL"
assert ( assert (
result[0].status_extended result[0].status_extended
== f"SNS topic {topic_name} is publicly accesible" == f"SNS topic {topic_name} is public because its policy allows public access"
) )
assert result[0].resource_id == topic_name assert result[0].resource_id == topic_name
assert result[0].resource_arn == topic_arn assert result[0].resource_arn == topic_arn