Back to Checks
aws

Kafka cluster has TLS authentication enabled

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

Amazon MSK clusters enforce client authentication on client-to-broker connections. Serverless clusters use TLS-based authentication by default; provisioned clusters must have mutual TLS (mTLS) explicitly enabled.

Prowler logoProwlerprowler-provider-aws

Risk

Without mTLS, adversaries can impersonate clients or intercept sessions, compromising confidentiality and integrity. Unauthorized producers/consumers can read or alter topics, poison data streams, and flood brokers, degrading availability and impacting downstream systems.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler aws --checks kafka_cluster_mutual_tls_authentication_enabled

Recommendation

Enable mutual TLS for client-broker traffic and disable PLAINTEXT listeners. Issue short-lived client certificates from a managed CA with rotation. Apply least privilege using Kafka ACLs, restrict network access to trusted sources, and monitor authentication events as part of defense in depth.

Remediation

CLI

aws kafka update-security --cluster-arn <CLUSTER_ARN> --current-version <CURRENT_VERSION> --client-authentication 'Tls={CertificateAuthorityArnList=["<ACM_PCA_ARN>"]}' --encryption-info 'EncryptionInTransit={ClientBroker=TLS}'

Terraform

# Terraform: Enable mTLS for an MSK cluster
resource "aws_msk_cluster" "<example_resource_name>" {
  cluster_name           = "<example_resource_name>"
  kafka_version          = "<example_kafka_version>"
  number_of_broker_nodes = 2

  broker_node_group_info {
    instance_type  = "kafka.m5.large"
    client_subnets = ["<subnet_id_1>", "<subnet_id_2>"]
  }

  client_authentication {
    tls {
      certificate_authority_arns = ["<acm_pca_arn>"]  # CRITICAL: Enables mutual TLS with this Private CA
    }
  }

  encryption_info {
    encryption_in_transit {
      client_broker = "TLS"  # CRITICAL: Required when enabling mTLS
    }
  }
}

NativeIaC

# CloudFormation: Enable mTLS for an MSK cluster
Resources:
  <example_resource_name>:
    Type: AWS::MSK::Cluster
    Properties:
      ClusterName: <example_resource_name>
      KafkaVersion: <example_kafka_version>
      NumberOfBrokerNodes: 2
      BrokerNodeGroupInfo:
        InstanceType: kafka.m5.large
        ClientSubnets:
          - <subnet_id_1>
          - <subnet_id_2>
      ClientAuthentication:
        Tls:
          CertificateAuthorityArnList:
            - <acm_pca_arn>  # CRITICAL: Enables mutual TLS using this Private CA
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS  # CRITICAL: Required when enabling mTLS

Other

1In the AWS Console, go to Amazon MSK > Clusters and select the provisioned cluster (state must be ACTIVE)
2Choose Actions > Update security (or Security > Edit)
3Under Client authentication, enable TLS and add your AWS Private CA ARN(s)
4Under Encryption in transit, set Client-broker to TLS
5Save/Update and wait for the update to complete

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_mutual_tls_authentication_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 mutual TLS authentication enabled."

            # Serverless clusters always have TLS authentication enabled by default
            if cluster.kafka_version == "SERVERLESS":
                report.status = "PASS"
                report.status_extended = f"Kafka cluster '{cluster.name}' is serverless and always has TLS authentication enabled by default."
            # For provisioned clusters, check the TLS configuration
            elif cluster.tls_authentication:
                report.status = "PASS"
                report.status_extended = f"Kafka cluster '{cluster.name}' has mutual TLS authentication enabled."

            findings.append(report)

        return findings

References