Back to Checks
aws

DMS replication task has logging enabled and SOURCE_CAPTURE and SOURCE_UNLOAD components set to at least Default severity

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

AWS DMS replication tasks have logging enabled and configure SOURCE_CAPTURE and SOURCE_UNLOAD with severity at least LOGGER_SEVERITY_DEFAULT (or higher: LOGGER_SEVERITY_DEBUG, LOGGER_SEVERITY_DETAILED_DEBUG).

Prowler logoProwlerprowler-provider-aws

Risk

Missing or low-severity source logs hinder visibility into CDC and full-load activity, risking undetected errors, stalls, or tampering. This can cause silent data drift, broken lineage, and failed recoveries, undermining integrity and availability and weakening auditability during investigations.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler aws --checks dms_replication_task_source_logging_enabled

Recommendation

Enable and standardize task logging for SOURCE_CAPTURE and SOURCE_UNLOAD at LOGGER_SEVERITY_DEFAULT or higher.

  • Centralize logs and alert on anomalies
  • Enforce least privilege for log access
  • Set retention to support audits
  • Avoid prolonged DEBUG levels, except during troubleshooting, to balance visibility and cost

Remediation

CLI

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

Terraform

# Enable DMS source logging at minimum DEFAULT severity
resource "aws_dms_replication_task" "<example_resource_name>" {
  replication_instance_arn = "<example_resource_arn>"
  source_endpoint_arn      = "<example_resource_arn>"
  target_endpoint_arn      = "<example_resource_arn>"
  migration_type           = "full-load"
  table_mappings           = "{\"rules\":[]}"

  # Critical: Enables logging and sets SOURCE components to at least DEFAULT
  replication_task_settings = <<JSON
{
  "Logging": {
    "EnableLogging": true,
    "LogComponents": [
      {"Id": "SOURCE_CAPTURE", "Severity": "LOGGER_SEVERITY_DEFAULT"},
      {"Id": "SOURCE_UNLOAD", "Severity": "LOGGER_SEVERITY_DEFAULT"}
    ]
  }
}
JSON
}

NativeIaC

# CloudFormation: enable DMS source logging at minimum DEFAULT severity
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":[]}'
      # Critical: Enables logging and sets SOURCE components to at least DEFAULT
      ReplicationTaskSettings: |
        {
          "Logging": {
            "EnableLogging": true,
            "LogComponents": [
              {"Id": "SOURCE_CAPTURE", "Severity": "LOGGER_SEVERITY_DEFAULT"},
              {"Id": "SOURCE_UNLOAD", "Severity": "LOGGER_SEVERITY_DEFAULT"}
            ]
          }
        }

Other

1In the AWS console, go to Database Migration Service > Database migration tasks
2Select the task and choose Modify
3Click Modify task logging
4Turn on Enable logging
5For SOURCE_CAPTURE and SOURCE_UNLOAD, set Severity to Default (or higher)
6Save/Modify to apply

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_source_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 SOURCE_CAPTURE and SOURCE_UNLOAD 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 SOURCE_CAPTURE and SOURCE_UNLOAD 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(
                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 source events."
            else:
                missing_components = []
                source_capture_compliant = False
                source_unload_compliant = False

                for component in replication_task.log_components:
                    if (
                        component["Id"] == "SOURCE_CAPTURE"
                        and component["Severity"] in MINIMUM_SEVERITY_LEVELS
                    ):
                        source_capture_compliant = True
                    elif (
                        component["Id"] == "SOURCE_UNLOAD"
                        and component["Severity"] in MINIMUM_SEVERITY_LEVELS
                    ):
                        source_unload_compliant = True

                if not source_capture_compliant:
                    missing_components.append("Source Capture")
                if not source_unload_compliant:
                    missing_components.append("Source Unload")

                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 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 {' and '.join(missing_components)} events."

            findings.append(report)

        return findings

References