kakfa 安全機制

在0.9版本以後，kafka 增長了身份認證和權限控制 兩種安全機制。

身份認證：指客戶端與服務端鏈接進行身份認證；包括 客戶端與kafka代理之間、代理與代理之間、代理與zookeeper之間的鏈接認證；目前支持 SSL 、 SASL/Kerberos 、SASL/PLAIN 這3種認證機制；

權限控制：指對客戶端的讀寫操做進行權限控制。

一、利用 SASL/PLAIN 進行身份認證：

　　1）修改 server.properties 文件，開啓 SASL認證配置：

　　　　listeners=SASL_PLAINTEXT://0.0.0.0:9092　　#配置一個SASL端口

　　　　security.inter.broker.protocol=SASL_PLAINTEXT　　#設置代理之間通訊協議

　　　　sasl.enable.mechanisms=PLAIN　　#啓用SASL機制

　　　　sasl.mechanism.inter.broker.protocol=PLAIN　　#配置SASL機制

　　2）建立服務端JAAS文件，配置PLAIN：

　　　　在config目錄下建立一個名爲 kafka_server_jaas.conf 的文件，而後文件內容爲：　

　　　　　　KafkaServer {
　　　　　　　　org.apache.kafka.common.security.plain.PlainLoginModule required
　　　　　　　　username="kafka"
　　　　　　　　password="kafkapswd"
　　　　　　　　user_kafka="kafkapswd"
　　　　　　　　user_morton="mortonpswd"
　　　　　　}

　　　　username 和 password 指定該代理與集羣中其餘代理初始化鏈接的用戶名和密碼；

　　　　經過以 "user_"爲前綴後接用戶名的方式建立鏈接代理的用戶名和密碼。

　　3）建立和配置客戶端JAAS文件：

　　　　在config目錄下建立一個名爲 kafka_client_jaas.conf 的文件，而後文件內容爲：

　　　　　　KafkaClient {
　　　　　　　　org.apache.kafka.common.security.plain.PlainLoginModule required
　　　　　　　　username="morton"
　　　　　　　　password="mortonpswd"
　　　　　　}

　　4）將JAAS配置文件加入相應的配置文件：

　　　　修改 kafka-server-start.sh 腳本，在該腳本中引入服務端 JAAS 文件：　

　　　　　　if [ "x$KAFKA_OPTS" = "x" ]; then
　　　　　　　　export KAFKA_OPTS="-DJava.security.auth.login.config=../config/kafka_server_jaas.config"
　　　　　　fi

　　　　修改 kafak-console-producer.sh 和 kafka-console-consumer.sh 甲苯，在該腳本中引入客戶端 JAAS 文件

　　　　　　if [ "x$KAFKA_OPTS" = "x" ]; then
　　　　　　　　export KAFKA_OPTS="-DJava.security.auth.login.config=../config/kafka_client_jaas.config"
　　　　　　fi

　　5）啓動服務端、生產者 和 消費者：

　　　　kafka-server-start.sh --deamon ../config/server.properties

　　　　kafka-console-producer.sh --broker-list localhost:9092 --topic kafka-action --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN

　　　　kafka-console-consumer.sh --bootstrap-server localhost:9092 topic kafka-action --consumer.property security.protocol=SASL_PLAINTEXT --consumer.property sasl.mechanism=PLAIN

 

二、權限控制

　　kafka 提供了 kafka-acls.sh 腳本支持查詢（list）、添加（add）、移除（remove）這3類權限控制的操做；　　

　　要啓用 kafka ACL 權限控制，首先須要在 server.properties 文件中增長權限控制實現類的設置：

　　　　authorizer.class.name=kafka.security.auth.SimpleAuthorizer

　　當啓用了 Kafka ACL 權限控制後，默認條件下除了超級用戶以外，全部用戶均沒有任何權限；

　　在 server.properties 文件設置超級用戶的格式爲：super.users=User:user1;User:user2

　　在 server.properties 文件給全部用戶開啓權限的配置格式爲：allow.everyone.if.no.acl.found=true

　　因爲客戶端啓動都須要鏈接到 kafka，所以須要假如 Java.security.auth.login.config 環境變量設置，不然即便進行了受權，客戶端依然鏈接不上kafka。因爲客戶端執行的腳本都會調用 kafka-run-class.sh 腳本，所以在此腳本中加入 Java.security.suth.login.config 環境變量：

# Launch mode
KAFKA_SASL_OPTS='-DJava.security.suth.login.config=../config/kafka_server_jaas.conf'
if [ "x$DAEMON_MODE" = "xtrue" ]; then
  nohup $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_SASL_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
else
  exec $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_SASL_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@"
fi

　　查詢權限列表：

　　　　經過 kafka-acls.sh 腳本能夠查看某個主題（--topic）、某個消費組（--group）、集羣（--cluster）當前的權限列表；

　　　　kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --list --cluster

　　爲生產者受權：

　　　　kafka-acls.sh --add --authorizer-properties zookeeper.connect=localhost:2181 --allow-principal User:morton --producer --topic=*

　　　　參數 --allow-principal 指定給某個用戶受權（支持正則）；

　　　　參數 --deny-principal 指定某些用戶不具備權限（支持正則）；

　　　　參數 --producer 指定爲該用戶生產者角色受權，至關於經過 --operation 參數賦予 Write 和 Describe 權限；

　　　　參數 --topic 參數指定哪些主題具備賦予的權限（支持正則）；

　　　　參數 --allow-host 和 參數 --deny-host 指定容許和禁止生產者訪問IP；

　　爲消費者受權：

　　　　kafka-acls.sh -add --authorizer-properties zookeeper.connect=localhost:2181 --allow-principal User:morton --consumer --topic=*  --group acls-group

　　　　參數 --group 指定該消費者所屬的消費組；

　　刪除權限：

　　　　經過 remove 參數刪除相應的權限信息，能夠刪除某用戶對主題（--topic）、集羣（cluster）、消費組（--group）的操做權限；

　　　　參數 --operation 能夠指定刪除具體的權限；

　　　　參數 --force 能夠強制刪除；

　　　　kafka-acls.sh --authorizer-properties zookeeper-connect=localhost:2181 --remove --topic acls-foo --force