.svg)
.svg)
We will cover how to set up mutual TLS authentication and authorization on Amazon MSK.
Amazon MSK is a fully managed service that makes it easy to build and run applications that use Apache Kafka to process streaming data. You can enable client authentication with TLS for connections and client authorization from your applications to your Amazon MSK brokers and ZooKeeper nodes.
To use client authentication with TLS on MSK, you need to create the following resources:
AWS Private CA can be either in the same AWS account as your cluster, or in a different account. For information about AWS Private CAs, see Creating and Managing a AWS Private CA. In this setup, we will use Terraform to create a private CA.
| terraform { | |
| required_providers { | |
| aws = { | |
| source = "hashicorp/aws" | |
| version = "~> 4.0" | |
| } | |
| } | |
| } | |
| resource "aws_acmpca_certificate_authority" "root_ca" { | |
| certificate_authority_configuration { | |
| key_algorithm = "RSA_4096" | |
| signing_algorithm = "SHA512WITHRSA" | |
| subject { | |
| #Update the attributes as per your need | |
| common_name = "exp-msk-ca" | |
| country = "US" | |
| locality = "Seattle" | |
| organization = "Example Corp" | |
| organizational_unit = "Sales" | |
| state = "WA" | |
| } | |
| } | |
| type = "ROOT" | |
| } |
Steps to install the certificate.
Note: We highly recommend using independent AWS Private CA for each MSK cluster when you use mutual TLS to control access. Doing so will ensure that TLS certificates signed by PCAs only authenticate with a single MSK cluster.
Run the below Terraform code to create MSK cluster
Note: Update attributes as per the requirement and configurations.
| terraform { | |
| required_providers { | |
| aws = { | |
| source = "hashicorp/aws" | |
| version = "~> 4.0" | |
| } | |
| } | |
| } | |
| module "kafka" { | |
| source = "cloudposse/msk-apache-kafka-cluster/aws" | |
| # Cloud Posse recommends pinning every module to a specific version | |
| version = "2.3.0" | |
| name = "test-msk-cluster" #Change MSK cluster name as per your need | |
| vpc_id = "<VPC_ID>" | |
| subnet_ids = ["SUBNET1a","SUNBNET2b"] # Minimum 2 subnets required. | |
| kafka_version = "3.4.0" #recommended version by AWS as of 19 Sep 2022 | |
| broker_per_zone = 1 #Number of broker per availability zone | |
| broker_instance_type = "kafka.t3.small" #MSK instance types | |
| broker_volume_size = 10 #Broker disk size | |
| certificate_authority_arns = ["<CA_ARN>"] #arn of the CA that you have created in the earlier step | |
| client_tls_auth_enabled = true | |
| encryption_in_cluster = true | |
| client_broker = "TLS" # Enables TLS encryption | |
| enhanced_monitoring = "PER_TOPIC_PER_BROKER" | |
| cloudwatch_logs_enabled = false # Enable if you need cloudwatch logs | |
| jmx_exporter_enabled = false # Enable if you need jmx metrics | |
| node_exporter_enabled = false # Enable if you need node metrics | |
| associated_security_group_ids = ["${aws_security_group.kafka_sg.id}"] | |
| allowed_security_group_ids = ["${aws_security_group.kafka_sg.id}"] | |
| create_security_group = false | |
| } | |
| #-----------------------End--------------------# | |
| resource "aws_security_group" "kafka_sg" { | |
| name = "test-msk-cluster-sg" #Change the name as per your need | |
| description = "Security Group for kafka cluster" | |
| vpc_id = "<VPC_ID>" | |
| egress { | |
| from_port = 0 | |
| to_port = 0 | |
| protocol = "-1" | |
| cidr_blocks = ["0.0.0.0/0"] | |
| ipv6_cidr_blocks = ["::/0"] | |
| } | |
| ingress { | |
| from_port = 2181 | |
| to_port = 2181 | |
| protocol = "tcp" | |
| cidr_blocks = ["0.0.0.0/0"] | |
| ipv6_cidr_blocks = ["::/0"] | |
| } | |
| ingress { | |
| from_port = 9094 | |
| to_port = 9094 | |
| protocol = "tcp" | |
| cidr_blocks = ["0.0.0.0/0"] | |
| ipv6_cidr_blocks = ["::/0"] | |
| } | |
| # Enable if you need to add tags to MSK cluster | |
| #tags = var.tags | |
| # Enable if you need cloudwatch logs | |
| # depends_on = [ | |
| # aws_cloudwatch_log_group.cw_log_group | |
| #] | |
| } | |
| # Required for cloudwatch logs | |
| # resource "aws_cloudwatch_log_group" "cw_log_group" { | |
| # name = "blog-msk-cluster" | |
| # #tags = var.tags | |
| # } | |
| output "bootstrap_url" { | |
| value = module.kafka.bootstrap_brokers_tls | |
| description = "Comma separated list of one or more DNS names (or IP addresses) and TLS port pairs for access to the Kafka cluster using TLS" | |
| } |
It will take 15-20 minutes to create the MSK cluster.
Note: Since the bootstrap URL will be used to communicate with the MSK cluster using the Kafka CLI or SDKs, save it from the Terraform output.
We will create two certificates, one is for admin access, and the other one is for client access. For creating certificates, a common name (CN) is required. The CN is used as a principal while granting permissions through kafka ACLs
| openssl req -newkey rsa:2048 -keyout key.pem -out cert.csr -batch -nodes -subj '/CN=admin' |
| certArn=$(aws acm-pca issue-certificate --region <region> --certificate-authority-arn "<CA_ARN>" | |
| --csr fileb://cert.csr | |
| --signing-algorithm 'SHA256WITHRSA' --validity Value=180,Type='DAYS' --query | |
| 'CertificateArn' --output text) |
| aws acm-pca get-certificate --region <region> --certificate-authority-arn | |
| "<CA_ARN>" --certificate-arn "${certArn}" --output text | sed 's/\t/\n/g' > | |
| cert.pem |
| openssl pkcs12 -export -in cert.pem -inkey key.pem -name ssl-configurator | |
| -password pass: -out admin.p12 |
| openssl req -newkey rsa:2048 -keyout key.pem -out cert.csr -batch -nodes -subj | |
| '/CN=client' |
| certArn=$(aws acm-pca issue-certificate --region <region> | |
| --certificate-authority-arn "<CA_ARN>" --csr fileb://cert.csr | |
| --signing-algorithm 'SHA256WITHRSA' --validity Value=180,Type='DAYS' --query | |
| 'CertificateArn' --output text) |
| aws acm-pca get-certificate --region <region> --certificate-authority-arn | |
| "<CA_ARN>" --certificate-arn "${certArn}" --output text | sed 's/\t/\n/g' > | |
| cert.pem |
| openssl pkcs12 -export -in cert.pem -inkey key.pem -name ssl-configurator | |
| -password pass: -out client.p12 |
https://archive.apache.org/dist/kafka/3.4.0/kafka_2.13-3.4.0.tgz
| cat <<EOF> admin.propertie | |
| bootstrap.servers="<BOOTSTRAP_URL>" | |
| security.protocol=SSL | |
| ssl.keystore.location=./admin.p12 | |
| ssl.keystore.type=PKCS12 | |
| ssl.keystore.password= | |
| EOF | |
| cat <<EOF> client.propertie | |
| bootstrap.servers="<BOOTSTRAP_URL>" | |
| security.protocol=SSL | |
| ssl.keystore.location=./client.p12 | |
| ssl.keystore.type=PKCS12 | |
| ssl.keystore.password= | |
| EOF |
By default, the MSK cluster will allow everyone if no ACL is found. Here the Admin ACL will be the first ACL. The Admin user ("User:CN=admin") will leverage on Admin ACL to grant permissions to Client User("User:CN=client").
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=admin" \ | |
| --operation All \ | |
| --cluster \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=admin" \ | |
| --operation All \ | |
| --topic "*" \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=admin" \ | |
| --operation All \ | |
| --group "*" \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
Create a topic.
| ./kafka_2.13-3.5.0/bin/kafka-topics.sh --bootstrap-server "<BOOTSTRAP_URL>" | |
| --create --topic test-topic --command-config admin.properties |
List topic and check the topic is created.
| ./kafka_2.13-3.5.0/bin/kafka-topics.sh --bootstrap-server "<BOOTSTRAP_URL>" | |
| --list --command-config admin.properties |
Grant write permission to the topic so that client (producer) can publish messages to the topic (Use admin user for granting access to client).
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=client" \ | |
| --operation Write \ | |
| --topic "test-topic" \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
Publish messages to the topic using client user.
| for x in {1..10}; do echo "message $x"; done | | |
| ./kafka_2.13-3.5.0/bin/kafka-console-producer.sh --bootstrap-server | |
| "<BOOTSTRAP_URL>" --producer.config kafka-admin-client.properties --topic | |
| test-topic --producer-property enable.idempotence=false |
Consume messages.
Note: If you try to consume messages from the topic using a consumer group, you will get a group authorization error since the client user is not authorized to access groups.
| ./kafka_2.13-3.5.0/bin/kafka-console-consumer.sh --bootstrap-server | |
| "<BOOTSTRAP_URL>" --topic test-topic --max-messages 2 --consumer-property | |
| enable.auto.commit=false --consumer-property group.id=consumer-test | |
| --from-beginning --consumer.config client.properties |
Grant group permission to the client user.
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=client" \ | |
| --operation Read \ | |
| --resource-pattern-type prefixed \ | |
| --group 'consumer-' \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
After providing group access, the client user should be able to consume messages from the topic using a consumer group.
This way, you can manage client access to the topics and groups.
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --list \ | |
| --command-config admin.properties |
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --remove \ | |
| --allow-principal "User:CN=admin" \ | |
| --operation All \ | |
| --cluster \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
AWS MSK eases the effort of managing independently hosted Kafka clusters. Users can scale Kafka brokers and storage as necessary. MSK supports TLS encryption and allows users to create TLS connections from the application to Amazon MSK brokers and ZooKeeper nodes with the help of the AWS Private CA, which enables users to create certificates for authentication.
Did you like the blog? If yes, we're sure you'll also like to work with the people who write them - our best-in-class engineering team.
We're looking for talented developers who are passionate about new emerging technologies. If that's you, get in touch with us.
We will cover how to set up mutual TLS authentication and authorization on Amazon MSK.
Amazon MSK is a fully managed service that makes it easy to build and run applications that use Apache Kafka to process streaming data. You can enable client authentication with TLS for connections and client authorization from your applications to your Amazon MSK brokers and ZooKeeper nodes.
To use client authentication with TLS on MSK, you need to create the following resources:
AWS Private CA can be either in the same AWS account as your cluster, or in a different account. For information about AWS Private CAs, see Creating and Managing a AWS Private CA. In this setup, we will use Terraform to create a private CA.
| terraform { | |
| required_providers { | |
| aws = { | |
| source = "hashicorp/aws" | |
| version = "~> 4.0" | |
| } | |
| } | |
| } | |
| resource "aws_acmpca_certificate_authority" "root_ca" { | |
| certificate_authority_configuration { | |
| key_algorithm = "RSA_4096" | |
| signing_algorithm = "SHA512WITHRSA" | |
| subject { | |
| #Update the attributes as per your need | |
| common_name = "exp-msk-ca" | |
| country = "US" | |
| locality = "Seattle" | |
| organization = "Example Corp" | |
| organizational_unit = "Sales" | |
| state = "WA" | |
| } | |
| } | |
| type = "ROOT" | |
| } |
Steps to install the certificate.
Note: We highly recommend using independent AWS Private CA for each MSK cluster when you use mutual TLS to control access. Doing so will ensure that TLS certificates signed by PCAs only authenticate with a single MSK cluster.
Run the below Terraform code to create MSK cluster
Note: Update attributes as per the requirement and configurations.
| terraform { | |
| required_providers { | |
| aws = { | |
| source = "hashicorp/aws" | |
| version = "~> 4.0" | |
| } | |
| } | |
| } | |
| module "kafka" { | |
| source = "cloudposse/msk-apache-kafka-cluster/aws" | |
| # Cloud Posse recommends pinning every module to a specific version | |
| version = "2.3.0" | |
| name = "test-msk-cluster" #Change MSK cluster name as per your need | |
| vpc_id = "<VPC_ID>" | |
| subnet_ids = ["SUBNET1a","SUNBNET2b"] # Minimum 2 subnets required. | |
| kafka_version = "3.4.0" #recommended version by AWS as of 19 Sep 2022 | |
| broker_per_zone = 1 #Number of broker per availability zone | |
| broker_instance_type = "kafka.t3.small" #MSK instance types | |
| broker_volume_size = 10 #Broker disk size | |
| certificate_authority_arns = ["<CA_ARN>"] #arn of the CA that you have created in the earlier step | |
| client_tls_auth_enabled = true | |
| encryption_in_cluster = true | |
| client_broker = "TLS" # Enables TLS encryption | |
| enhanced_monitoring = "PER_TOPIC_PER_BROKER" | |
| cloudwatch_logs_enabled = false # Enable if you need cloudwatch logs | |
| jmx_exporter_enabled = false # Enable if you need jmx metrics | |
| node_exporter_enabled = false # Enable if you need node metrics | |
| associated_security_group_ids = ["${aws_security_group.kafka_sg.id}"] | |
| allowed_security_group_ids = ["${aws_security_group.kafka_sg.id}"] | |
| create_security_group = false | |
| } | |
| #-----------------------End--------------------# | |
| resource "aws_security_group" "kafka_sg" { | |
| name = "test-msk-cluster-sg" #Change the name as per your need | |
| description = "Security Group for kafka cluster" | |
| vpc_id = "<VPC_ID>" | |
| egress { | |
| from_port = 0 | |
| to_port = 0 | |
| protocol = "-1" | |
| cidr_blocks = ["0.0.0.0/0"] | |
| ipv6_cidr_blocks = ["::/0"] | |
| } | |
| ingress { | |
| from_port = 2181 | |
| to_port = 2181 | |
| protocol = "tcp" | |
| cidr_blocks = ["0.0.0.0/0"] | |
| ipv6_cidr_blocks = ["::/0"] | |
| } | |
| ingress { | |
| from_port = 9094 | |
| to_port = 9094 | |
| protocol = "tcp" | |
| cidr_blocks = ["0.0.0.0/0"] | |
| ipv6_cidr_blocks = ["::/0"] | |
| } | |
| # Enable if you need to add tags to MSK cluster | |
| #tags = var.tags | |
| # Enable if you need cloudwatch logs | |
| # depends_on = [ | |
| # aws_cloudwatch_log_group.cw_log_group | |
| #] | |
| } | |
| # Required for cloudwatch logs | |
| # resource "aws_cloudwatch_log_group" "cw_log_group" { | |
| # name = "blog-msk-cluster" | |
| # #tags = var.tags | |
| # } | |
| output "bootstrap_url" { | |
| value = module.kafka.bootstrap_brokers_tls | |
| description = "Comma separated list of one or more DNS names (or IP addresses) and TLS port pairs for access to the Kafka cluster using TLS" | |
| } |
It will take 15-20 minutes to create the MSK cluster.
Note: Since the bootstrap URL will be used to communicate with the MSK cluster using the Kafka CLI or SDKs, save it from the Terraform output.
We will create two certificates, one is for admin access, and the other one is for client access. For creating certificates, a common name (CN) is required. The CN is used as a principal while granting permissions through kafka ACLs
| openssl req -newkey rsa:2048 -keyout key.pem -out cert.csr -batch -nodes -subj '/CN=admin' |
| certArn=$(aws acm-pca issue-certificate --region <region> --certificate-authority-arn "<CA_ARN>" | |
| --csr fileb://cert.csr | |
| --signing-algorithm 'SHA256WITHRSA' --validity Value=180,Type='DAYS' --query | |
| 'CertificateArn' --output text) |
| aws acm-pca get-certificate --region <region> --certificate-authority-arn | |
| "<CA_ARN>" --certificate-arn "${certArn}" --output text | sed 's/\t/\n/g' > | |
| cert.pem |
| openssl pkcs12 -export -in cert.pem -inkey key.pem -name ssl-configurator | |
| -password pass: -out admin.p12 |
| openssl req -newkey rsa:2048 -keyout key.pem -out cert.csr -batch -nodes -subj | |
| '/CN=client' |
| certArn=$(aws acm-pca issue-certificate --region <region> | |
| --certificate-authority-arn "<CA_ARN>" --csr fileb://cert.csr | |
| --signing-algorithm 'SHA256WITHRSA' --validity Value=180,Type='DAYS' --query | |
| 'CertificateArn' --output text) |
| aws acm-pca get-certificate --region <region> --certificate-authority-arn | |
| "<CA_ARN>" --certificate-arn "${certArn}" --output text | sed 's/\t/\n/g' > | |
| cert.pem |
| openssl pkcs12 -export -in cert.pem -inkey key.pem -name ssl-configurator | |
| -password pass: -out client.p12 |
https://archive.apache.org/dist/kafka/3.4.0/kafka_2.13-3.4.0.tgz
| cat <<EOF> admin.propertie | |
| bootstrap.servers="<BOOTSTRAP_URL>" | |
| security.protocol=SSL | |
| ssl.keystore.location=./admin.p12 | |
| ssl.keystore.type=PKCS12 | |
| ssl.keystore.password= | |
| EOF | |
| cat <<EOF> client.propertie | |
| bootstrap.servers="<BOOTSTRAP_URL>" | |
| security.protocol=SSL | |
| ssl.keystore.location=./client.p12 | |
| ssl.keystore.type=PKCS12 | |
| ssl.keystore.password= | |
| EOF |
By default, the MSK cluster will allow everyone if no ACL is found. Here the Admin ACL will be the first ACL. The Admin user ("User:CN=admin") will leverage on Admin ACL to grant permissions to Client User("User:CN=client").
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=admin" \ | |
| --operation All \ | |
| --cluster \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=admin" \ | |
| --operation All \ | |
| --topic "*" \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=admin" \ | |
| --operation All \ | |
| --group "*" \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
Create a topic.
| ./kafka_2.13-3.5.0/bin/kafka-topics.sh --bootstrap-server "<BOOTSTRAP_URL>" | |
| --create --topic test-topic --command-config admin.properties |
List topic and check the topic is created.
| ./kafka_2.13-3.5.0/bin/kafka-topics.sh --bootstrap-server "<BOOTSTRAP_URL>" | |
| --list --command-config admin.properties |
Grant write permission to the topic so that client (producer) can publish messages to the topic (Use admin user for granting access to client).
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=client" \ | |
| --operation Write \ | |
| --topic "test-topic" \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
Publish messages to the topic using client user.
| for x in {1..10}; do echo "message $x"; done | | |
| ./kafka_2.13-3.5.0/bin/kafka-console-producer.sh --bootstrap-server | |
| "<BOOTSTRAP_URL>" --producer.config kafka-admin-client.properties --topic | |
| test-topic --producer-property enable.idempotence=false |
Consume messages.
Note: If you try to consume messages from the topic using a consumer group, you will get a group authorization error since the client user is not authorized to access groups.
| ./kafka_2.13-3.5.0/bin/kafka-console-consumer.sh --bootstrap-server | |
| "<BOOTSTRAP_URL>" --topic test-topic --max-messages 2 --consumer-property | |
| enable.auto.commit=false --consumer-property group.id=consumer-test | |
| --from-beginning --consumer.config client.properties |
Grant group permission to the client user.
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --add \ | |
| --allow-principal "User:CN=client" \ | |
| --operation Read \ | |
| --resource-pattern-type prefixed \ | |
| --group 'consumer-' \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
After providing group access, the client user should be able to consume messages from the topic using a consumer group.
This way, you can manage client access to the topics and groups.
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --list \ | |
| --command-config admin.properties |
| ./kafka_2.13-3.5.0/bin/kafka-acls.sh \ | |
| --remove \ | |
| --allow-principal "User:CN=admin" \ | |
| --operation All \ | |
| --cluster \ | |
| --bootstrap-server "<BOOTSTRAP_URL>" \ | |
| --command-config admin.properties |
AWS MSK eases the effort of managing independently hosted Kafka clusters. Users can scale Kafka brokers and storage as necessary. MSK supports TLS encryption and allows users to create TLS connections from the application to Amazon MSK brokers and ZooKeeper nodes with the help of the AWS Private CA, which enables users to create certificates for authentication.
.svg)
.svg)