From d0ef75d8d96bcd8e4d57d17f3e0ad232eb9aec9e Mon Sep 17 00:00:00 2001 From: Daniel Barranquero <74871504+danibarranqueroo@users.noreply.github.com> Date: Thu, 7 Nov 2024 16:19:44 +0100 Subject: [PATCH] feat(dms): add new check `dms_replication_task_target_logging_enabled` (#5631) Co-authored-by: Sergio --- ...replication_task_source_logging_enabled.py | 2 +- .../__init__.py | 0 ..._task_target_logging_enabled.metadata.json | 32 ++ ...replication_task_target_logging_enabled.py | 79 +++ .../providers/aws/services/dms/dms_service.py | 1 + ...cation_task_source_logging_enabled_test.py | 2 +- ...cation_task_target_logging_enabled_test.py | 475 ++++++++++++++++++ 7 files changed, 589 insertions(+), 2 deletions(-) create mode 100644 prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/__init__.py create mode 100644 prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled.metadata.json create mode 100644 prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled.py create mode 100644 tests/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled_test.py diff --git a/prowler/providers/aws/services/dms/dms_replication_task_source_logging_enabled/dms_replication_task_source_logging_enabled.py b/prowler/providers/aws/services/dms/dms_replication_task_source_logging_enabled/dms_replication_task_source_logging_enabled.py index fce11658a6..d2e2a1c316 100644 --- a/prowler/providers/aws/services/dms/dms_replication_task_source_logging_enabled/dms_replication_task_source_logging_enabled.py +++ b/prowler/providers/aws/services/dms/dms_replication_task_source_logging_enabled/dms_replication_task_source_logging_enabled.py @@ -72,7 +72,7 @@ class dms_replication_task_source_logging_enabled(Check): report.status_extended = f"DMS Replication Task {replication_task.id} has logging enabled with the minimum severity level in source events." else: report.status = "FAIL" - report.status_extended = f"DMS Replication Task {replication_task.id} does not meet the minimum severity level of logging in {', '.join(missing_components)} events." + report.status_extended = f"DMS Replication Task {replication_task.id} does not meet the minimum severity level of logging in {' and '.join(missing_components)} events." findings.append(report) diff --git a/prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/__init__.py b/prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled.metadata.json b/prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled.metadata.json new file mode 100644 index 0000000000..852b690e14 --- /dev/null +++ b/prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled.metadata.json @@ -0,0 +1,32 @@ +{ + "Provider": "aws", + "CheckID": "dms_replication_task_target_logging_enabled", + "CheckTitle": "Check if DMS replication tasks for the target database have logging enabled.", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices" + ], + "ServiceName": "dms", + "SubServiceName": "", + "ResourceIdTemplate": "arn:aws:dms:region:account-id:task/task-id", + "Severity": "medium", + "ResourceType": "AwsDmsReplicationTask", + "Description": "This control checks whether logging is enabled with the minimum severity level of LOGGER_SEVERITY_DEFAULT for DMS replication tasks TARGET_APPLY and TARGET_LOAD. The control fails if logging isn't enabled for these tasks or if the minimum severity level is less than LOGGER_SEVERITY_DEFAULT.", + "Risk": "Without logging enabled, issues in data migration may go undetected, affecting the integrity and compliance of replicated data.", + "RelatedUrl": "https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Monitoring.html#CHAP_Monitoring.ManagingLogs", + "Remediation": { + "Code": { + "CLI": "aws dms modify-replication-task --replication-task-arn --task-settings '{\"Logging\":{\"EnableLogging\":true,\"LogComponents\":[{\"Id\":\"TARGET_APPLY\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TARGET_LOAD\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"}]}}'", + "NativeIaC": "", + "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/dms-controls.html#dms-7", + "Terraform": "" + }, + "Recommendation": { + "Text": "Enable logging for target database DMS replication tasks with a minimum severity level of LOGGER_SEVERITY_DEFAULT.", + "Url": "https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Tasks.CustomizingTasks.TaskSettings.Logging.html" + } + }, + "Categories": [], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled.py b/prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled.py new file mode 100644 index 0000000000..464657c81c --- /dev/null +++ b/prowler/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled.py @@ -0,0 +1,79 @@ +from typing import List + +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.dms.dms_client import dms_client + + +class dms_replication_task_target_logging_enabled(Check): + """ + Check if AWS DMS replication tasks have logging enabled with the required + logging components and severity levels. + + This class verifies that each DMS replication task has logging enabled + and that the components TARGET_APPLY and TARGET_LOAD are configured with + at least LOGGER_SEVERITY_DEFAULT severity level. If either component is missing + or does not meet the minimum severity requirement, the check will fail. + """ + + def execute(self) -> List[Check_Report_AWS]: + """ + Execute the DMS replication task logging requirements check. + + Iterates over all DMS replication tasks and generates a report indicating + whether each task has logging enabled and meets the logging requirements + for TARGET_APPLY and TARGET_LOAD components. + + Returns: + List[Check_Report_AWS]: A list of report objects with the results of the check. + """ + MINIMUM_SEVERITY_LEVELS = [ + "LOGGER_SEVERITY_DEFAULT", + "LOGGER_SEVERITY_DEBUG", + "LOGGER_SEVERITY_DETAILED_DEBUG", + ] + findings = [] + for ( + replication_task_arn, + replication_task, + ) in dms_client.replication_tasks.items(): + report = Check_Report_AWS(self.metadata()) + report.resource_id = replication_task.id + report.resource_arn = replication_task_arn + report.region = replication_task.region + report.resource_tags = replication_task.tags + + if not replication_task.logging_enabled: + report.status = "FAIL" + report.status_extended = f"DMS Replication Task {replication_task.id} does not have logging enabled for target events." + else: + missing_components = [] + source_capture_compliant = False + source_unload_compliant = False + + for component in replication_task.log_components: + if ( + component["Id"] == "TARGET_APPLY" + and component["Severity"] in MINIMUM_SEVERITY_LEVELS + ): + source_capture_compliant = True + elif ( + component["Id"] == "TARGET_LOAD" + and component["Severity"] in MINIMUM_SEVERITY_LEVELS + ): + source_unload_compliant = True + + if not source_capture_compliant: + missing_components.append("Target Apply") + if not source_unload_compliant: + missing_components.append("Target Load") + + if source_capture_compliant and source_unload_compliant: + report.status = "PASS" + report.status_extended = f"DMS Replication Task {replication_task.id} has logging enabled with the minimum severity level in target events." + else: + report.status = "FAIL" + report.status_extended = f"DMS Replication Task {replication_task.id} does not meet the minimum severity level of logging in {' and '.join(missing_components)} events." + + findings.append(report) + + return findings diff --git a/prowler/providers/aws/services/dms/dms_service.py b/prowler/providers/aws/services/dms/dms_service.py index a8aa2e0570..97764dd439 100644 --- a/prowler/providers/aws/services/dms/dms_service.py +++ b/prowler/providers/aws/services/dms/dms_service.py @@ -18,6 +18,7 @@ class DMS(AWSService): self.__threading_call__(self._describe_replication_instances) self.__threading_call__(self._list_tags, self.instances) self.__threading_call__(self._describe_endpoints) + self.__threading_call__(self._describe_replication_tasks) self.__threading_call__(self._list_tags, self.endpoints.values()) self.__threading_call__(self._describe_replication_tasks) self.__threading_call__(self._list_tags, self.replication_tasks.values()) diff --git a/tests/providers/aws/services/dms/dms_replication_task_source_logging_enabled/dms_replication_task_source_logging_enabled_test.py b/tests/providers/aws/services/dms/dms_replication_task_source_logging_enabled/dms_replication_task_source_logging_enabled_test.py index 3ba90bccca..885b86af10 100644 --- a/tests/providers/aws/services/dms/dms_replication_task_source_logging_enabled/dms_replication_task_source_logging_enabled_test.py +++ b/tests/providers/aws/services/dms/dms_replication_task_source_logging_enabled/dms_replication_task_source_logging_enabled_test.py @@ -403,7 +403,7 @@ class Test_dms_replication_task_source_logging_enabled: assert len(result) == 1 assert result[0].status == "FAIL" assert result[0].status_extended == ( - "DMS Replication Task rep-task does not meet the minimum severity level of logging in Source Capture, Source Unload events." + "DMS Replication Task rep-task does not meet the minimum severity level of logging in Source Capture and Source Unload events." ) assert result[0].resource_id == "rep-task" assert result[0].resource_arn == dms_replication_task_arn diff --git a/tests/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled_test.py b/tests/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled_test.py new file mode 100644 index 0000000000..507334c75f --- /dev/null +++ b/tests/providers/aws/services/dms/dms_replication_task_target_logging_enabled/dms_replication_task_target_logging_enabled_test.py @@ -0,0 +1,475 @@ +from unittest import mock + +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, +) + +DMS_ENDPOINT_NAME = "dms-endpoint" +DMS_ENDPOINT_ARN = f"arn:aws:dms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:endpoint:{DMS_ENDPOINT_NAME}" +DMS_INSTANCE_NAME = "rep-instance" +DMS_INSTANCE_ARN = ( + f"arn:aws:dms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:rep:{DMS_INSTANCE_NAME}" +) + + +class Test_dms_replication_task_target_logging_enabled: + @mock_aws + def test_no_dms_replication_tasks(self): + dms_client = client("dms", region_name=AWS_REGION_US_EAST_1) + dms_client.replication_tasks = {} + + from prowler.providers.aws.services.dms.dms_service import DMS + + 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, + ), mock.patch( + "prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled.dms_client", + new=DMS(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled import ( + dms_replication_task_target_logging_enabled, + ) + + check = dms_replication_task_target_logging_enabled() + result = check.execute() + + assert len(result) == 0 + + @mock_aws + def test_dms_replication_task_logging_not_enabled(self): + dms_client = client("dms", region_name=AWS_REGION_US_EAST_1) + dms_client.create_replication_task( + ReplicationTaskIdentifier="rep-task", + SourceEndpointArn=DMS_ENDPOINT_ARN, + TargetEndpointArn=DMS_ENDPOINT_ARN, + MigrationType="full-load", + ReplicationTaskSettings=""" + { + "Logging": { + "EnableLogging": false, + "LogComponents": [ + { + "Id": "TARGET_LOAD", + "Severity": "LOGGER_SEVERITY_DEFAULT" + } + ] + } + } + """, + TableMappings="", + ReplicationInstanceArn=DMS_INSTANCE_ARN, + ) + + dms_replication_task_arn = dms_client.describe_replication_tasks()[ + "ReplicationTasks" + ][0]["ReplicationTaskArn"] + + from prowler.providers.aws.services.dms.dms_service import DMS + + 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, + ), mock.patch( + "prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled.dms_client", + new=DMS(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled import ( + dms_replication_task_target_logging_enabled, + ) + + check = dms_replication_task_target_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "DMS Replication Task rep-task does not have logging enabled for target events." + ) + assert result[0].resource_id == "rep-task" + assert result[0].resource_arn == dms_replication_task_arn + assert result[0].resource_tags == [] + assert result[0].region == "us-east-1" + + @mock_aws + def test_dms_replication_task_logging_enabled_source_load_only(self): + dms_client = client("dms", region_name=AWS_REGION_US_EAST_1) + dms_client.create_replication_task( + ReplicationTaskIdentifier="rep-task", + SourceEndpointArn=DMS_ENDPOINT_ARN, + TargetEndpointArn=DMS_ENDPOINT_ARN, + MigrationType="full-load", + ReplicationTaskSettings=""" + { + "Logging": { + "EnableLogging": true, + "LogComponents": [ + { + "Id": "TARGET_LOAD", + "Severity": "LOGGER_SEVERITY_DEFAULT" + } + ] + } + } + """, + TableMappings="", + ReplicationInstanceArn=DMS_INSTANCE_ARN, + ) + + dms_replication_task_arn = dms_client.describe_replication_tasks()[ + "ReplicationTasks" + ][0]["ReplicationTaskArn"] + + from prowler.providers.aws.services.dms.dms_service import DMS + + 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, + ), mock.patch( + "prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled.dms_client", + new=DMS(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled import ( + dms_replication_task_target_logging_enabled, + ) + + check = dms_replication_task_target_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "DMS Replication Task rep-task does not meet the minimum severity level of logging in Target Apply events." + ) + assert result[0].resource_id == "rep-task" + assert result[0].resource_arn == dms_replication_task_arn + assert result[0].resource_tags == [] + assert result[0].region == "us-east-1" + + @mock_aws + def test_dms_replication_task_logging_enabled_source_apply_only(self): + dms_client = client("dms", region_name=AWS_REGION_US_EAST_1) + dms_client.create_replication_task( + ReplicationTaskIdentifier="rep-task", + SourceEndpointArn=DMS_ENDPOINT_ARN, + TargetEndpointArn=DMS_ENDPOINT_ARN, + MigrationType="full-load", + ReplicationTaskSettings=""" + { + "Logging": { + "EnableLogging": true, + "LogComponents": [ + { + "Id": "TARGET_APPLY", + "Severity": "LOGGER_SEVERITY_DEFAULT" + } + ] + } + } + """, + TableMappings="", + ReplicationInstanceArn=DMS_INSTANCE_ARN, + ) + + dms_replication_task_arn = dms_client.describe_replication_tasks()[ + "ReplicationTasks" + ][0]["ReplicationTaskArn"] + + from prowler.providers.aws.services.dms.dms_service import DMS + + 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, + ), mock.patch( + "prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled.dms_client", + new=DMS(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled import ( + dms_replication_task_target_logging_enabled, + ) + + check = dms_replication_task_target_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "DMS Replication Task rep-task does not meet the minimum severity level of logging in Target Load events." + ) + assert result[0].resource_id == "rep-task" + assert result[0].resource_arn == dms_replication_task_arn + assert result[0].resource_tags == [] + assert result[0].region == "us-east-1" + + @mock_aws + def test_dms_replication_task_logging_enabled_target_load_apply_with_not_enough_severity_on_load( + self, + ): + dms_client = client("dms", region_name=AWS_REGION_US_EAST_1) + dms_client.create_replication_task( + ReplicationTaskIdentifier="rep-task", + SourceEndpointArn=DMS_ENDPOINT_ARN, + TargetEndpointArn=DMS_ENDPOINT_ARN, + MigrationType="full-load", + ReplicationTaskSettings=""" + { + "Logging": { + "EnableLogging": true, + "LogComponents": [ + { + "Id": "TARGET_LOAD", + "Severity": "LOGGER_SEVERITY_INFO" + }, + { + "Id": "TARGET_APPLY", + "Severity": "LOGGER_SEVERITY_DEFAULT" + } + ] + } + } + """, + TableMappings="", + ReplicationInstanceArn=DMS_INSTANCE_ARN, + ) + + dms_replication_task_arn = dms_client.describe_replication_tasks()[ + "ReplicationTasks" + ][0]["ReplicationTaskArn"] + + from prowler.providers.aws.services.dms.dms_service import DMS + + 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, + ), mock.patch( + "prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled.dms_client", + new=DMS(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled import ( + dms_replication_task_target_logging_enabled, + ) + + check = dms_replication_task_target_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "DMS Replication Task rep-task does not meet the minimum severity level of logging in Target Load events." + ) + assert result[0].resource_id == "rep-task" + assert result[0].resource_arn == dms_replication_task_arn + assert result[0].resource_tags == [] + assert result[0].region == "us-east-1" + + @mock_aws + def test_dms_replication_task_logging_enabled_target_load_apply_with_not_enough_severity_on_apply( + self, + ): + dms_client = client("dms", region_name=AWS_REGION_US_EAST_1) + dms_client.create_replication_task( + ReplicationTaskIdentifier="rep-task", + SourceEndpointArn=DMS_ENDPOINT_ARN, + TargetEndpointArn=DMS_ENDPOINT_ARN, + MigrationType="full-load", + ReplicationTaskSettings=""" + { + "Logging": { + "EnableLogging": true, + "LogComponents": [ + { + "Id": "TARGET_LOAD", + "Severity": "LOGGER_SEVERITY_DEFAULT" + }, + { + "Id": "TARGET_APPLY", + "Severity": "LOGGER_SEVERITY_INFO" + } + ] + } + } + """, + TableMappings="", + ReplicationInstanceArn=DMS_INSTANCE_ARN, + ) + + dms_replication_task_arn = dms_client.describe_replication_tasks()[ + "ReplicationTasks" + ][0]["ReplicationTaskArn"] + + from prowler.providers.aws.services.dms.dms_service import DMS + + 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, + ), mock.patch( + "prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled.dms_client", + new=DMS(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled import ( + dms_replication_task_target_logging_enabled, + ) + + check = dms_replication_task_target_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "DMS Replication Task rep-task does not meet the minimum severity level of logging in Target Apply events." + ) + assert result[0].resource_id == "rep-task" + assert result[0].resource_arn == dms_replication_task_arn + assert result[0].resource_tags == [] + assert result[0].region == "us-east-1" + + @mock_aws + def test_dms_replication_task_logging_enabled_target_load_apply_with_not_enough_severity_on_both( + self, + ): + dms_client = client("dms", region_name=AWS_REGION_US_EAST_1) + dms_client.create_replication_task( + ReplicationTaskIdentifier="rep-task", + SourceEndpointArn=DMS_ENDPOINT_ARN, + TargetEndpointArn=DMS_ENDPOINT_ARN, + MigrationType="full-load", + ReplicationTaskSettings=""" + { + "Logging": { + "EnableLogging": true, + "LogComponents": [ + { + "Id": "TARGET_LOAD", + "Severity": "LOGGER_SEVERITY_INFO" + }, + { + "Id": "TARGET_APPLY", + "Severity": "LOGGER_SEVERITY_INFO" + } + ] + } + } + """, + TableMappings="", + ReplicationInstanceArn=DMS_INSTANCE_ARN, + ) + + dms_replication_task_arn = dms_client.describe_replication_tasks()[ + "ReplicationTasks" + ][0]["ReplicationTaskArn"] + + from prowler.providers.aws.services.dms.dms_service import DMS + + 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, + ), mock.patch( + "prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled.dms_client", + new=DMS(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled import ( + dms_replication_task_target_logging_enabled, + ) + + check = dms_replication_task_target_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].status_extended == ( + "DMS Replication Task rep-task does not meet the minimum severity level of logging in Target Apply and Target Load events." + ) + assert result[0].resource_id == "rep-task" + assert result[0].resource_arn == dms_replication_task_arn + assert result[0].resource_tags == [] + assert result[0].region == "us-east-1" + + @mock_aws + def test_dms_replication_task_logging_enabled_target_load_apply_with_enough_severity_on_both( + self, + ): + dms_client = client("dms", region_name=AWS_REGION_US_EAST_1) + dms_client.create_replication_task( + ReplicationTaskIdentifier="rep-task", + SourceEndpointArn=DMS_ENDPOINT_ARN, + TargetEndpointArn=DMS_ENDPOINT_ARN, + MigrationType="full-load", + ReplicationTaskSettings=""" + { + "Logging": { + "EnableLogging": true, + "LogComponents": [ + { + "Id": "TARGET_LOAD", + "Severity": "LOGGER_SEVERITY_DEFAULT" + }, + { + "Id": "TARGET_APPLY", + "Severity": "LOGGER_SEVERITY_DEFAULT" + } + ] + } + } + """, + TableMappings="", + ReplicationInstanceArn=DMS_INSTANCE_ARN, + ) + + dms_replication_task_arn = dms_client.describe_replication_tasks()[ + "ReplicationTasks" + ][0]["ReplicationTaskArn"] + + from prowler.providers.aws.services.dms.dms_service import DMS + + 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, + ), mock.patch( + "prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled.dms_client", + new=DMS(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.dms.dms_replication_task_target_logging_enabled.dms_replication_task_target_logging_enabled import ( + dms_replication_task_target_logging_enabled, + ) + + check = dms_replication_task_target_logging_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].status_extended == ( + "DMS Replication Task rep-task has logging enabled with the minimum severity level in target events." + ) + assert result[0].resource_id == "rep-task" + assert result[0].resource_arn == dms_replication_task_arn + assert result[0].resource_tags == [] + assert result[0].region == "us-east-1"