SASL/PLAIN 是一種簡單的 username/password 認證機制, 通常與 TLS 加密一起使用, 用於實現安全認證. Kafka 提供了一個默認的 SASL/PLAIN 實現, 可以做擴展后在生產環境使用.
1 配置kafka server端（每個broker）
root@CN-GRI-IDDC-AIRCONDITING:/opt/kafka/kafka1/kafka_2.12-0.10.2.1/config#
vi server.properties
listeners=SASL_PLAINTEXT://172.17.102.126:9092
port=9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
authorizer.class.name = kafka.security.auth.SimpleAclAuthorizer
#allow.everyone.if.no.acl.found=false
super.users=User:admin;User:Yam
root@CN-GRI-IDDC-AIRCONDITING:/opt/kafka/kafka1/kafka_2.12-0.10.2.1/config#
vi kafka_server_jaas.conf
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin"
user_Yam="password";
};
JAAS文件作為每個broker的jvm參數，在kafka-server-start.sh腳本中增加如下配置:
root@CN-GRI-IDDC-AIRCONDITING:/opt/kafka/kafka1/kafka_2.12-0.10.2.1/bin#
vi kafka-server-start.sh
if [ "x$KAFKA_OPTS" ]; then
export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/kafka1/kafka_2.12-0.10.2.1/config/kafka_server_jaas.conf"
fi
vi kafka-run-class.sh (增加****標注的部分)
**KAFKA_SASL_OPTS='-Djava.security.auth.login.config=/opt/kafka/kafka1/kafka_2.12-0.10.2.1/config/kafka_server_jaas.conf'**
if [ "x$DAEMON_MODE" = "xtrue" ]; then
nohup $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS **$KAFKA_SASL_OPTS **$KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
else
exec $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS **$KAFKA_SASL_OPTS** $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@"
fi
每個broker做相同的操作，除IP Port 文件路徑不同其他相同。
2 配置kafka client端 PLAIN機制下kafka生產者/消費者
root@CN-GRI-IDDC-AIRCONDITING:/opt/kafka/kafka2/kafka_2.12-0.10.2.1/config#
vi kafka_client_jaas.conf
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="Yam"
password="password";
};
vi consumer.properties
vi producer.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
root@CN-GRI-IDDC-AIRCONDITING:/opt/kafka/kafka2/kafka_2.12-0.10.2.1/bin#
vi kafka-console-consumer.sh
vi kafka-console-producer.sh
if [ "x$KAFKA_OPTS" ]; then
export KAFKA_OPTS=" -Djava.security.auth.login.config=/opt/kafka/kafka2/kafka_2.12-0.10.2.1/config/kafka_client_jaas.conf"
fi
權限設置
為用戶Yam在sean-security（topic）上添加讀寫的權限
sh kafka-acls.sh --authorizer-properties zookeeper.connect=172.17.102.126:2181 --add --allow-principal User:Yam --operation Read --operation Write --topic sean-security
3 常用操作
3.1 add操作
為用戶 alice 在 test（topic）上添加讀寫的權限
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=data-rt-dev02:2181/kafka_test10 --add --allow-principal User:alice --operation Read --operation Write --topic test
對於 topic 為 test 的消息隊列，拒絕來自 ip 為198.51.100.3賬戶為 BadBob 進行 read 操作，其他用戶都允許
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=data-rt-dev02:2181/kafka_test10 --add --allow-principal User:* --allow-host * --deny-principal User:BadBob --deny-host 198.51.100.3 --operation Read --topic test
為bob 和 alice 添加all，以允許來自 ip 為198.51.100.0或者198.51.100.1的讀寫請求
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=data-rt-dev02:2181/kafka_test10 --add --allow-principal User:bob --allow-principal User:alice --allow-host 198.51.100.0 --allow-host 198.51.100.1 --operation Read --operation Write --topic test
3.2 list操作
列出 topic 為 test 的所有權限賬戶
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=data-rt-dev02:2181/kafka_test10 --list --topic test
3.3 remove操作
移除 acl
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=data-rt-dev02:2181/kafka_test10 --remove --allow-principal User:Bob --allow-principal User:Alice --allow-host 198.51.100.0 --allow-host 198.51.100.1 --operation Read --operation Write --topic test
3.4 producer和consumer操作
producer
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=data-rt-dev02:2181/kafka_test10 --add --allow-principal User:alice --producer --topic test
consumer
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=data-rt-dev02:2181/kafka_test10 --add --allow-principal User:alice --consumer --topic test —group test-group