Compartilhamento de tecnologia

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

Autor Comunidade Lewei (forum.lwops.cn) Xu Yuan
Ao construir sistemas distribuídos modernos, garantir a segurança da transmissão de dados é crucial. Apache Kafka e Zookeeper, como filas de mensagens distribuídas populares e serviços de coordenação, fornecem mecanismo de autenticação SSL (Secure Sockets Layer) para aumentar a segurança durante a transmissão de dados.
Este artigo apresentará detalhadamente todo o processo, desde a geração de um certificado SSL até a configuração do servidor e do cliente para garantir que os dados estejam totalmente protegidos durante a transmissão.

1. Configure a senha da conta Kafka:
1. Primeiro, você precisa modificar o arquivo de configuração kafka: vim /asop/kafka/kafka_2.11-2.1.0/config/server.properties

corretor.id=0
ouvintes=SASL_PLAINTEXT://:9092
anunciado.ouvintes=SASL_PLAINTEXT://10.176.31.137:9092
num.network.threads=3
num.io.threads=8
soquete.enviar.buffer.bytes=102400
soquete.receber.buffer.bytes=102400
soquete.request.max.bytes=104857600
log.dirs=/asop/kafka/logs
num.partições=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.fator=1
transação.estado.log.replicação.fator=1
transação.estado.log.min.isr=1
log.retenção.horas=168
log.segmento.bytes=1073741824
log.retenção.verificação.intervalo.ms=300000
zookeeper.conectar=localhost:2181
zookeeper.conexão.tempo limite.ms=6000
grupo.inicial.rebalanceamento.atraso.ms=0

#Protocolo de autenticação usado
security.inter.broker.protocol=SASL_PLAINTEXT
#SASLmecanismo
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#Class para completar a autenticação
autorizador.class.name=kafka.security.auth.SimpleAclAuthorizer
#Se nenhuma configuração de ACL (lista de controle de acesso) for encontrada, qualquer operação será permitida.
permitir.todos.se.nenhum.acl.encontrado=falso
#Você precisa habilitar a configuração de superadministrador e definir o usuário visitante como superadministrador.
super.users=Usuário:visitante

2. Em segundo lugar, crie um arquivo de verificação de login para o servidor. Você pode nomear o arquivo de acordo com sua preferência, como vim /asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf. segue

Servidor Kafka {
org.apache.kafka.common.security.plain.PlainLoginModule obrigatório
nome de usuário=“visitante”
senha=“qaz@123”
usuário_visitante=“qaz@123”;
};

3. Em seguida, modifique o diretório de instalação do kafka vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh e adicione variáveis ​​​​na parte superior do arquivo

exportar KAFKA_OPTS=" -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf"

4. Em seguida, crie um arquivo de verificação de login para o consumidor e o produtor. Você pode nomear o arquivo de acordo com sua preferência, como kafka_client_jaas.conf O conteúdo do arquivo é o seguinte (se for acesso ao programa, como acesso ao springboot. , não precisa ser configurado)
vim /asop/kafka/kafka_2.11-2.1.0/config/kafka_client_jaas.conf

Cliente Kafka {
org.apache.kafka.common.security.plain.PlainLoginModule obrigatório
nome de usuário=“visitante”
senha=“qaz@123”;
};

5. Adicione as seguintes configurações a consumer.properties e Producer.properties respectivamente:
vim /asop/kafka/kafka_2.11-2.1.0/config/consumer.properties
vim /asop/kafka/kafka_2.11-2.1.0/config/producer.properties

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

6. Modifique os diretórios de instalação do kafka bin/kafka-console-producer.sh e bin/kafka-console-consumer.sh e adicione variáveis ​​na parte superior dos arquivos.
vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-producer.sh
vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-consumer.sh

exportar KAFKA_OPTS=" -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_client_jaas.conf"
7. Inicie o zookeeper e o kafka respectivamente. Neste ponto, a configuração de verificação de login do usuário kafka do lado do servidor é concluída (primeiro feche o kafka e depois feche o zookeeper).

Desligar serviço kafka
/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-stop.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties

Iniciar serviço kafka
#/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties

Desligue o serviço zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh parar /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
Inicie o serviço zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh iniciar /asop/zk/zookeeper-3.4.13/conf/zoo.cfg

8. Crie e visualize temas

/asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-producer.sh --broker-list 10.176.31.137:9092 --topic cmdb --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN

receber mensagens

/asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-consumer.sh --bootstrap-server 10.176.31.137:9092 --topic cmdb --from-beginning --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=PLAIN

2. Configure a senha da conta ssal para zk e kafka:

1. Configuração do Zookeeper SASL
   1.1 Crie um novo arquivo zoo_jaas.conf
   Não há requisitos especiais para o nome e caminho do arquivo zoo_jaas.conf. Ele geralmente é colocado no diretório ${ZOOKEEPER_HOME}/conf vim /asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf.

Servidor {
org.apache.kafka.common.security.plain.PlainLoginModule obrigatório
nome de usuário=“admin”
senha=“admin@12”
usuário_kafka=“kafka@123”;
};

    Server.username、Server.password为 Zookeeper 内部通信的用户名和密码，因此保证每个 zk 节点该属性一致即可
    Server.user_xxx 中 xxx 为自定义用户名，用于 zkClient 连接所使用的用户名和密码，即为 kafka 创建的用户名
1
2

1.2 Configure o arquivo /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenovar=3600000
zookeeper.sasl.client=verdadeiro

Zookeeper.sasl.client é configurado como true para ativar a autenticação do cliente, caso contrário, o nome de usuário configurado em zoo_jaas.conf não funcionará.

1.3 Importar pacotes de dependências
Como a classe de verificação de permissão usada é: org.apache.kafka.common.security.plain.PlainLoginModule, os pacotes jar relacionados ao kafka são necessários. Crie uma nova pasta zk_sasl_lib da seguinte maneira: Copie os seguintes pacotes jar do diretório kafka/lib para. zookeeper lib e o diretório zk_sasl_lib recém-criado:

kafka-clientes-2.4.1.jar
lz4-java-1.6.0.jar
slf4j-api-1.7.28.jar
slf4j-log4j12-1.7.28.jar
snappy-java-1.1.7.3.jar

mkdir /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/kafka-clients-2.1.0.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/lz4-java-1.5.0.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-api-1.7.25.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-log4j12-1.7.25.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/snappy-java-1.1.7.2.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/kafka-clients-2.1.0.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/lz4-java-1.5.0.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-api-1.7.25.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-log4j12-1.7.25.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/snappy-java-1.1.7.2.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib

chmod 755 -R /asop/zk/zookeeper-3.4.13/zk_sasl_lib/
chmod 755 -R /asop/zk/zookeeper-3.4.13/zk_sasl_lib/

1.4 Modifique o arquivo zkEnv.sh/asop/zk/zookeeper-3.4.13/bin/zkEnv.sh
Antes da modificação: caso contrário, adicione-o diretamente

exportar SERVER_JVMFLAGS=“-Xmx${ZK_SERVER_HEAP}m $SERVER_JVMFLAGS”

Após modificação:

para jar em /asop/zk/zookeeper-3.4.13/zk_sasl_lib/*.jar;
fazer
CAMINHO DE CLASSE=“$simr:$“CAMINHO DE CLASSE”
feito

exportar SERVER_JVMFLAGS=" -Djava.security.auth.login.config=/asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf "

Basta reiniciar o serviço Zookeeper

Desligue o serviço zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh parar /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
Inicie o serviço zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh iniciar /asop/zk/zookeeper-3.4.13/conf/zoo.cfg

2. Kakfa configurar SASL
   2.1 Crie um novo arquivo kafka_server_jaas.conf
   Não há requisitos para o nome do arquivo kafka_server_jaas.conf e caminho de armazenamento. Ele geralmente é colocado no diretório ${KAFKA_HOME}/config/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf.

Servidor Kafka {
org.apache.kafka.common.security.plain.PlainLoginModule obrigatório
nome de usuário=“visitante”
senha=“qaz@123”
usuário_visitante=“qaz@123”;
};
Cliente{
org.apache.kafka.common.security.plain.PlainLoginModule obrigatório
nome de usuário=“kafka”
senha=“kafka@123”;
};

    KafkaServer.username、KafkaServer.password 为 broker 内部通信的用户名密码，同上
1

KafkaServer.user_xxx onde xxx deve ser consistente com o nome de usuário configurado em KafkaServer.username e a senha também deve ser consistente
KafkaServer.user_producer e KafkaServer.user_consumer se preparam para a ACL subsequente, de modo que consumidores e produtores usem contas diferentes, e as contas do consumidor só possam consumir dados e as contas do produtor só possam produzir dados.
Client.username e Client.password preenchem a senha da conta cadastrada no Zookeeper, que é utilizada para comunicação entre o corretor e o zookeeper (se o zookeeper não estiver configurado com SASL, pode ser ignorado. Se zookeeper.sasl.client for falso, pode também será ignorado. Basta incluí-lo. O log é o seguinte)

[2021-06-29 17:14:30,204] Falha na configuração WARN SASL: javax.security.auth.login.LoginException: Nenhuma seção de configuração JAAS chamada 'Client' foi encontrada no arquivo de configuração JAAS especificado: '/Users/wjun/env/kafka/config/kafka_server_jaas.conf'. Continuará a conexão com o servidor Zookeeper sem autenticação SASL, se o servidor Zookeeper permitir. (org.apache.zookeeper.ClientCnxn)

2.2 Modifique o arquivo server.properties
corretor.id=0
ouvintes=SASL_PLAINTEXT://:9092
anunciado.ouvintes=SASL_PLAINTEXT://192.168.157.198:9092
num.network.threads=3
num.io.threads=8
soquete.enviar.buffer.bytes=102400
soquete.receber.buffer.bytes=102400
soquete.request.max.bytes=104857600
log.dirs=/asop/kafka/logs
num.partições=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.fator=1
transação.estado.log.replicação.fator=1
transação.estado.log.min.isr=1
log.retenção.horas=168
log.segmento.bytes=1073741824
log.retenção.verificação.intervalo.ms=300000
tratador.conectar=127.0.0.1:2181
zookeeper.conexão.tempo limite.ms=6000
grupo.inicial.rebalanceamento.atraso.ms=0

#Protocolo de autenticação usado
security.inter.broker.protocol=SASL_PLAINTEXT
#SASLmecanismo
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#Class para completar a autenticação
autorizador.class.name=kafka.security.auth.SimpleAclAuthorizer
#Se nenhuma configuração de ACL (lista de controle de acesso) for encontrada, qualquer operação será permitida.
permitir.todos.se.nenhum.acl.encontrado=falso
#Você precisa habilitar a configuração de superadministrador e definir o usuário visitante como superadministrador.
super.users=Usuário:visitante

Onde localhost precisa ser alterado para um endereço IP

super.users configura o superusuário, que não é afetado pelas configurações subsequentes da ACL.

2.3 Modifique o script de inicialização
Modifique o arquivo kafka-server-start.sh para carregá-lo no arquivo kafka_server_jaas.conf/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh

antes de consertar:

se [ “x$KAFKA_HEAP_OPTS” = “x” ]; então
exportar KAFKA_HEAP_OPTS=“-Xmx1G -Xms1G”
sim

Após modificação:
(Adicione esta linha à primeira linha primeiro. Se você já a possui, não precisa adicioná-la.) export KAFKA_OPTS=" -Djava.security.auth.login.config=/asop/kafka/kafka_2.11- 2.1.0/config/kafka_server_jaas.conf "
se [ “x$KAFKA_HEAP_OPTS” = “x” ]; então
exportar KAFKA_HEAP_OPTS=“-Xmx1G -Xms1G -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf”
sim

Definir regras de ACL para o zookeeper
/asop/zk/zookeeper-3.4.13/bin/zkCli.sh #Entra no modo de linha de comando do zk

addauth digest admin:admin@12 #Alternar usuário de login (o superadministrador está no arquivo de configuração zk/asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf)

setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka@123:cdrwa #(Defina o IP e a senha da conta do usuário que pode ser logado. admin é o administrador definido no arquivo de configuração zk acima, e o usuário Kafka é /asop A conexão kafka zk do usuário definida no arquivo /kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf (em Cliente))

addauth digest kafka:kafka@123 #Mude para o usuário kafka e defina acl novamente
definirAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka@123:cdrwa

Nota: Se você deseja adicionar um IP à lista de permissões ou o usuário deve adicioná-lo na base original, caso contrário ele será sobrescrito.
setAcl / ip:127.0.0.1:cdrwa,autenticação:kafka:kafka@123:cdrwa,autenticação:admin:admin@12:cdrwa,ip:1.1.1.1

Precisa restaurar permissões, execute sem definir acl
setAcl / mundo:qualquer um:cdrwa

Basta reiniciar o serviço kafka

Desligar serviço kafka
#/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-stop.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties

Iniciar serviço kafka
#/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties

Neste ponto, a configuração da autenticação SSL do kafka e do zookeeper está concluída. Para obter mais habilidades de operação e manutenção, preste atenção à comunidade Lewei. Para mais dúvidas sobre operação e manutenção, deixe uma mensagem.