Back to Checks
aws

Kafka cluster has encryption in transit enabled

kafka_cluster_in_transit_encryption_enabled
Service: kafkaSeverity: high
OfficialBuilt-inVerified
Categories:encryption
Resource Type:AwsMskCluster

Amazon MSK clusters are evaluated for encryption in transit on both paths: clientbroker set to TLS only and inter-broker encryption enabled. Serverless clusters provide this by default.

The finding highlights clusters where client-broker traffic isn't TLS-only or inter-broker encryption is turned off.

Prowler logoProwlerprowler-provider-aws

Risk

Unencrypted or mixed (TLS_PLAINTEXT/PLAINTEXT) traffic enables interception of records, credentials, and metadata, supporting MITM, replay, and message tampering.

Plaintext inter-broker links expose replication data within the VPC, enabling lateral movement and topic poisoning, degrading data confidentiality and integrity.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler aws --checks kafka_cluster_in_transit_encryption_enabled

Recommendation

Enforce end-to-end transport protection:

  • Require client_broker=TLS for all clients
  • Enable in_cluster=true for broker-to-broker links

Apply defense in depth: restrict network paths, prefer private connectivity, and use strong client authentication with least privilege authorization to limit blast radius.

Remediation

Terraform

# Terraform: MSK cluster with encryption in transit enforced
resource "aws_msk_cluster" "<example_resource_name>" {
  cluster_name           = "<example_resource_name>"
  kafka_version          = "<VERSION>"
  number_of_broker_nodes = 3

  broker_node_group_info {
    instance_type  = "kafka.m5.large"
    client_subnets = [
      "subnet-<example_resource_id>",
      "subnet-<example_resource_id>",
    ]
  }

  encryption_info {
    encryption_in_transit {
      client_broker = "TLS"  # Critical: forces client-to-broker TLS only
      in_cluster    = true    # Critical: enables inter-broker encryption
    }
  }
}

NativeIaC

# CloudFormation: MSK cluster with encryption in transit enforced
Resources:
  <example_resource_name>:
    Type: AWS::MSK::Cluster
    Properties:
      ClusterName: <example_resource_name>
      KafkaVersion: <VERSION>
      NumberOfBrokerNodes: 3
      BrokerNodeGroupInfo:
        ClientSubnets:
          - <example_resource_id>
          - <example_resource_id>
        InstanceType: kafka.m5.large
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS  # Critical: forces client-to-broker TLS only
          InCluster: true    # Critical: enables inter-broker encryption

Other

1In the AWS Console, go to Amazon MSK > Clusters and select your cluster
2Click Edit (Security)
3Under Encryption in transit, set Client-broker to TLS only
4Save changes
5Verify Inter-broker (in-cluster) encryption is enabled; if it is disabled (immutable), create a new cluster with:
6- Encryption in transit: Client-broker = TLS only, Inter-broker encryption = Enabled
7- Migrate clients to the new cluster, then decommission the old one

Check Code

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


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

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

            # Serverless clusters always have encryption in transit enabled by default
            if cluster.kafka_version == "SERVERLESS":
                report.status = "PASS"
                report.status_extended = f"Kafka cluster '{cluster.name}' is serverless and always has encryption in transit enabled by default."
            # For provisioned clusters, check the encryption configuration
            elif (
                cluster.encryption_in_transit.client_broker == "TLS"
                and cluster.encryption_in_transit.in_cluster
            ):
                report.status = "PASS"
                report.status_extended = (
                    f"Kafka cluster '{cluster.name}' has encryption in transit enabled."
                )

            findings.append(report)

        return findings

References