kafka ssl sasl_ssl 配置_kafka sasl_ssl

ssl参考资料：
https://www.cnblogs.com/enhance/p/11233164.html
http://kafka.apache.org/documentation/#security_ssl
http://www.javacoder.cn/?p=867
https://www.cnblogs.com/lt-blogs/p/7154345.html

 

 

一、kafka ssl 配置

#两台机器 kafka_2.11-2.2.0.tgz  centos7.7
kafka server  kafka61.aaa.cn
kafka client  kafka61.aaa.cn

ca服务器 kafka63.ccc.cn

 

 

server 端：

# 生成服务器keystore(密钥和证书)

keytool -keystore server.keystore.jks -alias kafka61.aaa.cn -validity 365 -genkey -keyalg RSA -storetype pkcs12 -storepass 123456 -keypass 123456 -genkey -dname "C=cn,ST=shanghai,L=shanghai,O=sx,OU=zh,CN=kafka61.aaa.cn"

#上传至ca：
scp  *.jks  kafka63.ccc.cn:/root/ssl/

 

 

client 端：

# 生成客户端keystore(密钥和证书)
keytool -keystore client.keystore.jks -alias kafka61.aaa.cn -validity 365 -genkey -keyalg RSA -storetype pkcs12 -storepass 123456 -keypass 123456 -genkey -dname "C=cn,ST=shanghai,L=shanghai,O=sx,OU=zh,CN=kafka61.aaa.cn"

#上传至ca：
scp  *.jks  kafka63.ccc.cn:/root/ssl/

 

 

ca端：

# 创建CA证书
mkdir ssl;
cd ssl;
echo 01 > serial.txt
touch index.txt

openssl req -new -x509 -keyout ca.key -out ca.crt -days 365 -passout pass:123456 -subj "/C=cn/ST=shanghai/L=shanghai/O=sx/OU=zh/CN=kafka63.ccc.cn/emailAddress=13438230@qq.com"

# 将CA证书导入到服务器truststore秘钥库
keytool -keystore server.truststore.jks -alias kafka63.ccc.cn  -import -file ca.crt -storepass 123456

# 将CA证书导入到客户端truststore秘钥库
keytool -keystore client.truststore.jks -alias kafka63.ccc.cn  -import -file ca.crt -storepass 123456

 

# 从秘钥库导出服务器、客户端证书

keytool -keystore server.keystore.jks -alias kafka61.aaa.cn -certreq -file cert-file -storepass 123456
keytool -keystore client.keystore.jks -alias kafka61.aaa.cn -certreq -file client-cert-file -storepass 123456

# 用CA证书给服务器、客户端证书签名

openssl x509 -req -CA ca.crt -CAkey ca.key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:123456
openssl x509 -req -CA ca.crt -CAkey ca.key -in client-cert-file -out client-cert-signed -days 365 -CAcreateserial -passin pass:123456

# 将CA证书导入服务器keystore秘钥库

keytool -keystore server.keystore.jks -alias kafka63.ccc.cn  -import -file ca.crt -storepass 123456
keytool -keystore client.keystore.jks -alias kafka63.ccc.cn  -import -file ca.crt -storepass 123456

# 将已签名的服务器证书导入keystore秘钥库

keytool -keystore server.keystore.jks -alias kafka61.aaa.cn -import -file cert-signed -storepass 123456
keytool -keystore client.keystore.jks -alias kafka61.aaa.cn -import -file client-cert-signed -storepass 123456

#上传秘钥库到服务器、客户端
scp  *.jks  kafka61.aaa.cn:/home/kafka/kafka/

配置kafka服务器
    cat config/zookeeper.properties
        dataDir=/home/kafka/kafka/zk/data
        dataLogDir=/home/kafka/kafka/zk/logs
    
    cat config/server.properties
        listeners=PLAINTEXT://kafka61.aaa.cn:9092,SSL://kafka61.aaa.cn:9093
        advertised.listeners=PLAINTEXT://kafka61.aaa.cn:9092,SSL://kafka61.aaa.cn:9093    
        ssl.keystore.location=/home/kafka/kafka/server.keystore.jks
        ssl.keystore.password=123456
        ssl.key.password=123456
        ssl.truststore.location=/home/kafka/kafka/server.truststore.jks
        ssl.truststore.password=123456
        security.inter.broker.protocol=SSL
        ssl.client.auth=requested
    
配置kafka客户端
    cat clientssl.conf
        security.protocol=SSL
        ssl.truststore.location=/home/kafka/kafka/client.truststore.jks
        ssl.truststore.password=123456
        ssl.keystore.location=/home/kafka/kafka/client.keystore.jks
        ssl.keystore.password=123456
        ssl.key.password=123456

启动
    bin/zookeeper-server-start.sh config/zookeeper.properties
    bin/kafka-server-start.sh config/server.properties
验证ssl
    openssl s_client -debug -connect kafka61.aaa.cn:9093 -tls1
        
        Verify return code: 19 (self signed certificate in certificate chain)
        显示一堆秘钥，有以上语句，不影响正常使用
创建topic
    bin/kafka-topics.sh --zookeeper kafka61.aaa.cn:2181 --create --topic test --partitions 1 --replication-factor 1
生产者
    bin/kafka-console-producer.sh --broker-list kafka61.aaa.cn:9093 --topic test --producer.config clientssl.conf   
消费者
    bin/kafka-console-consumer.sh --bootstrap-server kafka61.aaa.cn:9093 --topic test --consumer.config clientssl.conf

 

 

 

 

 

 

 

sasl_ssl参考资料：
https://docs.vmware.com/en/VMware-Smart-Assurance/10.1.0/sa-ui-installation-config-guide-10.1.0/GUID-DF659094-60D3-4E1B-8D63-3DE3ED8B0EDF.html
https://www.orchome.com/1946
https://blog.csdn.net/ahzsg1314/article/details/54140909

二、sasl_ssl 配置

 

zookeeper配置
    cat config/kafka_zoo_jaas.conf
        ZKServer{
                org.apache.kafka.common.security.plain.PlainLoginModule required
                        username="admin"
                        password="admin"
                        user_admin="admin";
        };
        
        ZKClient {
            org.apache.zookeeper.server.auth.PlainLoginModule required
            username="admin"
            password="admin";
        };
        
    cat  config/zookeeper.properties
        authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
        requireClientAuthScheme=sasl
        jaasLoginRenew=3600000

    cat bin/zookeeper-server-start.sh
        export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/kafka/kafka/config/kafka_zoo_jaas.conf  -Dzookeeper.sasl.serverconfig=ZKServer -Dzookeeper.sasl.clientconfig=ZKClient"
    
    启动
        bin/zookeeper-server-start.sh config/zookeeper.properties
        
kafka server 配置
    cat config/kafka_server_jaas.conf
        KafkaServer {
                org.apache.kafka.common.security.scram.ScramLoginModule required
                        username="admin"
                        password="admin"
                        user_admin="admin"
                        user_test="test";
        };        
        ZKClient {
        org.apache.zookeeper.server.auth.PlainLoginModule required
        username="admin"
        password="admin";
        };        
        KafkaClient {  
        org.apache.kafka.common.security.scram.ScramLoginModule required  
            username="admin"  
            password="admin";  
        };
        Client {
                org.apache.kafka.common.security.scram.ScramLoginModule required
                        username="admin"
                        password="admin";
        };

     cat config/server.properties
        listeners=SASL_SSL://kafka61.aaa.cn:9093
        advertised.listeners=SASL_SSL://kafka61.aaa.cn:9093
        security.inter.broker.protocol=SASL_SSL
        sasl.enabled.mechanisms=SCRAM-SHA-512
        sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512        
        ssl.endpoint.identification.algorithm=HTTPS        
        authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
        allow.everyone.if.no.acl.found=true
        delete.topic.enable=true
        auto.create.topics.enable=false
        ssl.secure.random.implementation=SHA1PRNG
        
    cat bin/kafka-server-start.sh         //注意，声明变量，放在执行启动语句之前
        export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/kafka/kafka/config/kafka_server_jaas.conf -Dzookeeper.sasl.clientconfig=ZKClient "
        
    创建SCRAM证书(存储于zk)
        bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name admin
    查看证书
        bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name admin
        
    启动
        bin/kafka-server-start.sh config/server.properties
    
    
kafka producer 配置
    cat config/kafka_client_jaas.conf
        KafkaClient {  
        org.apache.kafka.common.security.scram.ScramLoginModule required
            username="admin"  
            password="admin";  
        };
        Client {  
        org.apache.kafka.common.security.scram.ScramLoginModule required
            username="admin"  
            password="admin";  
        };

    cat config/producer.properties
        security.protocol=SASL_SSL
        sasl.mechanism=SCRAM-SHA-512
        
    cat clientssl.conf
        security.protocol=SASL_SSL
        sasl.mechanism=SCRAM-SHA-512
    
    cat bin/kafka-console-producer.sh
        export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/kafka/kafka/config/kafka_client_jaas.conf"
        
    启动    
        bin/kafka-console-producer.sh --broker-list kafka61.aaa.cn:9093 --topic test --producer.config clientssl.conf
        
    
kafka consumer 配置
    cat config/consumer.properties
        security.protocol=SASL_SSL
        sasl.mechanism=SCRAM-SHA-512
        #sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin";
    
    cat config/consumer.properties
        export KAFKA_OPTS=" -Djava.security.auth.login.config=/home/kafka/kafka/config/kafka_client_jaas.conf"
        
    授权zk
        bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --allow-principal User:admin  --consumer --topic=test --group=*  --add
    
    查看zk
        bin/kafka-acls.sh --list --authorizer-properties zookeeper.connect=localhost:2181

        
    启动
        bin/kafka-console-consumer.sh --bootstrap-server kafka61.aaa.cn:9093 --topic test --consumer.config clientssl.conf