kafka SASL/PLAIN 認證及 ACL 權限控制

一、Zookeeper 配置 SASL/PLAIN 認證（每個zookeeper節點都要做）

1.1 在 zookeeper 的 conf 目錄下，創建 zk_server_jaas.conf 文件，內容如下

Server {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="admin"user_kafka="kafka";
};

username="admin" 是 zookeeper 實例之間通信的用戶；user_kafka="kafka" 是 kafka broker 與 zookeeper 連接的時候的認證用戶，密碼為=后面的值；

1.2 修改 zoo.cfg ，添加以下內容

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

1.3 因為認證的時候用到包org.apache.kafka.common.security.plain.PlainLoginModule, 這個是 kafka-client.jar 里面，所有需要將相應的 jar 包拷貝到 zookeeper 安裝根目錄的 lib 目錄下, 大概要 copy 以下這些 jar 包

mv /opt/module/kafka_2.12-2.4.1/libs/kafka-clients-2.4.1.jar /opt/module/zookeeper/lib/
mv /opt/module/kafka_2.12-2.4.1/libs/lz4-java-1.6.0.jar /opt/module/zookeeper/lib/
mv /opt/module/kafka_2.12-2.4.1/libs/osgi-resource-locator-1.0.1.jar /opt/module/zookeeper/lib/
mv /opt/module/kafka_2.12-2.4.1/libs/slf4j-api-1.7.28.jar /opt/module/zookeeper/lib/
mv /opt/module/kafka_2.12-2.4.1/libs/snappy-java-1.1.7.3.jar /opt/module/zookeeper/lib/

1.4 修改 zookeeper 啟動命令參數，在文件末尾追加以下內容

SERVER_JVMFLAGS="-Djava.security.auth.login.config=/opt/module/zookeeper/conf/zk_server_jaas.conf"

1.5 重啟 zookeeper 服務。

二、Kafka Server 配置 SASL/PLAIN 認證（每個kafka節點都要做）

2.1 修改 kafka server.properties 文件，添加以下內容（各個node修改為自己的hostname）

listeners=SASL_PLAINTEXT://node2:9092
advertised.listeners=SASL_PLAINTEXT://node2:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
# 配置ACL入口類
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer  
# default false | true to accept all the users to use it.
# allow.everyone.if.no.acl.found=true 
# 設置本例中admin為超級用戶
super.users=User:admin

2.2 在kafka config 目錄下創建 kafka_server_jass.conf 文件，內容如下

KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="admin"user_admin="admin"user_consumer="consumer"
};Client {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="kafka"password="kafka";
};

KafkaServer 段里面配置了 broker 之間的認證配置以及 client 和 broker 之間的認證配置

KafkaServer.username, KafkaServer.password 用于broker之間的相互認證

KafkaServer.user_admin和KafkaServer.user_consumer 用于 client 和broker 之間的認證, 下面我們 client 里面都用用戶 consumer 進行認證

Client 段里面定義 username 和 password 用于 broker 與 zookeeper 連接的認證；

2.3 修改 kafka 啟動命令腳本工具 kafka-server-start.sh，添加 java.security.auth.login.config 環境變量

將最后一句

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

修改為

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

2.4 修改完所有節點后，重啟 kafka 服務，查看啟動日志。

三、 Kafka client 的認證配置

3.1 在 Kafka 的 config 目錄下創建 kafka_client_jaas.conf 文件，內容如下

KafkaClient {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="consumer"password="consumer";
};

3.2 修改 kafka 的 consumer.properties 和 producer.properties，添加以下內容

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

3.3 修改 producer 啟動腳本參數，修改 kafka-console-producer.sh 文件

將最后一行

exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"

修改為

exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/opt/module/kafka_2.12-2.4.1/config/kafka_client_jaas.conf kafka.tools.ConsoleProducer "$@"

3.4 修改 consumer 啟動腳本參數，修改 kafka-console-consumer.sh 文件

將最后一行

exec $(dirname $0)/kafka-run-class.sh  kafka.tools.ConsoleConsumer "$@"

修改為

exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/opt/module/kafka_2.12-2.4.1/config/kafka_client_jaas.conf kafka.tools.ConsoleConsumer "$@"

四、ACL 配置

4.1 因為我們有個超級用戶 admin，所以可以用 admin 做生產者，admin不需要做 ACL 配置

4.2 為 test topic 添加消費者用戶 consumer

kafka-acls.sh --authorizer-properties zookeeper.connect=node2:2181/kafka241 --add --allow-principal User:consumer --consumer --topic test --group test_group

4.3 測試

生產者測試

kafka-console-producer.sh --broker-list 
node2 --topic test --producer.config /opt/module/kafka_2.12-2.4.1/config/producer.properties

消費者測試

kafka-console-consumer.sh --bootstrap-server 
node2:9092 --topic test --from-beginning --consumer.config 
/opt/module/kafka_2.12-2.4.1/config/consumer.properties

4.4 Flink 代碼消費測試

object TestKafkaSource {// Kafka Broker 地址private val BOOTSTRAP_SERVERS = "node2:9092"// SASL/PLAIN 認證配置private val SASL_USERNAME = "consumer" private val SASL_PASSWORD = "consumer"// Topic 名稱private val ALLOWED_TOPIC = "test"private val DENIED_TOPIC = "test1"def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentval props = new Propertiesprops.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS)props.put(ConsumerConfig.GROUP_ID_CONFIG, "test_group")props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName)props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName)props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")// SASL/PLAIN 配置props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT")props.put("sasl.mechanism", "PLAIN")props.put("sasl.jaas.config", String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";", SASL_USERNAME, SASL_PASSWORD))val dataSource: DataStreamSource[String] = env.addSource(new FlinkKafkaConsumer[String](ALLOWED_TOPIC , new SimpleStringSchema(), props))dataSource.print()env.execute()}
}

五、kafka-exporter 配置（使用了 kafka-exporter 監控 kafka 集群才做）

5.1 在kafka_server_jaas.conf 文件中加入 kafka-exporter 用戶信息，內容如下

KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="admin"user_admin="admin"user_consumer="consumer"user_kafka-exporter="kafka-exporter"
};Client {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="kafka"password="kafka";
};

5.2 分發到所有 kafka broker ，并重啟 kafka 服務

5.3 為 kafka-exporter 用戶配置權限

kafka-acls.sh --authorizer-properties zookeeper.connect=node2:2181/kafka241   --add --allow-principal User:kafka-exporter   --operation Describe --cluster
kafka-acls.sh --authorizer-properties zookeeper.connect=node2:2181/kafka241   --add --allow-principal User:kafka-exporter   --operation Describe --topic '*'
kafka-acls.sh --authorizer-properties zookeeper.connect=node2:2181/kafka241   --add --allow-principal User:kafka-exporter   --operation Describe --group '*'
kafka-acls.sh --authorizer-properties zookeeper.connect=node2:2181/kafka241   --add --allow-principal User:kafka-exporter   --operation Read --topic '*'
kafka-acls.sh --authorizer-properties zookeeper.connect=node2:2181/kafka241   --add --allow-principal User:kafka-exporter   --operation Read --topic __consumer_offsets

5.4 配置 kafka_export 服務啟動參數，在 kafka-export 安裝目錄下創建一個文件 start_kafka_export ，內容如下

#!/bin/bash
export KAFKA_SASL_USERNAME="kafka-exporter"
export KAFKA_SASL_PASSWORD="kafka-exporter"
export KAFKA_SASL_MECHANISM="PLAIN"/opt/module/kafka_exporter-1.7.0/kafka_exporter --kafka.server=node2:9092 --web.listen-address=:9308 --zookeeper.server=node2:2181/kafka241 \--sasl.enabled \--sasl.username=$KAFKA_SASL_USERNAME \--sasl.password=$KAFKA_SASL_PASSWORD \--sasl.mechanism=$KAFKA_SASL_MECHANISM

5.5 修改 kafka_export.service 文件，內容如下

[Unit]
Description=kafka_exporter
Wants=prometheus.service
After=network.target prometheus.service[Service]
Type=simple
User=bigdata
Group=bigdata
Restart=on-failure
WorkingDirectory=/opt/module/kafka_exporter-1.7.0
#ExecStart=/opt/module/kafka_exporter-1.7.0/kafka_exporter --kafka.server=node2:9092 --web.listen-address=:9308 --zookeeper.server=node2:2181
ExecStart=/opt/module/kafka_exporter-1.7.0/start_kafka_exporter
[Install]
WantedBy=multi-user.target

5.6 重新加載 systemctl 管理的服務