1.概述
在Kafka0.9版本之前纹因，Kafka集群時沒有安全機制的喷屋。Kafka Client應用可以通過連接Zookeeper地址，例如zk1:2181:zk2:2181,zk3:2181等瞭恰。來獲取存儲在Zookeeper中的Kafka元數(shù)據(jù)信息屯曹。拿到Kafka Broker地址后，連接到Kafka集群惊畏，就可以操作集群上的所有主題了恶耽。由于沒有權(quán)限控制，集群核心的業(yè)務主題時存在風險的颜启。
2.內(nèi)容
2.2 身份認證
Kafka的認證范圍包含如下：
Client與Broker之間
Broker與Broker之間
Broker與Zookeeper之間
當前Kafka系統(tǒng)支持多種認證機制偷俭，如SSL、SASL（Kerberos缰盏、PLAIN涌萤、SCRAM）。
2.3?SSL認證流程
在Kafka系統(tǒng)中口猜，SSL協(xié)議作為認證機制默認是禁止的负溪，如果需要使用，可以手動啟動SSL機制济炎。安裝和配置SSL協(xié)議的步驟笙以，如下所示：
在每個Broker中Create一個Tmp密鑰庫
創(chuàng)建CA
給證書簽名
配置Server和Client
執(zhí)行腳本如下所示：
#!?/bin/bash
2#?1.Create?rsa
3keytool?-keystore?server.keystore.jks?-alias?dn1?-validity?365?-genkey?-keyalg?RSA
4#?2.Create?CA
5openssl?req?-new?-x509?-keyout?ca-key?-out?ca-cert?-days?365
6#?3.Import?client
7keytool?-keystore?client.truststore.jks?-alias?CAROOT?-import?-file?ca-cert
8#?4.Import?server
9keytool?-keystore?server.truststore.jks?-alias?CAROOT?-import?-file?ca-cert
10#?5.Export
11keytool?-keystore?server.keystore.jks?-alias?dn1?-certreq?-file?cert-file
12#?6.Signed
13openssl?x509?-req?-CA?ca-cert?-CAkey?ca-key?-in?cert-file?-out?cert-signed?-days?365?-CAcreateserial?-passin?pass:123456
14#?7.Import?ca-cert
15keytool?-keystore?server.keystore.jks?-alias?CARoot?-import?-file?ca-cert
16#?8.Import?cert-signed
17keytool?-keystore?server.keystore.jks?-alias?dn1?-import?-file?cert-signed
2.4 SASL認證流程
在Kafka系統(tǒng)中，SASL機制包含三種冻辩，它們分別是Kerberos、PLAIN拆祈、SCRAM恨闪。以PLAIN認證為示例，下面給大家介紹PLAIN認證流程放坏。
2.4.1?配置Server
首先咙咽，在$KAFKA_HOME/config目錄中新建一個文件，名為kafka_server_jaas.conf淤年，配置內(nèi)容如下：
1KafkaServer?{
2???org.apache.kafka.common.security.plain.PlainLoginModule?required
3???username="smartloli"
4???password="smartloli-secret"
5???user_admin="smartloli-secret";
6};
7
8Client?{
9???org.apache.kafka.common.security.plain.PlainLoginModule?required
10???username="smartloli"
11???password="smartloli-secret";
12};
然后在Kafka啟動腳本（kafka-server-start.sh）中添加配置文件路徑钧敞，設(shè)置內(nèi)容如下：
1[hadoop@dn1?bin]$?vi?kafka-server-start.sh
2
3#?Add?jaas?file
4export?KAFKA_OPTS="-Djava.security.auth.login.config=/data/soft/new/kafka/config/kafka_server_
接下來蜡豹，配置server.properties文件，內(nèi)容如下：
1#?Set?ip?&?port
2listeners=SASL_PLAINTEXT://dn1:9092
3advertised.listeners=SASL_PLAINTEXT://dn1:9092
4#?Set?protocol
5security.inter.broker.protocol=SASL_PLAINTEXT
6sasl.enabled.mechanisms=PLAIN
7sasl.mechanism.inter.broker.protocol=PLAIN
8
9#?Add?acl
10allow.everyone.if.no.acl.found=true
11auto.create.topics.enable=false
12delete.topic.enable=true
13advertised.host.name=dn1
14super.users=User:admin
15
16#?Add?class
17authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
2.4.2 配置Client
當Kafka Server端配置啟用了SASL/PLAIN溉苛，那么Client連接的時候需要配置認證信息镜廉，Client配置一個kafka_client_jaas.conf文件，內(nèi)容如下：
1KafkaClient?{
2??org.apache.kafka.common.security.plain.PlainLoginModule?required
3??username="admin"
4??password="admin-secret";
5};
然后愚战，在producer.properties和consumer.properties文件中設(shè)置認證協(xié)議娇唯，內(nèi)容如下：
1security.protocol=SASL_PLAINTEXT?
2sasl.mechanism=PLAIN
最后，在kafka-console-producer.sh腳本和kafka-console-producer.sh腳本中添加JAAS文件的路徑寂玲，內(nèi)容如下：
1#?For?example:?kafka-console-producer.sh
2hadoop@dn1?bin]$?vi?kafka-console-producer.sh
3
4#?Add?jaas?file
5export?KAFKA_OPTS="-Djava.security.auth.login.config=/data/soft/new/kafka\
6/config/kafka_client_jaas.conf"
2.5 ACL操作
在配置好SASL后塔插，啟動Zookeeper集群和Kafka集群之后，就可以使用kafka-acls.sh腳本來操作ACL機制拓哟。
（1）查看：在kafka-acls.sh腳本中傳入list參數(shù)來查看ACL授權(quán)新
1[hadoop@dn1?bin]$?kafka-acls.sh?--list?--authorizer-properties?zookeeper.connect=dn1:2181
（2）創(chuàng)建：創(chuàng)建待授權(quán)主題之前想许，在kafka-acls.sh腳本中指定JAAS文件路徑，然后在執(zhí)行創(chuàng)建操作
1[hadoop@dn1?bin]$?kafka-topics.sh?--create?--zookeeper?dn1:2181?--replication-factor?1?--partitions?1?--topic?kafka_acl_topic
（3）生產(chǎn)者授權(quán)：對生產(chǎn)者執(zhí)行授權(quán)操作
1[hadoop@dn1?~]$?kafka-acls.sh?--authorizer?kafka.security.auth.SimpleAclAuthorizer?--authorizer-properties?zookeeper.connect=dn1:2181?--add?--allow-principalUser:producer?--operation?Write?--topic?kafka_acl_topic
（4）消費者授權(quán)：對生產(chǎn)者執(zhí)行授權(quán)后断序，通過消費者來進行驗證
1[hadoop@dn1?~]$?kafka-acls.sh?--authorizer?kafka.security.auth.SimpleAclAuthorizer?--authorizer-properties?zookeeper.connect=dn1:2181?--add?--allow-principalUser:consumer?--operation?Read?--topic?kafka_acl_topic
（5）刪除：通過remove參數(shù)來回收相關(guān)權(quán)限
1[hadoop@dn1?bin]$?kafka-acls.sh?--authorizer-properties?zookeeper.connect=dn1:2181?--remove?--allow-principal?User:producer?--operation?Write?--topic?kafka_acl_topic3
3.總結(jié)
在處理一些核心的業(yè)務數(shù)據(jù)時流纹，Kafka的ACL機制還是非常重要的，對核心業(yè)務主題進行權(quán)限管控逢倍，能夠避免不必要的風險捧颅。
歡迎工作一到五年的Java工程師朋友們加入Java架構(gòu)開發(fā)： 854393687
群內(nèi)提供免費的Java架構(gòu)學習資料（里面有高可用、高并發(fā)较雕、高性能及分布式碉哑、Jvm性能調(diào)優(yōu)、Spring源碼亮蒋，MyBatis扣典，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多個知識點的架構(gòu)資料）合理利用自己每一分每一秒的時間來學習提升自己，不要再用"沒有時間“來掩飾自己思想上的懶惰慎玖！趁年輕贮尖，使勁拼，給未來的自己一個交代趁怔！
