實戰Kafka ACL機制

1.概述

　　在Kafka0.9版本之前，Kafka叢集時沒有安全機制的。Kafka Client應用可以通過連接配接Zookeeper位址，例如zk1:2181:zk2:2181,zk3:2181等。來擷取存儲在Zookeeper中的Kafka中繼資料資訊。拿到Kafka Broker位址後，連接配接到Kafka叢集，就可以操作叢集上的所有主題了。由于沒有權限控制，叢集核心的業務主題時存在風險的。

2.内容

2.2 身份認證

　　Kafka的認證範圍包含如下：

• Client與Broker之間
• Broker與Broker之間
• Broker與Zookeeper之間

　　目前Kafka系統支援多種認證機制，如SSL、SASL（Kerberos、PLAIN、SCRAM）。

2.3  SSL認證流程

　　在Kafka系統中，SSL協定作為認證機制預設是禁止的，如果需要使用，可以手動啟動SSL機制。安裝和配置SSL協定的步驟，如下所示：

1. 在每個Broker中Create一個Tmp密鑰庫
2. 建立CA
3. 給證書簽名
4. 配置Server和Client

　　執行腳本如下所示：

#! /bin/bash

# 1.Create rsa
keytool -keystore server.keystore.jks -alias dn1 -validity 365 -genkey -keyalg RSA
# 2.Create CA
openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
# 3.Import client
keytool -keystore client.truststore.jks -alias CAROOT -import -file ca-cert
# 4.Import server
keytool -keystore server.truststore.jks -alias CAROOT -import -file ca-cert
# 5.Export
keytool -keystore server.keystore.jks -alias dn1 -certreq -file cert-file
# 6.Signed
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:123456
# 7.Import ca-cert
keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
# 8.Import cert-signed
keytool -keystore server.keystore.jks -alias dn1 -import -file cert-signed      

2.4 SASL認證流程

　　在Kafka系統中，SASL機制包含三種，它們分别是Kerberos、PLAIN、SCRAM。以PLAIN認證為示例，下面給大家介紹PLAIN認證流程。

2.4.1  配置Server

　　首先，在$KAFKA_HOME/config目錄中建立一個檔案，名為kafka_server_jaas.conf，配置内容如下：

KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="smartloli"
   password="smartloli-secret"
   user_admin="smartloli-secret";
};

Client {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="smartloli"
   password="smartloli-secret";
};      

　　然後在Kafka啟動腳本（kafka-server-start.sh）中添加配置檔案路徑，設定内容如下：

[hadoop@dn1 bin]$ vi kafka-server-start.sh

# Add jaas file
export KAFKA_OPTS="-Djava.security.auth.login.config=/data/soft/new/kafka/config/kafka_server_jaas.conf"
      

　　接下來，配置server.properties檔案，内容如下：

# Set ip & port
listeners=SASL_PLAINTEXT://dn1:9092
advertised.listeners=SASL_PLAINTEXT://dn1:9092
# Set protocol
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

# Add acl
allow.everyone.if.no.acl.found=true
auto.create.topics.enable=false
delete.topic.enable=true
advertised.host.name=dn1
super.users=User:admin

# Add class
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer      

2.4.2 配置Client

　　當Kafka Server端配置啟用了SASL/PLAIN，那麼Client連接配接的時候需要配置認證資訊，Client配置一個kafka_client_jaas.conf檔案，内容如下：

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="admin"
  password="admin-secret";
};      

　　然後，在producer.properties和consumer.properties檔案中設定認證協定，内容如下：

security.protocol=SASL_PLAINTEXT 
sasl.mechanism=PLAIN      

　　最後，在kafka-console-producer.sh腳本和kafka-console-producer.sh腳本中添加JAAS檔案的路徑，内容如下：

# For example: kafka-console-producer.sh
hadoop@dn1 bin]$ vi kafka-console-producer.sh

# Add jaas file
export KAFKA_OPTS="-Djava.security.auth.login.config=/data/soft/new/kafka\
/config/kafka_client_jaas.conf"
      

2.5 ACL操作

　　在配置好SASL後，啟動Zookeeper叢集和Kafka叢集之後，就可以使用kafka-acls.sh腳本來操作ACL機制。

　　（1）檢視：在kafka-acls.sh腳本中傳入list參數來檢視ACL授權新

[hadoop@dn1 bin]$ kafka-acls.sh --list --authorizer-properties zookeeper.connect=dn1:2181      

　　（2）建立：建立待授權主題之前，在kafka-acls.sh腳本中指定JAAS檔案路徑，然後在執行建立操作

[hadoop@dn1 bin]$ kafka-topics.sh --create --zookeeper dn1:2181 --replication-factor 1 --partitions 1 --topic kafka_acl_topic      

　　（3）生産者授權：對生産者執行授權操作

[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）消費者授權：對生産者執行授權後，通過消費者來進行驗證

[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參數來回收相關權限

[hadoop@dn1 bin]$ kafka-acls.sh --authorizer-properties zookeeper.connect=dn1:2181 --remove --allow-principal User:producer --operation Write --topic kafka_acl_topic3      

3.總結

　　在處理一些核心的業務資料時，Kafka的ACL機制還是非常重要的，對核心業務主題進行權限管控，能夠避免不必要的風險。

4.結束語

　　這篇部落格就和大家分享到這裡，如果大家在研究學習的過程當中有什麼問題，可以加群進行讨論或發送郵件給我，我會盡我所能為您解答，與君共勉！

聯系方式：

郵箱：[email protected]

Twitter：

https://twitter.com/smartloli

QQ群（Hadoop - 交流社群1）：

424769183

溫馨提示：請大家加群的時候寫上加群理由（姓名＋公司/學校），友善管理者稽核，謝謝！

熱愛生活，享受程式設計，與君共勉！

作者：哥不是小蘿莉 ［ 關于我 ］［ 犒賞 ］

出處： http://www.cnblogs.com/smartloli/

轉載請注明出處，謝謝合作！