kafka 认证机制

使用 SSL 或 SASL 对来自客户端（生产者和使用者）、其他代理和工具的代理连接进行身份验证。Kafka 支持以下 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 开始

kafka 侦听器

kafka 侦听器类型

PLAINTEXT：用于不加密的普通通信。 listeners=PLAINTEXT://:9092
SSL：用于加密通信，确保数据传输的安全性。 listeners=SSL://:9093
SASL_PLAINTEXT：在不加密的基础上，添加身份验证机制。listeners=SASL_PLAINTEXT://:9094
SASL_SSL：结合加密和身份验证，确保通信的机密性和完整性。listeners=SASL_SSL://:9095
CONTROLLER：用于 Kafka 集群控制器进行内部通信，管理 Broker 状态。listeners=CONTROLLER://:9096
EXTERNAL：专为外部客户端访问设计，通常用于跨网络的通信。listeners=EXTERNAL://:9097

kafka 侦听器配置

Kafka 服务器支持监听多个端口上的连接。这可以通过listeners服务器配置中的属性进行配置，该属性接受要启用的侦听器的逗号分隔列表。每个服务器上必须定义至少一个侦听器。
  {LISTENER_NAME}://{hostname}:{port}
  listeners=CLIENT://localhost:9092

每个侦听器的安全协议在单独的配置中定义： listener.security.protocol.map。该值是每个侦听器映射到其安全协议的逗号分隔列表。
  listener.security.protocol.map=CLIENT:SSL,BROKER:PLAINTEXT

使用 SASL/PLAIN 进行身份验证

SASL/PLAIN 是一种简单的用户名/密码身份验证机制，通常与 TLS 一起用于加密以实现安全身份验证。Kafka 支持 SASL/PLAIN 的默认实现。

创建kafka_server_jaas.conf文件

cat >> config/kafka_server_jaas.conf <<EOF
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_alice="alice-secret"
    user_producer="producer-secret"
    user_consumer="consumer-secret";

};
EOF

更新 bin/kafka-server-start.sh

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/data/kafka/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"

更新配置文件 config/kraft/server.properties

listeners=PLAINTEXT://:9092,CONTROLLER://:9093,SASL_PLAINTEXT://:9094
advertised.listeners=PLAINTEXT://192.168.174.108:9092,SASL_PLAINTEXT://192.168.174.108:9094
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

验证认证配置信息

9092 端口

# bin/kafka-topics.sh --list --bootstrap-server 192.168.174.108:9092

wgs-test-event

9094 端口

# bin/kafka-topics.sh --list --bootstrap-server 192.168.174.108:9094

Error while executing topic command : Timed out waiting for a node assignment. Call: listTopics
[2024-12-01 12:03:37,085] ERROR org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: listTopics
 (org.apache.kafka.tools.TopicCommand)

Dec 01 12:03:13 node01 kafka-server-start.sh[10720]: [2024-12-01 12:03:13,475] INFO [SocketServer listenerType=BROKER, nodeId=1] Failed authentication with /192.168.174.108 (channelId=192.168.174.108:9094-192.168.174.108:50542-11) (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)

客户端认证配置

生成消息

bin/kafka-console-producer.sh --topic wgs-test-event --bootstrap-server 192.168.174.109:9094 --producer-property security.protocol=SASL_PLAINTEXT  --producer-property sasl.mechanism=PLAIN --producer-property 'sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="producer" password="producer-secret";'

消费消息

bin/kafka-console-consumer.sh --topic wgs-test-event --from-beginning --bootstrap-server 192.168.174.109:9094 --consumer-property "security.protocol=SASL_PLAINTEXT" --consumer-property "sasl.mechanism=PLAIN" --consumer-property 'sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="consumer" password="consumer-secret";'

查看 topic 

创建配置文件

cat >> config/jaas.conf <<EOF
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret";
EOF

查看 topic

bin/kafka-topics.sh --list --bootstrap-server 192.168.174.108:9094 --command-config config/jaas.conf

__consumer_offsets
wgs-test-event
wgs2-test-event

删除 topic

bin/kafka-topics.sh --delete --topic wgs2-tst-event --bootstrap-server 192.168.174.108:9094 --command-config config/jaas.conf

创建 topic

bin/kafka-topics.sh --create --topic test-event --bootstrap-server 192.168.174.108:9094 --partitions 3 --replication-factor 2  --command-config config/jaas.conf

查看 topic 详细信息

bin/kafka-topics.sh --describe --topic test-event --bootstrap-server 192.168.174.108:9094   --command-config config/jaas.conf 

Topic: test-event	TopicId: nSU3EOklQM2zGyX6oiZO7Q	PartitionCount: 3	ReplicationFactor: 2	Configs: segment.bytes=1073741824
	Topic: test-event	Partition: 0	Leader: 3	Replicas: 3,1	Isr: 3,1	Elr: 	LastKnownElr: 
	Topic: test-event	Partition: 1	Leader: 1	Replicas: 1,2	Isr: 1,2	Elr: 	LastKnownElr: 
	Topic: test-event	Partition: 2	Leader: 2	Replicas: 2,3	Isr: 2,3	Elr: 	LastKnownElr: 

参考文档

https://kafka.apache.org/documentation/#security_sasl