fix(aws): false negative in iam_role_cross_service_confused_deputy_prevention (#9213)

Co-authored-by: shaun <shaun@snotra.cloud>
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>

prowler/CHANGELOG.md

@@ -53,6 +53,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 ### Fixed
 - Check `check_name` has no `resource_name` error for GCP provider [(#9169)](https://github.com/prowler-cloud/prowler/pull/9169)
 - Depth Truncation and parsing error in PowerShell queries [(#9181)](https://github.com/prowler-cloud/prowler/pull/9181)
+- False negative in `iam_role_cross_service_confused_deputy_prevention` check [(#9213)](https://github.com/prowler-cloud/prowler/pull/9213)
 - Fix M365 Teams `--sp-env-auth` connection error and enhanced timeout logging [(#9191)](https://github.com/prowler-cloud/prowler/pull/9191)
 - Rename `get_oci_assessment_summary` to `get_oraclecloud_assessment_summary` in HTML output [(#9200)](https://github.com/prowler-cloud/prowler/pull/9200)
 

prowler/providers/aws/services/iam/lib/policy.py

@@ -427,25 +427,33 @@ def is_policy_public(
                         has_public_access = True
 
                     # Check for cross-service confused deputy
-                    if check_cross_service_confused_deputy and (
+                    if check_cross_service_confused_deputy:
                         # 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", "")
+
+                        svc = principal.get("Service", [])
+                        if isinstance(svc, str):
+                            services = [svc]
+                        elif isinstance(svc, list):
+                            services = [s for s in svc if isinstance(s, str)]
+                        else:
+                            services = []
+
+                        is_cross_service = any(
+                            s == "*"
+                            or s.endswith(".amazonaws.com")
+                            or s.endswith(".amazon.com")
+                            for s in services
                         )
-                        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
+
+                        # AWS ensures that resources called by SecretsManager are executed in the same AWS account
+                        # AWS ensures that resources called by EKS are executed in the same AWS account
+                        is_exempt = any(
+                            s in {"secretsmanager.amazonaws.com", "eks.amazonaws.com"}
+                            for s in services
                         )
-                    ):
-                        has_public_access = True
+
+                        if is_cross_service and not is_exempt:
+                            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

tests/providers/aws/services/iam/iam_role_cross_service_confused_deputy_prevention/iam_role_cross_service_confused_deputy_prevention_test.py

@@ -338,3 +338,55 @@ class Test_iam_role_cross_service_confused_deputy_prevention:
             )
             assert result[0].resource_id == "test"
             assert result[0].resource_arn == response["Role"]["Arn"]
+
+    @mock_aws
+    def test_iam_service_role_with_cross_service_confused_deputy_prevention_service_list(
+        self,
+    ):
+        iam_client = client("iam", region_name=AWS_REGION)
+        policy_document = {
+            "Version": "2008-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {
+                        "Service": ["scheduler.amazonaws.com", "events.amazonaws.com"]
+                    },
+                    "Action": "sts:AssumeRole",
+                }
+            ],
+        }
+        response = iam_client.create_role(
+            RoleName="test",
+            AssumeRolePolicyDocument=dumps(policy_document),
+        )
+
+        from prowler.providers.aws.services.iam.iam_service import IAM
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        aws_provider.identity.account = AWS_ACCOUNT_ID
+        with (
+            mock.patch(
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=aws_provider,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.iam.iam_role_cross_service_confused_deputy_prevention.iam_role_cross_service_confused_deputy_prevention.iam_client",
+                new=IAM(aws_provider),
+            ),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.iam.iam_role_cross_service_confused_deputy_prevention.iam_role_cross_service_confused_deputy_prevention import (
+                iam_role_cross_service_confused_deputy_prevention,
+            )
+
+            check = iam_role_cross_service_confused_deputy_prevention()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "IAM Service Role test does not prevent against a cross-service confused deputy attack."
+            )
+            assert result[0].resource_id == "test"
+            assert result[0].resource_arn == response["Role"]["Arn"]