Back to Checks
aws

MSK Connect connector has encryption in transit enabled

kafka_connector_in_transit_encryption_enabled
Service: kafkaSeverity: high
OfficialBuilt-inVerified
Categories:encryption
Resource Type:Other

MSK Connect connectors are evaluated for in-transit encryption using TLS on client connections to Kafka brokers and connected systems.

Prowler logoProwlerprowler-provider-aws

Risk

Without TLS, data streams can be intercepted or modified in transit. Attackers on the path can perform man-in-the-middle, replay, or message tampering, exposing records and secrets. This degrades confidentiality and integrity and can enable unauthorized access to downstream systems.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler aws --checks kafka_connector_in_transit_encryption_enabled

Recommendation

Require TLS for all connector communications and disallow plaintext. Prefer private connectivity, validate certificates, and use modern cipher suites. Pair with mutual authentication and least privilege roles for defense-in-depth. Regularly review connector configs to avoid non-TLS endpoints.

Remediation

Terraform

# Terraform: MSK Connect connector with in-transit encryption enabled
resource "aws_mskconnect_connector" "<example_resource_name>" {
  name                  = "<example_resource_name>"
  kafkaconnect_version  = "<KAFKA_CONNECT_VERSION>"

  kafka_cluster {
    apache_kafka_cluster {
      bootstrap_servers = "<BOOTSTRAP_SERVERS>"
      vpc {
        security_groups = ["<example_resource_id>"]
        subnets         = ["<example_resource_id>"]
      }
    }
  }

  kafka_cluster_client_authentication {
    authentication_type = "NONE"
  }

  kafka_cluster_encryption_in_transit {
    encryption_type = "TLS"  # Critical: enables TLS encryption in transit
  }

  capacity {
    provisioned_capacity {
      mcu_count    = 1
      worker_count = 1
    }
  }

  service_execution_role_arn = "<example_resource_id>"

  connector_configuration = {
    "connector.class" = "<CONNECTOR_CLASS>"
    "tasks.max"       = "1"
  }

  plugin {
    custom_plugin {
      arn      = "<example_resource_id>"
      revision = 1
    }
  }
}

NativeIaC

# CloudFormation: MSK Connect connector with in-transit encryption enabled
Resources:
  <example_resource_name>:
    Type: AWS::KafkaConnect::Connector
    Properties:
      ConnectorName: <example_resource_name>
      KafkaCluster:
        ApacheKafkaCluster:
          BootstrapServers: <BOOTSTRAP_SERVERS>
          Vpc:
            SecurityGroups: [<example_resource_id>]
            Subnets: [<example_resource_id>]
      KafkaClusterClientAuthentication:
        AuthenticationType: NONE
      KafkaClusterEncryptionInTransit:
        EncryptionType: TLS  # Critical: enables TLS encryption in transit
      KafkaConnectVersion: <KAFKA_CONNECT_VERSION>
      Plugins:
        - CustomPlugin:
            CustomPluginArn: <example_resource_id>
            Revision: 1
      Capacity:
        ProvisionedCapacity:
          McuCount: 1
          WorkerCount: 1
      ServiceExecutionRoleArn: <example_resource_id>
      ConnectorConfiguration:
        connector.class: <CONNECTOR_CLASS>
        tasks.max: "1"

Other

1In the AWS console, go to Amazon MSK > MSK Connect > Connectors
2Select the non-TLS connector and choose Delete (encryption setting can't be changed)
3Choose Create connector and select your custom plugin and cluster
4In the Security section, set Encryption in transit to TLS (required)
5Complete other required fields and Create the connector

Check Code

from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.kafka.kafkaconnect_client import kafkaconnect_client


class kafka_connector_in_transit_encryption_enabled(Check):
    def execute(self):
        findings = []

        for connector in kafkaconnect_client.connectors.values():
            report = Check_Report_AWS(metadata=self.metadata(), resource=connector)
            report.status = "FAIL"
            report.status_extended = f"Kafka connector {connector.name} does not have encryption in transit enabled."

            if connector.encryption_in_transit == "TLS":
                report.status = "PASS"
                report.status_extended = f"Kafka connector {connector.name} has encryption in transit enabled."

            findings.append(report)

        return findings

References