Kafka Kerberos 安全认证

本主要介绍在 Kafka 中如何配置 Kerberos 认证，文中所使用到的软件版本： Java 1.8.0_261 、 Kafka_2.12-2.6.0 、 Kerberos 1.15.1 。

要使用 Kerberos 服务，需先安装 Kerberos，安装方法可参考： Kerberos 安装及使用 。

节点信息如下：

在安装 Kerberos 的机器上进入 kadmin（Kerberos 服务端上使用 kadmin.local ，安装了 Kerberos Client 的机器上可以使用 kadmin ），然后执行如下命令分别创建服务端和客户端的 keytab：

拷贝 krb5.conf 及 keytab 文件到所有安装 Kafka 的机器，把文件都放到 Kafka 的 config/kerveros 目录下(kerberos 目录需新建)。

复制 config/server.properties（复制动作能帮助我们在 认证/非认证 模式自由切换）重命名为 config/server-sasl.properties ，增加如下配置：

新建 kafka-server-jaas.conf 文件，该文件也放到 Kafka 的 config/kerveros 目录下：

复制 bin/kafka-server-start.sh 脚本重命名为 bin/kafka-server-start-sasl.sh ，倒数第二行增加如下配置：

该配置主要为了使用 bin/kafka-topics.sh、bin/kafka-console-consumer.sh、kafka-console-producer.sh 等命令。

新建 client.properties 文件，该文件也放到 Kafka 的 config/kerveros 目录下：

新建 kafka-client-jaas.conf 文件，该文件也放到 Kafka 的 config/kerveros 目录下：

复制 bin/kafka-topics.sh、kafka-console-producer.sh、bin/kafka-console-consumer.sh 脚本，分别重命名为 bin/kafka-topics-sasl.sh 、 kafka-console-producer-sasl.sh 、 bin/kafka-console-consumer-sasl.sh ，倒数第二行增加如下配置：

运行 bin/kafka-server-start-sasl.s 启动 Kafka 集群：

运行 bin/kafka-topics-sasl.sh 查看 topic 列表信息：

运行 kafka-console-producer-sasl.sh 生产消息：

运行 bin/kafka-console-consumer-sasl.sh 消费消息：

java 可以使用 JAAS 来进行 Kerberos 认证，需要 JAAS 配置文件、keytab 文件及 Kerberos 配置文件。

在 C:\Windows\System32\drivers\etc\hosts 文件中添加：

实际上，样例程序中的种种设置就是为了还原我们在上一步中改动 bin/kafka-console-consumer-sasl.sh + config/kerberos/client.properties 的过程，二者是对应的：

报错原因是 keytab 文件不正确，可能是 principle 不匹配或其它配置文件错误。本文中所犯错误是，Windows 的路径要写双斜线，否则配置文件无法正确读取。

报错的可能原因之一是未配置 hosts 文件。认证策略中肯能会使用缓存，配置了 hosts 文件后要重启 IDE。

Flink消费集成kerberos认证的kafka集群时，需要做一些配置才可以正常执行。

    Flink版本：1.8；kafka版本：2.0.1；Flink模式：Standalone

    //指示是否从 Kerberos ticket 缓存中读取

    security.kerberos.login.use-ticket-cache: false1

   //Kerberos 密钥表文件的绝对路径

    security.kerberos.login.keytab: /data/home/keytab/flink.keytab

   //认证主体名称

    security.kerberos.login.principal: flink@data.com

    //Kerberos登陆contexts

    security.kerberos.login.contexts: Client,KafkaClient

  val properties: Properties =new Properties()

  properties.setProperty("bootstrap.servers","broker:9092")

  properties.setProperty("group.id","testKafka")

  properties.setProperty("security.protocol","SASL_PLAINTEXT")

  properties.setProperty("sasl.mechanism","GSSAPI")

  properties.setProperty("sasl.kerberos.service.name","kafka")

  consumer =new   FlinkKafkaConsumer[String]("flink",new SimpleStringSchema(), properties)

    参数说明 ：security.protocol 

    运行参数可以配置为PLAINTEXT（可不配置）/SASL_PLAINTEXT/SSL/SASL_SSL四种协议，分别对应Fusion Insight Kafka集群的21005/21007/21008/21009端口。 如果配置了SASL，则必须配置sasl.kerberos.service.name为kafka，并在conf/flink-conf.yaml中配置security.kerberos.login相关配置项。如果配置了SSL，则必须配置ssl.truststore.location和ssl.truststore.password，前者表示truststore的位置，后者表示truststore密码。

web应用中使用到了kafka生产者，由于集群做了kerberos认证，

在Tomcat容器中运行时报错：

Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is /home/abc/apache-tomcat-7.0.82/temp/root.jaas.conf

检查了路径/home/abc/apache-tomcat-7.0.82/temp/root.jaas.conf下的文件，

KafkaClient实体是存在的。

应用中的user.keytab和krb5.conf配置文件也是存在的。

应用是基于tomcat的，用户通过在tomcat启动参数中增加-Djava.security.auth.login.config指定jaas.conf路径，解决了KafkaClient找不到的问题。

---