安全概述

Kafka目前支持以下的安全措施，可以组合使用也可以分开使用。

1. broker使用SSL或SASL(Kerberos)，验证客户端（生产者或消费者）、其他broker或工具的连接。支持以下的SASL机制：

   • SASL/GSSAPI (Kerberos) - 从0.9.0.0版本开始

   • SASL/PLAIN - 从0.10.0.0版本开始

   • SASL/SCRAM-SHA-256 和 SASL/SCRAM-SHA-512 - 从0.10.2.0版本开始

   • SASL/OAUTHBEARER - 从2.0版本开始

2. 从broker连接到Zookeeper的身份验证。

3. broker和client之间的数据传输，broker之间，或使用SSL的broker和工具之间的数据加密（注意，当SSL时，性能会降低，其幅度取决于CPU类型和JVM）。

4. client的read/write操作验证。

5. 验证是插拔的，支持外部认证服务集成。

值得注意的是，安全是可选的。支持非安全集群，以及经过身份验证，未认证，加密和未加密客户端的组合。下面的指南将介绍如何配置和使用client和broker的安全特性。

kafka使用SASL/PLAIN认证

SASL/PLAIN是一种简单的用户名/密码的认证机制，通常与TLS加密一起使用，以实现安全的认证。Kafka支持SASL/PLAIN的默认实现，可作为生产者的扩展使用。

1. 配置Kafka Brokers

   • 新建kafka_server_jaas.conf。

     在每个Kafka broker的config目录下添加一个类似于下面的修改后的JAAS文件

     vi config/kafka_server_jaas.conf KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret" user_admin="admin-secret" user_alice="alice-secret"; };

     此配置定义了2个用户（admin 和 alice）。 在KafkaServer中，username和password是broker用于初始化连接到其他的broker，在这个例子中，admin是broker之间通信的用户。user_userName定义了连接到broker的所有用户的密码，broker使用这些来验证所有客户端的连接，包括来自其他的broker的连接。

   • 在server.properties中配置SASL端口和SASL机制

     ~$ vi server.properties listeners=SASL_SSL://host.name:port security.inter.broker.protocol=SASL_SSL sasl.mechanism.inter.broker.protocol=PLAIN sasl.enabled.mechanisms=PLAIN

2. 配置kafka客户端

   • 为producer.properties或consumer.properties中的每个客户端配置JAAS。

     登录模块展示了客户端如何连接Broker的（和生产者和消费者一样）

     ~$ vi config/kafka_client_jaas.conf KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret"; };

     • 客户端选择用户名和密码为客户端配置连接的用户。 在此示例中，客户端以用户alice连接到broker。也可以通过在sasl.jaas.config中指定不同的用户名和密码，JVM中的不同客户端可以根据不同的用户来进行连接。
     • 客户端的JAAS配置可以指定为类似于这里描述的broker作为JVM参数。客户端使用的命名为KafkaClient。 此选项仅允许来自JVM的所有客户端连接中的一个用户。

   • 在producer.properties或consumer.properties中配置以下属性

     security.protocol=SASL_SSL sasl.mechanism=PLAIN ssl.truststore.location=/var/private/ssl/client.truststore.jks ssl.truststore.password=test1234

3. 在生产者中使用SASL/PLAIN

   • SASL/PLAIN应仅用SSL作为传输层，以确保在没有加密的情况下不会在线上明文传输。
   • Kafka中SASL / PLAIN的默认实现在JAAS配置文件中指定用户名和密码，如下所示。 从Kafka 2.0版开始，您可以通过使用配置sasl.server.callback.handler.class和sasl.client.callback.handler.class配置自己的回调处理程序来从外部源获取用户名和密码，从而避免在磁盘上存储明文密码。
   • 在生产系统中，外部认证服务器可以实现密码认证。从Kafka 2.0版开始，可以通过配置sasl.server.callback.handler.class使用外部身份验证服务器进行密码验证的自己的回调处理程序。

4. 实战Kafka SASL/PLAIN认证

   • 启动kafka

     export KAFKA_OPTS='-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf' bin/kafka-server-start.sh config/server.properties

   • kafka消费者和生产者

     export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf" bin/kafka-console-producer.sh --broker-list localhost:9093 --topic test --producer.config config/producer.properties export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf" bin/kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test --consumer.config config/consumer.properties

5. kafka 设置开机启动

~$ vi /etc/rc.d/init.d/kafka-plain
#!/bin/sh
#
# chkconfig: 2345 10 90 
# description: Kafka
#
# File : Kafka-plain
#
# Description: Starts and stops the Kafka server for SASL/PLAIN
#

source /etc/rc.d/init.d/functions

SERVER_HOME=/usr/local/server
KAFKA_HOME=${SERVER_HOME}/kafka_2.12-2.3.1/bin
LOG_PATH=${SERVER_HOME}/log
KAFKA_USER=root
export JRE_HOME=/usr/java/jdk1.8.0_172/jre
export JAVA_HOME=/usr/java/jdk1.8.0_172
export CLASSPATH=$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib
export PATH=$JAVA_HOME/bin:$PATH

[ -e /etc/sysconfig/kafka ] && . /etc/sysconfig/kafka

# See how we were called.
case "$1" in

  start)
    echo -n "Starting Kafka for SASL/PLAIN:"
    export KAFKA_OPTS='-Djava.security.auth.login.config=$KAFKA_HOME/config/kafka_server_jaas.conf'
    /sbin/runuser -s /bin/sh $KAFKA_USER -c "nohup $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties > $LOG_PATH/server.out 2> $LOG_PATH/server.err &"
    echo " done."
    exit 0
    ;;

  stop)
    echo -n "Stopping Kafka for SASL/PLAIN: "
    /sbin/runuser -s /bin/sh $KAFKA_USER  -c "ps -ef | grep kafka.Kafka | grep -v grep | awk '{print \$2}' | xargs kill"
    echo " done."
    exit 0
    ;;
  hardstop)
    echo -n "Stopping (hard) Kafka for SASL/PLAIN: "
    /sbin/runuser -s /bin/sh $KAFKA_USER  -c "ps -ef | grep kafka.Kafka | grep -v grep | awk '{print \$2}' | xargs kill -9"
    echo " done."
    exit 0
    ;;

  status)
    c_pid=`ps -ef | grep kafka.Kafka | grep -v grep | awk '{print $2}'`
    if [ "$c_pid" = "" ] ; then
      echo "Stopped"
      exit 3
    else
      echo "Running $c_pid"
      exit 0
    fi
    ;;

  restart)
    stop
    start
    ;;

  *)
    echo "Usage: kafka-plain {start|stop|hardstop|status|restart}"
    exit 1
    ;;

esac
~$ chmod +x /etc/rc.d/init.d/kafka-plain
~$ chkconfig --add kafka-plain
~$ chkconfig kafka-plain on