chore(aws): improve logic for determining if resources are publicly accessible (#5195)

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

@@ -20,7 +20,8 @@ class dynamodb_table_cross_account_access(Check):
                 report.status_extended = f"DynamoDB table {table.name} has a resource-based policy but is not cross account."
                 if is_policy_public(
                     table.policy,
-                    source_account=dynamodb_client.audited_account,
+                    dynamodb_client.audited_account,
+                    is_cross_account_allowed=False,
                 ):
                     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 @@ class ecr_repositories_not_publicly_accessible(Check):
                     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 @@ class efs_not_publicly_accessible(Check):
             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/eks/eks_cluster_not_publicly_accessible/eks_cluster_not_publicly_accessible.metadata.json

@@ -0,0 +1,38 @@
+{
+  "Provider": "aws",
+  "CheckID": "eks_cluster_not_publicly_accessible",
+  "CheckTitle": "Ensure EKS Clusters are not publicly accessible",
+  "CheckAliases": [
+    "eks_endpoints_not_publicly_accessible",
+    "eks_control_plane_endpoint_access_restricted"
+  ],
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices/Network Reachability"
+  ],
+  "ServiceName": "eks",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "high",
+  "ResourceType": "AwsEksCluster",
+  "Description": "Ensure that your Amazon EKS cluster's Kubernetes API server endpoint is not publicly accessible from the Internet in order to avoid exposing private data and minimizing security risks.",
+  "Risk": "By default, this API server endpoint is publicly accessible, meaning any machine on the internet can potentially connect to your EKS cluster using its public endpoint. This exposes your cluster to a higher risk of malicious activities and attacks.",
+  "RelatedUrl": "https://docs.aws.amazon.com/securityhub/latest/userguide/eks-controls.html#eks-1",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws eks update-cluster-config --region <region_name> --name <cluster_name> --resources-vpc-config endpointPublicAccess=false,endpointPrivateAccess=true,publicAccessCidrs=[\"123.123.123.123/32\"]",
+      "NativeIaC": "",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/EKS/endpoint-access.html",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Restricting public access to the Kubernetes API endpoint managed by the EKS cluster is a security best practice that helps protect your cluster from unauthorized access and potential security threats. By not allowing public access to the cluster's Kubernetes API endpoint, you ensure that only authorized entities can interact with your Amazon EKS cluster.",
+      "Url": "https://docs.aws.amazon.com/eks/latest/userguide/cluster-endpoint.html"
+    }
+  },
+  "Categories": [
+    "internet-exposed"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/eks/eks_cluster_not_publicly_accessible/eks_cluster_not_publicly_accessible.py

@@ -2,7 +2,7 @@ from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.eks.eks_client import eks_client
 
 
-class eks_endpoints_not_publicly_accessible(Check):
+class eks_cluster_not_publicly_accessible(Check):
     def execute(self):
         findings = []
         for cluster in eks_client.clusters:
@@ -13,12 +13,15 @@ class eks_endpoints_not_publicly_accessible(Check):
             report.resource_tags = cluster.tags
             report.status = "PASS"
             report.status_extended = (
-                f"Cluster endpoint access is private for EKS cluster {cluster.name}."
+                f"EKS cluster {cluster.name} is not publicly accessible."
             )
-            if cluster.endpoint_public_access:
+            if (
+                cluster.endpoint_public_access
+                and "0.0.0.0/0" in cluster.public_access_cidrs
+            ):
                 report.status = "FAIL"
                 report.status_extended = (
-                    f"Cluster endpoint access is public for EKS cluster {cluster.name}."
+                    f"EKS cluster {cluster.name} is publicly accessible."
                 )
             findings.append(report)
 

prowler/providers/aws/services/eks/eks_control_plane_endpoint_access_restricted/eks_control_plane_endpoint_access_restricted.metadata.json

@@ -1,34 +0,0 @@
-{
-  "Provider": "aws",
-  "CheckID": "eks_control_plane_endpoint_access_restricted",
-  "CheckTitle": "Restrict Access to the EKS Control Plane Endpoint",
-  "CheckType": [
-    "Infrastructure Security"
-  ],
-  "ServiceName": "eks",
-  "SubServiceName": "",
-  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
-  "Severity": "medium",
-  "ResourceType": "AwsEksCluster",
-  "Description": "Restrict Access to the EKS Control Plane Endpoint",
-  "Risk": "By default, this API server endpoint is public to the internet, and access to the API server is secured using a combination of AWS Identity and Access Management (IAM) and native Kubernetes Role Based Access Control (RBAC).",
-  "RelatedUrl": "",
-  "Remediation": {
-    "Code": {
-      "CLI": "aws eks update-cluster-config --region <region_name> --name <cluster_name> --resources-vpc-config endpointPublicAccess=false,endpointPrivateAccess=true,publicAccessCidrs=[\"123.123.123.123/32\"]",
-      "NativeIaC": "",
-      "Other": "",
-      "Terraform": ""
-    },
-    "Recommendation": {
-      "Text": "You should enable private access to the Kubernetes API server so that all communication between your nodes and the API server stays within your VPC. You can limit the IP addresses that can access your API server from the internet, or completely disable internet access to the API server.",
-      "Url": "https://docs.aws.amazon.com/eks/latest/userguide/cluster-endpoint.html"
-    }
-  },
-  "Categories": [
-    "internet-exposed"
-  ],
-  "DependsOn": [],
-  "RelatedTo": [],
-  "Notes": ""
-}

prowler/providers/aws/services/eks/eks_control_plane_endpoint_access_restricted/eks_control_plane_endpoint_access_restricted.py

@@ -1,26 +0,0 @@
-from prowler.lib.check.models import Check, Check_Report_AWS
-from prowler.providers.aws.services.eks.eks_client import eks_client
-
-
-class eks_control_plane_endpoint_access_restricted(Check):
-    def execute(self):
-        findings = []
-        for cluster in eks_client.clusters:
-            report = Check_Report_AWS(self.metadata())
-            report.region = cluster.region
-            report.resource_id = cluster.name
-            report.resource_arn = cluster.arn
-            report.resource_tags = cluster.tags
-            report.status = "PASS"
-            report.status_extended = (
-                f"Cluster endpoint access is private for EKS cluster {cluster.name}."
-            )
-            if cluster.endpoint_public_access:
-                if "0.0.0.0/0" in cluster.public_access_cidrs:
-                    report.status = "FAIL"
-                    report.status_extended = f"Cluster control plane access is not restricted for EKS cluster {cluster.name}."
-                else:
-                    report.status_extended = f"Cluster control plane access is restricted for EKS cluster {cluster.name}."
-            findings.append(report)
-
-        return findings

prowler/providers/aws/services/eks/eks_endpoints_not_publicly_accessible/eks_endpoints_not_publicly_accessible.metadata.json

@@ -1,36 +0,0 @@
-{
-  "Provider": "aws",
-  "CheckID": "eks_endpoints_not_publicly_accessible",
-  "CheckTitle": "Ensure EKS Clusters are created with Private Endpoint Enabled and Public Access Disabled",
-  "CheckType": [
-    "Protect",
-    "Secure network configuration",
-    "Resources not publicly accessible"
-  ],
-  "ServiceName": "eks",
-  "SubServiceName": "",
-  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
-  "Severity": "high",
-  "ResourceType": "AwsEksCluster",
-  "Description": "Ensure EKS Clusters are created with Private Endpoint Enabled and Public Access Disabled",
-  "Risk": "Publicly accessible services could expose sensitive data to bad actors.",
-  "RelatedUrl": "",
-  "Remediation": {
-    "Code": {
-      "CLI": "aws eks update-cluster-config --region <region_name> --name <cluster_name> --resources-vpc-config endpointPublicAccess=false,endpointPrivateAccess=true,publicAccessCidrs=[\"123.123.123.123/32\"]",
-      "NativeIaC": "",
-      "Other": "https://github.com/cloudmatos/matos/tree/master/remediations/aws/eks/eks-disable-public-endpoint",
-      "Terraform": ""
-    },
-    "Recommendation": {
-      "Text": "Enable private access to the Kubernetes API server so that all communication between your nodes and the API server stays within your VPC. Disable internet access to the API server.",
-      "Url": "https://docs.aws.amazon.com/eks/latest/userguide/infrastructure-security.html"
-    }
-  },
-  "Categories": [
-    "internet-exposed"
-  ],
-  "DependsOn": [],
-  "RelatedTo": [],
-  "Notes": ""
-}

prowler/providers/aws/services/eks/eks_service.py

@@ -107,6 +107,6 @@ class EKSCluster(BaseModel):
     security_group_id: str = None
     endpoint_public_access: bool = None
     endpoint_private_access: bool = None
-    public_access_cidrs: list[str] = None
+    public_access_cidrs: list[str] = []
     encryptionConfig: bool = None
     tags: Optional[list] = []

prowler/providers/aws/services/elbv2/elbv2_internet_facing/elbv2_internet_facing.py

@@ -1,4 +1,6 @@
 from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.ec2.ec2_client import ec2_client
+from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
 from prowler.providers.aws.services.elbv2.elbv2_client import elbv2_client
 
 
@@ -14,10 +16,18 @@ class elbv2_internet_facing(Check):
             report.status = "PASS"
             report.status_extended = f"ELBv2 ALB {lb.name} is not internet facing."
             if lb.scheme == "internet-facing":
-                report.status = "FAIL"
-                report.status_extended = (
-                    f"ELBv2 ALB {lb.name} is internet facing in {lb.dns}."
-                )
+                report.status_extended = f"ELBv2 ALB {lb.name} has an internet facing scheme with domain {lb.dns} but is not public."
+                for sg_id in getattr(lb, "security_groups", []):
+                    sg_arn = f"arn:{elbv2_client.audited_partition}:ec2:{lb.region}:{elbv2_client.audited_account}:security-group/{sg_id}"
+                    if sg_arn in ec2_client.security_groups:
+                        for ingress_rule in ec2_client.security_groups[
+                            sg_arn
+                        ].ingress_rules:
+                            if check_security_group(
+                                ingress_rule, "tcp", any_address=True
+                            ):
+                                report.status = "FAIL"
+                                report.status_extended = f"ELBv2 ALB {lb.name} is internet facing with domain {lb.dns} due to their security group {sg_id} is public."
 
             findings.append(report)
 

prowler/providers/aws/services/elbv2/elbv2_service.py

@@ -47,6 +47,7 @@ class ELBv2(AWSService):
                             type=elbv2["Type"],
                             dns=elbv2.get("DNSName", None),
                             scheme=elbv2.get("Scheme", None),
+                            security_groups=elbv2.get("SecurityGroups", []),
                             availability_zones={
                                 az["ZoneName"]: az["SubnetId"]
                                 for az in elbv2.get("AvailabilityZones", [])
@@ -204,6 +205,7 @@ class LoadBalancerv2(BaseModel):
     drop_invalid_header_fields: Optional[str]
     listeners: Dict[str, Listenerv2] = {}
     scheme: Optional[str]
+    security_groups: list[str] = []
     # Key: ZoneName, Value: SubnetId
     availability_zones: Dict[str, str] = {}
     tags: Optional[list] = []

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

@@ -2,7 +2,7 @@ from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.eventbridge.eventbridge_client import (
     eventbridge_client,
 )
-from prowler.providers.aws.services.iam.lib.policy import is_policy_cross_account
+from prowler.providers.aws.services.iam.lib.policy import is_policy_public
 
 
 class eventbridge_bus_cross_account_access(Check):
@@ -18,7 +18,11 @@ class eventbridge_bus_cross_account_access(Check):
             report.status_extended = (
                 f"EventBridge event bus {bus.name} does not allow cross-account access."
             )
-            if is_policy_cross_account(bus.policy, eventbridge_client.audited_account):
+            if is_policy_public(
+                bus.policy,
+                eventbridge_client.audited_account,
+                is_cross_account_allowed=False,
+            ):
                 report.status = "FAIL"
                 report.status_extended = (
                     f"EventBridge event bus {bus.name} allows cross-account access."

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

@@ -18,7 +18,7 @@ class eventbridge_bus_exposed(Check):
             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/eventbridge/eventbridge_schema_registry_cross_account_access/eventbridge_schema_registry_cross_account_access.py

@@ -1,6 +1,6 @@
 from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.eventbridge.schema_client import schema_client
-from prowler.providers.aws.services.iam.lib.policy import is_policy_cross_account
+from prowler.providers.aws.services.iam.lib.policy import is_policy_public
 
 
 class eventbridge_schema_registry_cross_account_access(Check):
@@ -14,7 +14,11 @@ class eventbridge_schema_registry_cross_account_access(Check):
             report.region = registry.region
             report.status = "PASS"
             report.status_extended = f"EventBridge schema registry {registry.name} does not allow cross-account access."
-            if is_policy_cross_account(registry.policy, schema_client.audited_account):
+            if is_policy_public(
+                registry.policy,
+                schema_client.audited_account,
+                is_cross_account_allowed=False,
+            ):
                 report.status = "FAIL"
                 report.status_extended = f"EventBridge schema registry {registry.name} allows cross-account access."
 

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

@@ -19,6 +19,7 @@ class iam_role_cross_service_confused_deputy_prevention(Check):
                     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

@@ -4,45 +4,6 @@ from prowler.lib.logger import logger
 from prowler.providers.aws.aws_provider import read_aws_regions_file
 
 
-def is_policy_cross_account(policy: dict, audited_account: str) -> bool:
-    """
-    is_policy_cross_account checks if the policy allows cross-account access.
-    Args:
-        policy (dict): The policy to check.
-        audited_account (str): The account to check if it has access.
-    Returns:
-        bool: True if the policy allows cross-account access, False otherwise.
-    """
-    if policy and "Statement" in policy:
-        if isinstance(policy["Statement"], list):
-            for statement in policy["Statement"]:
-                if statement["Effect"] == "Allow" and "AWS" in statement["Principal"]:
-                    if isinstance(statement["Principal"]["AWS"], list):
-                        for aws_account in statement["Principal"]["AWS"]:
-                            if audited_account not in aws_account or "*" == aws_account:
-                                return True
-                    else:
-                        if (
-                            audited_account not in statement["Principal"]["AWS"]
-                            or "*" == statement["Principal"]["AWS"]
-                        ):
-                            return True
-        else:
-            statement = policy["Statement"]
-            if statement["Effect"] == "Allow" and "AWS" in statement["Principal"]:
-                if isinstance(statement["Principal"]["AWS"], list):
-                    for aws_account in statement["Principal"]["AWS"]:
-                        if audited_account not in aws_account or "*" == aws_account:
-                            return True
-                else:
-                    if (
-                        audited_account not in statement["Principal"]["AWS"]
-                        or "*" == statement["Principal"]["AWS"]
-                    ):
-                        return True
-    return False
-
-
 def check_full_service_access(service: str, policy: dict) -> bool:
     """
     check_full_service_access checks if the policy allows full access to a service.
@@ -125,12 +86,15 @@ def is_condition_restricting_from_private_ip(condition_statement: dict) -> bool:
 
                 for ip in condition_statement[CONDITION_OPERATOR][CONDITION_KEY]:
                     # Select if IP address or IP network searching in the string for '/'
-                    if "/" in ip:
-                        if not ip_network(ip, strict=False).is_private:
-                            break
+                    if ip == "*" or ip == "0.0.0.0/0":
+                        break
                     else:
-                        if not ip_address(ip).is_private:
-                            break
+                        if "/" in ip:
+                            if not ip_network(ip, strict=False).is_private:
+                                break
+                        else:
+                            if not ip_address(ip).is_private:
+                                break
                 else:
                     is_from_private_ip = True
 
@@ -148,7 +112,7 @@ def is_condition_restricting_from_private_ip(condition_statement: dict) -> bool:
 def is_policy_public(
     policy: dict,
     source_account: str = "",
-    is_cross_account_allowed=False,
+    is_cross_account_allowed=True,
     not_allowed_actions: list = [],
 ) -> bool:
     """
@@ -157,89 +121,105 @@ def is_policy_public(
     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
+        is_cross_account_allowed (bool): If the policy can allow cross-account access, default: True
         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
-    for statement in policy.get("Statement", []):
-        # Only check allow statements
-        if statement["Effect"] == "Allow":
-            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"), 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", "")
-                        or (  # Check if function can be invoked by other AWS services
-                            (
-                                ".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
-                    )
-                )
-            ) and (
-                not not_allowed_actions  # If not_allowed_actions is empty, the function will not consider the actions in the policy
-                or (
-                    statement.get(
-                        "Action"
-                    )  # If the statement has no action, it is not public
-                    and (
-                        (
-                            (
-                                isinstance(statement.get("Action", ""), list)
-                                and "*" in statement["Action"]
+    if policy:
+        for statement in policy.get("Statement", []):
+            # Only check allow statements
+            if statement["Effect"] == "Allow":
+                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(statement.get("Action", ""), str)
-                                and statement.get("Action", "") == "*"
+                                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 (
-                            isinstance(statement.get("Action", ""), list)
-                            and any(
-                                action in not_allowed_actions
-                                for action in statement["Action"]
+                            or "*" in principal.get("CanonicalUser", "")
+                            or "arn:aws:iam::*:root"
+                            in principal.get("CanonicalUser", "")
+                            or (  # Check if function can be invoked by other AWS services
+                                (
+                                    ".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 (statement.get("Action", "") in not_allowed_actions)
                     )
-                )
-            ):
-                is_public = (
-                    not is_condition_block_restrictive(
-                        statement.get("Condition", {}),
-                        source_account,
-                        is_cross_account_allowed,
+                ) and (
+                    not not_allowed_actions  # If not_allowed_actions is empty, the function will not consider the actions in the policy
+                    or (
+                        statement.get(
+                            "Action"
+                        )  # If the statement has no action, it is not public
+                        and (
+                            (
+                                (
+                                    isinstance(statement.get("Action", ""), list)
+                                    and "*" in statement["Action"]
+                                )
+                                or (
+                                    isinstance(statement.get("Action", ""), str)
+                                    and statement.get("Action", "") == "*"
+                                )
+                            )
+                            or (
+                                isinstance(statement.get("Action", ""), list)
+                                and any(
+                                    action in not_allowed_actions
+                                    for action in statement["Action"]
+                                )
+                            )
+                            or (statement.get("Action", "") in not_allowed_actions)
+                        )
                     )
-                    and not is_condition_block_restrictive_organization(
-                        statement.get("Condition", {})
+                ):
+                    is_public = not statement.get("Condition", {}) or (
+                        not is_condition_block_restrictive(
+                            statement.get("Condition", {}),
+                            source_account,
+                            is_cross_account_allowed,
+                        )
+                        and not is_condition_block_restrictive_organization(
+                            statement.get("Condition", {})
+                        )
+                        and not is_condition_restricting_from_private_ip(
+                            statement.get("Condition", {})
+                        )
                     )
-                    and not is_condition_restricting_from_private_ip(
-                        statement.get("Condition", {})
-                    )
-                )
-                if is_public:
-                    break
+                    if is_public:
+                        break
     return is_public
 
 
@@ -275,23 +255,38 @@ def is_condition_block_restrictive(
         "StringEquals": [
             "aws:sourceaccount",
             "aws:sourceowner",
-            "s3:resourceaccount",
-            "aws:principalaccount",
-            "aws:resourceaccount",
             "aws:sourcearn",
+            "aws:principalaccount",
+            "aws:principalarn",
+            "aws:principalorgid",
+            "aws:principalorgpaths",
+            "aws:resourceaccount",
             "aws:sourcevpc",
             "aws:sourcevpce",
+            "aws:sourceorgid",
+            "aws:sourceorgpaths",
+            "aws:userid",
+            "aws:username",
+            "s3:resourceaccount",
             "lambda:eventsourcetoken",  # For Alexa Home functions, a token that the invoker must supply.
         ],
         "StringLike": [
             "aws:sourceaccount",
             "aws:sourceowner",
             "aws:sourcearn",
-            "aws:principalarn",
-            "aws:resourceaccount",
             "aws:principalaccount",
+            "aws:principalarn",
+            "aws:principalorgid",
+            "aws:principalorgpaths",
+            "aws:resourceaccount",
             "aws:sourcevpc",
             "aws:sourcevpce",
+            "aws:sourceorgid",
+            "aws:sourceorgpaths",
+            "aws:userid",
+            "aws:username",
+            "s3:resourceaccount",
+            "lambda:eventsourcetoken",
         ],
         "ArnLike": ["aws:sourcearn", "aws:principalarn"],
         "ArnEquals": ["aws:sourcearn", "aws:principalarn"],
@@ -319,9 +314,13 @@ 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:
-                                    is_condition_key_restrictive = False
-                                    break
+                                if (
+                                    "aws:sourcevpc" != value
+                                    and "aws:sourcevpce" != value
+                                ):
+                                    if source_account not in item:
+                                        is_condition_key_restrictive = False
+                                        break
 
                         if is_condition_key_restrictive:
                             is_condition_valid = True
@@ -331,14 +330,17 @@ def is_condition_block_restrictive(
                         condition_statement[condition_operator][value],
                         str,
                     ):
-                        if is_cross_account_allowed:
+                        if "aws:sourcevpc" == value or "aws:sourcevpce" == value:
                             is_condition_valid = True
                         else:
-                            if (
-                                source_account
-                                in condition_statement[condition_operator][value]
-                            ):
+                            if is_cross_account_allowed:
                                 is_condition_valid = True
+                            else:
+                                if (
+                                    source_account
+                                    in condition_statement[condition_operator][value]
+                                ):
+                                    is_condition_valid = True
 
     return is_condition_valid
 

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

@@ -18,7 +18,11 @@ class kms_key_not_publicly_accessible(Check):
                 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.py

@@ -8,7 +8,6 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered
 from prowler.providers.aws.lib.service.service import AWSService
 
 
-################################ OpenSearch
 class OpenSearchService(AWSService):
     def __init__(self, provider):
         # Call AWSService's __init__
@@ -83,19 +82,13 @@ class OpenSearchService(AWSService):
             regional_client = self.regional_clients[domain.region]
             describe_domain = regional_client.describe_domain(DomainName=domain.name)
             domain.arn = describe_domain["DomainStatus"]["ARN"]
-            domain.vpc_endpoints = None
-            if "Endpoints" in describe_domain["DomainStatus"]:
-                if "vpc" in describe_domain["DomainStatus"]["Endpoints"]:
-                    domain.vpc_endpoints = [
-                        vpc
-                        for vpc in describe_domain["DomainStatus"]["Endpoints"].values()
-                    ]
-
-            domain.vpc_id = None
-            if "VPCOptions" in describe_domain["DomainStatus"]:
-                domain.vpc_id = describe_domain["DomainStatus"]["VPCOptions"].get(
-                    "VPCId", None
-                )
+            if "vpc" in describe_domain["DomainStatus"].get("Endpoints", {}):
+                domain.vpc_endpoints = [
+                    vpc for vpc in describe_domain["DomainStatus"]["Endpoints"].values()
+                ]
+            domain.vpc_id = (
+                describe_domain["DomainStatus"].get("VPCOptions", {}).get("VPCId", "")
+            )
             domain.cognito_options = describe_domain["DomainStatus"][
                 "CognitoOptions"
             ].get("Enabled", False)
@@ -158,9 +151,9 @@ class PublishingLoggingOption(BaseModel):
 class OpenSearchDomain(BaseModel):
     name: str
     region: str
-    arn: str = None
+    arn: str
     logging: list[PublishingLoggingOption] = []
-    vpc_endpoints: list[str] = None
+    vpc_endpoints: list[str] = []
     vpc_id: str = None
     access_policy: dict = None
     cognito_options: bool = None

prowler/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible.metadata.json

@@ -1,28 +1,27 @@
 {
   "Provider": "aws",
   "CheckID": "opensearch_service_domains_not_publicly_accessible",
-  "CheckTitle": "Check if Amazon Opensearch/Elasticsearch domains are set as Public or if it has open policy access",
+  "CheckTitle": "Check if Amazon Opensearch/Elasticsearch domains are publicly accessible",
   "CheckType": [
-    "Protect",
-    "Secure Access Management"
+    "Effects/Data Exposure"
   ],
   "ServiceName": "opensearch",
   "SubServiceName": "",
   "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
   "Severity": "critical",
   "ResourceType": "AwsOpenSearchServiceDomain",
-  "Description": "Check if Amazon Opensearch/Elasticsearch domains are set as Public or if it has open policy access",
+  "Description": "Check if Amazon Opensearch/Elasticsearch domains are publicly accessible via their access policies.",
   "Risk": "Publicly accessible services could expose sensitive data to bad actors.",
   "RelatedUrl": "",
   "Remediation": {
     "Code": {
-      "CLI": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/Elasticsearch/elasticsearch-domain-exposed.html",
+      "CLI": "",
       "NativeIaC": "",
-      "Other": "https://docs.prowler.com/checks/aws/public-policies/public_3#fix---runtime",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/Elasticsearch/domain-exposed.html",
       "Terraform": ""
     },
     "Recommendation": {
-      "Text": "Use  VPC endpoints for internal services.",
+      "Text": "Modify the access policy attached to your Amazon OpenSearch domain and replace the 'Principal' element value (i.e. '*') with the ARN of the trusted AWS account. You can also add a Condition clause to the policy statement to limit the domain access to a specific (trusted) IP address/IP range only.",
       "Url": "https://docs.aws.amazon.com/elasticsearch-service/latest/developerguide/es-vpc.html"
     }
   },

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

@@ -1,4 +1,5 @@
 from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.iam.lib.policy import is_policy_public
 from prowler.providers.aws.services.opensearch.opensearch_client import (
     opensearch_client,
 )
@@ -15,43 +16,16 @@ class opensearch_service_domains_not_publicly_accessible(Check):
             report.resource_tags = domain.tags
             report.status = "PASS"
             report.status_extended = (
-                f"Opensearch domain {domain.name} does not allow anonymous access."
+                f"Opensearch domain {domain.name} is not publicly accessible."
             )
 
             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:
-                for statement in domain.access_policy["Statement"]:
-                    # look for open policies
-                    if (
-                        statement["Effect"] == "Allow"
-                        and (
-                            "AWS" in statement["Principal"]
-                            and "*" in statement["Principal"]["AWS"]
-                        )
-                        or (statement["Principal"] == "*")
-                    ):
-                        if "Condition" not in statement:
-                            report.status = "FAIL"
-                            report.status_extended = f"Opensearch domain {domain.name} policy allows access (Principal: '*')."
-                            break
-                        else:
-                            if (
-                                "IpAddress" in statement["Condition"]
-                                and "aws:SourceIp"
-                                in statement["Condition"]["IpAddress"]
-                            ):
-                                for ip in statement["Condition"]["IpAddress"][
-                                    "aws:SourceIp"
-                                ]:
-                                    if ip == "*":
-                                        report.status = "FAIL"
-                                        report.status_extended = f"Opensearch domain {domain.name} policy allows access (Principal: '*') and network *."
-                                        break
-                                    elif ip == "0.0.0.0/0":
-                                        report.status = "FAIL"
-                                        report.status_extended = f"Opensearch domain {domain.name} policy allows access (Principal: '*') and network 0.0.0.0/0."
-                                        break
+            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."
 
             findings.append(report)
 

prowler/providers/aws/services/rds/rds_instance_no_public_access/rds_instance_no_public_access.py

@@ -2,6 +2,7 @@ from prowler.lib.check.models import Check, Check_Report_AWS
 from prowler.providers.aws.services.ec2.ec2_client import ec2_client
 from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
 from prowler.providers.aws.services.rds.rds_client import rds_client
+from prowler.providers.aws.services.vpc.vpc_client import vpc_client
 
 
 class rds_instance_no_public_access(Check):
@@ -21,7 +22,7 @@ class rds_instance_no_public_access(Check):
                 report.status_extended = f"RDS Instance {db_instance.id} is set as publicly accessible, but is not publicly exposed."
                 # Check if any DB Instance Security Group is publicly open
                 if db_instance.security_groups:
-                    report.status = "PASS"
+                    public_sg = False
                     report.status_extended = f"RDS Instance {db_instance.id} is set as publicly accessible but filtered with security groups."
                     db_instance_port = db_instance.endpoint.get("Port")
                     if db_instance_port:
@@ -34,9 +35,21 @@ class rds_instance_no_public_access(Check):
                                         [db_instance_port],
                                         any_address=True,
                                     ):
-                                        report.status = "FAIL"
-                                        report.status_extended = f"RDS Instance {db_instance.id} is set as publicly accessible and security group {security_group.name} ({security_group.id}) has {db_instance.engine} port {db_instance_port} open to the Internet at endpoint {db_instance.endpoint.get('Address')}."
+                                        report.status_extended = f"RDS Instance {db_instance.id} is set as publicly accessible and security group {security_group.name} ({security_group.id}) has {db_instance.engine} port {db_instance_port} open to the Internet at endpoint {db_instance.endpoint.get('Address')} but is not in a public subnet."
+                                        public_sg = True
                                         break
+                            if public_sg:
+                                break
+                        if db_instance.subnet_ids:
+                            for subnet_id in db_instance.subnet_ids:
+                                if (
+                                    subnet_id in vpc_client.vpc_subnets
+                                    and vpc_client.vpc_subnets[subnet_id].public
+                                ):
+                                    report.status = "FAIL"
+                                    report.status_extended = f"RDS Instance {db_instance.id} is set as publicly accessible and security group {security_group.name} ({security_group.id}) has {db_instance.engine} port {db_instance_port} open to the Internet at endpoint {db_instance.endpoint.get('Address')} in a public subnet {subnet_id}."
+                                    break
+
             findings.append(report)
 
         return findings

prowler/providers/aws/services/rds/rds_service.py

@@ -103,6 +103,13 @@ class RDS(AWSService):
                                 ),
                                 port=instance.get("Endpoint", {}).get("Port"),
                                 vpc_id=instance.get("DBSubnetGroup", {}).get("VpcId"),
+                                subnet_ids=[
+                                    subnet_id["SubnetIdentifier"]
+                                    for subnet_id in instance.get(
+                                        "DBSubnetGroup", {}
+                                    ).get("Subnets", [])
+                                    if subnet_id["SubnetStatus"] == "Active"
+                                ],
                             )
         except Exception as error:
             logger.error(
@@ -527,6 +534,7 @@ class DBInstance(BaseModel):
     copy_tags_to_snapshot: Optional[bool]
     port: Optional[int]
     vpc_id: Optional[str]
+    subnet_ids: list[str] = []
 
 
 class DBCluster(BaseModel):

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

@@ -1,5 +1,8 @@
 from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.ec2.ec2_client import ec2_client
+from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
 from prowler.providers.aws.services.redshift.redshift_client import redshift_client
+from prowler.providers.aws.services.vpc.vpc_client import vpc_client
 
 
 class redshift_cluster_public_access(Check):
@@ -15,9 +18,30 @@ class redshift_cluster_public_access(Check):
             report.status_extended = (
                 f"Redshift Cluster {cluster.id} is not publicly accessible."
             )
+            # 1. Check if Redshift Cluster is publicly accessible
             if cluster.endpoint_address and cluster.public_access:
-                report.status = "FAIL"
-                report.status_extended = f"Redshift Cluster {cluster.id} is publicly accessible at endpoint {cluster.endpoint_address}."
+                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 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 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}"
+                        if sg_arn in ec2_client.security_groups:
+                            for ingress_rule in ec2_client.security_groups[
+                                sg_arn
+                            ].ingress_rules:
+                                if check_security_group(
+                                    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 a public subnet."
+                                    break
+                        if report.status == "FAIL":
+                            break
 
             findings.append(report)
 

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

@@ -16,9 +16,10 @@ class Redshift(AWSService):
         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...")
+        logger.info("Redshift - Describing Clusters...")
         try:
             list_clusters_paginator = regional_client.get_paginator("describe_clusters")
             for page in list_clusters_paginator.paginate():
@@ -30,6 +31,12 @@ class Redshift(AWSService):
                         cluster_to_append = Cluster(
                             arn=arn,
                             id=cluster["ClusterIdentifier"],
+                            vpc_id=cluster.get("VpcId"),
+                            vpc_security_groups=[
+                                sg["VpcSecurityGroupId"]
+                                for sg in cluster.get("VpcSecurityGroups")
+                                if sg["Status"] == "active"
+                            ],
                             endpoint_address=cluster.get("Endpoint", {}).get(
                                 "Address", ""
                             ),
@@ -48,6 +55,7 @@ class Redshift(AWSService):
                             parameter_group_name=cluster.get(
                                 "ClusterParameterGroups", [{}]
                             )[0].get("ParameterGroupName", ""),
+                            subnet_group=cluster.get("ClusterSubnetGroupName", ""),
                         )
                         self.clusters.append(cluster_to_append)
         except Exception as error:
@@ -55,8 +63,28 @@ class Redshift(AWSService):
                 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...")
+        logger.info("Redshift - Describing Logging Status...")
         try:
             regional_client = self.regional_clients[cluster.region]
             cluster_attributes = regional_client.describe_logging_status(
@@ -76,7 +104,7 @@ class Redshift(AWSService):
             )
 
     def _describe_cluster_snapshots(self, cluster):
-        logger.info("Redshift - describing logging status...")
+        logger.info("Redshift - Describing Cluster Status...")
         try:
             regional_client = self.regional_clients[cluster.region]
             cluster_snapshots = regional_client.describe_cluster_snapshots(
@@ -91,7 +119,7 @@ class Redshift(AWSService):
             )
 
     def _describe_cluster_parameters(self, cluster):
-        logger.info("Redshift - describing cluster parameter groups...")
+        logger.info("Redshift - Describing Cluster Parameter Groups...")
         try:
             regional_client = self.regional_clients[cluster.region]
             cluster_parameter_groups = regional_client.describe_cluster_parameters(
@@ -112,6 +140,8 @@ class Cluster(BaseModel):
     id: str
     arn: str
     region: str
+    vpc_id: str = None
+    vpc_security_groups: list = []
     public_access: bool = False
     encrypted: bool = False
     master_username: str = None
@@ -125,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

@@ -21,7 +21,7 @@ class s3_bucket_cross_account_access(Check):
                     f"S3 Bucket {bucket.name} does not have a bucket policy."
                 )
             elif is_policy_public(
-                bucket.policy, source_account=s3_client.audited_account
+                bucket.policy, s3_client.audited_account, is_cross_account_allowed=False
             ):
                 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 @@ class s3_bucket_policy_public_write_access(Check):
                 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 @@ class s3_bucket_public_access(Check):
                                     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)

tests/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_test.py

@@ -267,6 +267,7 @@ class Test_awslambda_function_not_publicly_accessible:
     def test_function_public_with_canonical(self):
         lambda_client = mock.MagicMock
         lambda_client.audited_account = AWS_ACCOUNT_NUMBER
+        lambda_client.audit_config = {}
         function_name = "test-lambda"
         function_runtime = "nodejs4.3"
         function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"
@@ -498,6 +499,7 @@ class Test_awslambda_function_not_publicly_accessible:
     def test_function_could_be_invoked_by_specific_aws_account(self):
         lambda_client = mock.MagicMock
         lambda_client.audited_account = AWS_ACCOUNT_NUMBER
+        lambda_client.audit_config = {}
         function_name = "test-lambda"
         function_runtime = "nodejs4.3"
         function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"
@@ -557,6 +559,7 @@ class Test_awslambda_function_not_publicly_accessible:
     def test_function_could_be_invoked_by_specific_other_aws_account(self):
         lambda_client = mock.MagicMock
         lambda_client.audited_account = AWS_ACCOUNT_NUMBER
+        lambda_client.audit_config = {}
         function_name = "test-lambda"
         function_runtime = "nodejs4.3"
         function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"
@@ -616,6 +619,7 @@ class Test_awslambda_function_not_publicly_accessible:
     def test_function_public_policy_with_several_statements(self):
         lambda_client = mock.MagicMock
         lambda_client.audited_account = AWS_ACCOUNT_NUMBER
+        lambda_client.audit_config = {}
         function_name = "test-lambda"
         function_runtime = "nodejs4.3"
         function_arn = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function/{function_name}"

tests/providers/aws/services/cloudfront/cloudfront_distributions_s3_origin_non_existent_bucket/cloudfront_distributions_s3_origin_non_existent_bucket_test.py

@@ -14,6 +14,10 @@ DISTRIBUTION_ARN = (
 
 
 class Test_cloudfront_s3_origin_non_existent_bucket:
+    @mock.patch(
+        "prowler.providers.aws.services.s3.s3_service.S3._head_bucket",
+        new=mock.MagicMock(return_value=False),
+    )
     def test_no_distributions(self):
         # Distributions
         cloudfront_client = mock.MagicMock

tests/providers/aws/services/dynamodb/dynamodb_table_cross_account_access/dynamodb_table_cross_account_access_test.py

@@ -124,6 +124,7 @@ class Test_dynamodb_table_cross_account_access:
         from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
 
         dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
+        dynamodb_client.audit_config = {}
         arn = test_table_arn
         dynamodb_client.tables = {
             arn: Table(
@@ -161,6 +162,7 @@ class Test_dynamodb_table_cross_account_access:
         from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
 
         dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
+        dynamodb_client.audit_config = {}
         arn = test_table_arn
         dynamodb_client.tables = {
             arn: Table(
@@ -198,6 +200,7 @@ class Test_dynamodb_table_cross_account_access:
         from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
 
         dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
+        dynamodb_client.audit_config = {}
         arn = test_table_arn
         dynamodb_client.tables = {
             arn: Table(
@@ -237,6 +240,7 @@ class Test_dynamodb_table_cross_account_access:
         from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
 
         dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
+        dynamodb_client.audit_config = {}
         arn = test_table_arn
         dynamodb_client.tables = {
             arn: Table(
@@ -276,6 +280,7 @@ class Test_dynamodb_table_cross_account_access:
         from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
 
         dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
+        dynamodb_client.audit_config = {}
         arn = test_table_arn
         dynamodb_client.tables = {
             arn: Table(
@@ -314,6 +319,7 @@ class Test_dynamodb_table_cross_account_access:
         from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
 
         dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
+        dynamodb_client.audit_config = {}
         arn = test_table_arn
         dynamodb_client.tables = {
             arn: Table(
@@ -352,6 +358,7 @@ class Test_dynamodb_table_cross_account_access:
         from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
 
         dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
+        dynamodb_client.audit_config = {}
         arn = test_table_arn
         dynamodb_client.tables = {
             arn: Table(

tests/providers/aws/services/ecr/ecr_repositories_not_publicly_accessible/ecr_repositories_not_publicly_accessible_test.py

@@ -87,6 +87,8 @@ class Test_ecr_repositories_not_publicly_accessible:
     def test_repository_not_public(self):
         ecr_client = mock.MagicMock
         ecr_client.registries = {}
+        ecr_client.audit_config = {}
+        ecr_client.audited_account = AWS_ACCOUNT_NUMBER
         ecr_client.registries[AWS_REGION_EU_WEST_1] = Registry(
             id=AWS_ACCOUNT_NUMBER,
             region=AWS_REGION_EU_WEST_1,
@@ -173,6 +175,8 @@ class Test_ecr_repositories_not_publicly_accessible:
     def test_repository_public(self):
         ecr_client = mock.MagicMock
         ecr_client.registries = {}
+        ecr_client.audit_config = {}
+        ecr_client.audited_account = AWS_ACCOUNT_NUMBER
         ecr_client.registries[AWS_REGION_EU_WEST_1] = Registry(
             id=AWS_ACCOUNT_NUMBER,
             region=AWS_REGION_EU_WEST_1,

tests/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible_test.py

@@ -90,7 +90,7 @@ class Test_efs_mount_target_not_publicly_accessible:
         route_table = ec2_client.create_route_table(VpcId=vpc["Vpc"]["VpcId"])
         ec2_client.create_route(
             RouteTableId=route_table["RouteTable"]["RouteTableId"],
-            DestinationCidrBlock="0.0.0.0/0",  # Ruta predeterminada para todo el tráfico
+            DestinationCidrBlock="0.0.0.0/0",
             GatewayId=igw["InternetGateway"]["InternetGatewayId"],
         )
         ec2_client.associate_route_table(

tests/providers/aws/services/eks/eks_cluster_not_publicly_accessible/eks_endpoints_not_publicly_accessible_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 
 from prowler.providers.aws.services.eks.eks_service import EKSCluster
@@ -10,7 +9,7 @@ cluster_arn = (
 )
 
 
-class Test_eks_endpoints_not_publicly_accessible:
+class Test_eks_cluster_not_publicly_accessible:
     def test_no_clusters(self):
         eks_client = mock.MagicMock
         eks_client.clusters = []
@@ -18,15 +17,15 @@ class Test_eks_endpoints_not_publicly_accessible:
             "prowler.providers.aws.services.eks.eks_service.EKS",
             eks_client,
         ):
-            from prowler.providers.aws.services.eks.eks_endpoints_not_publicly_accessible.eks_endpoints_not_publicly_accessible import (
-                eks_endpoints_not_publicly_accessible,
+            from prowler.providers.aws.services.eks.eks_cluster_not_publicly_accessible.eks_cluster_not_publicly_accessible import (
+                eks_cluster_not_publicly_accessible,
             )
 
-            check = eks_endpoints_not_publicly_accessible()
+            check = eks_cluster_not_publicly_accessible()
             result = check.execute()
             assert len(result) == 0
 
-    def test_endpoint_public_access(self):
+    def test_cluster_public_access(self):
         eks_client = mock.MagicMock
         eks_client.clusters = []
         eks_client.clusters.append(
@@ -37,6 +36,7 @@ class Test_eks_endpoints_not_publicly_accessible:
                 logging=None,
                 endpoint_public_access=True,
                 endpoint_private_access=False,
+                public_access_cidrs=["0.0.0.0/0"],
             )
         )
 
@@ -44,17 +44,16 @@ class Test_eks_endpoints_not_publicly_accessible:
             "prowler.providers.aws.services.eks.eks_service.EKS",
             eks_client,
         ):
-            from prowler.providers.aws.services.eks.eks_endpoints_not_publicly_accessible.eks_endpoints_not_publicly_accessible import (
-                eks_endpoints_not_publicly_accessible,
+            from prowler.providers.aws.services.eks.eks_cluster_not_publicly_accessible.eks_cluster_not_publicly_accessible import (
+                eks_cluster_not_publicly_accessible,
             )
 
-            check = eks_endpoints_not_publicly_accessible()
+            check = eks_cluster_not_publicly_accessible()
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "Cluster endpoint access is public for EKS cluster",
-                result[0].status_extended,
+            assert result[0].status_extended == (
+                f"EKS cluster {cluster_name} is publicly accessible."
             )
             assert result[0].resource_id == cluster_name
             assert result[0].resource_arn == cluster_arn
@@ -79,17 +78,16 @@ class Test_eks_endpoints_not_publicly_accessible:
             "prowler.providers.aws.services.eks.eks_service.EKS",
             eks_client,
         ):
-            from prowler.providers.aws.services.eks.eks_endpoints_not_publicly_accessible.eks_endpoints_not_publicly_accessible import (
-                eks_endpoints_not_publicly_accessible,
+            from prowler.providers.aws.services.eks.eks_cluster_not_publicly_accessible.eks_cluster_not_publicly_accessible import (
+                eks_cluster_not_publicly_accessible,
             )
 
-            check = eks_endpoints_not_publicly_accessible()
+            check = eks_cluster_not_publicly_accessible()
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "Cluster endpoint access is private for EKS cluster",
-                result[0].status_extended,
+            assert result[0].status_extended == (
+                f"EKS cluster {cluster_name} is not publicly accessible."
             )
             assert result[0].resource_id == cluster_name
             assert result[0].resource_arn == cluster_arn

tests/providers/aws/services/eks/eks_control_plane_endpoint_access_restricted/eks_control_plane_endpoint_access_restricted_test.py

@@ -1,171 +0,0 @@
-from re import search
-from unittest import mock
-
-from prowler.providers.aws.services.eks.eks_service import EKSCluster
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
-
-cluster_name = "cluster_test"
-cluster_arn = (
-    f"arn:aws:eks:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:cluster/{cluster_name}"
-)
-
-
-class Test_eks_control_plane_endpoint_access_restricted:
-    def test_no_clusters(self):
-        eks_client = mock.MagicMock
-        eks_client.clusters = []
-        with mock.patch(
-            "prowler.providers.aws.services.eks.eks_service.EKS",
-            eks_client,
-        ):
-            from prowler.providers.aws.services.eks.eks_control_plane_endpoint_access_restricted.eks_control_plane_endpoint_access_restricted import (
-                eks_control_plane_endpoint_access_restricted,
-            )
-
-            check = eks_control_plane_endpoint_access_restricted()
-            result = check.execute()
-            assert len(result) == 0
-
-    def test_control_plane_access_private(self):
-        eks_client = mock.MagicMock
-        eks_client.clusters = []
-        eks_client.clusters.append(
-            EKSCluster(
-                name=cluster_name,
-                arn=cluster_arn,
-                region=AWS_REGION_EU_WEST_1,
-                logging=None,
-                endpoint_public_access=False,
-                endpoint_private_access=True,
-                public_access_cidrs=["123.123.123.123/32"],
-            )
-        )
-
-        with mock.patch(
-            "prowler.providers.aws.services.eks.eks_service.EKS",
-            eks_client,
-        ):
-            from prowler.providers.aws.services.eks.eks_control_plane_endpoint_access_restricted.eks_control_plane_endpoint_access_restricted import (
-                eks_control_plane_endpoint_access_restricted,
-            )
-
-            check = eks_control_plane_endpoint_access_restricted()
-            result = check.execute()
-            assert len(result) == 1
-            assert result[0].status == "PASS"
-            assert search(
-                "Cluster endpoint access is private for EKS cluster",
-                result[0].status_extended,
-            )
-            assert result[0].resource_id == cluster_name
-            assert result[0].resource_arn == cluster_arn
-            assert result[0].resource_tags == []
-            assert result[0].region == AWS_REGION_EU_WEST_1
-
-    def test_control_plane_access_restricted(self):
-        eks_client = mock.MagicMock
-        eks_client.clusters = []
-        eks_client.clusters.append(
-            EKSCluster(
-                name=cluster_name,
-                arn=cluster_arn,
-                region=AWS_REGION_EU_WEST_1,
-                logging=None,
-                endpoint_public_access=True,
-                endpoint_private_access=False,
-                public_access_cidrs=["123.123.123.123/32"],
-            )
-        )
-
-        with mock.patch(
-            "prowler.providers.aws.services.eks.eks_service.EKS",
-            eks_client,
-        ):
-            from prowler.providers.aws.services.eks.eks_control_plane_endpoint_access_restricted.eks_control_plane_endpoint_access_restricted import (
-                eks_control_plane_endpoint_access_restricted,
-            )
-
-            check = eks_control_plane_endpoint_access_restricted()
-            result = check.execute()
-            assert len(result) == 1
-            assert result[0].status == "PASS"
-            assert search(
-                "Cluster control plane access is restricted for EKS cluster",
-                result[0].status_extended,
-            )
-            assert result[0].resource_id == cluster_name
-            assert result[0].resource_arn == cluster_arn
-            assert result[0].resource_tags == []
-            assert result[0].region == AWS_REGION_EU_WEST_1
-
-    def test_control_plane_public(self):
-        eks_client = mock.MagicMock
-        eks_client.clusters = []
-        eks_client.clusters.append(
-            EKSCluster(
-                name=cluster_name,
-                arn=cluster_arn,
-                region=AWS_REGION_EU_WEST_1,
-                logging=None,
-                endpoint_public_access=True,
-                endpoint_private_access=False,
-                public_access_cidrs=["123.123.123.123/32", "0.0.0.0/0"],
-            )
-        )
-
-        with mock.patch(
-            "prowler.providers.aws.services.eks.eks_service.EKS",
-            eks_client,
-        ):
-            from prowler.providers.aws.services.eks.eks_control_plane_endpoint_access_restricted.eks_control_plane_endpoint_access_restricted import (
-                eks_control_plane_endpoint_access_restricted,
-            )
-
-            check = eks_control_plane_endpoint_access_restricted()
-            result = check.execute()
-            assert len(result) == 1
-            assert result[0].status == "FAIL"
-            assert search(
-                "Cluster control plane access is not restricted for EKS cluster",
-                result[0].status_extended,
-            )
-            assert result[0].resource_id == cluster_name
-            assert result[0].resource_arn == cluster_arn
-            assert result[0].resource_tags == []
-            assert result[0].region == AWS_REGION_EU_WEST_1
-
-    def test_control_plane_public_and_private(self):
-        eks_client = mock.MagicMock
-        eks_client.clusters = []
-        eks_client.clusters.append(
-            EKSCluster(
-                name=cluster_name,
-                arn=cluster_arn,
-                region=AWS_REGION_EU_WEST_1,
-                logging=None,
-                endpoint_public_access=True,
-                endpoint_private_access=True,
-                public_access_cidrs=["123.123.123.123/32", "0.0.0.0/0"],
-            )
-        )
-
-        with mock.patch(
-            "prowler.providers.aws.services.eks.eks_service.EKS",
-            eks_client,
-        ):
-            from prowler.providers.aws.services.eks.eks_control_plane_endpoint_access_restricted.eks_control_plane_endpoint_access_restricted import (
-                eks_control_plane_endpoint_access_restricted,
-            )
-
-            check = eks_control_plane_endpoint_access_restricted()
-            result = check.execute()
-            assert len(result) == 1
-            assert result[0].status == "FAIL"
-            assert search(
-                "Cluster control plane access is not restricted for EKS cluster",
-                result[0].status_extended,
-            )
-            assert result[0].resource_id == cluster_name
-            assert result[0].resource_arn == cluster_arn
-            assert result[0].resource_tags == []
-            assert result[0].region == AWS_REGION_EU_WEST_1

tests/providers/aws/services/elbv2/elbv2_internet_facing/elbv2_internet_facing_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 
 from boto3 import client, resource
@@ -89,9 +88,8 @@ class Test_elbv2_internet_facing:
 
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "is not internet facing",
-                result[0].status_extended,
+            assert result[0].status_extended == (
+                "ELBv2 ALB my-lb is not internet facing."
             )
             assert result[0].resource_id == "my-lb"
             assert result[0].resource_arn == lb["LoadBalancerArn"]
@@ -144,10 +142,82 @@ class Test_elbv2_internet_facing:
             result = check.execute()
 
             assert len(result) == 1
-            assert result[0].status == "FAIL"
-            assert search(
-                "is internet facing",
-                result[0].status_extended,
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                f"ELBv2 ALB my-lb has an internet facing scheme with domain {lb['DNSName']} but is not public."
+            )
+            assert result[0].resource_id == "my-lb"
+            assert result[0].resource_arn == lb["LoadBalancerArn"]
+
+    @mock_aws
+    def test_elbv2_public_sg(self):
+        conn = client("elbv2", region_name=AWS_REGION_EU_WEST_1)
+        ec2 = resource("ec2", region_name=AWS_REGION_EU_WEST_1)
+        ec2_client = client("ec2", region_name=AWS_REGION_EU_WEST_1)
+
+        vpc = ec2.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
+        subnet1 = ec2.create_subnet(
+            VpcId=vpc.id,
+            CidrBlock="172.28.7.192/26",
+            AvailabilityZone=AWS_REGION_EU_WEST_1_AZA,
+        )
+        subnet2 = ec2.create_subnet(
+            VpcId=vpc.id,
+            CidrBlock="172.28.7.0/26",
+            AvailabilityZone=AWS_REGION_EU_WEST_1_AZB,
+        )
+        default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
+            "SecurityGroups"
+        ][0]
+
+        default_sg_id = default_sg["GroupId"]
+
+        # Authorize ingress rule
+        ec2_client.authorize_security_group_ingress(
+            GroupId=default_sg_id,
+            IpPermissions=[
+                {
+                    "IpProtocol": "-1",
+                    "IpRanges": [{"CidrIp": "0.0.0.0/0"}],
+                }
+            ],
+        )
+        lb = conn.create_load_balancer(
+            Name="my-lb",
+            Subnets=[subnet1.id, subnet2.id],
+            SecurityGroups=[default_sg_id],
+            Scheme="internet-facing",
+        )["LoadBalancers"][0]
+
+        from prowler.providers.aws.services.ec2.ec2_service import EC2
+        from prowler.providers.aws.services.elbv2.elbv2_service import ELBv2
+
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.elbv2.elbv2_internet_facing.elbv2_internet_facing.elbv2_client",
+            new=ELBv2(aws_provider),
+        ), mock.patch(
+            "prowler.providers.aws.services.elbv2.elbv2_internet_facing.elbv2_internet_facing.ec2_client",
+            new=EC2(aws_provider),
+        ):
+
+            from prowler.providers.aws.services.elbv2.elbv2_internet_facing.elbv2_internet_facing import (
+                elbv2_internet_facing,
+            )
+
+            check = elbv2_internet_facing()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == (
+                f"ELBv2 ALB my-lb is internet facing with domain {lb['DNSName']} due to their security group {default_sg_id} is public."
             )
             assert result[0].resource_id == "my-lb"
             assert result[0].resource_arn == lb["LoadBalancerArn"]

tests/providers/aws/services/elbv2/elbv2_service_test.py

@@ -101,6 +101,10 @@ class Test_ELBv2_Service:
             ]
             == subnet2.id
         )
+        assert (
+            elbv2.loadbalancersv2[lb["LoadBalancerArn"]].security_groups[0]
+            == security_group.id
+        )
 
     # Test ELBv2 Describe Listeners
     @mock_aws

tests/providers/aws/services/iam/lib/policy_test.py

@@ -4,10 +4,8 @@ from prowler.providers.aws.services.iam.lib.policy import (
     is_condition_block_restrictive,
     is_condition_block_restrictive_organization,
     is_condition_restricting_from_private_ip,
-    is_policy_cross_account,
     is_policy_public,
 )
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER
 
 TRUSTED_AWS_ACCOUNT_NUMBER = "123456789012"
 NON_TRUSTED_AWS_ACCOUNT_NUMBER = "111222333444"
@@ -1449,14 +1447,18 @@ class Test_Policy:
             "Statement": [
                 {
                     "Effect": "Allow",
-                    "Principal": {"AWS": ["arn:aws:iam::123456789012:root", "*"]},
+                    "Principal": {
+                        "AWS": [f"arn:aws:iam::{TRUSTED_AWS_ACCOUNT_NUMBER}:root", "*"]
+                    },
                     "Action": "s3:*",
                     "Resource": "arn:aws:s3:::example_bucket/*",
                 }
             ]
         }
-        assert is_policy_cross_account(
-            policy_allow_root_and_wildcard_principal, AWS_ACCOUNT_NUMBER
+        assert is_policy_public(
+            policy_allow_root_and_wildcard_principal,
+            TRUSTED_AWS_ACCOUNT_NUMBER,
+            is_cross_account_allowed=False,
         )
 
     def test_policy_does_not_allow_cross_account_access_with_specific_root_principal(
@@ -1466,14 +1468,18 @@ class Test_Policy:
             "Statement": [
                 {
                     "Effect": "Allow",
-                    "Principal": {"AWS": ["arn:aws:iam::123456789012:root"]},
+                    "Principal": {
+                        "AWS": [f"arn:aws:iam::{TRUSTED_AWS_ACCOUNT_NUMBER}:root"]
+                    },
                     "Action": "s3:*",
                     "Resource": "arn:aws:s3:::example_bucket/*",
                 }
             ]
         }
-        assert not is_policy_cross_account(
-            policy_allow_specific_root_principal, AWS_ACCOUNT_NUMBER
+        assert not is_policy_public(
+            policy_allow_specific_root_principal,
+            TRUSTED_AWS_ACCOUNT_NUMBER,
+            is_cross_account_allowed=False,
         )
 
     def test_policy_does_not_allow_cross_account_access_with_deny_effect(self):
@@ -1481,14 +1487,18 @@ class Test_Policy:
             "Statement": [
                 {
                     "Effect": "Deny",
-                    "Principal": {"AWS": ["arn:aws:iam::123456789012:root"]},
+                    "Principal": {
+                        "AWS": [f"arn:aws:iam::{TRUSTED_AWS_ACCOUNT_NUMBER}:root"]
+                    },
                     "Action": "s3:*",
                     "Resource": "arn:aws:s3:::example_bucket/*",
                 }
             ]
         }
-        assert not is_policy_cross_account(
-            policy_deny_specific_root_principal, AWS_ACCOUNT_NUMBER
+        assert not is_policy_public(
+            policy_deny_specific_root_principal,
+            TRUSTED_AWS_ACCOUNT_NUMBER,
+            is_cross_account_allowed=False,
         )
 
     def test_policy_allows_public_access_with_wildcard_principal(self):
@@ -1503,7 +1513,9 @@ class Test_Policy:
             ]
         }
         assert is_policy_public(
-            policy_allow_wildcard_principal, not_allowed_actions=["s3:*"]
+            policy_allow_wildcard_principal,
+            TRUSTED_AWS_ACCOUNT_NUMBER,
+            not_allowed_actions=["s3:*"],
         )
 
     def test_policy_allows_public_access_with_aws_wildcard_principal(self):
@@ -1518,7 +1530,9 @@ class Test_Policy:
             ]
         }
         assert is_policy_public(
-            policy_allow_aws_wildcard_principal, not_allowed_actions=["s3:*"]
+            policy_allow_aws_wildcard_principal,
+            TRUSTED_AWS_ACCOUNT_NUMBER,
+            not_allowed_actions=["s3:*"],
         )
 
     def test_policy_does_not_allow_public_access_with_specific_aws_principal(self):
@@ -1526,13 +1540,17 @@ class Test_Policy:
             "Statement": [
                 {
                     "Effect": "Allow",
-                    "Principal": {"AWS": "arn:aws:iam::123456789012:root"},
+                    "Principal": {
+                        "AWS": f"arn:aws:iam::{TRUSTED_AWS_ACCOUNT_NUMBER}:root"
+                    },
                     "Action": "s3:*",
                     "Resource": "arn:aws:s3:::example_bucket/*",
                 }
             ]
         }
-        assert not is_policy_public(policy_allow_specific_aws_principal)
+        assert not is_policy_public(
+            policy_allow_specific_aws_principal, TRUSTED_AWS_ACCOUNT_NUMBER
+        )
 
     def test_policy_does_not_allow_public_access_with_condition(self):
         policy_allow_aws_wildcard_principal_with_condition = {
@@ -1546,7 +1564,10 @@ class Test_Policy:
                 }
             ]
         }
-        assert not is_policy_public(policy_allow_aws_wildcard_principal_with_condition)
+        assert not is_policy_public(
+            policy_allow_aws_wildcard_principal_with_condition,
+            TRUSTED_AWS_ACCOUNT_NUMBER,
+        )
 
     def test_policy_allows_full_service_access_with_wildcard_action_and_resource(self):
         policy_allow_wildcard_action_and_resource = {
@@ -1718,7 +1739,7 @@ class Test_Policy:
         }
         assert not is_condition_restricting_from_private_ip(condition_from_invalid_ip)
 
-    def test__is_policy_public__(self):
+    def test_is_policy_public_(self):
         policy = {
             "Statement": [
                 {
@@ -1730,10 +1751,12 @@ class Test_Policy:
             ]
         }
         assert is_policy_public(
-            policy, not_allowed_actions=["elasticfilesystem:ClientMount"]
+            policy,
+            TRUSTED_AWS_ACCOUNT_NUMBER,
+            not_allowed_actions=["elasticfilesystem:ClientMount"],
         )
 
-    def test__is_policy_public__with_principal_dict(self):
+    def test_is_policy_public_with_principal_dict(self):
         policy = {
             "Statement": [
                 {
@@ -1745,10 +1768,12 @@ class Test_Policy:
             ]
         }
         assert is_policy_public(
-            policy, not_allowed_actions=["elasticfilesystem:ClientMount"]
+            policy,
+            TRUSTED_AWS_ACCOUNT_NUMBER,
+            not_allowed_actions=["elasticfilesystem:ClientMount"],
         )
 
-    def test__is_policy_public__with_secure_conditions_and_allowed_conditions(
+    def test_is_policy_public_with_secure_conditions_and_allowed_conditions(
         self,
     ):
         policy = {
@@ -1760,14 +1785,14 @@ class Test_Policy:
                     "Resource": "*",
                     "Condition": {
                         "Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"},
-                        "StringEquals": {"aws:SourceOwner": "123456789012"},
+                        "StringEquals": {"aws:SourceOwner": TRUSTED_AWS_ACCOUNT_NUMBER},
                     },
                 }
             ]
         }
-        assert not is_policy_public(policy)
+        assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
 
-    def test__is_policy_public__with_secure_conditions_and_allowed_conditions_nested(
+    def test_is_policy_public_with_secure_conditions_and_allowed_conditions_nested(
         self,
     ):
         policy = {
@@ -1779,7 +1804,7 @@ class Test_Policy:
                     "Resource": "*",
                     "Condition": {
                         "Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"},
-                        "StringEquals": {"aws:SourceOwner": "123456789012"},
+                        "StringEquals": {"aws:SourceOwner": TRUSTED_AWS_ACCOUNT_NUMBER},
                         "StringEqualsIfExists": {
                             "aws:SourceVpce": "vpce-1234567890abcdef0"
                         },
@@ -1787,9 +1812,9 @@ class Test_Policy:
                 }
             ]
         }
-        assert not is_policy_public(policy)
+        assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
 
-    def test__is_policy_public__with_secure_conditions_and_allowed_conditions_nested_dict(
+    def test_is_policy_public_with_secure_conditions_and_allowed_conditions_nested_dict(
         self,
     ):
         policy = {
@@ -1801,7 +1826,7 @@ class Test_Policy:
                     "Resource": "*",
                     "Condition": {
                         "Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"},
-                        "StringEquals": {"aws:SourceOwner": "123456789012"},
+                        "StringEquals": {"aws:SourceOwner": TRUSTED_AWS_ACCOUNT_NUMBER},
                         "StringEqualsIfExists": {
                             "aws:SourceVpce": {
                                 "vpce-1234567890abcdef0": "vpce-1234567890abcdef0"
@@ -1811,9 +1836,9 @@ class Test_Policy:
                 }
             ]
         }
-        assert not is_policy_public(policy)
+        assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
 
-    def test__is_policy_public__with_secure_conditions_and_allowed_conditions_nested_dict_key(
+    def test_is_policy_public_with_secure_conditions_and_allowed_conditions_nested_dict_key(
         self,
     ):
         policy = {
@@ -1825,7 +1850,7 @@ class Test_Policy:
                     "Resource": "*",
                     "Condition": {
                         "Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"},
-                        "StringEquals": {"aws:SourceOwner": "123456789012"},
+                        "StringEquals": {"aws:SourceOwner": TRUSTED_AWS_ACCOUNT_NUMBER},
                         "StringEqualsIfExists": {
                             "aws:SourceVpce": {
                                 "vpce-1234567890abcdef0": "vpce-1234567890abcdef0"
@@ -1835,9 +1860,9 @@ class Test_Policy:
                 }
             ]
         }
-        assert not is_policy_public(policy)
+        assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
 
-    def test__is_policy_public_with_action_wildcard(
+    def test_is_policy_public_with_action_wildcard(
         self,
     ):
         policy = {
@@ -1850,9 +1875,9 @@ class Test_Policy:
                 }
             ]
         }
-        assert is_policy_public(policy)
+        assert is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
 
-    def test__is_policy_public_allowing_all_actions(
+    def test_is_policy_public_allowing_all_actions(
         self,
     ):
         policy = {
@@ -1865,9 +1890,24 @@ class Test_Policy:
                 }
             ]
         }
-        assert is_policy_public(policy)
+        assert is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
 
-    def test__is_policy_public_secrets_manager(
+    def test_is_policy_public_allowing_other_account(self):
+        policy = {
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {
+                        "AWS": f"arn:aws:iam::{TRUSTED_AWS_ACCOUNT_NUMBER}:root"
+                    },
+                    "Action": "*",
+                    "Resource": "*",
+                }
+            ]
+        }
+        assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
+
+    def test_is_policy_public_secrets_manager(
         self,
     ):
         policy = {
@@ -1881,9 +1921,9 @@ class Test_Policy:
                 }
             ]
         }
-        assert not is_policy_public(policy)
+        assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
 
-    def test__is_policy_public_alexa_condition(
+    def test_is_policy_public_alexa_condition(
         self,
     ):
         policy = {
@@ -1898,7 +1938,41 @@ class Test_Policy:
                 }
             ]
         }
-        assert not is_policy_public(policy)
+        assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
+
+    def test_is_policy_private_org_s3_bucket(
+        self,
+    ):
+        policy = {
+            "Statement": [
+                {
+                    "Sid": "test",
+                    "Effect": "Allow",
+                    "Principal": {"Service": "personalize.amazonaws.com"},
+                    "Action": "*",
+                    "Resource": "*",
+                    "Condition": {"StringLike": {"aws:SourceOrgID": "o-123456"}},
+                }
+            ]
+        }
+        assert not is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
+
+    def test_is_policy_public_ip(
+        self,
+    ):
+        policy = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {"AWS": "*"},
+                    "Action": ["*"],
+                    "Condition": {"IpAddress": {"aws:SourceIp": ["0.0.0.0/0"]}},
+                    "Resource": "*",
+                }
+            ],
+        }
+        assert is_policy_public(policy, TRUSTED_AWS_ACCOUNT_NUMBER)
 
     def test_check_admin_access(self):
         policy = {

tests/providers/aws/services/opensearch/opensearch_service_domains_not_publicly_accessible/opensearch_service_domains_not_publicly_accessible_test.py

@@ -133,7 +133,7 @@ class Test_opensearch_service_domains_not_publicly_accessible:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"Opensearch domain {domain_name} does not allow anonymous access."
+                == f"Opensearch domain {domain_name} is not publicly accessible."
             )
             assert result[0].resource_id == domain_name
             assert result[0].resource_arn == domain_arn
@@ -174,7 +174,7 @@ class Test_opensearch_service_domains_not_publicly_accessible:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Opensearch domain {domain_name} policy allows access (Principal: '*')."
+                == f"Opensearch domain {domain_name} is publicly accessible via access policy."
             )
             assert result[0].resource_id == domain_name
             assert result[0].resource_arn == domain_arn
@@ -215,7 +215,7 @@ class Test_opensearch_service_domains_not_publicly_accessible:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Opensearch domain {domain_name} policy allows access (Principal: '*')."
+                == f"Opensearch domain {domain_name} is publicly accessible via access policy."
             )
             assert result[0].resource_id == domain_name
             assert result[0].resource_arn == domain_arn
@@ -256,7 +256,7 @@ class Test_opensearch_service_domains_not_publicly_accessible:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Opensearch domain {domain_name} policy allows access (Principal: '*') and network *."
+                == f"Opensearch domain {domain_name} is publicly accessible via access policy."
             )
             assert result[0].resource_id == domain_name
             assert result[0].resource_arn == domain_arn
@@ -297,53 +297,9 @@ class Test_opensearch_service_domains_not_publicly_accessible:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Opensearch domain {domain_name} policy allows access (Principal: '*') and network 0.0.0.0/0."
+                == f"Opensearch domain {domain_name} is publicly accessible via access policy."
             )
             assert result[0].resource_id == domain_name
             assert result[0].resource_arn == domain_arn
             assert result[0].region == AWS_REGION_US_WEST_2
             assert result[0].resource_tags == []
-
-    def test_domain_inside_vpc(self):
-        opensearch_client = mock.MagicMock
-        opensearch_client.opensearch_domains = {}
-
-        aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
-
-        with mock.patch(
-            "prowler.providers.common.provider.Provider.get_global_provider",
-            return_value=aws_provider,
-        ), mock.patch(
-            "prowler.providers.aws.services.opensearch.opensearch_service_domains_not_publicly_accessible.opensearch_service_domains_not_publicly_accessible.opensearch_client",
-            new=opensearch_client,
-        ):
-            from prowler.providers.aws.services.opensearch.opensearch_service import (
-                OpenSearchDomain,
-            )
-            from prowler.providers.aws.services.opensearch.opensearch_service_domains_not_publicly_accessible.opensearch_service_domains_not_publicly_accessible import (
-                opensearch_service_domains_not_publicly_accessible,
-            )
-
-            domain_arn = f"arn:aws:es:{AWS_REGION_US_WEST_2}:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
-            opensearch_client.opensearch_domains[domain_arn] = OpenSearchDomain(
-                name=domain_name,
-                region=AWS_REGION_US_WEST_2,
-                arn=domain_arn,
-                vpc_id="vpc-123456",
-            )
-
-            check = opensearch_service_domains_not_publicly_accessible()
-            result = check.execute()
-            assert len(result) == 1
-            assert result[0].status == "PASS"
-            assert (
-                result[0].status_extended
-                == f"Opensearch domain {domain_name} is in a VPC, then it is not publicly accessible."
-            )
-            assert result[0].resource_id == domain_name
-            assert (
-                result[0].resource_arn
-                == f"arn:aws:es:{AWS_REGION_US_WEST_2}:{AWS_ACCOUNT_NUMBER}:domain/{domain_name}"
-            )
-            assert result[0].region == AWS_REGION_US_WEST_2
-            assert result[0].resource_tags == []

tests/providers/aws/services/opensearch/opensearch_service_test.py

@@ -108,7 +108,7 @@ class Test_OpenSearchService_Service:
         assert opensearch.opensearch_domains[domain_arn].name == test_domain_name
         assert opensearch.opensearch_domains[domain_arn].region == AWS_REGION_EU_WEST_1
 
-    # Test OpenSearchService describ domain config
+    # Test OpenSearchService describe domain config
     def test_describe_domain_config(self):
         aws_provider = set_mocked_aws_provider([])
         opensearch = OpenSearchService(aws_provider)
@@ -129,7 +129,7 @@ class Test_OpenSearchService_Service:
         assert opensearch.opensearch_domains[domain_arn].logging[2].name == "AUDIT_LOGS"
         assert opensearch.opensearch_domains[domain_arn].logging[2].enabled
 
-    # Test OpenSearchService describ domain
+    # Test OpenSearchService describe domain
     @mock_aws
     def test_describe_domain(self):
         opensearch_client = client("opensearch", region_name=AWS_REGION_EU_WEST_1)

tests/providers/aws/services/rds/rds_instance_no_public_access/rds_instance_no_public_access_test.py

@@ -146,7 +146,7 @@ class Test_rds_instance_no_public_access:
                 assert result[0].resource_tags == []
 
     @mock_aws
-    def test_rds_instance_public_with_public_sg(self):
+    def test_rds_instance_public_with_public_sg_in_private_subnet(self):
         ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
         ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
         default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
@@ -201,10 +201,10 @@ class Test_rds_instance_no_public_access:
                 result = check.execute()
 
                 assert len(result) == 1
-                assert result[0].status == "FAIL"
+                assert result[0].status == "PASS"
                 assert (
                     result[0].status_extended
-                    == f"RDS Instance db-master-1 is set as publicly accessible and security group default ({default_sg_id}) has postgres port 5432 open to the Internet at endpoint db-master-1.aaaaaaaaaa.us-east-1.rds.amazonaws.com."
+                    == f"RDS Instance db-master-1 is set as publicly accessible and security group default ({default_sg_id}) has postgres port 5432 open to the Internet at endpoint db-master-1.aaaaaaaaaa.us-east-1.rds.amazonaws.com but is not in a public subnet."
                 )
                 assert result[0].resource_id == "db-master-1"
                 assert result[0].region == AWS_REGION_US_EAST_1
@@ -274,3 +274,111 @@ class Test_rds_instance_no_public_access:
                     == f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
                 )
                 assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_rds_instance_public_with_public_subnet(self):
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
+        # Get default security group from VPC
+        default_sg_id = (
+            ec2_client.describe_security_groups(
+                Filters=[
+                    {
+                        "Name": "vpc-id",
+                        "Values": [vpc_id],
+                    }
+                ]
+            )
+        )["SecurityGroups"][0]["GroupId"]
+        ec2_client.authorize_security_group_ingress(
+            GroupId=default_sg_id,
+            IpPermissions=[
+                {
+                    "IpProtocol": "-1",
+                    "IpRanges": [{"CidrIp": "0.0.0.0/0"}],
+                }
+            ],
+        )
+        subnet_id = ec2_client.create_subnet(
+            VpcId=vpc_id,
+            CidrBlock="10.0.0.0/16",
+            AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
+        )["Subnet"]["SubnetId"]
+        # add default route of subnet to an internet gateway to make it public
+        igw_id = ec2_client.create_internet_gateway()["InternetGateway"][
+            "InternetGatewayId"
+        ]
+        # attach internet gateway to subnet
+        ec2_client.attach_internet_gateway(InternetGatewayId=igw_id, VpcId=vpc_id)
+        # create route table
+        route_table_id = ec2_client.create_route_table(VpcId=vpc_id)["RouteTable"][
+            "RouteTableId"
+        ]
+        # associate route table with subnet
+        ec2_client.associate_route_table(
+            RouteTableId=route_table_id, SubnetId=subnet_id
+        )
+        # add route to route table
+        ec2_client.create_route(
+            RouteTableId=route_table_id,
+            DestinationCidrBlock="0.0.0.0/0",
+            GatewayId=igw_id,
+        )
+
+        conn = client("rds", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_subnet_group(
+            DBSubnetGroupName="subnet-group",
+            DBSubnetGroupDescription="subnet-group",
+            SubnetIds=[subnet_id],
+        )
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            AllocatedStorage=10,
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PubliclyAccessible=True,
+            DBSubnetGroupName="subnet-group",
+            VpcSecurityGroupIds=[default_sg_id],
+        )
+        from prowler.providers.aws.services.ec2.ec2_service import EC2
+        from prowler.providers.aws.services.rds.rds_service import RDS
+        from prowler.providers.aws.services.vpc.vpc_service import VPC
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access.rds_client",
+                new=RDS(aws_provider),
+            ), mock.patch(
+                "prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access.ec2_client",
+                new=EC2(aws_provider),
+            ), mock.patch(
+                "prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access import (
+                    rds_instance_no_public_access,
+                )
+
+                check = rds_instance_no_public_access()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"RDS Instance db-master-1 is set as publicly accessible and security group default ({default_sg_id}) has postgres port 5432 open to the Internet at endpoint db-master-1.aaaaaaaaaa.us-east-1.rds.amazonaws.com in a public subnet {subnet_id}."
+                )
+                assert result[0].resource_id == "db-master-1"
+                assert result[0].region == AWS_REGION_US_EAST_1
+                assert (
+                    result[0].resource_arn
+                    == f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
+                )
+                assert result[0].resource_tags == []

tests/providers/aws/services/rds/rds_service_test.py

@@ -33,35 +33,35 @@ def mock_make_api_call(self, operation_name, kwarg):
 
 @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
 class Test_RDS_Service:
-    # Test Dynamo Service
+    # Test RDS Service
     @mock_aws
     def test_service(self):
-        # Dynamo client for this test class
+        # RDS client for this test class
         aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
         rds = RDS(aws_provider)
         assert rds.service == "rds"
 
-    # Test Dynamo Client
+    # Test RDS Client
     @mock_aws
     def test_client(self):
-        # Dynamo client for this test class
+        # RDS client for this test class
         aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
         rds = RDS(aws_provider)
         for regional_client in rds.regional_clients.values():
             assert regional_client.__class__.__name__ == "RDS"
 
-    # Test Dynamo Session
+    # Test RDS Session
     @mock_aws
     def test__get_session__(self):
-        # Dynamo client for this test class
+        # RDS client for this test class
         aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
         rds = RDS(aws_provider)
         assert rds.session.__class__.__name__ == "Session"
 
-    # Test Dynamo Session
+    # Test RDS Session
     @mock_aws
     def test_audited_account(self):
-        # Dynamo client for this test class
+        # RDS client for this test class
         aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
         rds = RDS(aws_provider)
         assert rds.audited_account == AWS_ACCOUNT_NUMBER
@@ -120,6 +120,7 @@ class Test_RDS_Service:
             {"Key": "test", "Value": "test"},
         ]
         assert "test" in db_instance.parameter_groups
+        assert db_instance.subnet_ids == []
         assert db_instance.copy_tags_to_snapshot
         assert db_instance.port == 5432
 

tests/providers/aws/services/redshift/redshift_cluster_public_access/redshift_cluster_public_access_test.py

@@ -2,7 +2,11 @@ from unittest import mock
 from uuid import uuid4
 
 from prowler.providers.aws.services.redshift.redshift_service import Cluster
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    set_mocked_aws_provider,
+)
 
 CLUSTER_ID = str(uuid4())
 CLUSTER_ARN = (
@@ -14,22 +18,33 @@ class Test_redshift_cluster_public_access:
     def test_no_clusters(self):
         redshift_client = mock.MagicMock
         redshift_client.clusters = []
+        vpc_client = mock.MagicMock
+        ec2_client = mock.MagicMock
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
         with mock.patch(
-            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
-            redshift_client,
-        ), mock.patch(
-            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
-            redshift_client,
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
         ):
-            from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
-                redshift_cluster_public_access,
-            )
+            with mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.redshift_client",
+                new=redshift_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.vpc_client",
+                new=vpc_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.ec2_client",
+                new=ec2_client,
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
+                    redshift_cluster_public_access,
+                )
 
-            check = redshift_cluster_public_access()
-            result = check.execute()
-            assert len(result) == 0
+                check = redshift_cluster_public_access()
+                result = check.execute()
+                assert len(result) == 0
 
-    def test_cluster_is_public(self):
+    def test_cluster_with_public_endpoint(self):
         redshift_client = mock.MagicMock
         redshift_client.clusters = []
         redshift_client.clusters.append(
@@ -41,26 +56,37 @@ class Test_redshift_cluster_public_access:
                 endpoint_address="192.192.192.192",
             )
         )
-        with mock.patch(
-            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
-            redshift_client,
-        ), mock.patch(
-            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
-            redshift_client,
-        ):
-            from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
-                redshift_cluster_public_access,
-            )
+        vpc_client = mock.MagicMock
+        ec2_client = mock.MagicMock
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
 
-            check = redshift_cluster_public_access()
-            result = check.execute()
-            assert result[0].status == "FAIL"
-            assert (
-                result[0].status_extended
-                == f"Redshift Cluster {CLUSTER_ID} is publicly accessible at endpoint 192.192.192.192."
-            )
-            assert result[0].resource_id == CLUSTER_ID
-            assert result[0].resource_arn == CLUSTER_ARN
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.redshift_client",
+                new=redshift_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.vpc_client",
+                new=vpc_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.ec2_client",
+                new=ec2_client,
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
+                    redshift_cluster_public_access,
+                )
+
+                check = redshift_cluster_public_access()
+                result = check.execute()
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"Redshift Cluster {CLUSTER_ID} has the endpoint 192.192.192.192 set as publicly accessible but is not publicly exposed."
+                )
+                assert result[0].resource_id == CLUSTER_ID
+                assert result[0].resource_arn == CLUSTER_ARN
 
     def test_cluster_is_not_public1(self):
         redshift_client = mock.MagicMock
@@ -74,26 +100,37 @@ class Test_redshift_cluster_public_access:
                 endpoint_address="192.192.192.192",
             )
         )
-        with mock.patch(
-            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
-            redshift_client,
-        ), mock.patch(
-            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
-            redshift_client,
-        ):
-            from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
-                redshift_cluster_public_access,
-            )
+        vpc_client = mock.MagicMock
+        ec2_client = mock.MagicMock
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
 
-            check = redshift_cluster_public_access()
-            result = check.execute()
-            assert result[0].status == "PASS"
-            assert (
-                result[0].status_extended
-                == f"Redshift Cluster {CLUSTER_ID} is not publicly accessible."
-            )
-            assert result[0].resource_id == CLUSTER_ID
-            assert result[0].resource_arn == CLUSTER_ARN
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.redshift_client",
+                new=redshift_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.vpc_client",
+                new=vpc_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.ec2_client",
+                new=ec2_client,
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
+                    redshift_cluster_public_access,
+                )
+
+                check = redshift_cluster_public_access()
+                result = check.execute()
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"Redshift Cluster {CLUSTER_ID} is not publicly accessible."
+                )
+                assert result[0].resource_id == CLUSTER_ID
+                assert result[0].resource_arn == CLUSTER_ARN
 
     def test_cluster_is_not_public2(self):
         redshift_client = mock.MagicMock
@@ -106,23 +143,145 @@ class Test_redshift_cluster_public_access:
                 public_access=True,
             )
         )
-        with mock.patch(
-            "prowler.providers.aws.services.redshift.redshift_service.Redshift",
-            redshift_client,
-        ), mock.patch(
-            "prowler.providers.aws.services.redshift.redshift_client.redshift_client",
-            redshift_client,
-        ):
-            from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
-                redshift_cluster_public_access,
-            )
+        vpc_client = mock.MagicMock
+        ec2_client = mock.MagicMock
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
 
-            check = redshift_cluster_public_access()
-            result = check.execute()
-            assert result[0].status == "PASS"
-            assert (
-                result[0].status_extended
-                == f"Redshift Cluster {CLUSTER_ID} is not publicly accessible."
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.redshift_client",
+                new=redshift_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.vpc_client",
+                new=vpc_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.ec2_client",
+                new=ec2_client,
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
+                    redshift_cluster_public_access,
+                )
+
+                check = redshift_cluster_public_access()
+                result = check.execute()
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"Redshift Cluster {CLUSTER_ID} is not publicly accessible."
+                )
+                assert result[0].resource_id == CLUSTER_ID
+                assert result[0].resource_arn == CLUSTER_ARN
+
+    def test_cluster_is_in_public_subnet(self):
+        redshift_client = mock.MagicMock
+        redshift_client.clusters = []
+        redshift_client.clusters.append(
+            Cluster(
+                id=CLUSTER_ID,
+                arn=CLUSTER_ARN,
+                region=AWS_REGION_EU_WEST_1,
+                public_access=True,
+                vpc_id="vpc-123456",
+                subnets=["subnet-123456"],
+                endpoint_address="192.192.192.192",
             )
-            assert result[0].resource_id == CLUSTER_ID
-            assert result[0].resource_arn == CLUSTER_ARN
+        )
+        vpc_client = mock.MagicMock
+        vpc_client.subnets = {"subnet-123456": mock.MagicMock(public=True)}
+        ec2_client = mock.MagicMock
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.redshift_client",
+                new=redshift_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.vpc_client",
+                new=vpc_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.ec2_client",
+                new=ec2_client,
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
+                    redshift_cluster_public_access,
+                )
+
+                check = redshift_cluster_public_access()
+                result = check.execute()
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"Redshift Cluster {CLUSTER_ID} has the endpoint 192.192.192.192 set as publicly accessible in a public subnet but is not publicly exposed."
+                )
+                assert result[0].resource_id == CLUSTER_ID
+                assert result[0].resource_arn == CLUSTER_ARN
+
+    def test_cluster_with_public_vpc_sgs(self):
+        redshift_client = mock.MagicMock
+        redshift_client.audited_partition = "aws"
+        redshift_client.audited_account = AWS_ACCOUNT_NUMBER
+        redshift_client.clusters = []
+        redshift_client.clusters.append(
+            Cluster(
+                id=CLUSTER_ID,
+                arn=CLUSTER_ARN,
+                region=AWS_REGION_EU_WEST_1,
+                vpc_id="vpc-123456",
+                public_access=True,
+                vpc_security_groups=["sg-123456"],
+                endpoint_address="192.192.192.192",
+                subnets=["subnet-123456"],
+            )
+        )
+        vpc_client = mock.MagicMock
+        vpc_client.subnets = {"subnet-123456": mock.MagicMock(public=True)}
+        ec2_client = mock.MagicMock
+        ec2_client.security_groups = {
+            f"arn:aws:ec2:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:security-group/sg-123456": mock.MagicMock(
+                id="sg-123456",
+                ingress_rules=[
+                    {
+                        "IpProtocol": "-1",
+                        "IpRanges": [
+                            {
+                                "CidrIp": "0.0.0.0/0",
+                            },
+                        ],
+                    }
+                ],
+            )
+        }
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.redshift_client",
+                new=redshift_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.vpc_client",
+                new=vpc_client,
+            ), mock.patch(
+                "prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access.ec2_client",
+                new=ec2_client,
+            ):
+                from prowler.providers.aws.services.redshift.redshift_cluster_public_access.redshift_cluster_public_access import (
+                    redshift_cluster_public_access,
+                )
+
+                check = redshift_cluster_public_access()
+                result = check.execute()
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"Redshift Cluster {CLUSTER_ID} has the endpoint 192.192.192.192 set as publicly accessible and it is exposed to the Internet by security group (sg-123456) in a public subnet."
+                )
+                assert result[0].resource_id == CLUSTER_ID
+                assert result[0].resource_arn == CLUSTER_ARN

tests/providers/aws/services/redshift/redshift_service_test.py

@@ -120,6 +120,12 @@ class Test_Redshift_Service:
         assert redshift.clusters[0].id == cluster_id
         assert redshift.clusters[0].region == AWS_REGION_EU_WEST_1
         assert redshift.clusters[0].public_access
+        assert redshift.clusters[0].vpc_id == response["Cluster"].get("VpcId")
+        assert redshift.clusters[0].vpc_security_groups == [
+            sg["VpcSecurityGroupId"]
+            for sg in response["Cluster"]["VpcSecurityGroups"]
+            if sg["Status"] == "active"
+        ]
         assert (
             redshift.clusters[0].endpoint_address
             == response["Cluster"]["Endpoint"]["Address"]
@@ -233,3 +239,39 @@ class Test_Redshift_Service:
         ]
         assert redshift.clusters[0].parameter_group_name == "default.redshift-1.0"
         assert redshift.clusters[0].require_ssl is True
+
+    @mock_aws
+    def test_describe_cluster_subnets(self):
+        ec2_client = client("ec2", region_name=AWS_REGION_EU_WEST_1)
+        vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
+
+        subnet_id = ec2_client.create_subnet(
+            VpcId=vpc_id,
+            CidrBlock="10.0.1.0/24",
+            AvailabilityZone=f"{AWS_REGION_EU_WEST_1}a",
+        )["Subnet"]["SubnetId"]
+        redshift_client = client("redshift", region_name=AWS_REGION_EU_WEST_1)
+        redshift_client.create_cluster_subnet_group(
+            ClusterSubnetGroupName="test-subnet",
+            Description="Test Subnet",
+            SubnetIds=[subnet_id],
+        )
+        _ = redshift_client.create_cluster(
+            DBName="test",
+            ClusterIdentifier=cluster_id,
+            ClusterType="single-node",
+            NodeType="ds2.xlarge",
+            MasterUsername="user",
+            MasterUserPassword="password",
+            PubliclyAccessible=True,
+            VpcSecurityGroupIds=["sg-123456"],
+            ClusterSubnetGroupName="test-subnet",
+        )
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        redshift = Redshift(aws_provider)
+
+        assert len(redshift.clusters) == 1
+        assert redshift.clusters[0].id == cluster_id
+        assert redshift.clusters[0].region == AWS_REGION_EU_WEST_1
+        assert redshift.clusters[0].subnet_group == "test-subnet"
+        assert redshift.clusters[0].subnets[0] == subnet_id

tests/providers/aws/services/vpc/vpc_service_test.py

@@ -415,6 +415,52 @@ class Test_VPC_Service:
                 assert vpc.subnets[0].region == AWS_REGION_US_EAST_1
                 assert vpc.subnets[0].tags is None
 
+    @mock_aws
+    def test_vpc_subnet_with_open_nacl(self):
+        # Generate VPC Client
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        # Create VPC
+        vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
+        subnet_id = ec2_client.create_subnet(
+            VpcId=vpc_id,
+            CidrBlock="10.0.0.0/16",
+            AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
+        )["Subnet"]["SubnetId"]
+        nacl_id = ec2_client.create_network_acl(VpcId=vpc_id)["NetworkAcl"][
+            "NetworkAclId"
+        ]
+        ec2_client.create_network_acl_entry(
+            NetworkAclId=nacl_id,
+            RuleNumber=100,
+            Protocol="-1",
+            RuleAction="allow",
+            Egress=False,
+            CidrBlock="0.0.0.0/0",
+        )
+        ec2_client.create_network_acl_entry(
+            NetworkAclId=nacl_id,
+            RuleNumber=200,
+            Protocol="-1",
+            RuleAction="allow",
+            Egress=True,
+            CidrBlock="0.0.0.0/0",
+        )
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
+        )
+        from prowler.providers.aws.services.vpc.vpc_service import VPC
+
+        vpc = VPC(aws_provider)
+        assert (
+            len(vpc.vpcs) == 3
+        )  # Number of AWS regions + created VPC, one default VPC per region
+        for vpc in vpc.vpcs.values():
+            if vpc.cidr_block == "10.0.0.0/16":
+                assert vpc.subnets[0].id == subnet_id
+                assert vpc.subnets[0].vpc_id == vpc_id
+                assert vpc.subnets[0].availability_zone == f"{AWS_REGION_US_EAST_1}a"
+                assert vpc.subnets[0].region == AWS_REGION_US_EAST_1
+
     # Test VPC Describe VPN Connections
     @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
     def test_describe_vpn_connections(self):

tests/providers/aws/services/workspaces/workspaces_vpc_2private_1public_subnets_nat/workspaces_vpc_2private_1public_subnets_nat_test.py

@@ -365,6 +365,23 @@ class Test_workspaces_vpc_2private_1public_subnets_nat:
             RouteTableId=route_table_private["RouteTable"]["RouteTableId"],
             SubnetId=subnet_private["Subnet"]["SubnetId"],
         )
+        nacls = (
+            ec2_client.describe_network_acls(
+                Filters=[
+                    {
+                        "Name": "association.subnet-id",
+                        "Values": [subnet_private["Subnet"]["SubnetId"]],
+                    }
+                ]
+            )
+        )["NetworkAcls"]
+        # Remove public ingress rule from private subnet in default NACLs
+        for nacl in nacls:
+            ec2_client.delete_network_acl_entry(
+                NetworkAclId=nacl["NetworkAclId"],
+                Egress=True,
+                RuleNumber=100,
+            )
         # VPC Private 2
         subnet_private_2 = ec2_client.create_subnet(
             VpcId=vpc.id,