10.8.1 认证机制与安全配置（SASL/SSL）

本节聚焦Kafka在生产环境中的认证与传输安全配置，覆盖SASL与SSL的核心原理、部署步骤、命令示例与排错方法，确保客户端、Broker与控制器之间身份可信与链路加密。

一、认证与加密模型概览（含原理草图）

SASL：用于身份鉴别（PLAIN/SCRAM/GSSAPI）。
SSL/TLS：用于链路加密与可选双向证书认证（mTLS）。
组合策略：SASL负责认证，SSL负责加密（SASL_SSL）。

二、SSL/TLS 安装与证书准备（完整示例）

目标：自建CA，签发Broker证书与Client证书，生成keystore/truststore。

1）创建CA与证书（示例路径：/opt/kafka/security）：

mkdir -p /opt/kafka/security && cd /opt/kafka/security

# 1. 生成CA私钥与证书
openssl req -new -x509 -keyout ca.key -out ca.crt -days 3650 \
  -subj "/CN=kafka-ca" -passout pass:capass

# 2. 生成Broker私钥与CSR
keytool -genkeypair -alias kafka-broker \
  -keystore kafka.server.keystore.jks \
  -keyalg RSA -keysize 2048 -validity 3650 \
  -storepass storepass -keypass keypass \
  -dname "CN=kafka01,OU=OPS,O=ORG,L=BJ,ST=BJ,C=CN"

keytool -certreq -alias kafka-broker \
  -keystore kafka.server.keystore.jks \
  -file kafka-broker.csr -storepass storepass

# 3. CA签发Broker证书
openssl x509 -req -in kafka-broker.csr -CA ca.crt -CAkey ca.key \
  -CAcreateserial -out kafka-broker.crt -days 3650 -passin pass:capass

# 4. 导入CA与Broker证书
keytool -importcert -alias CARoot -keystore kafka.server.keystore.jks \
  -file ca.crt -storepass storepass -noprompt

keytool -importcert -alias kafka-broker -keystore kafka.server.keystore.jks \
  -file kafka-broker.crt -storepass storepass

# 5. 生成truststore（仅包含CA）
keytool -importcert -alias CARoot -keystore kafka.server.truststore.jks \
  -file ca.crt -storepass storepass -noprompt

2）关键命令解释：
- openssl req -x509：生成自签CA证书。
- keytool -genkeypair：生成JKS私钥与证书。
- keytool -certreq：生成CSR供CA签发。
- keytool -importcert：导入CA与签发证书，构建信任链。

三、Broker端SSL配置（完整可用）

/opt/kafka/config/server.properties：

listeners=SSL://0.0.0.0:9093
advertised.listeners=SSL://kafka01:9093
ssl.keystore.location=/opt/kafka/security/kafka.server.keystore.jks
ssl.keystore.password=storepass
ssl.key.password=keypass
ssl.truststore.location=/opt/kafka/security/kafka.server.truststore.jks
ssl.truststore.password=storepass
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.endpoint.identification.algorithm=HTTPS

预期效果：客户端通过SSL连接成功，未提供客户端证书将被拒绝（mTLS）。

四、SASL（SCRAM）启用与用户创建

1）Broker启用SCRAM（与SSL组合）：

listeners=SASL_SSL://0.0.0.0:9094
advertised.listeners=SASL_SSL://kafka01:9094
listener.security.protocol.map=SASL_SSL:SASL_SSL
sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_SSL

2）创建SCRAM用户（需使用Kafka命令）：

/opt/kafka/bin/kafka-configs.sh --bootstrap-server kafka01:9094 \
  --alter --add-config 'SCRAM-SHA-256=[password=AdminPass123]' \
  --entity-type users --entity-name admin

3）配置Broker JAAS（示例 /opt/kafka/config/kafka_server_jaas.conf）：

KafkaServer {
  org.apache.kafka.common.security.scram.ScramLoginModule required
  username="admin"
  password="AdminPass123";
};

4）启动Kafka时注入JAAS：

export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
systemctl restart kafka

五、客户端SASL_SSL配置与验证

1）Client配置文件 client.properties：

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="AdminPass123";
ssl.truststore.location=/opt/kafka/security/kafka.server.truststore.jks
ssl.truststore.password=storepass
ssl.endpoint.identification.algorithm=HTTPS

2）生产/消费验证：

# 创建topic
/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka01:9094 \
  --command-config client.properties --create \
  --topic sec-demo --partitions 3 --replication-factor 1

# 生产消息
/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server kafka01:9094 \
  --producer.config client.properties --topic sec-demo

# 消费消息
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka01:9094 \
  --consumer.config client.properties --topic sec-demo --from-beginning

六、排错清单（问题→命令→解决）

1）SSL握手失败
- 排查命令：

openssl s_client -connect kafka01:9093 -showcerts

解决要点：证书链不完整/域名不匹配/客户端系统时间漂移。

2）SASL认证失败
- 排查命令：

grep -i "SASL" /opt/kafka/logs/server.log

解决要点：用户未创建、JAAS用户名密码不一致、机制不一致。

3）连接被拒绝
- 排查命令：

ss -lntp | grep 9094

解决要点：监听器端口未启动或防火墙拦截。

4）mTLS客户端证书缺失
- 排查命令：

grep -i "SSLHandshakeException" /opt/kafka/logs/server.log

解决要点：客户端未配置keystore或证书未由CA签发。

七、练习与实战任务

1）基础练习
- 任务：用自建CA完成SSL单向认证；客户端消费sec-demo。
- 验证：kafka-console-consumer.sh能正常消费。

2）进阶练习
- 任务：启用ssl.client.auth=required，完成mTLS双向认证。
- 验证：未配置客户端证书时连接应失败。

3）排错练习
- 任务：故意修改ssl.endpoint.identification.algorithm为""，
观察错误与修复效果。
- 验证：恢复HTTPS后握手成功。

八、安全配置最佳实践

统一CA管理并定期轮换证书，避免过期中断。
禁用弱加密套件与过时TLS协议。
使用独立监听器区分内外网访问。
生产环境优先使用SASL_SSL + SCRAM/GSSAPI。