chore(sns): refactor method and add tests

.github/workflows/pull-request.yml

@@ -73,7 +73,7 @@ jobs:
       - name: Safety
         if: steps.are-non-ignored-files-changed.outputs.any_changed == 'true'
         run: |
-          poetry run safety check --ignore 70612
+          poetry run safety check --ignore 70612,40459
       - name: Vulture
         if: steps.are-non-ignored-files-changed.outputs.any_changed == 'true'
         run: |

.pre-commit-config.yaml

@@ -97,7 +97,7 @@ repos:
       - id: safety
         name: safety
         description: "Safety is a tool that checks your installed dependencies for known security vulnerabilities"
-        entry: bash -c 'safety check --ignore 70612'
+        entry: bash -c 'safety check --ignore 70612,40459'
         language: system
 
       - id: vulture

docs/tutorials/configuration_file.md

@@ -43,6 +43,7 @@ The following list includes all the AWS checks with configurable variables that
 | `ec2_securitygroup_allow_ingress_from_internet_to_any_port`   | `ec2_allowed_interface_types`                    | List of Strings |
 | `ec2_securitygroup_allow_ingress_from_internet_to_any_port`   | `ec2_allowed_instance_owners`                    | List of Strings |
 | `acm_certificates_expiration_check`                           | `days_to_expire_threshold`                       | Integer         |
+| `sns_topics_not_publicly_accessible`                          | `organization_id`                                | String          |
 
 
 ## Azure
@@ -355,6 +356,10 @@ aws:
   # aws.acm_certificates_expiration_check
   days_to_expire_threshold: 7
 
+  # AWS SNS Configuration
+  # aws.sns_topics_not_publicly_accessible
+  organization_id: null
+
 # Azure Configuration
 azure:
   # Azure Network Configuration

prowler/config/config.yaml

@@ -272,6 +272,10 @@ aws:
   # aws.acm_certificates_expiration_check
   days_to_expire_threshold: 7
 
+  # AWS SNS Configuration
+  # aws.sns_topics_not_publicly_accessible
+  organization_id: null
+
 # Azure Configuration
 azure:
   # Azure Network Configuration

prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py

@@ -1,11 +1,10 @@
-def is_condition_block_restrictive(
+def is_condition_block_restrictive_account(
     condition_statement: dict,
     source_account: str,
     is_cross_account_allowed=False,
-    org_id: str = None,
 ):
     """
-    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.
+    is_condition_block_restrictive_account parses the IAM Condition policy block and, by default, returns True if the source_account passed as argument is within, False if not.
 
     If argument is_cross_account_allowed is True it tests if the Condition block includes any of the operators allowlisted returning True if does, False if not.
 
@@ -21,7 +20,6 @@ def is_condition_block_restrictive(
 
     @param is_cross_account_allowed: bool to allow cross-account access, e.g.: True
 
-    @param org_id: str with AWS Organization ID, e.g.: o-123456789012
     """
     is_condition_valid = False
 
@@ -77,7 +75,7 @@ def is_condition_block_restrictive(
                             # if there is an arn/account without the source account -> we do not consider it safe
                             # here by default we assume is true and look for false entries
                             for item in condition_statement[condition_operator][value]:
-                                if source_account not in item or org_id not in item:
+                                if source_account not in item:
                                     is_condition_key_restrictive = False
                                     break
 
@@ -95,9 +93,80 @@ def is_condition_block_restrictive(
                             if (
                                 source_account
                                 in condition_statement[condition_operator][value]
-                                or org_id
-                                in condition_statement[condition_operator][value]
                             ):
                                 is_condition_valid = True
 
     return is_condition_valid
+
+
+def is_condition_block_restrictive_organization(
+    condition_statement: dict,
+    source_organization: str,
+):
+    """
+    is_condition_block_restrictive_organization parses the IAM Condition policy block and returns True if the source_organization passed as argument is within, False if not.
+
+
+    @param condition_statement: dict with an IAM Condition block, e.g.:
+        {
+            "StringLike": {
+                "AWS:PrincipalOrgID": "o-111122223333"
+            }
+        }
+
+    @param source_organization: str with a 12-digit AWS Organization ID, e.g.: 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_key_restrictive = True
+                        # if there is an arn/organization without the source organization -> we do not consider it safe
+                        # here by default we assume is true and look for false entries
+                        for item in condition_statement[condition_operator][value]:
+                            if source_organization not in item:
+                                is_condition_key_restrictive = False
+                                break
+
+                        if is_condition_key_restrictive:
+                            is_condition_valid = True
+
+                    # value is a string
+                    elif isinstance(
+                        condition_statement[condition_operator][value],
+                        str,
+                    ):
+                        if (
+                            source_organization
+                            in condition_statement[condition_operator][value]
+                        ):
+                            is_condition_valid = True
+
+    return is_condition_valid

prowler/providers/aws/services/dynamodb/dynamodb_table_cross_account_access/dynamodb_table_cross_account_access.py

@@ -1,6 +1,6 @@
 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_account,
 )
 from prowler.providers.aws.services.dynamodb.dynamodb_client import dynamodb_client
 
@@ -43,7 +43,7 @@ def execute(self):
                                     cross_account_access = True
                                     # Check if the condition block is restrictive
                                     conditions = statement.get("Condition", {})
-                                    if is_condition_block_restrictive(
+                                    if is_condition_block_restrictive_account(
                                         conditions, dynamodb_client.audited_account
                                     ):
                                         cross_account_access = False

prowler/providers/aws/services/iam/iam_role_cross_service_confused_deputy_prevention/iam_role_cross_service_confused_deputy_prevention.py

@@ -1,6 +1,6 @@
 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_account,
 )
 from prowler.providers.aws.services.iam.iam_client import iam_client
 
@@ -31,7 +31,7 @@ def execute(self) -> Check_Report_AWS:
                             and "Service" in statement["Principal"]
                             # Check to see if the appropriate condition statements have been implemented
                             and "Condition" in statement
-                            and is_condition_block_restrictive(
+                            and is_condition_block_restrictive_account(
                                 statement["Condition"], iam_client.audited_account
                             )
                         ):

prowler/providers/aws/services/s3/s3_bucket_public_access/s3_bucket_public_access.py

@@ -1,6 +1,6 @@
 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_account,
 )
 from prowler.providers.aws.services.iam.lib.policy import (
     is_condition_restricting_from_private_ip,
@@ -56,7 +56,7 @@ def execute(self):
                                 if (
                                     "Principal" in statement
                                     and statement["Effect"] == "Allow"
-                                    and not is_condition_block_restrictive(
+                                    and not is_condition_block_restrictive_account(
                                         statement.get("Condition", {}), "", True
                                     )
                                     and (

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_account,
+    is_condition_block_restrictive_organization,
 )
 from prowler.providers.aws.services.sns.sns_client import sns_client
 
@@ -9,6 +10,10 @@ class sns_topics_not_publicly_accessible(Check):
     def execute(self):
         findings = []
         for topic in sns_client.topics:
+            # Get the organization id from the provider if it is not available in the client
+            org_id = sns_client.provider.organizations_metadata.organization_id
+            if org_id is None:
+                sns_client.audit_config.get("organization_id", None)
             report = Check_Report_AWS(self.metadata())
             report.region = topic.region
             report.resource_id = topic.name
@@ -33,15 +38,32 @@ def execute(self):
                                 and "*" in statement["Principal"]["CanonicalUser"]
                             )
                         ):
+                            condition_account = False
+                            condition_org = False
                             if (
                                 "Condition" in statement
-                                and is_condition_block_restrictive(
+                                and is_condition_block_restrictive_account(
                                     statement["Condition"],
                                     sns_client.audited_account,
-                                    sns_client.audited_org_id,
                                 )
                             ):
-                                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 org_id is not None
+                                and is_condition_block_restrictive_organization(
+                                    statement["Condition"],
+                                    org_id,
+                                )
+                            ):
+                                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 organization {org_id}."
+                            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 the organization {org_id}."
                             else:
                                 report.status = "FAIL"
                                 report.status_extended = f"SNS topic {topic.name} is public because its policy allows public access."

prowler/providers/aws/services/sqs/sqs_queues_not_publicly_accessible/sqs_queues_not_publicly_accessible.py

@@ -1,6 +1,6 @@
 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_account,
 )
 from prowler.providers.aws.services.sqs.sqs_client import sqs_client
 
@@ -32,7 +32,7 @@ def execute(self):
                             )
                         ):
                             if "Condition" in statement:
-                                if is_condition_block_restrictive(
+                                if is_condition_block_restrictive_account(
                                     statement["Condition"],
                                     sqs_client.audited_account,
                                     True,

prowler/providers/aws/services/vpc/vpc_endpoint_connections_trust_boundaries/vpc_endpoint_connections_trust_boundaries.py

@@ -2,7 +2,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_account,
 )
 from prowler.providers.aws.services.vpc.vpc_client import vpc_client
 
@@ -35,7 +35,7 @@ def execute(self):
 
                         if "Condition" in statement:
                             for account_id in trusted_account_ids:
-                                if is_condition_block_restrictive(
+                                if is_condition_block_restrictive_account(
                                     statement["Condition"], account_id
                                 ):
                                     access_from_trusted_accounts = True
@@ -74,7 +74,7 @@ def execute(self):
                                 access_from_trusted_accounts = False
                                 if "Condition" in statement:
                                     for account_id in trusted_account_ids:
-                                        if is_condition_block_restrictive(
+                                        if is_condition_block_restrictive_account(
                                             statement["Condition"], account_id
                                         ):
                                             access_from_trusted_accounts = True
@@ -106,7 +106,7 @@ def execute(self):
 
                                 if "Condition" in statement:
                                     for account_id in trusted_account_ids:
-                                        if is_condition_block_restrictive(
+                                        if is_condition_block_restrictive_account(
                                             statement["Condition"], account_id
                                         ):
                                             access_from_trusted_accounts = True

tests/config/config_test.py

@@ -255,6 +255,7 @@ def mock_prowler_get_latest_release(_, **kwargs):
     ],
     "check_rds_instance_replicas": False,
     "days_to_expire_threshold": 7,
+    "organization_id": None,
 }
 
 config_azure = {

tests/config/fixtures/config.yaml

@@ -272,6 +272,10 @@ aws:
   # aws.acm_certificates_expiration_check
   days_to_expire_threshold: 7
 
+  # AWS SNS Configuration
+  # aws.sns_topics_not_publicly_accessible
+  organization_id: null
+
 # Azure Configuration
 azure:
   # Azure Network Configuration