Technologieaustausch

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

Autor Lewei Community (forum.lwops.cn) Xu Yuan
Beim Aufbau moderner verteilter Systeme ist die Gewährleistung der Sicherheit der Datenübertragung von entscheidender Bedeutung. Apache Kafka und Zookeeper stellen als beliebte verteilte Nachrichtenwarteschlangen und Koordinierungsdienste einen SSL-Authentifizierungsmechanismus (Secure Sockets Layer) bereit, um die Sicherheit bei der Datenübertragung zu erhöhen.
In diesem Artikel wird der gesamte Prozess von der Erstellung eines SSL-Zertifikats bis zur Konfiguration des Servers und Clients ausführlich vorgestellt, um sicherzustellen, dass die Daten während der Übertragung vollständig geschützt sind.

1. Konfigurieren Sie das Passwort für das Kafka-Konto:
1. Zuerst müssen Sie die Kafka-Konfigurationsdatei ändern: vim /asop/kafka/kafka_2.11-2.1.0/config/server.properties

Broker-ID = 0
Listener = SASL_PLAINTEXT://:9092
beworben.listeners=SASL_PLAINTEXT://10.176.31.137:9092
Anzahl.Netzwerkthreads=3
Anzahl.io.threads=8
socket.send.buffer.bytes=102400
socket.empfangen.puffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/asop/kafka/logs
Anzahl Partitionen = 1
Anzahl Wiederherstellungsthreads pro Datenverzeichnis = 1
Offsets.Thema.Replikationsfaktor=1
Transaktion.Status.Log.Replikation.Faktor=1
Transaktion.Status.log.min.isr=1
log.Aufbewahrungszeiten.Stunden=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.Verbindungstimeout.ms=6000
Gruppe.initial.rebalance.delay.ms=0

#Authentifizierungsprotokoll verwendet
Sicherheit.inter.broker.protocol=SASL_PLAINTEXT
#SASL-Mechanismus
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#Klasse zum Abschließen der Authentifizierung
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#Wenn keine ACL-Konfiguration (Zugriffskontrollliste) gefunden wird, ist jede Operation zulässig.
erlauben.jeder.wenn.keine.acl.gefunden.=false
#Sie müssen die Einstellung „Superadministrator“ aktivieren und den Besucherbenutzer als Superadministrator festlegen.
super.users=Benutzer:Besucher

2. Zweitens erstellen Sie eine Anmeldebestätigungsdatei für den Server. Sie können die Datei nach Ihren Wünschen benennen, z. B. vim /asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf folgt

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule erforderlich
Benutzername=“Besucher“
Passwort=“qaz@123”
Benutzer_Besucher=“qaz@123”;
};

3. Ändern Sie dann das Kafka-Installationsverzeichnis vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh und fügen Sie Variablen oben in der Datei hinzu

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

4. Erstellen Sie als Nächstes eine Anmeldebestätigungsdatei für den Verbraucher und den Produzenten. Sie können die Datei nach Ihren Wünschen benennen, z. B. kafka_client_jaas.conf. Der Inhalt der Datei lautet wie folgt (wenn es sich um einen Programmzugriff handelt, z. B. Springboot-Zugriff). , es muss nicht konfiguriert werden)
vim /asop/kafka/kafka_2.11-2.1.0/config/kafka_client_jaas.conf

KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule erforderlich
Benutzername=“Besucher“
Passwort=“qaz@123”;
};

5. Fügen Sie Consumer.properties bzw. Producer.properties die folgenden Konfigurationen hinzu:
vim /asop/kafka/kafka_2.11-2.1.0/config/consumer.properties
vim /asop/kafka/kafka_2.11-2.1.0/config/producer.properties

Sicherheit.Protokoll=SASL_PLAINTEXT
sasl.mechanism=PLAIN

6. Ändern Sie die Kafka-Installationsverzeichnisse bin/kafka-console-producer.sh und bin/kafka-console-consumer.sh und fügen Sie oben in den Dateien Variablen hinzu.
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

export KAFKA_OPTS=" -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_client_jaas.conf"
7. Starten Sie zookeeper bzw. kafka. Zu diesem Zeitpunkt ist die Konfiguration der serverseitigen Kafka-Benutzeranmeldeüberprüfung abgeschlossen (schließen Sie zuerst kafka und dann zookeeper).

Beenden Sie den Kafka-Dienst
/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

Starten Sie den Dienst 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

Beenden Sie den Dienst zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh stop /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
Starten Sie den Dienst zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh starte /asop/zk/zookeeper-3.4.13/conf/zoo.cfg

8. Erstellen Sie Themen und zeigen Sie sie an

/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

Nachrichten erhalten

/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. Konfigurieren Sie das SSAL-Kontokennwort für ZK und Kafka:

1. Zookeeper-Konfiguration SASL
   1.1 Erstellen Sie eine neue zoo_jaas.conf-Datei
   Es gibt keine besonderen Anforderungen an den Dateinamen und den Pfad der Datei zoo_jaas.conf. Sie wird im Allgemeinen im Verzeichnis ${ZOOKEEPER_HOME}/conf vim /asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf abgelegt

Server {
org.apache.kafka.common.security.plain.PlainLoginModule erforderlich
Benutzername=“admin”
Passwort=“admin@12”
Benutzer_kafka = „kafka@123“;
};

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

1.2 Konfigurieren Sie die Datei /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
zookeeper.sasl.client=true

Zookeeper.sasl.client ist auf „true“ gesetzt, um die Client-Authentifizierung zu aktivieren. Andernfalls funktioniert der in zoo_jaas.conf konfigurierte Benutzername nicht ohne eine Jaas-Datei, jedoch mit WARNUNG.

1.3 Abhängigkeitspakete importieren
Da die verwendete Berechtigungsüberprüfungsklasse org.apache.kafka.common.security.plain.PlainLoginModule ist, sind kafkabezogene JAR-Pakete wie folgt erforderlich: Erstellen Sie einen neuen Ordner zk_sasl_lib: Kopieren Sie die folgenden JAR-Pakete aus dem Verzeichnis kafka/lib zookeeper lib und das neu erstellte Verzeichnis zk_sasl_lib:

kafka-clients-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 Ändern Sie die Datei zkEnv.sh/asop/zk/zookeeper-3.4.13/bin/zkEnv.sh
Vor der Änderung: Wenn nicht, fügen Sie es direkt hinzu

export SERVER_JVMFLAGS="-Xmx${ZK_SERVER_HEAP}m $SERVER_JVMFLAGS"

Nach der Modifikation:

für Jar in /asop/zk/zookeeper-3.4.13/zk_sasl_lib/*.jar;
Tun
CLASSPATH=“$jaR:$KLASSENPFAD“
Erledigt

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

Starten Sie einfach den Zookeeper-Dienst neu

Beenden Sie den Dienst zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh stop /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
Starten Sie den Dienst zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh starte /asop/zk/zookeeper-3.4.13/conf/zoo.cfg

2. Kakfa konfiguriert SASL
   2.1 Erstellen Sie eine neue Datei kafka_server_jaas.conf
   Es gibt keine Anforderungen für den Dateinamen und den Speicherpfad von kafka_server_jaas.conf. Er wird im Allgemeinen im Verzeichnis ${KAFKA_HOME}/config/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf abgelegt

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule erforderlich
Benutzername=“Besucher“
Passwort=“qaz@123”
Benutzer_Besucher=“qaz@123”;
};
Klient{
org.apache.kafka.common.security.plain.PlainLoginModule erforderlich
Benutzername=“kafka”
Passwort = „kafka@123“;
};

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

KafkaServer.user_xxx, wobei xxx mit dem in KafkaServer.username konfigurierten Benutzernamen übereinstimmen muss und das Kennwort ebenfalls konsistent sein muss
KafkaServer.user_producer und KafkaServer.user_consumer bereiten sich auf die nachfolgende ACL vor, sodass Verbraucher und Produzenten unterschiedliche Konten verwenden und Verbraucherkonten nur Daten konsumieren können und Produzentenkonten nur Daten produzieren können.
Client.username und Client.password geben das in Zookeeper registrierte Kontokennwort ein, das für die Kommunikation zwischen Broker und Zookeeper verwendet wird (wenn Zookeeper nicht mit SASL konfiguriert ist, kann es ignoriert werden. Wenn zookeeper.sasl.client falsch ist, kann es ignoriert werden auch ignoriert werden. Das Protokoll lautet wie folgt.

[2021-06-29 17:14:30,204] WARN SASL-Konfiguration fehlgeschlagen: javax.security.auth.login.LoginException: In der angegebenen JAAS-Konfigurationsdatei wurde kein JAAS-Konfigurationsabschnitt mit dem Namen „Client“ gefunden: „/Users/wjun/env/kafka/config/kafka_server_jaas.conf“. Die Verbindung zum Zookeeper-Server wird ohne SASL-Authentifizierung fortgesetzt, wenn der Zookeeper-Server dies zulässt. (org.apache.zookeeper.ClientCnxn)

2.2 Ändern Sie die Datei server.properties
Broker-ID = 0
Listener = SASL_PLAINTEXT://:9092
beworben.listeners=SASL_PLAINTEXT://192.168.157.198:9092
Anzahl.Netzwerkthreads=3
Anzahl.io.threads=8
socket.send.buffer.bytes=102400
socket.empfangen.puffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/asop/kafka/logs
Anzahl Partitionen = 1
Anzahl Wiederherstellungsthreads pro Datenverzeichnis = 1
Offsets.Thema.Replikationsfaktor=1
Transaktion.Status.Log.Replikation.Faktor=1
Transaktion.Status.log.min.isr=1
log.Aufbewahrungszeiten.Stunden=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=127.0.0.1:2181
zookeeper.Verbindungstimeout.ms=6000
Gruppe.initial.rebalance.delay.ms=0

#Authentifizierungsprotokoll verwendet
Sicherheit.inter.broker.protocol=SASL_PLAINTEXT
#SASL-Mechanismus
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#Klasse zum Abschließen der Authentifizierung
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#Wenn keine ACL-Konfiguration (Zugriffskontrollliste) gefunden wird, ist jede Operation zulässig.
erlauben.jeder.wenn.keine.acl.gefunden.=false
#Sie müssen die Einstellung „Superadministrator“ aktivieren und den Besucherbenutzer als Superadministrator festlegen.
super.users=Benutzer:Besucher

Wobei localhost in eine IP-Adresse geändert werden muss

super.users konfiguriert den Superuser, der von nachfolgenden ACL-Konfigurationen nicht betroffen ist.

2.3 Ändern Sie das Startskript
Ändern Sie die Datei kafka-server-start.sh, um sie in die Datei kafka_server_jaas.conf/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh zu laden

vor der Reparatur:

wenn [ „x$KAFKA_HEAP_OPTS“ = „x“ ]; dann
export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi

Nach der Modifikation:
(Fügen Sie diese Zeile zuerst zur ersten Zeile hinzu. Wenn Sie sie bereits haben, müssen Sie sie nicht hinzufügen.) export KAFKA_OPTS=" -Djava.security.auth.login.config=/asop/kafka/kafka_2.11- 2.1.0/config/kafka_server_jaas.conf "
wenn [ „x$KAFKA_HEAP_OPTS“ = „x“ ]; dann
export KAFKA_HEAP_OPTS=-Xmx1G -Xms1G -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf”
fi

Legen Sie ACL-Regeln für zookeeper fest
/asop/zk/zookeeper-3.4.13/bin/zkCli.sh #Geben Sie den Befehlszeilenmodus von zk ein

addauth-Digest admin:admin@12 #Anmeldebenutzer wechseln (der Superadministrator befindet sich in der zk-Konfigurationsdatei/asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf)

setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka@123:cdrwa #(Legen Sie die IP-Adresse und das Benutzerkontokennwort fest, mit denen Sie sich anmelden können. Admin ist der in der ZK-Konfigurationsdatei oben definierte Administrator und der Kafka-Benutzer ist /asop Der Kafka-Verbindungs-zk-Benutzer, der in der Datei /kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf (unter Client) definiert ist.

Addauth Digest kafka:kafka@123 #Wechseln Sie zum Kafka-Benutzer und setzen Sie acl erneut
setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka@123:cdrwa

Hinweis: Wenn Sie eine Whitelist-IP hinzufügen möchten oder der Benutzer sie auf der ursprünglichen Basis hinzufügen möchte, wird sie sonst überschrieben.
setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka@123:cdrwa,auth:admin:admin@12:cdrwa,ip:1.1.1.1

Berechtigungen müssen wiederhergestellt werden, ohne ACL auszuführen
setAcl / Welt:jeder:cdrwa

Starten Sie einfach den Kafka-Dienst neu

Beenden Sie den Kafka-Dienst
#/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

Starten Sie den Dienst 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

Zu diesem Zeitpunkt ist die SSL-Authentifizierungskonfiguration von Kafka und Zookeeper abgeschlossen. Für weitere Betriebs- und Wartungskenntnisse wenden Sie sich bitte an die Lewei-Community. Bei weiteren Betriebs- und Wartungsfragen hinterlassen Sie bitte eine Nachricht.