Back to Checks
aws

DMS replication task has TARGET_APPLY and TARGET_LOAD logging enabled with at least default severity

dms_replication_task_target_logging_enabled
Service: dmsSeverity: medium
OfficialBuilt-inVerified
Categories:logging
Resource Type:AwsDmsReplicationTask

AWS DMS replication tasks have target logging enabled, including TARGET_APPLY and TARGET_LOAD, each set to at least LOGGER_SEVERITY_DEFAULT.

Prowler logoProwlerprowler-provider-aws

Risk

Insufficient target logging limits visibility into load/apply activity, masking failures and anomalies. This risks data integrity (silent drift, partial loads) and availability (longer incident resolution), and reduces auditability of migration events.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler aws --checks dms_replication_task_target_logging_enabled

Recommendation

Enable and maintain CloudWatch logging at LOGGER_SEVERITY_DEFAULT or higher for target components:

  • Configure TARGET_APPLY and TARGET_LOAD
  • Enforce least-privilege log access
  • Monitor logs/alerts for anomalies
  • Standardize task settings and validate data for defense in depth

Remediation

CLI

aws dms modify-replication-task --replication-task-arn <task-arn> --replication-task-settings '{"Logging":{"EnableLogging":true,"LogComponents":[{"Id":"TARGET_APPLY","Severity":"LOGGER_SEVERITY_DEFAULT"},{"Id":"TARGET_LOAD","Severity":"LOGGER_SEVERITY_DEFAULT"}]}}'

Terraform

# Enable DMS task logging for target components
resource "aws_dms_replication_task" "<example_resource_name>" {
  replication_task_id      = "<example_resource_id>"
  replication_instance_arn = "<example_resource_arn>"
  source_endpoint_arn      = "<example_resource_arn>"
  target_endpoint_arn      = "<example_resource_arn>"
  migration_type           = "full-load"
  table_mappings           = jsonencode({ rules = [{
    "rule-type" : "selection", "rule-id" : "1", "rule-name" : "1",
    "object-locator" : { "schema-name" : "%", "table-name" : "%" },
    "rule-action" : "include"
  }]} )

  # Critical: enables logging and sets TARGET_APPLY and TARGET_LOAD to minimum required severity
  replication_task_settings = jsonencode({
    Logging = {
      EnableLogging = true
      LogComponents = [
        { Id = "TARGET_APPLY", Severity = "LOGGER_SEVERITY_DEFAULT" },
        { Id = "TARGET_LOAD",  Severity = "LOGGER_SEVERITY_DEFAULT" }
      ]
    }
  })
}

NativeIaC

# CloudFormation: enable DMS task logging for target components
Resources:
  <example_resource_name>:
    Type: AWS::DMS::ReplicationTask
    Properties:
      ReplicationInstanceArn: <example_resource_arn>
      SourceEndpointArn: <example_resource_arn>
      TargetEndpointArn: <example_resource_arn>
      MigrationType: full-load
      TableMappings: |
        {"rules":[{"rule-type":"selection","rule-id":"1","rule-name":"1","object-locator":{"schema-name":"%","table-name":"%"},"rule-action":"include"}]}
      ReplicationTaskSettings: |
        {"Logging":{"EnableLogging":true, "LogComponents":[
          {"Id":"TARGET_APPLY","Severity":"LOGGER_SEVERITY_DEFAULT"},  # Critical: ensure TARGET_APPLY logging at default
          {"Id":"TARGET_LOAD","Severity":"LOGGER_SEVERITY_DEFAULT"}    # Critical: ensure TARGET_LOAD logging at default
        ]}}

Other

1Open the AWS DMS console and go to Database migration tasks
2Select the replication task and choose Modify
3Expand Task settings (JSON) or Logging
4Enable CloudWatch logs (EnableLogging = true)
5Set log components:
6- TARGET_APPLY severity: DEFAULT
7- TARGET_LOAD severity: DEFAULT
8Save changes (Modify task), then rerun the task if required

Check Code

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 = []
        # Check if replication_tasks is not None before iterating
        if dms_client.replication_tasks:
            for (
                replication_task_arn,
                replication_task,
            ) in dms_client.replication_tasks.items():
                report = Check_Report_AWS(
                    metadata=self.metadata(), resource=replication_task
                )
                report.resource_arn = replication_task_arn

                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

References