feat(aws): Add Fault Tolerance Checks (#5488)

Co-authored-by: Sergio <sergio@prowler.com>
2026-07-24 04:51:51 +00:00 · 2024-10-23 17:07:43 -04:00
parent fef99fd5fb
commit 28f8915f6f
17 changed files with 612 additions and 2 deletions
@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "documentdb_cluster_multi_az_enabled",
+  "CheckTitle": "Ensure DocumentDB Cluster have Multi-AZ enabled.",
+  "CheckType": [],
+  "ServiceName": "documentdb",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:rds:region:account-id:db-cluster",
+  "Severity": "medium",
+  "ResourceType": "AwsRdsDbCluster",
+  "Description": "Ensure DocumentDB Cluster have Multi-AZ enabled.",
+  "Risk": "Ensure that your Amazon DocumentDB Clusters are using Multi-AZ deployment configurations to provide High Availability (HA) through automatic failover to standby replicas in the event of a failure such as an Availability Zone (AZ) outage, an internal hardware or network outage, a software failure or in case of a planned maintenance session.",
+  "RelatedUrl": "https://docs.aws.amazon.com/documentdb/latest/developerguide/failover.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable Multi-AZ for all DocumentDB Clusters.",
+      "Url": "https://docs.aws.amazon.com/documentdb/latest/developerguide/failover.html"
+    }
+  },
+  "Categories": [
+    "redundancy"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,28 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.documentdb.documentdb_client import (
+    documentdb_client,
+)
+
+
+class documentdb_cluster_multi_az_enabled(Check):
+    def execute(self):
+        findings = []
+        for db_cluster in documentdb_client.db_clusters.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = db_cluster.region
+            report.resource_id = db_cluster.id
+            report.resource_arn = db_cluster.arn
+            report.resource_tags = db_cluster.tags
+            report.status = "FAIL"
+            report.status_extended = (
+                f"DocumentDB Cluster {db_cluster.id} does not have Multi-AZ enabled."
+            )
+            if db_cluster.multi_az:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"DocumentDB Cluster {db_cluster.id} has Multi-AZ enabled."
+                )
+
+            findings.append(report)
+
+        return findings
@@ -23,7 +23,9 @@
      "Url": "https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DAX.concepts.cluster.html#DAX.concepts.regions-and-azs"
    }
  },
-  "Categories": [],
+  "Categories": [
+    "redundancy"
+  ],
  "DependsOn": [],
  "RelatedTo": [],
  "Notes": ""
@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "efs_multi_az_enabled",
+  "CheckTitle": "Check if EFS File systems are configured with Multi-AZ.",
+  "CheckType": [],
+  "ServiceName": "efs",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "medium",
+  "ResourceType": "AwsEfsFileSystem",
+  "Description": "Check if EFS File systems has Multi-AZ enabled.",
+  "Risk": "Ensure that your Amazon EFS are using Multi-AZ deployment configurations to provide High Availability (HA).",
+  "RelatedUrl": "https://docs.aws.amazon.com/efs/latest/ug/creating-using-create-fs.html#availabiltydurability",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable EFS file systems Multi-AZ with mount targets in multiple AZs.",
+      "Url": "https://docs.aws.amazon.com/efs/latest/ug/creating-using-create-fs.html#availabiltydurability"
+    }
+  },
+  "Categories": [
+    "redundancy"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.efs.efs_client import efs_client
+
+
+class efs_multi_az_enabled(Check):
+    def execute(self):
+        findings = []
+        for fs in efs_client.filesystems.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = fs.region
+            report.resource_id = fs.id
+            report.resource_arn = fs.arn
+            report.resource_tags = fs.tags
+            if fs.availability_zone_id:
+                report.status = "FAIL"
+                report.status_extended = f"EFS {fs.id} is a Single-AZ file system."
+            else:
+                if fs.number_of_mount_targets <= 1:
+                    report.status = "FAIL"
+                    report.status_extended = f"EFS {fs.id} is a Multi-AZ file system but with only one mount target."
+                else:
+                    report.status = "PASS"
+                    report.status_extended = f"EFS {fs.id} is a Multi-AZ file system with more than one mount target."
+
+            findings.append(report)
+
+        return findings
@@ -39,6 +39,8 @@ class EFS(AWSService):
                            id=efs_id,
                            arn=efs_arn,
                            region=regional_client.region,
+                            availability_zone_id=efs.get("AvailabilityZoneId", ""),
+                            number_of_mount_targets=efs["NumberOfMountTargets"],
                            encrypted=efs["Encrypted"],
                            tags=efs.get("Tags"),
                        )
@@ -166,6 +168,8 @@ class FileSystem(BaseModel):
    policy: Optional[dict] = {}
    backup_policy: Optional[str] = "DISABLED"
    encrypted: bool
+    availability_zone_id: Optional[str]
+    number_of_mount_targets: int
    mount_targets: list[MountTarget] = []
    access_points: list[AccessPoint] = []
    tags: Optional[list] = []
@@ -23,7 +23,9 @@
      "Url": "https://docs.aws.amazon.com/redshift/latest/mgmt/managing-cluster-multi-az.html"
    }
  },
-  "Categories": [],
+  "Categories": [
+    "redundancy"
+  ],
  "DependsOn": [],
  "RelatedTo": [],
  "Notes": ""
@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "vpc_endpoint_multi_az_enabled",
+  "CheckTitle": "Amazon VPC Interface Endpoints should have ENIs in more than one subnet.",
+  "CheckType": [],
+  "ServiceName": "vpc",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "medium",
+  "ResourceType": "AwsVpcEndpointService",
+  "Description": "Ensure that all vpc interface endpoints have ENIs in multiple subnets. If a VPC endpoint has an ENI in only a single subnet then this check will fail. You cannot create VPC Endpoints in 2 different subnets in the same AZ. So, for the purpose of VPC endpoints, having multiple subnets implies multiple AZs.",
+  "Risk": "Without VPC endpoints ENIs in multiple subnets an AZ impacting event could lead to increased downtime or your network traffic between your VPC and Amazon services may traverse the public internet.",
+  "RelatedUrl": "https://docs.aws.amazon.com/vpc/latest/privatelink/interface-endpoints.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "To improve the availability of your services residing in your VPC, configure multiple subnets for VPC Interface Endpoints.",
+      "Url": "https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/interface-vpc-endpoints.html"
+    }
+  },
+  "Categories": [
+    "redundancy"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,23 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.vpc.vpc_client import vpc_client
+
+
+class vpc_endpoint_multi_az_enabled(Check):
+    def execute(self):
+        findings = []
+        for endpoint in vpc_client.vpc_endpoints:
+            if endpoint.vpc_id in vpc_client.vpcs and endpoint.type == "Interface":
+                report = Check_Report_AWS(self.metadata())
+                report.region = endpoint.region
+                report.resource_tags = endpoint.tags
+                report.resource_id = endpoint.id
+                report.resource_arn = endpoint.arn
+                report.status = "FAIL"
+                report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} has subnets in different AZs."
+                if len(endpoint.subnet_ids) > 1:
+                    report.status = "PASS"
+                    report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} does not have subnets in different AZs."
+
+                findings.append(report)
+
+        return findings
@@ -240,6 +240,7 @@ class VPC(AWSService):
                                    service_name=endpoint["ServiceName"],
                                    state=endpoint["State"],
                                    policy_document=endpoint_policy,
+                                    subnet_ids=endpoint.get("SubnetIds", []),
                                    owner_id=endpoint["OwnerId"],
                                    type=endpoint["VpcEndpointType"],
                                    region=regional_client.region,
@@ -484,6 +485,7 @@ class VpcEndpoint(BaseModel):
    vpc_id: str
    service_name: str
    state: str
+    subnet_ids: Optional[list] = []
    policy_document: Optional[dict]
    owner_id: str
    type: str
@@ -0,0 +1,106 @@
+from unittest import mock
+
+from prowler.providers.aws.services.documentdb.documentdb_service import DBCluster
+
+AWS_ACCOUNT_NUMBER = "123456789012"
+AWS_REGION = "us-east-1"
+
+DOC_DB_CLUSTER_NAME = "test-cluster"
+DOC_DB_CLUSTER_ARN = (
+    f"arn:aws:rds:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:cluster:{DOC_DB_CLUSTER_NAME}"
+)
+DOC_DB_ENGINE_VERSION = "5.0.0"
+
+
+class Test_documentdb_cluster_multi_az_enabled:
+    def test_documentdb_no_clusters(self):
+        documentdb_client = mock.MagicMock
+        documentdb_client.db_clusters = {}
+
+        with mock.patch(
+            "prowler.providers.aws.services.documentdb.documentdb_service.DocumentDB",
+            new=documentdb_client,
+        ):
+            from prowler.providers.aws.services.documentdb.documentdb_cluster_multi_az_enabled.documentdb_cluster_multi_az_enabled import (
+                documentdb_cluster_multi_az_enabled,
+            )
+
+            check = documentdb_cluster_multi_az_enabled()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_documentdb_cluster_not_encrypted(self):
+        documentdb_client = mock.MagicMock
+        documentdb_client.db_clusters = {
+            DOC_DB_CLUSTER_ARN: DBCluster(
+                id=DOC_DB_CLUSTER_NAME,
+                arn=DOC_DB_CLUSTER_ARN,
+                engine="docdb",
+                status="available",
+                backup_retention_period=1,
+                encrypted=False,
+                cloudwatch_logs=[],
+                multi_az=False,
+                parameter_group="default.docdb3.6",
+                deletion_protection=True,
+                region=AWS_REGION,
+                tags=[],
+            )
+        }
+
+        with mock.patch(
+            "prowler.providers.aws.services.documentdb.documentdb_service.DocumentDB",
+            new=documentdb_client,
+        ):
+            from prowler.providers.aws.services.documentdb.documentdb_cluster_multi_az_enabled.documentdb_cluster_multi_az_enabled import (
+                documentdb_cluster_multi_az_enabled,
+            )
+
+            check = documentdb_cluster_multi_az_enabled()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"DocumentDB Cluster {DOC_DB_CLUSTER_NAME} does not have Multi-AZ enabled."
+            )
+            assert result[0].region == AWS_REGION
+            assert result[0].resource_id == DOC_DB_CLUSTER_NAME
+            assert result[0].resource_arn == DOC_DB_CLUSTER_ARN
+
+    def test_documentdb_cluster_with_encryption(self):
+        documentdb_client = mock.MagicMock
+        documentdb_client.db_clusters = {
+            DOC_DB_CLUSTER_ARN: DBCluster(
+                id=DOC_DB_CLUSTER_NAME,
+                arn=DOC_DB_CLUSTER_ARN,
+                engine="docdb",
+                status="available",
+                backup_retention_period=1,
+                encrypted=True,
+                cloudwatch_logs=[],
+                multi_az=True,
+                parameter_group="default.docdb3.6",
+                deletion_protection=True,
+                region=AWS_REGION,
+                tags=[],
+            )
+        }
+        with mock.patch(
+            "prowler.providers.aws.services.documentdb.documentdb_service.DocumentDB",
+            new=documentdb_client,
+        ):
+            from prowler.providers.aws.services.documentdb.documentdb_cluster_multi_az_enabled.documentdb_cluster_multi_az_enabled import (
+                documentdb_cluster_multi_az_enabled,
+            )
+
+            check = documentdb_cluster_multi_az_enabled()
+            result = check.execute()
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"DocumentDB Cluster {DOC_DB_CLUSTER_NAME} has Multi-AZ enabled."
+            )
+            assert result[0].region == AWS_REGION
+            assert result[0].resource_id == DOC_DB_CLUSTER_NAME
+            assert result[0].resource_arn == DOC_DB_CLUSTER_ARN
@@ -0,0 +1,150 @@
+from unittest import mock
+
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_EU_WEST_1,
+    set_mocked_aws_provider,
+)
+
+CREATION_TOKEN = "fs-123"
+
+
+class Test_efs_multi_az_enabled:
+    @mock_aws
+    def test_no_efs_filesystems(self):
+        from prowler.providers.aws.services.efs.efs_service import EFS
+
+        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.efs.efs_multi_az_enabled.efs_multi_az_enabled.efs_client",
+                new=EFS(aws_provider),
+            ):
+                from prowler.providers.aws.services.efs.efs_multi_az_enabled.efs_multi_az_enabled import (
+                    efs_multi_az_enabled,
+                )
+
+                check = efs_multi_az_enabled()
+                result = check.execute()
+                assert len(result) == 0
+
+    @mock_aws
+    def test_efs_multi_az_availability_zone_id_present(self):
+        efs_client = client("efs", region_name=AWS_REGION_EU_WEST_1)
+        file_system = efs_client.create_file_system(
+            CreationToken=CREATION_TOKEN, Backup=False
+        )
+        efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
+        from prowler.providers.aws.services.efs.efs_service import EFS
+
+        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.efs.efs_multi_az_enabled.efs_multi_az_enabled.efs_client",
+                new=EFS(aws_provider),
+            ) as service_client:
+                from prowler.providers.aws.services.efs.efs_multi_az_enabled.efs_multi_az_enabled import (
+                    efs_multi_az_enabled,
+                )
+
+                service_client.filesystems[efs_arn].availability_zone_id = "az-123"
+                check = efs_multi_az_enabled()
+                result = check.execute()
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"EFS {file_system['FileSystemId']} is a Single-AZ file system."
+                )
+                assert result[0].resource_id == file_system["FileSystemId"]
+                assert (
+                    result[0].resource_arn
+                    == f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
+                )
+
+    @mock_aws
+    def test_efs_multi_az_single_mount_target(self):
+        efs_client = client("efs", region_name=AWS_REGION_EU_WEST_1)
+        file_system = efs_client.create_file_system(
+            CreationToken=CREATION_TOKEN, Backup=False
+        )
+        efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
+        from prowler.providers.aws.services.efs.efs_service import EFS
+
+        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.efs.efs_multi_az_enabled.efs_multi_az_enabled.efs_client",
+                new=EFS(aws_provider),
+            ) as service_client:
+                from prowler.providers.aws.services.efs.efs_multi_az_enabled.efs_multi_az_enabled import (
+                    efs_multi_az_enabled,
+                )
+
+                service_client.filesystems[efs_arn].number_of_mount_targets = 1
+                check = efs_multi_az_enabled()
+                result = check.execute()
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"EFS {file_system['FileSystemId']} is a Multi-AZ file system but with only one mount target."
+                )
+                assert result[0].resource_id == file_system["FileSystemId"]
+                assert (
+                    result[0].resource_arn
+                    == f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
+                )
+
+    @mock_aws
+    def test_efs_multi_az_enabled(self):
+        efs_client = client("efs", region_name=AWS_REGION_EU_WEST_1)
+        file_system = efs_client.create_file_system(
+            CreationToken=CREATION_TOKEN, Backup=False
+        )
+        efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
+        from prowler.providers.aws.services.efs.efs_service import EFS
+
+        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.efs.efs_multi_az_enabled.efs_multi_az_enabled.efs_client",
+                new=EFS(aws_provider),
+            ) as service_client:
+                from prowler.providers.aws.services.efs.efs_multi_az_enabled.efs_multi_az_enabled import (
+                    efs_multi_az_enabled,
+                )
+
+                service_client.filesystems[efs_arn].number_of_mount_targets = 12
+                check = efs_multi_az_enabled()
+                result = check.execute()
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"EFS {file_system['FileSystemId']} is a Multi-AZ file system with more than one mount target."
+                )
+                assert result[0].resource_id == file_system["FileSystemId"]
+                assert (
+                    result[0].resource_arn
+                    == f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}"
+                )
@@ -37,6 +37,8 @@ def mock_make_api_call(self, operation_name, kwarg):
                    "FileSystemId": FILE_SYSTEM_ID,
                    "Encrypted": True,
                    "Tags": [{"Key": "test", "Value": "test"}],
+                    "AvailabilityZoneId": "az-12345",
+                    "NumberOfMountTargets": 123,
                    "BackupPolicy": {"Status": "ENABLED"},
                    "Policy": json.dumps(FILESYSTEM_POLICY),
                }
@@ -108,6 +110,8 @@ class Test_EFS:
        assert len(efs.filesystems) == 1
        assert efs.filesystems[efs_arn].id == FILE_SYSTEM_ID
        assert efs.filesystems[efs_arn].encrypted
+        assert efs.filesystems[efs_arn].availability_zone_id == "az-12345"
+        assert efs.filesystems[efs_arn].number_of_mount_targets == 123
        assert efs.filesystems[efs_arn].tags == [
            {"Key": "test", "Value": "test"},
        ]
@@ -0,0 +1,166 @@
+from unittest import mock
+
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
+
+
+class Test_vpc_endpoint_for_multi_az:
+    @mock_aws
+    def test_vpc_no_endpoints(self):
+        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.vpc.vpc_endpoint_multi_az_enabled.vpc_endpoint_multi_az_enabled.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.vpc.vpc_endpoint_multi_az_enabled.vpc_endpoint_multi_az_enabled import (
+                    vpc_endpoint_multi_az_enabled,
+                )
+
+                check = vpc_endpoint_multi_az_enabled()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_vpc_no_multi_az_endpoint(self):
+        # Create VPC Mocked Resources
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        vpc = ec2_client.create_vpc(
+            CidrBlock="172.28.7.0/24",
+            InstanceTenancy="default",
+            TagSpecifications=[
+                {
+                    "ResourceType": "vpc",
+                    "Tags": [
+                        {"Key": "Name", "Value": "vpc_name"},
+                    ],
+                },
+            ],
+        )["Vpc"]
+        # VPC AZ 1
+        subnet_one = ec2_client.create_subnet(
+            VpcId=vpc["VpcId"],
+            CidrBlock="172.28.7.192/26",
+            AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
+        )["Subnet"]
+
+        route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
+        vpc_endpoint = ec2_client.create_vpc_endpoint(
+            VpcId=vpc["VpcId"],
+            ServiceName="com.amazonaws.vpce.us-east-1.ssmmessages",
+            RouteTableIds=[route_table["RouteTableId"]],
+            SubnetIds=[subnet_one["SubnetId"]],
+            VpcEndpointType="Interface",
+        )["VpcEndpoint"]
+        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.vpc.vpc_endpoint_multi_az_enabled.vpc_endpoint_multi_az_enabled.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.vpc.vpc_endpoint_multi_az_enabled.vpc_endpoint_multi_az_enabled import (
+                    vpc_endpoint_multi_az_enabled,
+                )
+
+                check = vpc_endpoint_multi_az_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].region == AWS_REGION_US_EAST_1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"VPC Endpoint {vpc_endpoint['VpcEndpointId']} in VPC {vpc['VpcId']} has subnets in different AZs."
+                )
+                assert (
+                    result[0].resource_arn
+                    == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:123456789012:vpc-endpoint/{vpc_endpoint['VpcEndpointId']}"
+                )
+
+    @mock_aws
+    def test_vpc_endpoint_multi_az_enabled(self):
+        # Create VPC Mocked Resources
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        vpc = ec2_client.create_vpc(
+            CidrBlock="172.28.7.0/24",
+            InstanceTenancy="default",
+            TagSpecifications=[
+                {
+                    "ResourceType": "vpc",
+                    "Tags": [
+                        {"Key": "Name", "Value": "vpc_name"},
+                    ],
+                },
+            ],
+        )["Vpc"]
+        # VPC AZ 1
+        subnet_one = ec2_client.create_subnet(
+            VpcId=vpc["VpcId"],
+            CidrBlock="172.28.7.192/26",
+            AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
+        )["Subnet"]
+
+        # VPC AZ 2
+        subnet_two = ec2_client.create_subnet(
+            VpcId=vpc["VpcId"],
+            CidrBlock="172.28.7.0/26",
+            AvailabilityZone=f"{AWS_REGION_US_EAST_1}b",
+        )["Subnet"]
+
+        route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
+        vpc_endpoint = ec2_client.create_vpc_endpoint(
+            VpcId=vpc["VpcId"],
+            ServiceName="com.amazonaws.vpce.us-east-1.ssmmessages",
+            RouteTableIds=[route_table["RouteTableId"]],
+            SubnetIds=[subnet_one["SubnetId"], subnet_two["SubnetId"]],
+            VpcEndpointType="Interface",
+        )["VpcEndpoint"]
+
+        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.vpc.vpc_endpoint_multi_az_enabled.vpc_endpoint_multi_az_enabled.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.vpc.vpc_endpoint_multi_az_enabled.vpc_endpoint_multi_az_enabled import (
+                    vpc_endpoint_multi_az_enabled,
+                )
+
+                check = vpc_endpoint_multi_az_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].region == AWS_REGION_US_EAST_1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"VPC Endpoint {vpc_endpoint['VpcEndpointId']} in VPC {vpc['VpcId']} does not have subnets in different AZs."
+                )
+                assert (
+                    result[0].resource_arn
+                    == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:123456789012:vpc-endpoint/{vpc_endpoint['VpcEndpointId']}"
+                )