Ensure Kafka Cluster Has Unrestricted Access Disabled

Overview

This check verifies that your Amazon MSK (Managed Streaming for Apache Kafka) clusters require authentication for all client connections. Clusters should not allow unauthenticated access, which would let anyone connect without credentials.

MSK Serverless clusters always require authentication. For provisioned MSK clusters, you must explicitly disable unauthenticated access and enable at least one authentication method.

Risk

Allowing unauthenticated access to your Kafka cluster creates serious security vulnerabilities:

• Data theft: Anyone can read sensitive messages from your topics
• Data tampering: Attackers can publish malicious messages or modify existing data
• Service disruption: Bad actors can overwhelm brokers and consumers
• Compliance violations: Most security frameworks require authentication for data access

This is rated as a critical severity finding because it directly exposes your streaming data to unauthorized access.

Remediation Steps

Prerequisites

You need:

• Access to the AWS Console with permissions to modify MSK clusters, OR
• AWS CLI configured with appropriate IAM permissions

Required IAM permissions

Your IAM user or role needs these permissions:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kafka:DescribeCluster",
        "kafka:UpdateSecurity",
        "kafka:ListClusters"
      ],
      "Resource": "*"
    }
  ]
}

AWS Console Method

1. Open the Amazon MSK console
2. Make sure you are in the correct region (e.g., US East (N. Virginia))
3. In the left navigation, click Clusters
4. Click on the name of your cluster to open its details
5. Select the Properties tab
6. Scroll down to the Security settings section
7. Click Edit next to Access control methods
8. Under Access control methods:
   • Uncheck "Unauthenticated access" (if checked)
   • Check at least one authentication method:
     • IAM role-based authentication (recommended for AWS workloads)
     • SASL/SCRAM authentication (for username/password access)
     • TLS client authentication (for certificate-based access)
9. Click Save changes

The cluster will update its configuration. This may take several minutes.

AWS CLI

First, get your cluster ARN and current version:

# List all MSK clusters
aws kafka list-clusters --region us-east-1

# Get details for a specific cluster (including current version)
aws kafka describe-cluster \
  --cluster-arn <your-cluster-arn> \
  --region us-east-1

Note the CurrentVersion value from the output.

Then disable unauthenticated access and enable IAM authentication:

aws kafka update-security \
  --cluster-arn <your-cluster-arn> \
  --current-version <current-version> \
  --client-authentication 'Sasl={Iam={Enabled=true}},Unauthenticated={Enabled=false}' \
  --region us-east-1

Replace:

• <your-cluster-arn> with your cluster's ARN (e.g., arn:aws:kafka:us-east-1:123456789012:cluster/my-cluster/abc123)
• <current-version> with the version string from describe-cluster output

Alternative: Enable SASL/SCRAM authentication

aws kafka update-security \
  --cluster-arn <your-cluster-arn> \
  --current-version <current-version> \
  --client-authentication 'Sasl={Scram={Enabled=true}},Unauthenticated={Enabled=false}' \
  --region us-east-1

CloudFormation

This template creates a new MSK cluster with IAM authentication enabled and unauthenticated access disabled:

AWSTemplateFormatVersion: '2010-09-09'
Description: Amazon MSK Cluster with authentication required (no unauthenticated access)

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'
  SubnetIds:
    Type: List<AWS::EC2::Subnet::Id>
    Description: List of subnet IDs (at least 2 in different AZs)
  SecurityGroupId:
    Type: AWS::EC2::SecurityGroup::Id
    Description: Security group ID for the MSK cluster

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

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

Key configuration:

• ClientAuthentication.Sasl.Iam.Enabled: true - Enables IAM authentication
• ClientAuthentication.Unauthenticated.Enabled: false - Disables unauthenticated access
• EncryptionInfo.EncryptionInTransit.ClientBroker: TLS - Enforces encrypted connections

Terraform

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

provider "aws" {
  region = "us-east-1"
}

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 "subnet_ids" {
  description = "List of subnet IDs for the MSK cluster"
  type        = list(string)
}

variable "security_group_ids" {
  description = "List of security group IDs for the MSK cluster"
  type        = list(string)
}

resource "aws_msk_cluster" "this" {
  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 = var.security_group_ids

    storage_info {
      ebs_storage_info {
        volume_size = 100
      }
    }
  }

  client_authentication {
    sasl {
      iam = true
    }
    unauthenticated = false
  }

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

  enhanced_monitoring = "PER_TOPIC_PER_BROKER"

  tags = {
    Environment = "production"
    ManagedBy   = "Terraform"
  }
}

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

output "bootstrap_brokers_sasl_iam" {
  description = "IAM connection host:port pairs"
  value       = aws_msk_cluster.this.bootstrap_brokers_sasl_iam
}

Key configuration:

• client_authentication.sasl.iam = true - Enables IAM authentication
• client_authentication.unauthenticated = false - Disables unauthenticated access
• encryption_info.encryption_in_transit.client_broker = "TLS" - Enforces encrypted connections

Verification

After making changes, verify that unauthenticated access is disabled:

1. In the AWS Console, go to your MSK cluster's Properties tab
2. Under Security settings, confirm:
   • "Unauthenticated access" shows as Disabled or is not listed
   • At least one authentication method (IAM, SASL/SCRAM, or TLS) is enabled

CLI verification

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

Expected output should show Unauthenticated as disabled or absent, and at least one authentication method enabled:

{
  "Sasl": {
    "Iam": {
      "Enabled": true
    }
  },
  "Unauthenticated": {
    "Enabled": false
  }
}

Additional Resources

• Amazon MSK Authentication Documentation
• IAM Access Control for Amazon MSK
• SASL/SCRAM Authentication for Amazon MSK
• Amazon MSK Security Best Practices
• Prowler Check Documentation

Notes

• Update time: Security configuration changes can take 10-30 minutes to apply. The cluster remains available during the update.
• Client impact: After disabling unauthenticated access, all clients must use the configured authentication method. Update your client applications before making this change.
• Serverless clusters: MSK Serverless clusters always require IAM authentication and cannot have unauthenticated access enabled.
• Multiple authentication methods: You can enable multiple authentication methods simultaneously (e.g., both IAM and SASL/SCRAM) to support different client types.
• Encryption: When using IAM or SASL/SCRAM authentication, TLS encryption in transit is required. The examples above include this configuration.