kafka集群单独开启kerberos详细步骤

Kafka 2.8新特性
可以不使用zk。使用自我管理的仲裁机制。
增加describe集群的API。
支持SASL、SSL双向TLS身份认证。
根据IP，限制broker connection的创建速率。

JDK 1.8
zookeeper（可以使用DP的zk，但需指定zk的namespace，避免跟DP的kafka冲突。）
kafka自带的zookeeper的启动：nohup ./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 2>&1 & @XY

一、kafka安装部署
tar -xzf kafka_2.13-2.8.0.tgz
$ cd kafka_2.13-2.8.0/config

1、配置 server.properties
broker.id=0 ##依次递增
listeners=PLAINTEXT://demo2:9092 ##访问端口
advertised.listeners=PLAINTEXT://demo2:9092
zookeeper.connect=localhost:2181
zookeeper.connect=demo2:2181/kafka-2.8 ##设置zk Chroot path。一个zk可以管理多个kafka集群。也称kafka的namespace

log.dirs=/data/kafka/kafka_2.12-2.8.0/kafka-logs
2、启动kafka
./bin/kafka-server-start.sh -daemon ./config/server.properties

3、测试kafka
./bin/kafka-topics.sh --create --zookeeper demo2:2181/kafka-2.8 --replication-factor 1 --partitions 1 --topic test
./bin/kafka-topics.sh --list --zookeeper demo2:2181/kafka-2.8

./bin/kafka-console-producer.sh --broker-list demo2:9092 --topic test
./bin/kafka-console-consumer.sh --bootstrap-server demo2:9092 --topic test --from-beginning

二、Kafka配置kerberos认证 https://blog.csdn.net/man_earth/article/details/87256449
1、准备工作：
(1)、安装无限制的jce
(2)、手动创建每个broker server的pricipals，并导出keytab，然后分发到每个broker server节点。
addprinc -randkey kafka/kafkahost1@EXAMPLE.COM
xst -norandkey -k /opt/kafkahost1/kafka.keytab kafka/kafkahost1@EXAMPLE.COM
xst -norandkey -k /data/kafka/kafka_2.12-2.8.0-krb/config/kafka.keytab kafka/demo2@HADOOP.COM

2、部署zk的kerberos认证（单独下载的zk）
下载apache-zookeeper-3.5.9-bin.tar.gz ,并解压
cp zoo_sample.cfg zoo.cfg

修改zoo.cfg
dataDir=/data/kafka/apache-zookeeper-3.5.9-bin/zk-data
clientPort=12181
admin.serverPort=38080
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000

创建zk_server_jaas.conf文件
在zookeeper/conf目录下，新建zk_server_jaas.conf文件（vi zk_server_jaas.conf）
Server {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
useTicketCache=false
keyTab=“/data/kafka/apache-zookeeper-3.5.9-bin/conf/zk.keytab”
principal=“zookeeper/demo2@HADOOP.COM”;
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
useTicketCache=false
keyTab=“/data/kafka/apache-zookeeper-3.5.9-bin/conf/zk.keytab”
principal=“zookeeper/demo2@HADOOP.COM”;
};

修改启动文件zkEnv.sh
启动文件添加如下内容：
export JVMFLAGS=" -Djava.security.auth.login.config=$ZOOCFGDIR/zk_server_jaas.conf "

启动zk: ./zkServer.sh start

3、配置kafka/config/server.properties
listeners=PLAINTEXT://demo2:9093,SASL_PLAINTEXT://demo2:9092
advertised.listeners=SASL_PLAINTEXT://demo2:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=kafka //这里的service.name要跟上面建立的principal相对应，kafka/kafkahost3@EXAMPLE.COM的principal服务名就是kafka

log.dirs=/data/kafka/kafka_2.12-2.8.0-krb/kafka-logs
zookeeper.connect=demo2:12181

3、ACL配置
kafka自带了一个开箱即用的ACL授权工具。它基于zk来存储所有的acl授权信息。可通过authorizer.class.name属性来配置：authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=true
super.users=User:kafka;User:@@@ ##跟kafka.keytab的pricipals的用户名一致。

#ACL ： topic 的用户权限配置，比如org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer
super.users=User:kafka;User:@@@ //acl相关，broker服务本身是采用kafka这个服务身份进行交互，只有配置成superuser才能获取集群内的metadata信息。
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer //acl相关，配置后才能启用acl
#设置为true，ACL机制为黑名单机制，只有黑名单中的用户无法访问
allow.everyone.if.no.acl.found=true
#默认为false，ACL机制为白名单机制，只有白名单中的用户可以访问。启动kafka时，需添加zookeeper.sasl.client.username=zookeeper，需要有该pricipals。
#allow.everyone.if.no.acl.found=false

3、配置kafka/config/kafka_jaas.conf
KafkaServer {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab=“/data/kafka/kafka_2.12-2.8.0-krb/config/kafka.keytab”
storeKey=true
useTicketCache=false
serviceName=“kafka”
principal=“kafka/demo2@HADOOP.COM”;
};
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useTicketCache=false
useKeyTab=true
seKeyTab=true
keyTab=“/data/kafka/kafka_2.12-2.8.0-krb/config/kafka.keytab”
principal=“kafka/demo2@HADOOP.COM”
renewTicket=true
serviceName=“kafka”;
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useTicketCache=false
useKeyTab=true
seKeyTab=true
keyTab=“/data/kafka/kafka_2.12-2.8.0-krb/config/zk.keytab”
principal=“zookeeper/demo2@HADOOP.COM”
renewTicket=true
serviceName=“zookeeper”;
};

4、修改bin目录下kafka-run-class.sh，增加kerberos启动参数：
export KAFKA_OPTS=“-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/data/kafka/kafka_2.12-2.8.0-krb/config/kafka_jaas.conf -Dsun.security.krb5.debug=true”

-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf
-Dzookeeper.sasl.client=Client #默认是Client，对应jaas中的Client
-Dzookeeper.sasl.client.username=zk ##默认是zookeeper，跟pricipals一致。

5、启动kafka
./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
./bin/kafka-server-start.sh -daemon ./config/server.properties

常见问题：/etc/krb5.conf中配置的default realm必须和使用的realm一致，否则必须先做kdc互信。
##注意kafka broker与kdc的时间同步。
查看错误日志：kafkaServer.out

6、测试创建topic：
./bin/kafka-topics.sh --create --zookeeper demo2:12181 --replication-factor 1 --partitions 1 --topic test
./bin/kafka-topics.sh --list --zookeeper demo2:12181

6、客户端使用kafka
启用kerberos后，部分kafka管理脚本需要增加额外的参数才能使用

首先建立配置文件client.properties:
security.protocol=SASL_PLAINTEXT
sasl.kerberos.service.name=kafka
sasl.mechanism=GSSAPI

所以新命令的使用方式为:
./bin/kafka-consumer-groups.sh --bootstrap-server demo2:9092 --list --command-config ./client.properties
./bin/kafka-console-producer.sh --broker-list demo2:9092 --topic test --producer.config ./client.properties
./bin/kafka-console-consumer.sh --bootstrap-server demo2:9092 --topic test --consumer.config ./client.properties

7、登录zk： ./bin/zookeeper-shell.sh demo2:12181
三、Kakfa配置SSL安全认证
http://kafka.apache.org/documentation/#security_ssl
https://docs.confluent.io/platform/current/kafka/authentication_ssl.html

单向SSL认证：仅仅是在客户端，校验kafka服务器端的合法性。ssl.client.auth=requested。只需拷贝服务器端的kafka.client.truststore.jks即可。
双向SSL认证：除了客户端去校验kafka服务器端的合法性，服务器端还要校验客户端的合法性。ssl.client.auth=required。需要拷贝服务器端的kafka.client.truststore.jks和kafka.server.keystore.jks。

(一)、SSL认证之生成证书
https://blog.csdn.net/justry_deng/article/details/88383081 中文配置
SSL的基础知识、kafka证书生成过程，参考《 SSL安全认证是什么？》

证书生成脚本参考：

集群内选择其中一台broker节点，执行以下命令：
#! /bin/bash

CLUSTER_NAME=kafak2.8-ssl
BASE_DIR=/data/kafka/kafka_2.12-2.8.0/ssl
export PASSWORD=“@@@@123”

KEY_STORE=“ $BASE_DIR/server/kafka.server.keystore.jks" CERT_OUTPUT_PATH="$ BASE_DIR/ca”
CERT_AUTH_FILE=“ $CERT_OUTPUT_PATH/ca-cert.pem" TRUST_STORE_PATH="$ BASE_DIR/trust”
CLUSTER_CERT_FILE=“ $BASE_DIR/server/cluster-$ {CLUSTER_NAME}-cert”

echo “Step1: Config env”
STORE_PASSWORD= $PASSWORD KEY_PASSWORD=$ PASSWORD
TRUST_KEY_PASSWORD= $PASSWORD TRUST_STORE_PASSWORD=$ PASSWORD
DAYS_VALID=3650
export Local_HOSTNAME=hostname -f
D_NAME=“CN=$Local_HOSTNAME, OU=asiainfo”

mkdir -p ${BASE_DIR}/{ca,server,client,trust}

echo “Step2: Create certificate to keystore.每个broker节点执行”
keytool -genkey -keystore $KEY_STORE -alias $CLUSTER_NAME -validity $DAYS_VALID -keyalg RSA -storetype pkcs12 -storepass $STORE_PASSWORD -keypass $KEY_PASSWORD -dname "$ D_NAME" -ext “SAN=DNS:$Local_HOSTNAME”

echo “Step3: Create CA（证书签发机构）的证书.只需在其中一个节点上执行即可，执行完成后，会生成了两个文件cat-key、ca-cert，然后将这两个文件分别拷贝到所有broker节点上，后面需要用到。”
openssl req -new -x509 -keyout $CERT_OUTPUT_PATH/ca-key -out "$ CERT_AUTH_FILE" -days “ $DAYS_VALID" -passin pass:"$ PASSWORD” -passout pass:“ $P A SS W OR D " - s u bj "/ C = CN / ST = b e ijin g / L = b e ijin g / O = a s iain f o / O U = a s iain f o / CN =$ Local_HOSTNAME”
#openssl req -x509 -config openssl-ca.cnf -newkey rsa:4096 -sha256 -nodes -out cacert.pem -outform PEM

echo “Step4: Import CA into kafka client truststore.每个broker节点执行”
keytool -keystore “ $TRUST_STORE_PATH/kafka.client.truststore.jks" -alias CARoot -import -file "$ CERT_AUTH_FILE” -storepass “ $TRUST_STORE_PASSWORD" -keypass "$ TRUST_KEY_PASS” -noprompt

echo “Step4: Import CA into kafka server truststore.每个broker节点执行”
keytool -keystore “ $TRUST_STORE_PATH/kafka.server.truststore.jks" -alias CARoot -import -file "$ CERT_AUTH_FILE” -storepass “ $TRUST_STORE_PASSWORD" -keypass "$ TRUST_KEY_PASS” -noprompt

echo “Step5: Export certificate from (Step2)keystore, prepare for signing.导出证书，准备签名.每个broker节点执行”
keytool -keystore “ $KEY_STORE" -alias "$ CLUSTER_NAME” -certreq -file “ $CLUSTER_CERT_FILE" -storepass "$ STORE_PASSWORD” -keypass “$KEY_PASSWORD” -noprompt

echo “Step6: Signing the certificate.证书签名.每个broker节点执行”
openssl x509 -req -CA “$CERT_AUTH_FILE” -CAkey $CERT_OUTPUT_PATH/ca-key -in "$ CLUSTER_CERT_FILE" -out “ ${CLUSTER_CERT_FILE}-signed" -days "$ DAYS_VALID” -CAcreateserial -passin pass:“$PASSWORD”

echo “Setp7: Import CA into keystore”
keytool -keystore “ $KEY_STORE" -alias CARoot -import -file "$ CERT_AUTH_FILE.每个broker节点执行” -storepass “ $STORE_PASSWORD" -keypass "$ KEY_PASSWORD” -noprompt

echo “Setp8: Import signed certificate into keystore.每个broker节点执行”
keytool -keystore “ $KEY_STORE" -alias "$ {CLUSTER_NAME}” -import -file “ ${CLUSTER_CERT_FILE}-signed" -storepass "$ STORE_PASSWORD” -keypass “$KEY_PASSWORD” -noprompt

echo “done,list keystore”
keytool -list -v -keystore $KEY_STORE -storepass ${KEY_PASSWORD}

注意：以上步骤，只有步骤3，生成CA认证证书时，在其中一台主机上执行即可。集群内的所有主机使用同一个CA证书签发机构。

集群内其他broker节点，执行以下命令：执行除了Step3: Create CA（证书签发机构)以外的其他的命令。

（二）、修改config/server.properties
注意：SSL配置最好写在配置文件的最上面，否者可能导致Kafka配置SSL失败。

1、如果配置了SSL认证，那么原来的port和advertised.listeners可以注释掉。也可以保留PLAINTEXT配置，另外增加SSL协议的端口吗？

listeners=PLAINTEXT://host.name:port,SSL://host.name:9093
advertised.listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093
#advertised.listeners=SSL://kafka-single:9095
ssl.keystore.location=/…/ssl/server/kafka.server.keystore.jks
ssl.keystore.password=@@@@123
ssl.key.password=@@@@123
ssl.truststore.location=/…/ssl/trust/kafka.server.truststore.jks
ssl.truststore.password=@@@@123##注意：ssl.truststore.password 在技术上是可选的，但强烈推荐设置。如果未设置密码，则仍可访问信任库，但会禁用完整性检查。
ssl.client.auth=required ##是否需要客户端身份验证。如果设置为required，是双向认证，服务器端会验证客户端身份。如果设置为requested，是单向认证，则没有keystore证书的客户端也可连接broker。不建议设置为reqeusted。
ssl.keystore.type=PKCS12
ssl.truststore.type=JKS

2、其他可选参数
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1 ##至少要列出broker端的协议之一。
ssl.cipher.suites= ##TLS or SSL
ssl.secure.random.implementation=SHA1PRNG

配置主机名验证。
kafka2.0.x开始，默认情况下客户端连接以及broker之间的连接启用了服务器的主机名验证，为防止中间人攻击。
将ssl.endpoint.identification.algorithm设置为了HTTPS，即:需要验证主机名。

如果不需要验证主机名，那么可以这么设置ssl.endpoint.identification.algorithm=即可。

ssl.endpoint.identification.algorithm=HTTPS

设置内部访问也用SSL，默认值为security.inter.broker.protocol=PLAINTEXT

security.inter.broker.protocol=SSL
注：如果设置的内部broker的通讯协议PLAINTEXT，那么监听PLAINTEXT的时候就需要作相应的配置listeners=PLAINTEXT://host.name:port,SSL://host.name:port。

(三)、重启kafka
./bin/kafka-server-start.sh -daemon ./config/server.ssl.properties

如果配置SSL之前，存在Kafka数据，那么建议重新换一个位置来存放数据；
如果确保之前的数据已经没什么用了，也可以直接删除之前的数据

(四)、本机测试
1、测试9093端口可以访问：
openssl s_client -debug -connect demo3:9093 -tls1
以下说明连接成功。

2、测试创建Topic
./bin/kafka-topics.sh --list --zookeeper demo3:2181/kafka-2.8
./bin/kafka-topics.sh --create --zookeeper demo3:2181/kafka-2.8 --replication-factor 2 --partitions 10 --topic test2

3、测试生产、消费数据
首先，创建consumer.properties
group.id=demo3-group
client.id=demo3
security.protocol=SSL
ssl.truststore.location=/data/kafka/kafka_2.12-2.8.0/ssl/trust/kafka.client.truststore.jks
ssl.truststore.password=@@@@123
##如果服务器端设置了：ssl.client.auth=required，则必须提供kafka.server.keystore.jks文件。如果ssl.client.auth=requested，则可以只提供kafka.client.truststore.jks文件。
ssl.keystore.password=@@@@123
ssl.keystore.location=/data/kafka/kafka_2.12-2.8.0/ssl/server/kafka.server.keystore.jks

然后生产、消费数据：
./bin/kafka-console-producer.sh --broker-list demo3:9093 --topic test --producer.config ./ssl/producer.properties
./bin/kafka-console-consumer.sh --bootstrap-server demo3:9093 --topic test --from-beginning --consumer.config ./ssl/consumer.properties

三、远程Kakfa客户端配置https://docs.confluent.io/platform/current/kafka/authentication_ssl.html#clients
任意要与broker进行交互的主机，都要提供安全配置信息。包括：副本、kafka客户端，java客户端。
如果直接在其他主机上执行，消费、生产数据的命令时，会报错Broker连接失败：Bootstrap broker demo3:9093 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)

1、将Kafka集群中的ssl/server/kafka.server.keystore.jks文件和ssl/server/kafka.client.truststore.jks文件，拷贝至客户端运行环境。
2、配置consumer.properties（参考以上配置）
#group.id=demo2-consumer-group
client.id=demo2

security.protocol=SSL
ssl.truststore.location=/data/kafka/kafka_2.12-2.8.0/demo3-ssl/trust/kafka.client.truststore.jks
ssl.truststore.password=@@@@123
##如果服务器端设置了：ssl.client.auth=required，则必须提供kafka.server.keystore.jks文件。如果ssl.client.auth=requested，则可以只提供kafka.client.truststore.jks文件。
ssl.keystore.password=@@@@123
ssl.keystore.location=/data/kafka/kafka_2.12-2.8.0/ssl/server/kafka.server.keystore.jks

3、执行消费命令
./bin/kafka-console-consumer.sh --bootstrap-server demo3:9093 --topic test --from-beginning --consumer.config ./ssl/consumer.properties

四、java访问使用SSL加密的kafka集群。
1、将Kafka集群中的ssl/server/kafka.server.keystore.jks文件和ssl/server/kafka.client.truststore.jks文件，拷贝至客户端
2、配置Properties类的参数
Properties props = new Properties();
props.put(“security.protocol”, “SSL”);
props.put(“ssl.truststore.location”, “E:\kafka\safe\client.truststore.jks”);
props.put(“ssl.truststore.password”, “test1234”);

props.put("ssl.keystore.location", "E:\\kafka\\safe\\client.keystore.jks");
props.put("ssl.keystore.password", "p2****02");
props.put("ssl.key.password", "p((((2");KafkaConsumer consumer=new KafkaConsumer(props);