kafka动态权限认证（SASL SCRAM + ACL）

创建三个测试用户

bin/kafka-configs.sh --zookeeper 192.168.x.x:2181 --alter --add-config ‘SCRAM-SHA-256=[iterations=8192,password=admin],SCRAM-SHA-512=[password=admin]’ --entity-type users --entity-name admin

PS：用户 admin 这里配置admin用户用于实现broker间的通讯。

测试用户 writer
bin/kafka-configs.sh --zookeeper 192.168.x.x:2181 --alter --add-config ‘SCRAM-SHA-256=[iterations=8192,password=writer],SCRAM-SHA-512=[password=writer]’ --entity-type users --entity-name writer

测试用户 reader
bin/kafka-configs.sh --zookeeper 192.168.x.x:2181 --alter --add-config ‘SCRAM-SHA-256=[iterations=8192,password=reader],SCRAM-SHA-512=[password=reader]’ --entity-type users --entity-name reader

查看创建用户信息

bin/kafka-configs.sh --zookeeper 192.168.2.6:2181 --describe --entity-type users （可以单独指定某个用户–entity-name writer）

创建配置文件kafka-broker-jaas.conf

保存至 /opt/kafka/config 下 (每台主机)
KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username=“admin”
password=“admin”;
};

配置broker端的server.properties

启用ACL
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#设置本例中admin为超级用户
super.users=User:admin
启用SCRAM机制，采用SCRAM-SHA-512算法
sasl.enabled.mechanisms=SCRAM-SHA-512
#为broker间通讯开启SCRAM机制，采用SCRAM-SHA-512算法
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
#broker间通讯使用PLAINTEXT
security.inter.broker.protocol=SASL_PLAINTEXT
#配置listeners使用SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://n6.aa-data.cn:9092（指定当前主机）
#配置advertised.listeners
advertised.listeners=SASL_PLAINTEXT://n6.aa-data.cn:9092（指定当前主机）

配置环境变量，引入jaas

export KAFKA_OPTS=’-Djava.security.auth.login.config=/opt/kafka/config/kafka-broker-jaas.conf’

生产

创建 producer.conf

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=“writer” password=“writer”;

给writer 提供写权限

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.6:2181 --add --allow-principal User:writer --operation Write --topic testcon

生产消息

bin/kafka-console-producer.sh --broker-list n6.aa-data.cn:9092 --topic testcon --producer.config /opt/kafka/config/producer.conf

消费

创建 consumer.conf

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=“reader” password=“reader”;

给reader 提供读取权限

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.6:2181 --add --allow-principal User:reader --operation Read --topic testcon

给reader 添加访问consumer group的权限
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=n6.aa-data.cn:2181 --add --allow-principal User:reader --operation Read --group test

消费 testcon消息
bin/kafka-console-consumer.sh --bootstrap-server 192.168.2.6:9092 --topic testcon --from-beginning --consumer.config /opt/kafka/config/consumer.conf

动态添加生成用户

测试用户 writer1

bin/kafka-configs.sh --zookeeper 192.168.2.6:2181 --alter --add-config ‘SCRAM-SHA-256=[iterations=8192,password=writer],SCRAM-SHA-512=[password=writer1]’ --entity-type users --entity-name writer1

测试用户 reader1
bin/kafka-configs.sh --zookeeper 192.168.2.6:2181 --alter --add-config ‘SCRAM-SHA-256=[iterations=8192,password=reader],SCRAM-SHA-512=[password=reader1]’ --entity-type users --entity-name reader1

赋予权限
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.6:2181 --add --allow-principal User:writer1 --operation Write --topic testcon

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.2.6:2181 --add --allow-principal User:reader1 --operation Read --topic testcon

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=n6.aa-data.cn:2181 --add --allow-principal User:reader --operation Read --group test1

删除原来的用户
bin/kafka-configs.sh --zookeeper 192.168.2.6:2181 --alter --delete-config ‘SCRAM-SHA-256’ --entity-type users --entity-name writer
bin/kafka-configs.sh --zookeeper 192.168.2.6:2181 --alter --delete-config ‘SCRAM-SHA-512’ --entity-type users --entity-name writer

Ps：用户删除后需修改 producer.conf consumer.conf