kafka ssl每客户端单独证书与acl之内外单独鉴权-EW帮帮网

接 https://blog.51cto.com/mapengfei/14087247

配置完Kafka SSL 认证与 ACL后，可以实现kafka对外访问每个客户端单独证书和acl限制topic

但需要能实现对外还是SSL每客户端单独证书认证，但内部不需要认证或者不需要分别用证书认证，

kafka ssl每客户端单独证书与acl之内外单独鉴权_bootstrap

主要修改/opt/kafka/config/server.properties 以下几点：

1、监听器配置变更

将 BROKER 拆分为 INTERNAL(9091/PLAINTEXT) 和 EXTERNAL(9093/SSL)
实现内外网分离：内网明文，外网加密

修改前配置：

listeners=CONTROLLER://10.0.70.189:9092,BROKER://10.0.70.189:9093

inter.broker.listener.name=BROKER

advertised.listeners=CONTROLLER://10.0.70.189:9092,BROKER://10.0.70.189:9093

listener.security.protocol.map=CONTROLLER:SSL,BROKER:SSL

修改后配置：

listeners=CONTROLLER://10.0.70.189:9092,INTERNAL://10.0.70.189:9091,EXTERNAL://10.0.70.189:9093

inter.broker.listener.name=INTERNAL

advertised.listeners=CONTROLLER://10.0.70.189:9092,INTERNAL://10.0.70.189:9091,EXTERNAL://10.0.70.189:9093

listener.security.protocol.map=CONTROLLER:SSL,INTERNAL:PLAINTEXT,EXTERNAL:SSL

2. 控制器引导服务器变更

参考文章： controller.quorum.bootstrap.servers=10.0.70.189:9093

当前配置： controller.quorum.bootstrap.servers=10.0.70.189:9092

变更说明：修正为控制器专用端口 9092

3. 超级用户权限变更

参考文章： super.users=User:root;User:CN=10.0.70.189,OU=test,O=test,L=BJ,ST=BJ,C=CN

当前配置： super.users=User:root;User:ANONYMOUS;User:CN=10.0.70.189,OU=test,O=test,L=BJ,ST=BJ,C=CN

变更说明：增加 User:ANONYMOUS 使 9091 明文连接具备超级权限

4. 单节点集群参数新增

新增配置：

# single-node replication settings

offsets.topic.replication.factor=1

transaction.state.log.replication.factor=1

transaction.state.log.min.isr=1

# allow auto creation for internal topics like __consumer_offsets in single-node

auto.create.topics.enable=true

变更说明：为单节点部署优化，解决 __consumer_offsets 创建问题

所有配置修改完成记得重启下kafka服务

5. 效果验证

1)发送内部9091非认证端口，随意发一个topic，没有任何限制

/opt/kafka/bin/kafka-producer-perf-test.sh --producer.config /dev/null --topic other_topic --num-records 1 --record-size 16 --throughput -1 --producer-props bootstrap.servers=10.0.70.189:9091 acks=all

[2025-09-05 19:18:29,364] WARN [Producer clientId=perf-producer-client] The metadata response from the cluster reported a recoverable issue with correlation id 1 : {other_topic=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient)

1 records sent, 2.0 records/sec (0.00 MB/sec), 497.00 ms avg latency, 497.00 ms max latency, 497 ms 50th, 497 ms 95th, 497 ms 99th, 497 ms 99.9th.

2)发送外部9093认证端口，配置acl的t_mafei的topic，正常发

/root/kafka_test_amd64 --ca=/etc/kafka/server/ca.crt --cert=/etc/kafka/clientA/clientA.crt --key=/etc/kafka/clientA/clientA.key --topic t_mafei --broker 10.0.70.189:9093

2025/09/05 19:20:11 消息发送成功，分区: 0, 偏移: 20

3)发送外部9093认证端口，配置非acl允许的topic，认证失败

[root@xsiem-master kafka]# /root/kafka_test_amd64 --ca=/etc/kafka/server/ca.crt --cert=/etc/kafka/clientA/clientA.crt --key=/etc/kafka/clientA/clientA.key --topic other_topic --broker 10.0.70.189:9093

2025/09/05 19:20:22 发送消息失败: kafka server: The client is not authorized to access this topic

6.修改后完整配置 /opt/kafka/config/server.properties

process.roles=broker,controller

node.id=1

controller.quorum.bootstrap.servers=10.0.70.189:9092

listeners=CONTROLLER://10.0.70.189:9092,INTERNAL://10.0.70.189:9091,EXTERNAL://10.0.70.189:9093

inter.broker.listener.name=INTERNAL

advertised.listeners=CONTROLLER://10.0.70.189:9092,INTERNAL://10.0.70.189:9091,EXTERNAL://10.0.70.189:9093

controller.listener.names=CONTROLLER

listener.security.protocol.map=CONTROLLER:SSL,INTERNAL:PLAINTEXT,EXTERNAL:SSL

ssl.keystore.type=PKCS12

ssl.keystore.location=/etc/kafka/server/kafka.server.p12

ssl.keystore.password=mafei2025

ssl.key.password=mafei2025

ssl.truststore.type=PKCS12

ssl.truststore.location=/etc/kafka/server/kafka.server.truststore.p12

ssl.truststore.password=mafei2025

ssl.client.auth=required

authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer

super.users=User:root;User:ANONYMOUS;User:CN=10.0.70.189,OU=test,O=test,L=BJ,ST=BJ,C=CN

log.dirs=/data/kafka

num.partitions=1

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

# single-node replication settings
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

# allow auto creation for internal topics like __consumer_offsets in single-node
auto.create.topics.enable=true

kafka ssl每客户端单独证书与acl之内外单独鉴权