Back to Checks
aws

Kafka cluster has encryption at rest enabled with a customer managed key (CMK) or is serverless

kafka_cluster_encryption_at_rest_uses_cmk
Service: kafkaSeverity: medium
OfficialBuilt-inVerified
Categories:encryption
Resource Type:AwsMskCluster

Amazon MSK clusters are inspected for encryption at rest using a customer-managed KMS key for data volumes. Serverless clusters are inherently encrypted. Provisioned clusters are recognized only when the configured DataVolumeKMSKeyId corresponds to a customer-managed key.

Prowler logoProwlerprowler-provider-aws

Risk

Relying on service-managed keys weakens confidentiality and accountability: you can't enforce granular key policies, separation of duties, or independent rotation. This limits incident response (e.g., disabling the key for crypto-shredding) and reduces auditability, increasing impact of credential misuse or broker compromise.

Run this check with Prowler

Run in Prowler Cloud
Run
prowler aws --checks kafka_cluster_encryption_at_rest_uses_cmk

Recommendation

Use a customer-managed KMS key for MSK at-rest encryption. Apply least privilege in key policies and grants, enable key rotation, and log key use for auditing. Enforce separation of duties between MSK admins and KMS key custodians, and regularly review access, aliases, and pending-deletion states.

Remediation

Terraform

# MSK cluster using a customer managed KMS key for encryption at rest
resource "aws_msk_cluster" "<example_resource_name>" {
  cluster_name           = "<example_resource_name>"
  kafka_version          = "<KAFKA_VERSION>"
  number_of_broker_nodes = 2

  broker_node_group_info {
    instance_type  = "kafka.m5.large"
    client_subnets = ["<example_subnet_id_a>", "<example_subnet_id_b>"]
    security_groups = ["<example_security_group_id>"]
  }

  encryption_info {
    encryption_at_rest_kms_key_arn = "<example_kms_key_arn>" # Critical: customer managed KMS key to pass the check
  }
}

NativeIaC

# CloudFormation: MSK cluster using a customer managed KMS key for encryption at rest
Resources:
  <example_resource_name>:
    Type: AWS::MSK::Cluster
    Properties:
      ClusterName: <example_resource_name>
      KafkaVersion: <KAFKA_VERSION>
      NumberOfBrokerNodes: 2
      BrokerNodeGroupInfo:
        InstanceType: kafka.m5.large
        ClientSubnets:
          - <example_subnet_id_a>
          - <example_subnet_id_b>
        SecurityGroups:
          - <example_security_group_id>
      EncryptionInfo:
        EncryptionAtRest:
          DataVolumeKMSKeyId: <example_kms_key_arn>  # Critical: use a customer managed KMS key ARN to enable CMK encryption at rest

Other

1In the AWS Console, go to Amazon MSK > Clusters
2Click Create cluster
3Choose Provisioned (or choose Serverless to pass by default)
4In Encryption settings, for At-rest encryption, select Customer managed key and choose your CMK (not alias/aws/kafka)
5Create the cluster, migrate clients to it, then delete the old cluster that used the AWS managed key

Check Code

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


class kafka_cluster_encryption_at_rest_uses_cmk(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 at rest enabled with a CMK."

            # Serverless clusters always have encryption at rest 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 at rest enabled by default."
            # For provisioned clusters, check if they use a customer managed KMS key
            elif any(
                (
                    cluster.data_volume_kms_key_id == key.arn
                    and getattr(key, "manager", "") == "CUSTOMER"
                )
                for key in kms_client.keys
            ):
                report.status = "PASS"
                report.status_extended = f"Kafka cluster '{cluster.name}' has encryption at rest enabled with a CMK."

            findings.append(report)

        return findings

References