0
0

Kafka配置动态SASL_SCRAM认证

Kafka配置动态SASL_SCRAM认证

Kafka中需要加上认证，并动态新增用户，SASL/SCRAM验证可以支持
本文章是对https://blog.csdn.net/qq_38616503/article/details/117529690中的内容整理与重新记录

1.启动Zookeeper和Kafka

第一步，在没有设置任何权限的配置下启动Kafka和Zookeeper，如需要从头安装Kafka，可参见Kafka的安装单机安装以及集群安装

2.创建SCRAM证书

(1)创建broker通信用户：admin(在使用sasl之前必须先创建,否则启动报错)

  1. bin/kafka-configs.sh --zookeeper 192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--alter --add-config 'SCRAM-SHA-256=[password=admin-sec],SCRAM-SHA-512=[password=admin-sec]' --entity-type users --entity-name admin

(2)创建生产用户producer

  1. bin/kafka-configs.sh --zookeeper 192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=prod-sec],SCRAM-SHA-512=[password=prod-sec]' --entity-type users --entity-name producer

(2)创建消费用户：consumer

  1. bin/kafka-configs.sh --zookeeper 192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=cons-sec],SCRAM-SHA-512=[password=cons-sec]' --entity-type users --entity-name consumer

SCRAM-SHA-256/SCRAM-SHA-512是对密码加密的算法，二者有其一即可

3.维护SCRAM证书

3.1查看SCRAM证书

  1. bin/kafka-configs.sh --zookeeper 192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--describe --entity-type users --entity-name consumer
  2. bin/kafka-configs.sh --zookeeper 192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--describe --entity-type users --entity-name producer

3.1删除SCRAM证书

  1. bin/kafka-configs.sh --zookeeper localhost:2181--alter --delete-config 'SCRAM-SHA-512'--delete-config 'SCRAM-SHA-256'--entity-type users --entity-name producer

4.服务端配置

在用户证书创建完毕之后开始Kafka服务端的配置
（1）创建JAAS文件：

  1. cat > kafka_server_jaas.conf <<EOF
  2. KafkaServer {
  3. org.apache.kafka.common.security.scram.ScramLoginModule required
  4. username="admin"
  5. password="admin-sec";};EOF

（2）将JAAS配置文件位置作为JVM参数传递给每个Kafka Broker【bin/kafka-server-start.sh】添加-Djava.security.auth.login.config=/opt/kafka_2.12-2.4.1/config/kafka_server_jaas.conf

  1. -Djava.security.auth.login.config=/opt/kafka_2.12-2.4.1/config/kafka_server_jaas.conf
  1. exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/opt/kafka_2.12-2.4.1/config/kafka_server_jaas.conf kafka.Kafka "$@"

（3）配置server.properties【config/server.properties】

  1. #认证配置
  2. listeners=SASL_PLAINTEXT://IP:9092
  3. security.inter.broker.protocol=SASL_PLAINTEXT
  4. sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
  5. sasl.enabled.mechanisms=SCRAM-SHA-256
  6. advertised.listeners=SASL_PLAINTEXT://IP:9092#ACL配置
  7. allow.everyone.if.no.acl.found=false
  8. super.users=User:admin
  9. authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

可以根据自己的需求选择SASL_SSL或SASL_PLAINTEXT, PLAINTEXT为不加密明文传输，性能好一点。配置完后重启Kafka和Zookeeper

5.客户端配置

(1）创建的三个用户的三个JAAS文件:
kafka_client_scram_admin_jaas.conf
kafka_client_scram_producer_jaas.conf
kafka_client_scram_consumer_jaas.conf

  1. cat > kafka_client_scram_admin_jaas.conf <<EOF
  2. KafkaClient {
  3. org.apache.kafka.common.security.scram.ScramLoginModule required
  4. username="admin"
  5. password="admin-sec";};EOF
  7. cat > kafka_client_scram_producer_jaas.conf <<EOF
  8. KafkaClient {
  9. org.apache.kafka.common.security.scram.ScramLoginModule required
  10. username="producer"
  11. password="prod-sec";};EOF
  13. cat > kafka_client_scram_consumer_jaas.conf <<EOF
  14. KafkaClient {
  15. org.apache.kafka.common.security.scram.ScramLoginModule required
  16. username="consumer"
  17. password="cons-sec";};EOF

(2)修改启动脚本引入JAAS文件

  1. ###生产者配置bin/kafka-console-producer.sh
  2. exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/opt/kafka_2.12-2.4.1/config/kafka_client_scram_producer_jaas.conf
  4. ###消费者配置bin/kafka-console-consumer.sh
  5. exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/opt/kafka_2.12-2.4.1/config/kafka_client_scram_consumer_jaas.conf

(3)配置consumer.properties和producer.properties,都加入以下配置

  1. security.protocol=SASL_PLAINTEXT
  2. sasl.mechanism=SCRAM-SHA-256
  3. bootstrap.servers=192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092

(4)创建主题

  1. bin/kafka-topics.sh --zookeeper 192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--create --topic topictest --partitions 3--replication-factor 1

(5)启动生产

  1. bin/kafka-console-producer.sh --broker-list 192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092--topic topictest --producer.config config/producer.properties

(6)对生产者赋予写的权限

  1. bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--add --allow-principal User:producer --operation Write --topic topictest

(7)查看权限

  1. bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--list

(8)对消费者赋予读的权限

  1. bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--add --allow-principal User:consumer --operation Read --topic topictest

(9)对消费者赋予组的权限

  1. bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.56.101:2181,192.168.56.102:2181,192.168.56.103:2181--add --allow-principal User:consumer --operation Read --group test-consumer-group

(10)启动消费者

  1. bin/kafka-console-consumer.sh --bootstrap-server 192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092--topic topictest --from-beginning --consumer.config config/consumer.properties

6.Java代码测试

6.1生产者

maven的pom.xml

  1. <dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.3.0</version><!--<version>0.10.2.0</version>--></dependency>

kafka_client_scram_producer_jaas.conf

  1. KafkaClient {
  2. org.apache.kafka.common.security.scram.ScramLoginModule required
  3. username="producer"
  4. password="prod-sec";};

代码：

  1. import java.util.Properties;
  2. import java.util.concurrent.Future;
  4. import org.apache.kafka.clients.producer.Callback;
  5. import org.apache.kafka.clients.producer.KafkaProducer;
  6. import org.apache.kafka.clients.producer.Producer;
  7. import org.apache.kafka.clients.producer.ProducerRecord;
  8. import org.apache.kafka.clients.producer.RecordMetadata;
  10. public class MySaslScramProducer {
  11.     
  12.     public static MySaslScramProducer ins ;
  13.     
  14.     private Producer<String, String> producer;
  15.     
  16.     private MySaslScramProducer(){
  17.         
  18.         Properties props = new Properties();
  19.         props.put("bootstrap.servers","192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092");
  20.         props.put("acks","1");
  21.         props.put("retries",3);
  22.         props.put("batch.size",16384);
  23.         props.put("linger.ms",1);
  24.         props.put("buffer.memory",33554432);//props.put("compression.type","gzip");//props.put("max.request.size","5242880");
  25.         props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
  26.         props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");//配置文件采用项目相对路径访问，plan-text鉴权将以下注解开放即可
  27.         System.out.println(MySaslScramProducer.class.getResource("/").getPath()+"kafka_client_scram_producer_jaas.conf");
  28.           System.setProperty("java.security.auth.login.config", MySaslScramProducer.class.getResource("/").getPath()+"kafka_client_scram_producer_jaas.conf");
  29.         props.put("security.protocol","SASL_PLAINTEXT");
  30.         props.put("sasl.mechanism","SCRAM-SHA-256");
  31.         
  32.         producer = new KafkaProducer<>(props);}
  33.     
  34.     public static MySaslScramProducer getIns(){if(ins == null){synchronized(MySaslScramProducer.class){if(ins == null){
  35.                     ins = new MySaslScramProducer();}}}return ins;}
  36.     
  37.     public Future<RecordMetadata>send(String topic, String valueStr){//采用异步发送，在失败时打印出失败的日志，备核查
  38.         Future<RecordMetadata> meta = producer.send(new ProducerRecord<String, String>(topic, valueStr), new Callback(){
  39.             
  40.             @Override
  41.             public voidonCompletion(RecordMetadata metadata, Exception exception){if(exception != null){//发送失败的打印出来到error.log
  42.                     System.out.println("sendi failed--->>> "+ valueStr);}else{
  43.                     System.out.println("topic:"+ metadata.topic()+" ,partition:"+metadata.partition()+" , offset:"+ metadata.offset()+" -> "+ valueStr);}}});return meta;}
  44.     
  45.     public voidclose(){if(producer != null) producer.close();}
  46.     
  47.     public staticvoidmain(String[] args) throws InterruptedException {
  48.         
  49.         String valueStr ="{\"metric\":\"host.mem.pused\",\"value\":\"97.781098\",\"tags\":{\"resCi\":\"TA_RES_PHYSICALHOST\",\"dataType\":0,\"ip\":\"132.121.93.69\",\"cmd\":\"\",\"resId\":\"auto217A77657DDC70403B949090D3EA5543\",\"itemKey\":\"vm.memory.size[pavailable]\"},\"timestamp\":\"1617673320000\"}";
  50.         MySaslScramProducer.getIns().send("topictest", valueStr);
  51.         MySaslScramProducer.getIns().close();}}

6.2消费者

kafka_client_scram_consumer_jaas.conf

  1. KafkaClient {
  2. org.apache.kafka.common.security.scram.ScramLoginModule required
  3. username="consumer"
  4. password="cons-sec";};

代码：

  1. package cn.gzsendi;
  3. import java.util.Arrays;
  4. import java.util.HashMap;
  5. import java.util.Map;
  6. import java.util.Properties;
  7. import java.util.Set;
  9. import org.apache.kafka.clients.consumer.ConsumerRecord;
  10. import org.apache.kafka.clients.consumer.ConsumerRecords;
  11. import org.apache.kafka.clients.consumer.KafkaConsumer;
  12. import org.apache.kafka.clients.consumer.OffsetAndMetadata;
  13. import org.apache.kafka.common.TopicPartition;
  14. import org.slf4j.Logger;
  15. import org.slf4j.LoggerFactory;
  17. public class SaslScramTopicTest {
  18.     
  19.     public static boolean stop = false;
  20.     
  21.     private static Logger logger = LoggerFactory.getLogger(SaslScramTopicTest.class);
  23.     public staticvoidmain(String[] args){
  24.         
  25.         
  26.         KafkaConsumer<String, String> consumer = null;
  27.         Properties props = new Properties();
  28.         props.put("bootstrap.servers","192.168.56.101:9092,192.168.56.102:9092,192.168.56.103:9092");
  29.         props.put("group.id","liutest");
  30.         props.put("enable.auto.commit","true");// 自动提交
  31.         props.put("auto.offset.reset","latest");
  32.         props.put("auto.commit.interval.ms","1000");
  33.         props.put("session.timeout.ms","300000");
  34.         props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  35.         props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
  36.         
  37.         System.setProperty("java.security.auth.login.config", SaslScramTopicTest.class.getResource("/").getPath()+"kafka_client_scram_consumer_jaas.conf");
  38.         props.put("security.protocol","SASL_PLAINTEXT");
  39.         props.put("sasl.mechanism","SCRAM-SHA-256");
  40.         
  41.         consumer = new KafkaConsumer<>(props);
  42.         
  43.         String topicName ="topictest";
  44.         consumer.subscribe(Arrays.asList(topicName));while(!stop){
  45.             try {
  46.                 ConsumerRecords<String, String> records = consumer.poll(100);for(ConsumerRecord<String, String> record : records){
  47.                     String valueStr = record.value();
  48.                     
  49.                     try {
  50.                         logger.info(valueStr);
  51.                         logger.info("topic:"+ record.topic()+" ,partition:"+ record.partition()+" ,offset:"+record.offset()+"  -> "+ record.value());}catch(Exception e){
  52.                         System.out.println("error------->>> "+ valueStr);}}}catch(Exception e){
  53.                 e.printStackTrace();}}if(consumer != null)
  54.             consumer.close();}/**
  55.      *
  56.      * <跳过历史数据，从最新的数据开始消费>
  57.      *
  58.      * @param consumer
  59.      * @throws
  60.      */
  61.     public staticvoidassignOffset(KafkaConsumer<String, String> consumer){if(consumer == null){return;}
  62.         
  63.         Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<TopicPartition, OffsetAndMetadata>();
  64.         consumer.poll(100);
  65.         Set<TopicPartition> assignment = consumer.assignment();
  66.         consumer.seekToEnd(assignment);//consumer.seekToBeginning(assignment);for(TopicPartition topicPartition : assignment){long position = consumer.position(topicPartition);
  67.             offsetMap.put(topicPartition, new OffsetAndMetadata(position));
  68.             consumer.commitSync(offsetMap);}}}
标签： kafka java 分布式
本文转载自: https://blog.csdn.net/jxlhljh/article/details/127286603
版权归原作者 冰之杍 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Kafka配置动态SASL_SCRAM认证”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？