fix(aws): remove rds certificate expiration false positives

prowler/changelog.d/rds-certificate-expiration-neptune-fp.fixed.md

@@ -0,0 +1 @@
+`rds_instance_certificate_expiration` no longer reports false positives on Neptune resources, which are now excluded from the RDS inventory, or on instances whose CA certificate could not be retrieved, which are reported as MANUAL instead of a critical failure

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

@@ -16,6 +16,17 @@ class rds_instance_certificate_expiration(Check):
         findings = []
         for db_instance in rds_client.db_instances.values():
             report = Check_Report_AWS(metadata=self.metadata(), resource=db_instance)
+
+            # An empty certificate list means the CA certificate could not be
+            # retrieved, not that the certificate is expired. Reporting it as a
+            # failure would be a false positive, so it is flagged for manual review.
+            if not db_instance.cert:
+                report.status = "MANUAL"
+                report.check_metadata.Severity = Severity.informational
+                report.status_extended = f"RDS Instance {db_instance.id} certificate expiration could not be determined because no CA certificate information was retrieved."
+                findings.append(report)
+                continue
+
             report.status = "FAIL"
             report.check_metadata.Severity = Severity.critical
             report.status_extended = (

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

@@ -8,6 +8,12 @@ from prowler.lib.logger import logger
 from prowler.lib.scan_filters.scan_filters import is_resource_filtered
 from prowler.providers.aws.lib.service.service import AWSService
 
+# Engines that are returned by the RDS API but belong to a different AWS service.
+# DocumentDB and Neptune share the RDS control plane, so their resources show up in
+# describe_db_* calls even though RDS checks do not apply to them. Each one has its
+# own dedicated Prowler service (docdb, neptune) that audits them properly.
+NON_RDS_ENGINES = ["docdb", "neptune"]
+
 
 class RDS(AWSService):
     def __init__(self, provider):
@@ -52,7 +58,7 @@ class RDS(AWSService):
                     if not self.audit_resources or (
                         is_resource_filtered(arn, self.audit_resources)
                     ):
-                        if instance["Engine"] != "docdb":
+                        if instance["Engine"] not in NON_RDS_ENGINES:
                             self.db_instances[arn] = DBInstance(
                                 id=instance["DBInstanceIdentifier"],
                                 arn=arn,
@@ -199,7 +205,7 @@ class RDS(AWSService):
                     if not self.audit_resources or (
                         is_resource_filtered(arn, self.audit_resources)
                     ):
-                        if snapshot["Engine"] != "docdb":
+                        if snapshot["Engine"] not in NON_RDS_ENGINES:
                             self.db_snapshots.append(
                                 DBSnapshot(
                                     id=snapshot["DBSnapshotIdentifier"],
@@ -253,7 +259,7 @@ class RDS(AWSService):
                                     db_cluster_arn, self.audit_resources
                                 )
                             ):
-                                if cluster["Engine"] != "docdb":
+                                if cluster["Engine"] not in NON_RDS_ENGINES:
                                     db_cluster = DBCluster(
                                         id=cluster["DBClusterIdentifier"],
                                         arn=db_cluster_arn,
@@ -376,7 +382,7 @@ class RDS(AWSService):
                             self.audit_resources,
                         )
                     ):
-                        if snapshot["Engine"] != "docdb":
+                        if snapshot["Engine"] not in NON_RDS_ENGINES:
                             self.db_cluster_snapshots.append(
                                 ClusterSnapshot(
                                     id=snapshot["DBClusterSnapshotIdentifier"],

tests/providers/aws/services/rds/rds_instance_certificate_expiration/rds_instance_certificate_expiration_test.py

@@ -805,3 +805,116 @@ class Test_rds_instance_certificate_expiration:
                 == f"arn:aws:rds:{AWS_REGION}:{AWS_ACCOUNT_NUMBER_CON}:db:db-master-1"
             )
             assert result[0].resource_tags == []
+
+    def test_rds_instance_without_ca_certificate_identifier(self):
+        rds_client = mock.MagicMock
+        instance_arn = (
+            f"arn:aws:rds:{AWS_REGION}:{AWS_ACCOUNT_NUMBER_CON}:db:db-master-1"
+        )
+        rds_client.db_instances = {
+            instance_arn: DBInstance(
+                id="db-master-1",
+                arn=instance_arn,
+                engine="postgres",
+                engine_version="aurora14",
+                status="available",
+                public=False,
+                encrypted=True,
+                deletion_protection=True,
+                auto_minor_version_upgrade=False,
+                multi_az=True,
+                username="test",
+                iam_auth=False,
+                region=AWS_REGION,
+                ca_cert=None,
+                endpoint={},
+                cert=[],
+            )
+        }
+
+        with (
+            mock.patch(
+                "prowler.providers.aws.services.rds.rds_service.RDS",
+                new=rds_client,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.rds.rds_client.rds_client",
+                new=rds_client,
+            ),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.rds.rds_instance_certificate_expiration.rds_instance_certificate_expiration import (
+                rds_instance_certificate_expiration,
+            )
+
+            check = rds_instance_certificate_expiration()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "MANUAL"
+            assert result[0].check_metadata.Severity == "informational"
+            assert (
+                result[0].status_extended
+                == "RDS Instance db-master-1 certificate expiration could not be determined because no CA certificate information was retrieved."
+            )
+            assert result[0].resource_id == "db-master-1"
+            assert result[0].region == AWS_REGION
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:rds:{AWS_REGION}:{AWS_ACCOUNT_NUMBER_CON}:db:db-master-1"
+            )
+            assert result[0].resource_tags == []
+
+    def test_rds_instance_with_unretrievable_certificate(self):
+        rds_client = mock.MagicMock
+        instance_arn = (
+            f"arn:aws:rds:{AWS_REGION}:{AWS_ACCOUNT_NUMBER_CON}:db:db-master-1"
+        )
+        rds_client.db_instances = {
+            instance_arn: DBInstance(
+                id="db-master-1",
+                arn=instance_arn,
+                engine="postgres",
+                engine_version="aurora14",
+                status="available",
+                public=False,
+                encrypted=True,
+                deletion_protection=True,
+                auto_minor_version_upgrade=False,
+                multi_az=True,
+                username="test",
+                iam_auth=False,
+                region=AWS_REGION,
+                # The instance reports a CA identifier but DescribeCertificates raised
+                # CertificateNotFound, so no certificate could be collected.
+                ca_cert="rds-ca-rsa2048-g1",
+                endpoint={},
+                cert=[],
+            )
+        }
+
+        with (
+            mock.patch(
+                "prowler.providers.aws.services.rds.rds_service.RDS",
+                new=rds_client,
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.rds.rds_client.rds_client",
+                new=rds_client,
+            ),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.rds.rds_instance_certificate_expiration.rds_instance_certificate_expiration import (
+                rds_instance_certificate_expiration,
+            )
+
+            check = rds_instance_certificate_expiration()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "MANUAL"
+            assert result[0].check_metadata.Severity == "informational"
+            assert (
+                result[0].status_extended
+                == "RDS Instance db-master-1 certificate expiration could not be determined because no CA certificate information was retrieved."
+            )

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

@@ -146,6 +146,55 @@ class Test_RDS_Service:
         assert db_instance.copy_tags_to_snapshot
         assert db_instance.port == 5432
 
+    # Neptune shares the RDS control plane, so its resources are returned by the RDS
+    # API. They must be skipped because they are audited by the neptune service.
+    @mock_aws
+    def test_neptune_resources_are_excluded(self):
+        conn = client("rds", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_cluster(
+            DBClusterIdentifier="neptune-cluster",
+            Engine="neptune",
+            DBClusterInstanceClass="db.r5.large",
+            MasterUsername="root",
+            MasterUserPassword="hunter2000",
+        )
+        conn.create_db_instance(
+            DBInstanceIdentifier="neptune-instance-1",
+            DBInstanceClass="db.r5.large",
+            Engine="neptune",
+            DBClusterIdentifier="neptune-cluster",
+        )
+        conn.create_db_cluster_snapshot(
+            DBClusterIdentifier="neptune-cluster",
+            DBClusterSnapshotIdentifier="neptune-snapshot-1",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="postgres-cluster",
+            Engine="postgres",
+            DBClusterInstanceClass="db.m1.small",
+            MasterUsername="root",
+            MasterUserPassword="hunter2000",
+        )
+        conn.create_db_instance(
+            DBInstanceIdentifier="postgres-instance-1",
+            DBInstanceClass="db.m1.small",
+            Engine="postgres",
+            AllocatedStorage=10,
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        rds = RDS(aws_provider)
+
+        assert [instance.id for instance in rds.db_instances.values()] == [
+            "postgres-instance-1"
+        ]
+        assert [cluster.id for cluster in rds.db_clusters.values()] == [
+            "postgres-cluster"
+        ]
+        assert all(
+            "neptune" not in snapshot.id for snapshot in rds.db_cluster_snapshots
+        )
+
     @mock_aws
     def test_describe_db_parameters(self):
         conn = client("rds", region_name=AWS_REGION_US_EAST_1)