feat(Elasticache): Additional Elasticache checks (#4317)

Co-authored-by: Sergio <sergio@prowler.com>
2026-07-24 04:51:51 +00:00 · 2024-06-27 18:07:22 -04:00
parent 630d980861
commit ba726b205d
17 changed files with 888 additions and 1 deletions
@@ -0,0 +1,30 @@
+{
+  "Provider": "aws",
+  "CheckID": "elasticache_cluster_backup_enabled",
+  "CheckTitle": "Ensure Elasticache Cluster has automatic backups enabled.",
+  "CheckType": [],
+  "ServiceName": "elasticache",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "high",
+  "ResourceType": "AWSElastiCacheClusters",
+  "Description": "Ensure Elasticache Cluster has automatic backups enabled.",
+  "Risk": "Ensure that your Amazon ElastiCache Redis cache clusters have a sufficient backup retention period set in order to fulfill your organization's compliance requirements. The retention period represents the number of days for which Amazon ElastiCache service retains automatic Redis cluster backups before deleting them.",
+  "RelatedUrl": "https://docs.aws.amazon.com/securityhub/latest/userguide/elasticache-controls.html#elasticache-1",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws elasticache modify-replication-group --region <value> --replication-group-id <value> --snapshot-retention-limit <value> --apply-immediately",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": "https://docs.prowler.com/checks/aws/general-policies/ensure-that-amazon-elasticache-redis-clusters-have-automatic-backup-turned-on/"
+    },
+    "Recommendation": {
+      "Text": "Ensure Elasticache Cluster has automatic backups enabled.",
+      "Url": "https://docs.aws.amazon.com/securityhub/latest/userguide/elasticache-controls.html#elasticache-1"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,30 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.elasticache.elasticache_client import (
+    elasticache_client,
+)
+
+
+class elasticache_cluster_backup_enabled(Check):
+    def execute(self):
+        findings = []
+        for repl_group in elasticache_client.replication_groups.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = repl_group.region
+            report.resource_id = repl_group.id
+            report.resource_arn = repl_group.arn
+            report.status = "FAIL"
+            report.status_extended = f"Elasticache Cluster {repl_group.id} does not have automated snapshot backups enabled."
+            if repl_group.snapshot_retention > elasticache_client.audit_config.get(
+                "minimum_snapshot_retention_period", 7
+            ):
+                report.status = "PASS"
+                report.status_extended = f"Elasticache Cluster {repl_group.id} has automated snapshot backups enabled with retention period {repl_group.snapshot_retention} days."
+            else:
+                if repl_group.snapshot_retention > 0:
+                    report.status = "FAIL"
+                    report.check_metadata.Severity = "low"
+                    report.status_extended = f"Elasticache Cluster {repl_group.id} has automated snapshot backups enabled with retention period {repl_group.snapshot_retention} days. Recommended to increase the snapshot retention period to a minimum of 7 days."
+
+            findings.append(report)
+
+        return findings
@@ -0,0 +1,30 @@
+{
+  "Provider": "aws",
+  "CheckID": "elasticache_cluster_multi_az_enabled",
+  "CheckTitle": "Ensure Elasticache Cluster has Multi-AZ enabled.",
+  "CheckType": [],
+  "ServiceName": "elasticache",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "medium",
+  "ResourceType": "AWSElastiCacheClusters",
+  "Description": "Ensure Elasticache Cluster has Multi-AZ enabled.",
+  "Risk": "Ensure that your Amazon ElastiCache Redis cache clusters has Multi-AZ enabled.",
+  "RelatedUrl": "https://www.trendmicro.com/cloudoneconformity-staging/knowledge-base/aws/ElastiCache/elasticache-multi-az.html#",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws elasticache modify-replication-group --region <value> --replication-group-id <value> --multi-az-enabled --apply-immediately",
+      "NativeIaC": "https://www.trendmicro.com/cloudoneconformity-staging/knowledge-base/aws/ElastiCache/elasticache-multi-az.html#",
+      "Other": "",
+      "Terraform": "https://docs.prowler.com/checks/aws/general-policies/ensure-aws-elasticache-redis-cluster-with-multi-az-automatic-failover-feature-set-to-enabled/"
+    },
+    "Recommendation": {
+      "Text": "Ensure Elasticache Cluster has Multi-AZ enabled.",
+      "Url": "https://www.trendmicro.com/cloudoneconformity-staging/knowledge-base/aws/ElastiCache/elasticache-multi-az.html#"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.elasticache.elasticache_client import (
+    elasticache_client,
+)
+
+
+class elasticache_cluster_multi_az_enabled(Check):
+    def execute(self):
+        findings = []
+        for repl_group in elasticache_client.replication_groups.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = repl_group.region
+            report.resource_id = repl_group.id
+            report.resource_arn = repl_group.arn
+            report.status = "FAIL"
+            report.status_extended = (
+                f"Elasticache Cluster {repl_group.id} does not have Multi-AZ enabled."
+            )
+            if repl_group.multi_az == "enabled":
+                report.status = "PASS"
+                report.status_extended = (
+                    f"Elasticache Cluster {repl_group.id} has Multi-AZ enabled."
+                )
+
+            findings.append(report)
+
+        return findings
@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "elasticache_replication_group_encrypted_at_rest",
+  "CheckTitle": "Ensure Elasticache Replication Groups have at rest encryption enabled.",
+  "CheckType": [],
+  "ServiceName": "elasticache",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "medium",
+  "ResourceType": "AWSElastiCacheReplicationGroups",
+  "Description": "Ensure Elasticache Replication Groups have at rest encryption enabled.",
+  "Risk": "There is a risk of exposing sensitive data if Elasticache Replication Group does not have at rest encryption enabled.",
+  "RelatedUrl": "https://docs.aws.amazon.com/securityhub/latest/userguide/elasticache-controls.html#elasticache-4",
+  "Remediation": {
+    "Code": {
+      "CLI": "https://docs.prowler.com/checks/aws/general-policies/general_9/",
+      "NativeIaC": "https://docs.prowler.com/checks/aws/general-policies/general_9/",
+      "Other": "",
+      "Terraform": "https://docs.prowler.com/checks/aws/general-policies/general_9/"
+    },
+    "Recommendation": {
+      "Text": "Ensure your Elasticache replication groups have at rest encryption enabled.",
+      "Url": "https://docs.aws.amazon.com/AmazonElastiCache/latest/red-ug/at-rest-encryption.html#at-rest-encryption-enable"
+    }
+  },
+  "Categories": [
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,23 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.elasticache.elasticache_client import (
+    elasticache_client,
+)
+
+
+class elasticache_replication_group_encrypted_at_rest(Check):
+    def execute(self):
+        findings = []
+        for repl_group in elasticache_client.replication_groups.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = repl_group.region
+            report.resource_id = repl_group.id
+            report.resource_arn = repl_group.arn
+            report.status = "FAIL"
+            report.status_extended = f"Elasticache Replication Group {repl_group.id} does not have at rest encryption enabled."
+            if repl_group.encrypted:
+                report.status = "PASS"
+                report.status_extended = f"Elasticache Replication Group {repl_group.id} has at rest encryption enabled."
+
+            findings.append(report)
+
+        return findings
@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "elasticache_replication_group_encrypted_in_transit",
+  "CheckTitle": "Ensure Elasticache Replication Groups have in transit encryption enabled.",
+  "CheckType": [],
+  "ServiceName": "elasticache",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "medium",
+  "ResourceType": "AWSElastiCacheReplicationGroups",
+  "Description": "Ensure Elasticache Replication Groups have in transit encryption enabled.",
+  "Risk": "There is a risk of exposing sensitive data if Elasticache Replication Group does not have in transit encryption enabled.",
+  "RelatedUrl": "https://docs.aws.amazon.com/securityhub/latest/userguide/elasticache-controls.html#elasticache-5",
+  "Remediation": {
+    "Code": {
+      "CLI": "https://docs.prowler.com/checks/aws/general-policies/general_10/",
+      "NativeIaC": "https://docs.prowler.com/checks/aws/general-policies/general_10/",
+      "Other": "",
+      "Terraform": "https://docs.prowler.com/checks/aws/general-policies/general_10/"
+    },
+    "Recommendation": {
+      "Text": "Ensure your Elasticache replication groups have in transit encryption enabled.",
+      "Url": "https://docs.aws.amazon.com/AmazonElastiCache/latest/red-ug/in-transit-encryption.html"
+    }
+  },
+  "Categories": [
+    "encryption"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,23 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.elasticache.elasticache_client import (
+    elasticache_client,
+)
+
+
+class elasticache_replication_group_encrypted_in_transit(Check):
+    def execute(self):
+        findings = []
+        for repl_group in elasticache_client.replication_groups.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = repl_group.region
+            report.resource_id = repl_group.id
+            report.resource_arn = repl_group.arn
+            report.status = "FAIL"
+            report.status_extended = f"Elasticache Replication Group {repl_group.id} does not have in transit encryption enabled."
+            if repl_group.transit_encryption:
+                report.status = "PASS"
+                report.status_extended = f"Elasticache Replication Group {repl_group.id} has in transit encryption enabled."
+
+            findings.append(report)
+
+        return findings
@@ -14,9 +14,11 @@ class ElastiCache(AWSService):
        # Call AWSService's __init__
        super().__init__(__class__.__name__, provider)
        self.clusters = {}
+        self.replication_groups = {}
        self.__threading_call__(self.__describe_cache_clusters__)
        self.__threading_call__(self.__describe_cache_subnet_groups__)
        self.__list_tags_for_resource__()
+        self.__threading_call__(self.__describe_replication_groups__)

    def __describe_cache_clusters__(self, regional_client):
        logger.info("Elasticache - Describing Cache Clusters...")
@@ -32,6 +34,11 @@ class ElastiCache(AWSService):
                        id=cache_cluster["CacheClusterId"],
                        arn=cluster_arn,
                        region=regional_client.region,
+                        security_groups=[
+                            sg["SecurityGroupId"]
+                            for sg in cache_cluster["SecurityGroups"]
+                            if sg["Status"] == "active"
+                        ],
                        cache_subnet_group_id=cache_cluster.get(
                            "CacheSubnetGroupName", None
                        ),
@@ -95,11 +102,48 @@ class ElastiCache(AWSService):
                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

+    def __describe_replication_groups__(self, regional_client):
+        logger.info("Elasticache - Describing Replication Groups...")
+        try:
+            for repl_group in regional_client.describe_replication_groups()[
+                "ReplicationGroups"
+            ]:
+                replication_arn = repl_group["ARN"]
+                if not self.audit_resources or (
+                    is_resource_filtered(replication_arn, self.audit_resources)
+                ):
+                    self.replication_groups[replication_arn] = ReplicationGroup(
+                        id=repl_group["ReplicationGroupId"],
+                        arn=replication_arn,
+                        region=regional_client.region,
+                        status=repl_group["Status"],
+                        snapshot_retention=repl_group.get("SnapshotRetentionLimit", 0),
+                        encrypted=repl_group["AtRestEncryptionEnabled"],
+                        transit_encryption=repl_group["TransitEncryptionEnabled"],
+                        multi_az=repl_group["MultiAZ"],
+                    )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+

 class Cluster(BaseModel):
    id: str
    arn: str
    region: str
+    security_groups: list[str] = []
    cache_subnet_group_id: Optional[str]
    subnets: list = []
    tags: Optional[list]
+
+
+class ReplicationGroup(BaseModel):
+    id: str
+    arn: str
+    region: str
+    status: str
+    snapshot_retention: int
+    encrypted: bool
+    transit_encryption: bool
+    multi_az: str
@@ -0,0 +1,173 @@
+from unittest import mock
+
+import botocore
+from mock import MagicMock
+from moto import mock_aws
+
+from prowler.providers.aws.services.elasticache.elasticache_service import (
+    ReplicationGroup,
+)
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+REPLICATION_GROUP_ID = "clustered-redis"
+REPLICATION_GROUP_ARN = f"arn:aws:elasticache:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:replicationgroup:{REPLICATION_GROUP_ID}"
+REPLICATION_GROUP_STATUS = "available"
+REPLICATION_GROUP_SNAPSHOT_RETENTION = "0"
+REPLICATION_GROUP_ENCRYPTION = True
+REPLICATION_GROUP_TRANSIT_ENCRYPTION = True
+REPLICATION_GROUP_MULTI_AZ = "enabled"
+
+# Patch every AWS call using Boto3
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+class Test_elasticache_cluster_backup_enabled:
+    @mock_aws
+    def test_elasticache_no_replication_groups(self):
+
+        # Mock ElastiCache Service
+        elasticache_client = MagicMock
+        elasticache_client.replication_groups = {}
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_client,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_cluster_backup_enabled.elasticache_cluster_backup_enabled import (
+                elasticache_cluster_backup_enabled,
+            )
+
+            check = elasticache_cluster_backup_enabled()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_elasticache_cluster_backup_disabled(self):
+        # Mock ElastiCache Service
+        elasticache_client = MagicMock
+        elasticache_client.replication_groups = {}
+
+        elasticache_client.replication_groups[REPLICATION_GROUP_ARN] = ReplicationGroup(
+            arn=REPLICATION_GROUP_ARN,
+            id=REPLICATION_GROUP_ID,
+            region=AWS_REGION_US_EAST_1,
+            status=REPLICATION_GROUP_STATUS,
+            snapshot_retention=0,
+            encrypted=False,
+            transit_encryption=False,
+            multi_az=REPLICATION_GROUP_MULTI_AZ,
+        )
+
+        elasticache_client.audit_config = {"minimum_snapshot_retention_period": 7}
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_client,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_cluster_backup_enabled.elasticache_cluster_backup_enabled import (
+                elasticache_cluster_backup_enabled,
+            )
+
+            check = elasticache_cluster_backup_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Cluster {REPLICATION_GROUP_ID} does not have automated snapshot backups enabled."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
+
+    def test_elasticache_cluster_backup_enabled(self):
+        # Mock ElastiCache Service
+        elasticache_client = MagicMock
+        elasticache_client.replication_groups = {}
+
+        elasticache_client.replication_groups[REPLICATION_GROUP_ARN] = ReplicationGroup(
+            arn=REPLICATION_GROUP_ARN,
+            id=REPLICATION_GROUP_ID,
+            region=AWS_REGION_US_EAST_1,
+            status=REPLICATION_GROUP_STATUS,
+            snapshot_retention=9,
+            encrypted=REPLICATION_GROUP_ENCRYPTION,
+            transit_encryption=REPLICATION_GROUP_TRANSIT_ENCRYPTION,
+            multi_az=REPLICATION_GROUP_MULTI_AZ,
+        )
+
+        elasticache_client.audit_config = {"minimum_snapshot_retention_period": 7}
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_client,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_cluster_backup_enabled.elasticache_cluster_backup_enabled import (
+                elasticache_cluster_backup_enabled,
+            )
+
+            check = elasticache_cluster_backup_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Cluster {REPLICATION_GROUP_ID} has automated snapshot backups enabled with retention period 9 days."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
+
+    def test_elasticache_cluster_backup_enabled_modified_retention(self):
+        # Mock ElastiCache Service
+        elasticache_client = MagicMock
+        elasticache_client.replication_groups = {}
+
+        elasticache_client.replication_groups[REPLICATION_GROUP_ARN] = ReplicationGroup(
+            arn=REPLICATION_GROUP_ARN,
+            id=REPLICATION_GROUP_ID,
+            region=AWS_REGION_US_EAST_1,
+            status=REPLICATION_GROUP_STATUS,
+            snapshot_retention=3,
+            encrypted=REPLICATION_GROUP_ENCRYPTION,
+            transit_encryption=REPLICATION_GROUP_TRANSIT_ENCRYPTION,
+            multi_az=REPLICATION_GROUP_MULTI_AZ,
+        )
+
+        elasticache_client.audit_config = {"minimum_snapshot_retention_period": 1}
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_client,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_cluster_backup_enabled.elasticache_cluster_backup_enabled import (
+                elasticache_cluster_backup_enabled,
+            )
+
+            check = elasticache_cluster_backup_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Cluster {REPLICATION_GROUP_ID} has automated snapshot backups enabled with retention period 3 days."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
@@ -0,0 +1,133 @@
+from unittest import mock
+
+import botocore
+from mock import MagicMock
+from moto import mock_aws
+
+from prowler.providers.aws.services.elasticache.elasticache_service import (
+    ReplicationGroup,
+)
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+REPLICATION_GROUP_ID = "clustered-redis"
+REPLICATION_GROUP_ARN = f"arn:aws:elasticache:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:replicationgroup:{REPLICATION_GROUP_ID}"
+REPLICATION_GROUP_STATUS = "available"
+REPLICATION_GROUP_SNAPSHOT_RETENTION = "0"
+REPLICATION_GROUP_ENCRYPTION = True
+REPLICATION_GROUP_TRANSIT_ENCRYPTION = True
+REPLICATION_GROUP_MULTI_AZ = "enabled"
+
+# Patch every AWS call using Boto3
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+class Test_elasticache_replication_group_multi_az_enabled:
+    @mock_aws
+    def test_elasticache_no_replication_groups(self):
+
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_cluster_multi_az_enabled.elasticache_cluster_multi_az_enabled import (
+                elasticache_cluster_multi_az_enabled,
+            )
+
+            check = elasticache_cluster_multi_az_enabled()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_elasticache_cluster_multi_az_disabled(self):
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        elasticache_service.replication_groups[REPLICATION_GROUP_ARN] = (
+            ReplicationGroup(
+                arn=REPLICATION_GROUP_ARN,
+                id=REPLICATION_GROUP_ID,
+                region=AWS_REGION_US_EAST_1,
+                status=REPLICATION_GROUP_STATUS,
+                snapshot_retention=REPLICATION_GROUP_SNAPSHOT_RETENTION,
+                encrypted=False,
+                transit_encryption=False,
+                multi_az="disabled",
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_cluster_multi_az_enabled.elasticache_cluster_multi_az_enabled import (
+                elasticache_cluster_multi_az_enabled,
+            )
+
+            check = elasticache_cluster_multi_az_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Cluster {REPLICATION_GROUP_ID} does not have Multi-AZ enabled."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
+
+    def test_elasticache_cluster_multi_az_enabled(self):
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        elasticache_service.replication_groups[REPLICATION_GROUP_ARN] = (
+            ReplicationGroup(
+                arn=REPLICATION_GROUP_ARN,
+                id=REPLICATION_GROUP_ID,
+                region=AWS_REGION_US_EAST_1,
+                status=REPLICATION_GROUP_STATUS,
+                snapshot_retention=REPLICATION_GROUP_SNAPSHOT_RETENTION,
+                encrypted=REPLICATION_GROUP_ENCRYPTION,
+                transit_encryption=REPLICATION_GROUP_TRANSIT_ENCRYPTION,
+                multi_az=REPLICATION_GROUP_MULTI_AZ,
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_cluster_multi_az_enabled.elasticache_cluster_multi_az_enabled import (
+                elasticache_cluster_multi_az_enabled,
+            )
+
+            check = elasticache_cluster_multi_az_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Cluster {REPLICATION_GROUP_ID} has Multi-AZ enabled."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
@@ -0,0 +1,133 @@
+from unittest import mock
+
+import botocore
+from mock import MagicMock
+from moto import mock_aws
+
+from prowler.providers.aws.services.elasticache.elasticache_service import (
+    ReplicationGroup,
+)
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+REPLICATION_GROUP_ID = "clustered-redis"
+REPLICATION_GROUP_ARN = f"arn:aws:elasticache:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:replicationgroup:{REPLICATION_GROUP_ID}"
+REPLICATION_GROUP_STATUS = "available"
+REPLICATION_GROUP_SNAPSHOT_RETENTION = "0"
+REPLICATION_GROUP_ENCRYPTION = True
+REPLICATION_GROUP_TRANSIT_ENCRYPTION = True
+REPLICATION_GROUP_MULTI_AZ = True
+
+# Patch every AWS call using Boto3
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+class Test_elasticache_replication_group_at_rest_encryption_enabled:
+    @mock_aws
+    def test_elasticache_no_replication_groups(self):
+
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_replication_group_encrypted_at_rest.elasticache_replication_group_encrypted_at_rest import (
+                elasticache_replication_group_encrypted_at_rest,
+            )
+
+            check = elasticache_replication_group_encrypted_at_rest()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_elasticache_replication_groups_at_rest_encryption_disabled(self):
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        elasticache_service.replication_groups[REPLICATION_GROUP_ARN] = (
+            ReplicationGroup(
+                arn=REPLICATION_GROUP_ARN,
+                id=REPLICATION_GROUP_ID,
+                region=AWS_REGION_US_EAST_1,
+                status=REPLICATION_GROUP_STATUS,
+                snapshot_retention=REPLICATION_GROUP_SNAPSHOT_RETENTION,
+                encrypted=False,
+                transit_encryption=False,
+                multi_az=REPLICATION_GROUP_MULTI_AZ,
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_replication_group_encrypted_at_rest.elasticache_replication_group_encrypted_at_rest import (
+                elasticache_replication_group_encrypted_at_rest,
+            )
+
+            check = elasticache_replication_group_encrypted_at_rest()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Replication Group {REPLICATION_GROUP_ID} does not have at rest encryption enabled."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
+
+    def test_elasticache_replication_groups_at_rest_encryption_enabled(self):
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        elasticache_service.replication_groups[REPLICATION_GROUP_ARN] = (
+            ReplicationGroup(
+                arn=REPLICATION_GROUP_ARN,
+                id=REPLICATION_GROUP_ID,
+                region=AWS_REGION_US_EAST_1,
+                status=REPLICATION_GROUP_STATUS,
+                snapshot_retention=REPLICATION_GROUP_SNAPSHOT_RETENTION,
+                encrypted=REPLICATION_GROUP_ENCRYPTION,
+                transit_encryption=REPLICATION_GROUP_TRANSIT_ENCRYPTION,
+                multi_az=REPLICATION_GROUP_MULTI_AZ,
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_replication_group_encrypted_at_rest.elasticache_replication_group_encrypted_at_rest import (
+                elasticache_replication_group_encrypted_at_rest,
+            )
+
+            check = elasticache_replication_group_encrypted_at_rest()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Replication Group {REPLICATION_GROUP_ID} has at rest encryption enabled."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
@@ -0,0 +1,133 @@
+from unittest import mock
+
+import botocore
+from mock import MagicMock
+from moto import mock_aws
+
+from prowler.providers.aws.services.elasticache.elasticache_service import (
+    ReplicationGroup,
+)
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+REPLICATION_GROUP_ID = "clustered-redis"
+REPLICATION_GROUP_ARN = f"arn:aws:elasticache:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:replicationgroup:{REPLICATION_GROUP_ID}"
+REPLICATION_GROUP_STATUS = "available"
+REPLICATION_GROUP_SNAPSHOT_RETENTION = "0"
+REPLICATION_GROUP_ENCRYPTION = True
+REPLICATION_GROUP_TRANSIT_ENCRYPTION = True
+REPLICATION_GROUP_MULTI_AZ = "enabled"
+
+# Patch every AWS call using Boto3
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+class Test_elasticache_replication_group_in_transit_encryption_enabled:
+    @mock_aws
+    def test_elasticache_no_replication_groups(self):
+
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_replication_group_encrypted_in_transit.elasticache_replication_group_encrypted_in_transit import (
+                elasticache_replication_group_encrypted_in_transit,
+            )
+
+            check = elasticache_replication_group_encrypted_in_transit()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_elasticache_replication_groups_in_transit_encryption_disabled(self):
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        elasticache_service.replication_groups[REPLICATION_GROUP_ARN] = (
+            ReplicationGroup(
+                arn=REPLICATION_GROUP_ARN,
+                id=REPLICATION_GROUP_ID,
+                region=AWS_REGION_US_EAST_1,
+                status=REPLICATION_GROUP_STATUS,
+                snapshot_retention=REPLICATION_GROUP_SNAPSHOT_RETENTION,
+                encrypted=REPLICATION_GROUP_ENCRYPTION,
+                transit_encryption=False,
+                multi_az=REPLICATION_GROUP_MULTI_AZ,
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_replication_group_encrypted_in_transit.elasticache_replication_group_encrypted_in_transit import (
+                elasticache_replication_group_encrypted_in_transit,
+            )
+
+            check = elasticache_replication_group_encrypted_in_transit()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Replication Group {REPLICATION_GROUP_ID} does not have in transit encryption enabled."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
+
+    def test_elasticache_replication_groups_in_transit_encryption_enabled(self):
+        # Mock ElastiCache Service
+        elasticache_service = MagicMock
+        elasticache_service.replication_groups = {}
+
+        elasticache_service.replication_groups[REPLICATION_GROUP_ARN] = (
+            ReplicationGroup(
+                arn=REPLICATION_GROUP_ARN,
+                id=REPLICATION_GROUP_ID,
+                region=AWS_REGION_US_EAST_1,
+                status=REPLICATION_GROUP_STATUS,
+                snapshot_retention=REPLICATION_GROUP_SNAPSHOT_RETENTION,
+                encrypted=REPLICATION_GROUP_ENCRYPTION,
+                transit_encryption=REPLICATION_GROUP_TRANSIT_ENCRYPTION,
+                multi_az=REPLICATION_GROUP_MULTI_AZ,
+            )
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_aws_provider([AWS_REGION_US_EAST_1]),
+        ), mock.patch(
+            "prowler.providers.aws.services.elasticache.elasticache_service.ElastiCache",
+            new=elasticache_service,
+        ):
+            from prowler.providers.aws.services.elasticache.elasticache_replication_group_encrypted_in_transit.elasticache_replication_group_encrypted_in_transit import (
+                elasticache_replication_group_encrypted_in_transit,
+            )
+
+            check = elasticache_replication_group_encrypted_in_transit()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Elasticache Replication Group {REPLICATION_GROUP_ID} has in transit encryption enabled."
+            )
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_id == REPLICATION_GROUP_ID
+            assert result[0].resource_arn == REPLICATION_GROUP_ARN
@@ -4,6 +4,7 @@ from mock import patch
 from prowler.providers.aws.services.elasticache.elasticache_service import (
    Cluster,
    ElastiCache,
+    ReplicationGroup,
 )
 from tests.providers.aws.utils import (
    AWS_ACCOUNT_NUMBER,
@@ -25,6 +26,14 @@ ELASTICACHE_CLUSTER_TAGS = [
    {"Key": "environment", "Value": "test"},
 ]

+REPLICATION_GROUP_ID = "clustered-redis"
+REPLICATION_GROUP_ARN = f"arn:aws:elasticache:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:replicationgroup:{REPLICATION_GROUP_ID}"
+REPLICATION_GROUP_STATUS = "available"
+REPLICATION_GROUP_SNAPSHOT_RETENTION = "0"
+REPLICATION_GROUP_ENCRYPTION = True
+REPLICATION_GROUP_TRANSIT_ENCRYPTION = True
+REPLICATION_GROUP_MULTI_AZ = "enabled"
+
 # Mocking Access Analyzer Calls
 make_api_call = botocore.client.BaseClient._make_api_call

@@ -44,6 +53,7 @@ def mock_make_api_call(self, operation_name, kwargs):
                    "CacheClusterId": ELASTICACHE_CLUSTER_NAME,
                    "CacheSubnetGroupName": SUBNET_GROUP_NAME,
                    "ARN": ELASTICACHE_CLUSTER_ARN,
+                    "SecurityGroups": [],
                },
            ]
        }
@@ -77,7 +87,20 @@ def mock_make_api_call(self, operation_name, kwargs):
        }
    if operation_name == "ListTagsForResource":
        return {"TagList": ELASTICACHE_CLUSTER_TAGS}
-
+    if operation_name == "DescribeReplicationGroups":
+        return {
+            "ReplicationGroups": [
+                {
+                    "ReplicationGroupId": REPLICATION_GROUP_ID,
+                    "Status": REPLICATION_GROUP_STATUS,
+                    "SnapshotRetentionLimit": REPLICATION_GROUP_SNAPSHOT_RETENTION,
+                    "MultiAZ": REPLICATION_GROUP_MULTI_AZ,
+                    "TransitEncryptionEnabled": REPLICATION_GROUP_TRANSIT_ENCRYPTION,
+                    "AtRestEncryptionEnabled": REPLICATION_GROUP_ENCRYPTION,
+                    "ARN": REPLICATION_GROUP_ARN,
+                },
+            ]
+        }
    return make_api_call(self, operation_name, kwargs)


@@ -132,7 +155,28 @@ class Test_ElastiCache_Service:
            name=ELASTICACHE_CLUSTER_NAME,
            id=ELASTICACHE_CLUSTER_NAME,
            region=AWS_REGION_US_EAST_1,
+            security_groups=[],
            cache_subnet_group_id=SUBNET_GROUP_NAME,
            subnets=[SUBNET_1, SUBNET_2],
            tags=ELASTICACHE_CLUSTER_TAGS,
        )
+
+    # Test ElastiCache Replication Groups
+    def test_describe_replication_groups(self):
+        aws_provider = set_mocked_aws_provider()
+        elasticache = ElastiCache(aws_provider)
+
+        assert len(elasticache.replication_groups) == 1
+        assert elasticache.replication_groups[REPLICATION_GROUP_ARN]
+        assert elasticache.replication_groups[
+            REPLICATION_GROUP_ARN
+        ] == ReplicationGroup(
+            id=REPLICATION_GROUP_ID,
+            arn=REPLICATION_GROUP_ARN,
+            region=AWS_REGION_US_EAST_1,
+            status=REPLICATION_GROUP_STATUS,
+            snapshot_retention=REPLICATION_GROUP_SNAPSHOT_RETENTION,
+            encrypted=REPLICATION_GROUP_ENCRYPTION,
+            transit_encryption=REPLICATION_GROUP_TRANSIT_ENCRYPTION,
+            multi_az=REPLICATION_GROUP_MULTI_AZ,
+        )