ubuntu上kafka安全设置怎么做

Ubuntu上Kafka安全设置实操指南

一安全目标与总体架构

启用传输加密：优先使用 SASL_SSL 或 SSL 保护数据链路，禁用明文 PLAINTEXT。
启用身份认证：推荐 SASL/SCRAM（如 SCRAM-SHA-256/512），便于动态管理用户；不建议在集群内长期使用 SASL/PLAIN。
启用授权控制：开启 ACL，按用户/客户端主机对 Topic/Group/Cluster 进行精细化授权。
网络与系统加固：仅开放必要端口，限制来源网段，开启防火墙与系统日志审计。

二生成证书与密钥库

准备证书目录并生成 Broker 密钥库（JKS）：

KAFKA_HOME=/opt/kafka
mkdir -p $KAFKA_HOME/config/certs

keytool -genkey \
  -alias kafka \
  -keyalg RSA \
  -keystore $KAFKA_HOME/config/certs/kafka.server.keystore.jks \
  -validity 3650 \
  -keysize 2048

导出并分发证书，构建信任库（每个 Broker 和客户端都需要导入其他 Broker 的证书）：

# 导出
keytool -export \
  -alias kafka \
  -file $KAFKA_HOME/config/certs/kafka.server.crt \
  -keystore $KAFKA_HOME/config/certs/kafka.server.keystore.jks

# 导入到信任库（客户端与集群内各 Broker 均执行）
keytool -import \
  -alias kafka \
  -file $KAFKA_HOME/config/certs/kafka.server.crt \
  -keystore $KAFKA_HOME/config/certs/kafka.server.truststore.jks

提示：生产环境建议使用由内部 CA 签发的证书，并在所有节点统一信任该 CA。

三配置 Broker 安全参数

编辑 $KAFKA_HOME/config/server.properties，建议仅保留安全监听器并开启双向认证：

# 监听与对外地址（按实际主机名/IP填写）
listeners=SASL_SSL://0.0.0.0:9093
advertised.listeners=SASL_SSL://<your.kafka.host>:9093

# 内部通信安全
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
sasl.enabled.mechanisms=SCRAM-SHA-512

# SSL 参数
ssl.keystore.location=/opt/kafka/config/certs/kafka.server.keystore.jks
ssl.keystore.password=YourKeystorePass
ssl.key.password=YourKeyPass
ssl.truststore.location=/opt/kafka/config/certs/kafka.server.truststore.jks
ssl.truststore.password=YourTruststorePass
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.3
# 可选：限制加密套件
# ssl.cipher.suites=TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384

# 授权器（ACL）
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
# 超级用户（谨慎授予，运维/初始化用）
super.users=User:admin

说明：
- 如暂未启用 SASL，可先用 SSL 监听器（将 listeners 改为 SSL://），但生产建议“加密+认证”同时使用。
- 若使用 KRaft 模式，请将上述安全配置同样应用于 controller.listeners 与 inter.broker.listener.name。

四配置 JAAS 与用户凭证

创建 JAAS 配置文件 $KAFKA_HOME/config/kafka_server_jaas.conf：

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

# 如果仍使用 Zookeeper，需为 ZK 连接配置身份（示例）
Client {
  org.apache.zookeeper.server.auth.DigestLoginModule required
  username="zkadmin"
  password="ZkPass!";
};

在 Broker 启动脚本中注入 JAAS（以 kafka-server-start.sh 为例，在 exec java 前加入）：

export KAFKA_OPTS="-Djava.security.auth.login.config=$KAFKA_HOME/config/kafka_server_jaas.conf"

创建 SCRAM 用户（无需重启 Broker）：

# 创建 admin（如上面 JAAS 已配置，可跳过或用于变更密码）
$KAFKA_HOME/bin/kafka-configs.sh --alter \
  --add-config 'SCRAM-SHA-512=[iterations=8192,password=AdminPass!]' \
  --entity-type users --entity-name admin

# 创建业务用户
$KAFKA_HOME/bin/kafka-configs.sh --alter \
  --add-config 'SCRAM-SHA-512=[iterations=8192,password=WriterPass!' \
  --entity-type users --entity-name writer

$KAFKA_HOME/bin/kafka-configs.sh --alter \
  --add-config 'SCRAM-SHA-512=[iterations=8192,password=ReaderPass!' \
  --entity-type users --entity-name reader

提示：SCRAM 凭证可通过命令行随时增删改，适合生产运维。

五客户端与 ACL 验证及网络加固

客户端配置示例（producer.properties/consumer.properties 或命令行 --command-config）：

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="writer" \
  password="WriterPass!";

ssl.truststore.location=/opt/kafka/config/certs/kafka.server.truststore.jks
ssl.truststore.password=YourTruststorePass
# 双向认证时再配置客户端密钥库
# ssl.keystore.location=/path/to/client.keystore.jks
# ssl.keystore.password=YourClientKeystorePass
# ssl.key.password=YourClientKeyPass

命令行快速测试：

# 列出 Topic（使用 SCRAM 凭证）
$KAFKA_HOME/bin/kafka-topics.sh --list \
  --bootstrap-server <your.kafka.host>:9093 \
  --command-config <(echo -e "security.protocol=SASL_SSL\nsasl.mechanism=SCRAM-SHA-512\nsasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"writer\" password=\"WriterPass!\";")

# 生产消息
$KAFKA_HOME/bin/kafka-console-producer.sh --topic test-topic \
  --bootstrap-server <your.kafka.host>:9093 \
  --producer.config producer.properties

# 消费消息
$KAFKA_HOME/bin/kafka-console-consumer.sh --topic test-topic \
  --from-beginning \
  --bootstrap-server <your.kafka.host>:9093 \
  --consumer.config consumer.properties

配置 ACL（示例：授予 writer 对 test-topic 的生产权限，reader 的消费权限）：

# 写权限
$KAFKA_HOME/bin/kafka-acls.sh --add \
  --topic test-topic \
  --principal User:writer \
  --operation Write --operation Create \
  --host <client.ip.or.cidr> \
  --bootstrap-server <your.kafka.host>:9093

# 读权限
$KAFKA_HOME/bin/kafka-acls.sh --add \
  --topic test-topic \
  --principal User:reader \
  --operation Read --operation Describe \
  --group '*' \
  --host <client.ip.or.cidr> \
  --bootstrap-server <your.kafka.host>:9093

# 查看 ACL
$KAFKA_HOME/bin/kafka-acls.sh --list \
  --topic test-topic \
  --bootstrap-server <your.kafka.host>:9093

网络与防火墙（UFW 示例，仅开放 9093）：

sudo ufw allow 9093/tcp
sudo ufw enable
# 云环境请同时配置安全组仅放通可信网段

验证与排错要点：
- 使用 TLSv1.2+，禁用 PLAINTEXT；必要时开启 ssl.client.auth=required 做双向认证。
- 客户端报 “SASL authentication failed” 时，核对 sasl.jaas.config 的用户名/密码与 SCRAM 配置是否一致。
- ACL 无权限时，检查 principal（User:xxx）、host 段、以及是否对 Topic/Group/Cluster 授予了对应 operation。

最新问答

相关标签