chore: solve comments

prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible.py

@@ -17,6 +17,7 @@ def execute(self):
             report.status_extended = f"Lambda function {function.name} has a policy resource-based policy not public."
             if is_policy_public(
                 function.policy,
+                awslambda_client.audited_account,
                 is_cross_account_allowed=True,
             ):
                 report.status = "FAIL"

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

@@ -20,6 +20,7 @@ def execute(self):
                 report.status_extended = f"DynamoDB table {table.name} has a resource-based policy but is not cross account."
                 if is_policy_public(
                     table.policy,
+                    dynamodb_client.audited_account,
                 ):
                     report.status = "FAIL"
                     report.status_extended = f"DynamoDB table {table.name} has a resource-based policy allowing cross account access."

prowler/providers/aws/services/ecr/ecr_repositories_not_publicly_accessible/ecr_repositories_not_publicly_accessible.py

@@ -18,7 +18,7 @@ def execute(self):
                     f"Repository {repository.name} is not publicly accessible."
                 )
                 if repository.policy:
-                    if is_policy_public(repository.policy):
+                    if is_policy_public(repository.policy, ecr_client.audited_account):
                         report.status = "FAIL"
                         report.status_extended = (
                             f"Repository {repository.name} is publicly accessible."

prowler/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible.py

@@ -17,7 +17,7 @@ def execute(self):
             if not fs.policy:
                 report.status = "FAIL"
                 report.status_extended = f"EFS {fs.id} doesn't have any policy which means it grants full access to any client within the VPC."
-            elif is_policy_public(fs.policy) and any(
+            elif is_policy_public(fs.policy, efs_client.audited_account) and any(
                 statement.get("Condition", {})
                 .get("Bool", {})
                 .get("elasticfilesystem:AccessedViaMountTarget", "false")

prowler/providers/aws/services/eventbridge/eventbridge_bus_exposed/eventbridge_bus_exposed.py

@@ -18,7 +18,7 @@ def execute(self):
             report.resource_arn = bus.arn
             report.resource_tags = bus.tags
             report.region = bus.region
-            if is_policy_public(bus.policy):
+            if is_policy_public(bus.policy, eventbridge_client.audited_account):
                 report.status = "FAIL"
                 report.status_extended = (
                     f"EventBridge event bus {bus.name} is exposed to everyone."

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

@@ -19,6 +19,7 @@ def execute(self) -> Check_Report_AWS:
                     report.status_extended = f"IAM Service Role {role.name} does not prevent against a cross-service confused deputy attack."
                     if not is_policy_public(
                         role.assume_role_policy,
+                        iam_client.audited_account,
                         not_allowed_actions=["sts:AssumeRole", "sts:*"],
                     ):
                         report.status = "PASS"

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

@@ -1,4 +1,3 @@
-import inspect
 from ipaddress import ip_address, ip_network
 
 from prowler.lib.logger import logger
@@ -151,6 +150,7 @@ def is_condition_restricting_from_private_ip(condition_statement: dict) -> bool:
 # TODO: Add logic for deny statements
 def is_policy_public(
     policy: dict,
+    source_account: str = "",
     is_cross_account_allowed=False,
     not_allowed_actions: list = [],
 ) -> bool:
@@ -159,20 +159,13 @@ def is_policy_public(
     If the policy gives access to an AWS service principal is considered public if the policy is not pair with conditions since it can be invoked by AWS services in other accounts.
     Args:
         policy (dict): The AWS policy to check
+        source_account (str): The account to check if the access is restricted to it, default: ""
         is_cross_account_allowed (bool): If the policy can allow cross-account access, default: False
         not_allowed_actions (list): List of actions that are not allowed, default: []. If not_allowed_actions is empty, the function will not consider the actions in the policy.
     Returns:
         bool: True if the policy allows public access, False otherwise
     """
     is_public = False
-    source_account = ""
-    trusted_account_ids = []
-    # Get service client from the check to get the trusted account IDs and source account
-    for var_name, value in inspect.stack()[1].frame.f_globals.items():
-        if "_client" in var_name:
-            trusted_account_ids = value.audit_config.get("trusted_account_ids", [])
-            source_account = value.audited_account
-            break
     for statement in policy.get("Statement", []):
         # Only check allow statements
         if statement["Effect"] == "Allow":
@@ -240,7 +233,6 @@ def is_policy_public(
                         statement.get("Condition", {}),
                         source_account,
                         is_cross_account_allowed,
-                        trusted_account_ids,
                     )
                     and not is_condition_block_restrictive_organization(
                         statement.get("Condition", {})
@@ -258,7 +250,6 @@ def is_condition_block_restrictive(
     condition_statement: dict,
     source_account: str = "",
     is_cross_account_allowed=False,
-    trusted_account_ids: list = [],
 ):
     """
     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.
@@ -275,8 +266,6 @@ def is_condition_block_restrictive(
 
         source_account: str with a 12-digit AWS Account number, e.g.: 111122223333, default: ""
 
-        trusted_account_ids: list with a list of trusted AWS Account numbers, e.g.: ["111122223333"], default: []
-
         is_cross_account_allowed: bool to allow cross-account access, e.g.: True, default: False
 
     """
@@ -345,20 +334,16 @@ def is_condition_block_restrictive(
                         # if cross account is not allowed check for each condition block looking for accounts
                         # different than default
                         if not is_cross_account_allowed:
-                            # if there is an arn/account without the source account or trusted accounts -> we do not consider it safe
+                            # 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 (
                                     "aws:sourcevpc" != value
                                     and "aws:sourcevpce" != value
                                 ):
                                     if source_account not in item:
-                                        if not any(
-                                            trusted_account_id in item
-                                            for trusted_account_id in trusted_account_ids
-                                        ):
-                                            is_condition_key_restrictive = False
-                                            break
+                                        is_condition_key_restrictive = False
+                                        break
 
                         if is_condition_key_restrictive:
                             is_condition_valid = True
@@ -377,10 +362,6 @@ def is_condition_block_restrictive(
                                 if (
                                     source_account
                                     in condition_statement[condition_operator][value]
-                                ) or any(
-                                    trusted_account_id
-                                    in condition_statement[condition_operator][value]
-                                    for trusted_account_id in trusted_account_ids
                                 ):
                                     is_condition_valid = True
 

prowler/providers/aws/services/kms/kms_key_not_publicly_accessible/kms_key_not_publicly_accessible.py

@@ -18,7 +18,11 @@ def execute(self):
                 report.resource_tags = key.tags
                 report.region = key.region
                 # If the "Principal" element value is set to { "AWS": "*" } and the policy statement is not using any Condition clauses to filter the access, the selected AWS KMS master key is publicly accessible.
-                if is_policy_public(key.policy, not_allowed_actions=["kms:*"]):
+                if is_policy_public(
+                    key.policy,
+                    kms_client.audited_account,
+                    not_allowed_actions=["kms:*"],
+                ):
                     report.status = "FAIL"
                     report.status_extended = (
                         f"KMS key {key.id} may be publicly accessible."

prowler/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible.py

@@ -21,7 +21,9 @@ def execute(self):
 
             if domain.vpc_id:
                 report.status_extended = f"Opensearch domain {domain.name} is in a VPC, then it is not publicly accessible."
-            elif domain.access_policy and is_policy_public(domain.access_policy):
+            elif domain.access_policy and is_policy_public(
+                domain.access_policy, opensearch_client.audited_account
+            ):
                 report.status = "FAIL"
                 report.status_extended = f"Opensearch domain {domain.name} is publicly accessible via access policy."
 

prowler/providers/aws/services/redshift/redshift_cluster_public_access/redshift_cluster_public_access.py

@@ -21,13 +21,12 @@ def execute(self):
             # 1. Check if Redshift Cluster is publicly accessible
             if cluster.endpoint_address and cluster.public_access:
                 report.status_extended = f"Redshift Cluster {cluster.id} has the endpoint {cluster.endpoint_address} set as publicly accessible but is not publicly exposed."
-                # 2. Check if Redshift Cluster is in a public VPC
-                if (
-                    cluster.vpc_id
-                    and cluster.vpc_id in vpc_client.vpcs
-                    and vpc_client.vpcs[cluster.vpc_id].public
+                # 2. Check if Redshift Cluster is in a public subnet
+                if any(
+                    subnet in vpc_client.subnets and vpc_client.subnets[subnet].public
+                    for subnet in cluster.subnets
                 ):
-                    report.status_extended = f"Redshift Cluster {cluster.id} has the endpoint {cluster.endpoint_address} set as publicly accessible in the public VPC {cluster.vpc_id} but is not publicly exposed."
+                    report.status_extended = f"Redshift Cluster {cluster.id} has the endpoint {cluster.endpoint_address} set as publicly accessible in a public subnet but is not publicly exposed."
                     # 3. Check if any Redshift Cluster Security Group is publicly open
                     for sg_id in getattr(cluster, "vpc_security_groups", []):
                         sg_arn = f"arn:{redshift_client.audited_partition}:ec2:{cluster.region}:{redshift_client.audited_account}:security-group/{sg_id}"
@@ -39,7 +38,7 @@ def execute(self):
                                     ingress_rule, "tcp", any_address=True
                                 ):
                                     report.status = "FAIL"
-                                    report.status_extended = f"Redshift Cluster {cluster.id} has the endpoint {cluster.endpoint_address} set as publicly accessible and it is exposed to the Internet by security group ({sg_id}) in public VPC {cluster.vpc_id}."
+                                    report.status_extended = f"Redshift Cluster {cluster.id} has the endpoint {cluster.endpoint_address} set as publicly accessible and it is exposed to the Internet by security group ({sg_id}) in a public subnet."
                                     break
                         if report.status == "FAIL":
                             break

prowler/providers/aws/services/redshift/redshift_service.py

@@ -16,6 +16,7 @@ def __init__(self, provider):
         self.__threading_call__(self._describe_logging_status, self.clusters)
         self.__threading_call__(self._describe_cluster_snapshots, self.clusters)
         self.__threading_call__(self._describe_cluster_parameters, self.clusters)
+        self.__threading_call__(self._describe_cluster_subnets, self.clusters)
 
     def _describe_clusters(self, regional_client):
         logger.info("Redshift - Describing Clusters...")
@@ -54,13 +55,34 @@ def _describe_clusters(self, regional_client):
                             parameter_group_name=cluster.get(
                                 "ClusterParameterGroups", [{}]
                             )[0].get("ParameterGroupName", ""),
+                            subnet_group=cluster.get("ClusterSubnetGroupName", ""),
                         )
                         self.clusters.append(cluster_to_append)
         except Exception as error:
             logger.error(
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _describe_cluster_subnets(self, cluster):
+        logger.info("Redshift - Describing Cluster Subnets...")
+        try:
+            regional_client = self.regional_clients[cluster.region]
+            if cluster.subnet_group:
+                subnet_group_details = regional_client.describe_cluster_subnet_groups(
+                    ClusterSubnetGroupName=cluster.subnet_group
+                )
+                subnets = [
+                    subnet["SubnetIdentifier"]
+                    for subnet in subnet_group_details["ClusterSubnetGroups"][0][
+                        "Subnets"
+                    ]
+                ]
+                cluster.subnets = subnets
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
     def _describe_logging_status(self, cluster):
         logger.info("Redshift - Describing Logging Status...")
         try:
@@ -133,3 +155,5 @@ class Cluster(BaseModel):
     enhanced_vpc_routing: bool = False
     parameter_group_name: str = None
     require_ssl: bool = False
+    subnet_group: str = None
+    subnets: list[str] = []

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

@@ -20,7 +20,7 @@ def execute(self):
                 report.status_extended = (
                     f"S3 Bucket {bucket.name} does not have a bucket policy."
                 )
-            elif is_policy_public(bucket.policy):
+            elif is_policy_public(bucket.policy, s3_client.audited_account):
                 report.status = "FAIL"
                 report.status_extended = f"S3 Bucket {bucket.name} has a bucket policy allowing cross account access."
 

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

@@ -40,6 +40,7 @@ def execute(self):
                 report.status_extended = f"S3 Bucket {bucket.name} does not allow public write access in the bucket policy."
                 if is_policy_public(
                     bucket.policy,
+                    s3_client.audited_account,
                     not_allowed_actions=[
                         "s3:PutObject",
                         "s3:DeleteObject",

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

@@ -46,7 +46,7 @@ def execute(self):
                                     report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket ACL."
 
                         # 4. Check bucket policy
-                        if is_policy_public(bucket.policy):
+                        if is_policy_public(bucket.policy, s3_client.audited_account):
                             report.status = "FAIL"
                             report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket policy."
                     findings.append(report)