kafka sasl/plain安全認證

1.SASL認證機制版本支持

SASL/GSSAPI (Kerberos) - starting at version 0.9.0.0
SASL/PLAIN - starting at version 0.10.0.0
SASL/SCRAM-SHA-256 and SASL/SCRAM-SHA-512 - starting at version 0.10.2.0

2.以下采用SASL/PLAIN進行認證操作

　　zookeeper配置

1）修改zoo.cfg增加兩行配置：
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl

2）配置JAAS文件：conf目錄下創建zk_server_jaas.conf（定義了需要鏈接到Zookeeper服務器的用戶名和密碼）
Server {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin-sec";
};

3）加入需要的包：（從kafka下拷貝）
kafka-clients-0.10.0.1.jar
lz4-1.3.0.jar
slf4j-api-1.7.21.jar
slf4j-log4j12-1.7.21.jar
snappy-java-1.1.2.6.jar

3）修改zkEnv.sh
最后一行添加
export SERVER_JVMFLAGS=" -Djava.security.auth.login.config=/usr/local/zookeeper/conf/zk_server_jaas.con"
4）啟動Zookeeper　　

 kafka服務的配置

1）kafka增加認證信息：conf/kafka_server_jaas.conf
創建JAAS文件：
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin-sec"
user_admin="admin-sec"
user_producer="prod-sec"
user_consumer="cons-sec";
};

2）配置server.properties
listeners=SASL_PLAINTEXT://主機名稱:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true //當沒有找到ACL配置時，允許所有的訪問操作。


3）修改啟動腳本 bin/kafka-server-start.sh
修改
exec $base_dir/kafka-run-class.sh kafka.Kafka "$@"
為
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/usr/local/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"

　　kafka客戶端配置

1）創建JAAS文件：
消費者：conf/kafka-consumer-jaas.conf
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="consumer"
password="cons-sec";
};
生產者：conf/kafka-producer-jaas.conf
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="producer"
password="prod-sec";
};

2）修改客戶端配置信息：
分別在conf/producer.properties和conf/consumer.properties添加認證機制
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
consumer.properties中額外加入分組配置
group.id=test-group
3）修改客戶端腳本指定JAAS文件加載：

生產者bin/kafka-console-producer.sh：
修改
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
為
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/usr/local/kafka/config/kafka-producer-jaas.conf kafka.tools.ConsoleProducer "$@"

消費者bin/kafka-console-consumer.sh：
修改
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"
為
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/usr/local/kafka/config/kafka-consumer-jaas.conf kafka.tools.ConsoleConsumer "$@"

　　進行授權

1）創建主題 test
2）增加生產權限
./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:producer --operation Write --topic test
3）配置消費權限
./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:consumer --operation Read --topic test
4）配置消費分組權限
./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:consumer --operation Read --group test-group
5）查看配置的權限
./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --list

6）取消權限
./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --remove --allow-principal User:producer --operation Write --topic test

測試
1）生產數據
./bin/kafka-console-producer-jaas.sh --topic test --broker-list 192.168.1.20:9092 --producer.config config/producer-jaas.properties

2）消費數據
./bin/kafka-console-consumer-jaas.sh --topic test --bootstrap-server 192.168.1.20:9092 --consumer.config config/consumer-jaas.properties