diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/__init__.py b/prowler/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs.metadata.json b/prowler/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs.metadata.json new file mode 100644 index 0000000000..12f90b666e --- /dev/null +++ b/prowler/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs.metadata.json @@ -0,0 +1,32 @@ +{ + "Provider": "aws", + "CheckID": "neptune_cluster_integration_cloudwatch_logs", + "CheckTitle": "Check if Neptune Clusters have audit cloudwatch logs enabled.", + "CheckType": [ + "Software and Configuration Checks, AWS Security Best Practices" + ], + "ServiceName": "neptune", + "SubServiceName": "", + "ResourceIdTemplate": "arn:aws:rds:region:account-id:db-cluster", + "Severity": "medium", + "ResourceType": "AwsRdsDbCluster", + "Description": "Check if Neptune Clusters have audit cloudwatch logs enabled.", + "Risk": "If audit logs are not enabled, it is difficult to determine the root cause of security incidents.", + "RelatedUrl": "https://docs.aws.amazon.com/neptune/latest/userguide/auditing.html", + "Remediation": { + "Code": { + "CLI": "aws neptune modify-db-cluster --db-cluster-identifier --cloudwatch-logs-export-configuration '{\"EnableLogTypes\":[\"audit\"]}'", + "NativeIaC": "", + "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/neptune-controls.html#neptune-2", + "Terraform": "" + }, + "Recommendation": { + "Text": "Enable audit logs for Neptune Clusters.", + "Url": "https://docs.aws.amazon.com/neptune/latest/userguide/cloudwatch-logs.html" + } + }, + "Categories": [], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs.py b/prowler/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs.py new file mode 100644 index 0000000000..90d6fbbb33 --- /dev/null +++ b/prowler/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs.py @@ -0,0 +1,24 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.neptune.neptune_client import neptune_client + + +class neptune_cluster_integration_cloudwatch_logs(Check): + def execute(self): + findings = [] + for cluster in neptune_client.clusters.values(): + report = Check_Report_AWS(self.metadata()) + report.region = cluster.region + report.resource_id = cluster.name + report.resource_arn = cluster.arn + report.resource_tags = cluster.tags + report.status = "FAIL" + report.status_extended = f"Neptune Cluster {cluster.name} does not have cloudwatch audit logs enabled." + if "audit" in cluster.cloudwatch_logs: + report.status = "PASS" + report.status_extended = ( + f"Neptune Cluster {cluster.name} has cloudwatch audit logs enabled." + ) + + findings.append(report) + + return findings diff --git a/prowler/providers/aws/services/neptune/neptune_service.py b/prowler/providers/aws/services/neptune/neptune_service.py index d6d706961b..106fd9caa2 100644 --- a/prowler/providers/aws/services/neptune/neptune_service.py +++ b/prowler/providers/aws/services/neptune/neptune_service.py @@ -46,6 +46,7 @@ class Neptune(AWSService): backup_retention_period=cluster.get("BackupRetentionPeriod", 0), encrypted=cluster.get("StorageEncrypted", False), kms_key=cluster.get("KmsKeyId", ""), + cloudwatch_logs=cluster.get("EnabledCloudwatchLogsExports", []), multi_az=cluster["MultiAZ"], iam_auth=cluster.get("IAMDatabaseAuthenticationEnabled", False), deletion_protection=cluster.get("DeletionProtection", False), @@ -172,6 +173,7 @@ class Cluster(BaseModel): db_subnet_group_id: str subnets: Optional[list] tags: Optional[list] + cloudwatch_logs: Optional[list] class ClusterSnapshot(BaseModel): diff --git a/tests/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs_test.py b/tests/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs_test.py new file mode 100644 index 0000000000..1d3dc746f5 --- /dev/null +++ b/tests/providers/aws/services/neptune/neptune_cluster_integration_cloudwatch_logs/neptune_cluster_integration_cloudwatch_logs_test.py @@ -0,0 +1,205 @@ +from unittest import mock + +from boto3 import client +from moto import mock_aws + +from prowler.providers.aws.services.neptune.neptune_service import Cluster +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + + +class Test_neptune_cluster_integration_cloudwatch_logs: + @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_integration_cloudwatch_logs.neptune_cluster_integration_cloudwatch_logs.neptune_client", + new=Neptune(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.neptune.neptune_cluster_integration_cloudwatch_logs.neptune_cluster_integration_cloudwatch_logs import ( + neptune_cluster_integration_cloudwatch_logs, + ) + + check = neptune_cluster_integration_cloudwatch_logs() + result = check.execute() + + assert len(result) == 0 + + @mock_aws + def test_neptune_cluster_without_integration_cloudwatch_logs(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=[], + EnableCloudwatchLogsExports=[], + ) + 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_integration_cloudwatch_logs.neptune_cluster_integration_cloudwatch_logs.neptune_client", + new=Neptune(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.neptune.neptune_cluster_integration_cloudwatch_logs.neptune_cluster_integration_cloudwatch_logs import ( + neptune_cluster_integration_cloudwatch_logs, + ) + + check = neptune_cluster_integration_cloudwatch_logs() + 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 cloudwatch audit logs 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_integration_cloudwatch_logs_not_audit(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", + BackupRetentionPeriod=0, + StorageEncrypted=True, + Tags=[], + EnableCloudwatchLogsExports=["slowquery"], + ) + 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_integration_cloudwatch_logs.neptune_cluster_integration_cloudwatch_logs.neptune_client", + new=Neptune(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.neptune.neptune_cluster_integration_cloudwatch_logs.neptune_cluster_integration_cloudwatch_logs import ( + neptune_cluster_integration_cloudwatch_logs, + ) + + check = neptune_cluster_integration_cloudwatch_logs() + 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 cloudwatch audit logs 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 == [] + + def test_neptune_cluster_with_integration_cloudwatch_logs_audit(self): + neptune_client = mock.MagicMock + cluster_arn = f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1" + neptune_client.clusters = { + cluster_arn: Cluster( + arn=cluster_arn, + name="db-cluster-1", + id="db-cluster-1", + backup_retention_period=7, + encrypted=True, + kms_key="clave-kms", + multi_az=False, + iam_auth=True, + deletion_protection=False, + region="us-east-1", + db_subnet_group_id="subnet-grupo-id", + subnets=[ + { + "SubnetIdentifier": "subnet-123", + "SubnetAvailabilityZone": {"Name": "us-east-1a"}, + "SubnetStatus": "Active", + } + ], + tags=[], + cloudwatch_logs=["audit"], + ) + } + + with mock.patch( + "prowler.providers.aws.services.neptune.neptune_service.Neptune", + new=neptune_client, + ), mock.patch( + "prowler.providers.aws.services.neptune.neptune_cluster_integration_cloudwatch_logs.neptune_cluster_integration_cloudwatch_logs.neptune_client", + new=neptune_client, + ): + # Test Check + from prowler.providers.aws.services.neptune.neptune_cluster_integration_cloudwatch_logs.neptune_cluster_integration_cloudwatch_logs import ( + neptune_cluster_integration_cloudwatch_logs, + ) + + check = neptune_cluster_integration_cloudwatch_logs() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == "Neptune Cluster db-cluster-1 has cloudwatch audit logs 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 == [] diff --git a/tests/providers/aws/services/neptune/neptune_service_test.py b/tests/providers/aws/services/neptune/neptune_service_test.py index c804c6853e..802b261569 100644 --- a/tests/providers/aws/services/neptune/neptune_service_test.py +++ b/tests/providers/aws/services/neptune/neptune_service_test.py @@ -183,6 +183,7 @@ class Test_Neptune_Service: db_subnet_group_id=SUBNET_GROUP_NAME, subnets=[SUBNET_1, SUBNET_2], tags=NEPTUNE_CLUSTER_TAGS, + cloudwatch_logs=[], ) def test_describe_db_cluster_snapshots(self):