ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-20 11:01:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sns): add condition to sns topics (#4498)

Browse Source 
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
This commit is contained in:
Pedro Martín
2024-08-01 17:54:36 +02:00
committed by GitHub
parent 32da86f393
commit bbc51114b0
4 changed files with 289 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py
+66 -1
View File
@@ -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
prowler/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible.py
+20 -2
View File
@@ -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."
tests/providers/aws/lib/policy_condition_parser/policy_condition_parser_test.py
+48
View File
@@ -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)
tests/providers/aws/services/sns/sns_topics_not_publicly_accessible/sns_topics_not_publicly_accessible_test.py
+155 -2
View File
@@ -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 == []