feat(neptune): Additional Neptune checks (#4243)

tests/providers/aws/services/neptune/neptune_cluster_backup_enabled/neptune_cluster_backup_enabled_test.py

@@ -0,0 +1,282 @@
+from unittest import mock
+
+import botocore
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+minimum_backup_retention_period = 2
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "DescribeDBEngineVersions":
+        return {
+            "DBEngineVersions": [
+                {
+                    "Engine": "mysql",
+                    "EngineVersion": "8.0.32",
+                    "DBEngineDescription": "description",
+                    "DBEngineVersionDescription": "description",
+                },
+            ]
+        }
+
+    return make_api_call(self, operation_name, kwarg)
+
+
+@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+class Test_neptune_cluster_backup_enabled:
+    @mock_aws
+    def test_neptune_no_instances(self):
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled import (
+                    neptune_cluster_backup_enabled,
+                )
+
+                check = neptune_cluster_backup_enabled()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_neptune_cluster_without_backup(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=False,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled import (
+                    neptune_cluster_backup_enabled,
+                )
+
+                check = neptune_cluster_backup_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 does not have backup enabled."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_neptune_cluster_with_backup_less_than_recommended(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=True,
+            BackupRetentionPeriod=4,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        db_cluster = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled.neptune_client",
+                new=Neptune(aws_provider),
+            ) as service_client:
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled import (
+                    neptune_cluster_backup_enabled,
+                )
+
+                service_client.clusters[db_cluster].iam_auth = True
+                check = neptune_cluster_backup_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 has backup enabled with retention period 4 days. Recommended to increase the backup retention period to a minimum of 7 days."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_neptune_cluster_with_backup(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=True,
+            BackupRetentionPeriod=9,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled import (
+                    neptune_cluster_backup_enabled,
+                )
+
+                check = neptune_cluster_backup_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 has backup enabled with retention period 9 days."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_neptune_cluster_with_backup_modified_retention(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=True,
+            BackupRetentionPeriod=2,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled.neptune_client",
+                new=Neptune(aws_provider),
+            ) as service_client:
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_backup_enabled.neptune_cluster_backup_enabled import (
+                    neptune_cluster_backup_enabled,
+                )
+
+                service_client.audit_config = {"minimum_backup_retention_period": 1}
+
+                check = neptune_cluster_backup_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 has backup enabled with retention period 2 days."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []

tests/providers/aws/services/neptune/neptune_cluster_deletion_protection/neptune_cluster_deletion_protection_test.py

@@ -0,0 +1,168 @@
+from unittest import mock
+
+import botocore
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "DescribeDBEngineVersions":
+        return {
+            "DBEngineVersions": [
+                {
+                    "Engine": "mysql",
+                    "EngineVersion": "8.0.32",
+                    "DBEngineDescription": "description",
+                    "DBEngineVersionDescription": "description",
+                },
+            ]
+        }
+
+    return make_api_call(self, operation_name, kwarg)
+
+
+@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+class Test_neptune_cluster_deletion_protection:
+    @mock_aws
+    def test_neptune_no_instances(self):
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_deletion_protection.neptune_cluster_deletion_protection.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_deletion_protection.neptune_cluster_deletion_protection import (
+                    neptune_cluster_deletion_protection,
+                )
+
+                check = neptune_cluster_deletion_protection()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_neptune_cluster_without_deletion_protection(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=False,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=False,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=False,
+            Tags=[],
+        )
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_deletion_protection.neptune_cluster_deletion_protection.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_deletion_protection.neptune_cluster_deletion_protection import (
+                    neptune_cluster_deletion_protection,
+                )
+
+                check = neptune_cluster_deletion_protection()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 does not have deletion protection enabled."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_neptune_cluster_with_deletion_protection(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=True,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        db_cluster = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_deletion_protection.neptune_cluster_deletion_protection.neptune_client",
+                new=Neptune(aws_provider),
+            ) as service_client:
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_deletion_protection.neptune_cluster_deletion_protection import (
+                    neptune_cluster_deletion_protection,
+                )
+
+                service_client.clusters[db_cluster].deletion_protection = True
+                check = neptune_cluster_deletion_protection()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 has deletion protection enabled."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []

tests/providers/aws/services/neptune/neptune_cluster_iam_authentication_enabled/neptune_cluster_iam_authentication_enabled_test.py

@@ -0,0 +1,168 @@
+from unittest import mock
+
+import botocore
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "DescribeDBEngineVersions":
+        return {
+            "DBEngineVersions": [
+                {
+                    "Engine": "mysql",
+                    "EngineVersion": "8.0.32",
+                    "DBEngineDescription": "description",
+                    "DBEngineVersionDescription": "description",
+                },
+            ]
+        }
+
+    return make_api_call(self, operation_name, kwarg)
+
+
+@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+class Test_neptune_cluster_iam_authentication_enabled:
+    @mock_aws
+    def test_neptune_no_instances(self):
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_iam_authentication_enabled.neptune_cluster_iam_authentication_enabled.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_iam_authentication_enabled.neptune_cluster_iam_authentication_enabled import (
+                    neptune_cluster_iam_authentication_enabled,
+                )
+
+                check = neptune_cluster_iam_authentication_enabled()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_neptune_cluster_without_iam_auth(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=False,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_iam_authentication_enabled.neptune_cluster_iam_authentication_enabled.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_iam_authentication_enabled.neptune_cluster_iam_authentication_enabled import (
+                    neptune_cluster_iam_authentication_enabled,
+                )
+
+                check = neptune_cluster_iam_authentication_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 does not have IAM authentication enabled."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_neptune_cluster_with_iam_auth(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=True,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        db_cluster = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_iam_authentication_enabled.neptune_cluster_iam_authentication_enabled.neptune_client",
+                new=Neptune(aws_provider),
+            ) as service_client:
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_iam_authentication_enabled.neptune_cluster_iam_authentication_enabled import (
+                    neptune_cluster_iam_authentication_enabled,
+                )
+
+                service_client.clusters[db_cluster].iam_auth = True
+                check = neptune_cluster_iam_authentication_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 has IAM authentication enabled."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []

tests/providers/aws/services/neptune/neptune_cluster_multi_az/neptune_cluster_multi_az_test.py

@@ -0,0 +1,168 @@
+from unittest import mock
+
+import botocore
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "DescribeDBEngineVersions":
+        return {
+            "DBEngineVersions": [
+                {
+                    "Engine": "mysql",
+                    "EngineVersion": "8.0.32",
+                    "DBEngineDescription": "description",
+                    "DBEngineVersionDescription": "description",
+                },
+            ]
+        }
+
+    return make_api_call(self, operation_name, kwarg)
+
+
+@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+class Test_neptune_cluster_multi_az:
+    @mock_aws
+    def test_neptune_no_instances(self):
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_multi_az.neptune_cluster_multi_az.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_multi_az.neptune_cluster_multi_az import (
+                    neptune_cluster_multi_az,
+                )
+
+                check = neptune_cluster_multi_az()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_neptune_cluster_without_multi_az(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=False,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_multi_az.neptune_cluster_multi_az.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_multi_az.neptune_cluster_multi_az import (
+                    neptune_cluster_multi_az,
+                )
+
+                check = neptune_cluster_multi_az()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 does not have Multi-AZ enabled."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_neptune_cluster_with_multi_az(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=True,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        db_cluster = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_multi_az.neptune_cluster_multi_az.neptune_client",
+                new=Neptune(aws_provider),
+            ) as service_client:
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_multi_az.neptune_cluster_multi_az import (
+                    neptune_cluster_multi_az,
+                )
+
+                service_client.clusters[db_cluster].multi_az = True
+                check = neptune_cluster_multi_az()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 has Multi-AZ enabled."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []

tests/providers/aws/services/neptune/neptune_cluster_storage_encrypted/neptune_cluster_storage_encrypted_test.py

@@ -0,0 +1,166 @@
+from unittest import mock
+
+import botocore
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "DescribeDBEngineVersions":
+        return {
+            "DBEngineVersions": [
+                {
+                    "Engine": "mysql",
+                    "EngineVersion": "8.0.32",
+                    "DBEngineDescription": "description",
+                    "DBEngineVersionDescription": "description",
+                },
+            ]
+        }
+
+    return make_api_call(self, operation_name, kwarg)
+
+
+@mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+class Test_neptune_cluster_storage_encrypted:
+    @mock_aws
+    def test_neptune_no_instances(self):
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_storage_encrypted.neptune_cluster_storage_encrypted.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_storage_encrypted.neptune_cluster_storage_encrypted import (
+                    neptune_cluster_storage_encrypted,
+                )
+
+                check = neptune_cluster_storage_encrypted()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_neptune_cluster_without_storage_encrypted(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=False,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=False,
+            Tags=[],
+        )
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_storage_encrypted.neptune_cluster_storage_encrypted.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_storage_encrypted.neptune_cluster_storage_encrypted import (
+                    neptune_cluster_storage_encrypted,
+                )
+
+                check = neptune_cluster_storage_encrypted()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 is not encrypted at rest."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []
+
+    @mock_aws
+    def test_neptune_cluster_with_storage_encrypted(self):
+        conn = client("neptune", region_name=AWS_REGION_US_EAST_1)
+        conn.create_db_parameter_group(
+            DBParameterGroupName="test",
+            DBParameterGroupFamily="default.neptune",
+            Description="test parameter group",
+        )
+        conn.create_db_cluster(
+            DBClusterIdentifier="db-cluster-1",
+            Engine="neptune",
+            DatabaseName="test-1",
+            DeletionProtection=True,
+            DBClusterParameterGroupName="test",
+            MasterUsername="test",
+            MasterUserPassword="password",
+            EnableIAMDatabaseAuthentication=True,
+            BackupRetentionPeriod=0,
+            StorageEncrypted=True,
+            Tags=[],
+        )
+        from prowler.providers.aws.services.neptune.neptune_service import Neptune
+
+        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.neptune.neptune_cluster_storage_encrypted.neptune_cluster_storage_encrypted.neptune_client",
+                new=Neptune(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.neptune.neptune_cluster_storage_encrypted.neptune_cluster_storage_encrypted import (
+                    neptune_cluster_storage_encrypted,
+                )
+
+                check = neptune_cluster_storage_encrypted()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == "Neptune Cluster db-cluster-1 is encrypted at rest."
+                )
+                assert result[0].resource_id == "db-cluster-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}:cluster:db-cluster-1"
+                )
+                assert result[0].resource_tags == []

tests/providers/aws/services/neptune/neptune_service_test.py

@@ -127,8 +127,8 @@ class Test_Neptune_Service:
             DBClusterIdentifier=NEPTUNE_CLUSTER_NAME,
             Port=123,
             Tags=NEPTUNE_CLUSTER_TAGS,
-            StorageEncrypted=False,
-            DeletionProtection=True | False,
+            EnableIAMDatabaseAuthentication=False,
+            DeletionProtection=False,
         )["DBCluster"]
 
         cluster_arn = cluster["DBClusterArn"]
@@ -143,6 +143,12 @@ class Test_Neptune_Service:
             arn=cluster_arn,
             name=NEPTUNE_CLUSTER_NAME,
             id=cluster_id,
+            backup_retention_period=1,
+            encrypted=True,
+            kms_key="default_kms_key_id",
+            multi_az=False,
+            iam_auth=False,
+            deletion_protection=False,
             region=AWS_REGION_US_EAST_1,
             db_subnet_group_id=SUBNET_GROUP_NAME,
             subnets=[SUBNET_1, SUBNET_2],