Ensure Mutual TLS Authentication is Enabled for Kafka Cluster

Overview

This check verifies that your Amazon MSK (Managed Streaming for Apache Kafka) clusters use mutual TLS (mTLS) authentication for client connections. With mTLS, both clients and the Kafka brokers must prove their identity using certificates, providing strong two-way authentication.

Risk

Without mutual TLS authentication, your Kafka cluster faces several security risks:

• Unauthorized access: Any client that can reach the network could connect to your cluster
• Man-in-the-middle attacks: Attackers could intercept or modify data in transit
• Data breaches: Unauthorized producers or consumers could read sensitive topics or inject malicious data
• Compliance violations: Many regulatory frameworks require strong authentication for data systems

Remediation Steps

Prerequisites

• An existing Amazon MSK provisioned cluster (must be in ACTIVE state)
• An AWS Private Certificate Authority (ACM Private CA) to issue client certificates
• Permission to modify MSK cluster security settings

Setting up AWS Private CA (if needed)

If you don't have a Private CA, you'll need to create one first:

1. Go to AWS Certificate Manager > Private Certificate Authority
2. Click Create a private CA
3. Choose Root CA for a standalone setup, or Subordinate CA if you have an existing hierarchy
4. Complete the setup wizard and activate the CA

Note: Private CA has associated costs. See AWS Private CA pricing.

AWS Console Method

1. Open the Amazon MSK console at https://console.aws.amazon.com/msk/
2. Select Clusters from the left navigation
3. Click on your cluster name (cluster must be in ACTIVE state)
4. Choose Actions > Edit security settings
5. Under Access control methods, select TLS client authentication via AWS Certificate Manager (ACM)
6. Click Add ACM PCA and enter your Private CA ARN
7. Under Encryption, set Between clients and brokers to TLS encryption
8. Click Save changes

The update takes approximately 15-30 minutes. Your cluster will remain available during this time.

AWS CLI (optional)

First, get your cluster's current version:

aws kafka describe-cluster \
  --cluster-arn <your-cluster-arn> \
  --region us-east-1 \
  --query 'ClusterInfo.CurrentVersion' \
  --output text

Then enable mTLS authentication:

aws kafka update-security \
  --cluster-arn <your-cluster-arn> \
  --current-version <current-version> \
  --client-authentication '{
    "Tls": {
      "CertificateAuthorityArnList": ["<your-private-ca-arn>"],
      "Enabled": true
    },
    "Unauthenticated": {
      "Enabled": false
    }
  }' \
  --encryption-info '{
    "EncryptionInTransit": {
      "ClientBroker": "TLS"
    }
  }' \
  --region us-east-1

Replace:

• <your-cluster-arn> with your MSK cluster ARN
• <current-version> with the version string from the first command
• <your-private-ca-arn> with your ACM Private CA ARN

CloudFormation (optional)

Use this template to create a new MSK cluster with mTLS enabled:

AWSTemplateFormatVersion: '2010-09-09'
Description: Amazon MSK Cluster with Mutual TLS Authentication enabled

Parameters:
  ClusterName:
    Type: String
    Description: Name for the MSK cluster
    Default: my-msk-cluster

  KafkaVersion:
    Type: String
    Description: Apache Kafka version
    Default: '3.5.1'

  VpcId:
    Type: AWS::EC2::VPC::Id
    Description: VPC where MSK cluster will be deployed

  SubnetIds:
    Type: List<AWS::EC2::Subnet::Id>
    Description: Subnet IDs for MSK broker nodes (minimum 2 subnets in different AZs)

  PrivateCAArn:
    Type: String
    Description: ARN of the AWS Private Certificate Authority for client authentication

Resources:
  MSKSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Security group for MSK cluster
      VpcId: !Ref VpcId
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 9094
          ToPort: 9094
          CidrIp: 10.0.0.0/8
          Description: TLS client access
      Tags:
        - Key: Name
          Value: !Sub '${ClusterName}-sg'

  MSKCluster:
    Type: AWS::MSK::Cluster
    Properties:
      ClusterName: !Ref ClusterName
      KafkaVersion: !Ref KafkaVersion
      NumberOfBrokerNodes: 3
      BrokerNodeGroupInfo:
        InstanceType: kafka.m5.large
        ClientSubnets: !Ref SubnetIds
        SecurityGroups:
          - !Ref MSKSecurityGroup
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 100
      ClientAuthentication:
        Tls:
          CertificateAuthorityArnList:
            - !Ref PrivateCAArn
          Enabled: true
        Unauthenticated:
          Enabled: false
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS
          InCluster: true
      EnhancedMonitoring: PER_TOPIC_PER_BROKER
      Tags:
        Environment: production

Outputs:
  ClusterArn:
    Description: ARN of the MSK cluster
    Value: !Ref MSKCluster

  SecurityGroupId:
    Description: Security group ID for the MSK cluster
    Value: !Ref MSKSecurityGroup

Deploy with:

aws cloudformation deploy \
  --template-file template.yaml \
  --stack-name msk-mtls-cluster \
  --parameter-overrides \
    ClusterName=my-secure-cluster \
    VpcId=vpc-xxxxxxxx \
    SubnetIds=subnet-aaaaaaaa,subnet-bbbbbbbb,subnet-cccccccc \
    PrivateCAArn=arn:aws:acm-pca:us-east-1:123456789012:certificate-authority/xxxxx \
  --region us-east-1

Terraform (optional)

terraform {
  required_version = ">= 1.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = ">= 5.0"
    }
  }
}

variable "cluster_name" {
  description = "Name of the MSK cluster"
  type        = string
  default     = "my-msk-cluster"
}

variable "kafka_version" {
  description = "Apache Kafka version"
  type        = string
  default     = "3.5.1"
}

variable "vpc_id" {
  description = "VPC ID where MSK cluster will be deployed"
  type        = string
}

variable "subnet_ids" {
  description = "List of subnet IDs for MSK broker nodes"
  type        = list(string)
}

variable "private_ca_arn" {
  description = "ARN of the AWS Private Certificate Authority for mTLS"
  type        = string
}

resource "aws_security_group" "msk" {
  name_prefix = "${var.cluster_name}-msk-"
  description = "Security group for MSK cluster"
  vpc_id      = var.vpc_id

  ingress {
    description = "TLS client access"
    from_port   = 9094
    to_port     = 9094
    protocol    = "tcp"
    cidr_blocks = ["10.0.0.0/8"]
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }

  tags = {
    Name = "${var.cluster_name}-sg"
  }
}

resource "aws_msk_cluster" "main" {
  cluster_name           = var.cluster_name
  kafka_version          = var.kafka_version
  number_of_broker_nodes = 3

  broker_node_group_info {
    instance_type   = "kafka.m5.large"
    client_subnets  = var.subnet_ids
    security_groups = [aws_security_group.msk.id]

    storage_info {
      ebs_storage_info {
        volume_size = 100
      }
    }
  }

  client_authentication {
    tls {
      certificate_authority_arns = [var.private_ca_arn]
    }
    unauthenticated = false
  }

  encryption_info {
    encryption_in_transit {
      client_broker = "TLS"
      in_cluster    = true
    }
  }

  enhanced_monitoring = "PER_TOPIC_PER_BROKER"

  tags = {
    Environment = "production"
  }
}

output "cluster_arn" {
  description = "ARN of the MSK cluster"
  value       = aws_msk_cluster.main.arn
}

output "bootstrap_brokers_tls" {
  description = "TLS connection string for Kafka clients"
  value       = aws_msk_cluster.main.bootstrap_brokers_tls
}

Apply with:

terraform apply \
  -var="cluster_name=my-secure-cluster" \
  -var="vpc_id=vpc-xxxxxxxx" \
  -var='subnet_ids=["subnet-aaaaaaaa","subnet-bbbbbbbb","subnet-cccccccc"]' \
  -var="private_ca_arn=arn:aws:acm-pca:us-east-1:123456789012:certificate-authority/xxxxx"

Verification

After the update completes, verify that mTLS is enabled:

1. Go to the Amazon MSK console
2. Select your cluster
3. Check the Security settings section
4. Confirm that TLS client authentication shows your Private CA ARN
5. Confirm that Encryption between clients and brokers shows TLS

CLI verification

aws kafka describe-cluster \
  --cluster-arn <your-cluster-arn> \
  --region us-east-1 \
  --query 'ClusterInfo.{
    TlsEnabled: ClientAuthentication.Tls.Enabled,
    CertificateAuthorities: ClientAuthentication.Tls.CertificateAuthorityArnList,
    ClientBrokerEncryption: EncryptionInfo.EncryptionInTransit.ClientBroker
  }'

Expected output:

{
    "TlsEnabled": true,
    "CertificateAuthorities": [
        "arn:aws:acm-pca:us-east-1:123456789012:certificate-authority/xxxxx"
    ],
    "ClientBrokerEncryption": "TLS"
}

Additional Resources

• Amazon MSK Authentication Documentation
• Mutual TLS Client Authentication
• AWS Private Certificate Authority User Guide
• Prowler Check Documentation

Notes

• Client updates required: After enabling mTLS, all Kafka clients must present valid certificates signed by your Private CA. Update your client applications before enabling mTLS to avoid connectivity issues.
• Certificate rotation: Plan for certificate rotation before they expire. Use short-lived certificates (90 days or less) and automate renewal.
• Cost considerations: AWS Private CA has ongoing costs. Consider using a single Private CA for multiple clusters if appropriate.
• Serverless clusters: MSK Serverless clusters automatically use IAM-based authentication and do not support mTLS configuration.
• Broker port: With TLS enabled, clients connect on port 9094 (TLS) instead of 9092 (plaintext).
• Security groups: Ensure your security groups allow inbound traffic on port 9094 from your client networks.