mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 20:42:02 +00:00
fix(aws): sns_topics_not_publicly_accessible false positive with aws:SourceArn conditions (#8340)
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>
This commit is contained in:
Andoni Alonso
@@ -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)
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
+22
-36
@@ -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
|
||||
|
||||
+1
-1
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
+249
-1
@@ -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",
|
||||
[
|
||||
|
||||
Reference in New Issue
Block a user