CentOS 上 Kafka 安全设置方法

一 安全基线

• 启用传输加密：优先使用 SSL/TLS；如为内网且暂无法启用 TLS，可先用 SASL 做身份鉴别，再尽快迁移到 SASL_SSL。
• 启用身份认证：推荐 SASL/SCRAM-SHA-256（可动态增删用户、无需落地明文口令）；避免使用 SASL/PLAIN 明文口令于生产。
• 启用授权控制：开启 ACL，仅授予最小权限；为运维与集群通信配置 超级用户。
• 网络与系统加固：仅开放必要端口（如 9092/9093），限制来源 IP；为 ZooKeeper 启用 SASL 与 ACL；配置 JMX 访问控制与日志轮转；避免不安全的参数（如 unclean.leader.election.enable=true）；设置 min.insync.replicas 提升数据可靠性。

二 启用 SSL TLS 加密

• 生成密钥与证书（JKS）
  • 服务器密钥库：
    keytool -genkey -alias kafka-server -keystore /etc/kafka/kafka.server.keystore.jks -storepass YourPass -validity 365 -keysize 2048
  • 导出并导入信任库：
    keytool -export -alias kafka-server -file kafka.server.crt -keystore /etc/kafka/kafka.server.keystore.jks -storepass YourPass
    keytool -import -alias kafka-server -file kafka.server.crt -keystore /etc/kafka/kafka.server.truststore.jks -storepass YourPass
• 服务端配置 server.properties
  • 监听与协议：
    listeners=SSL://:9093
    advertised.listeners=SSL://your-hostname:9093
  • SSL 参数：
    ssl.keystore.location=/etc/kafka/kafka.server.keystore.jks
    ssl.keystore.password=YourPass
    ssl.key.password=YourPass
    ssl.truststore.location=/etc/kafka/kafka.server.truststore.jks
    ssl.truststore.password=YourPass
    ssl.enabled.protocols=TLSv1.2,TLSv1.3
    ssl.client.auth=required（双向校验客户端证书）
• 客户端连接示例
  • 命令行：
    kafka-topics.sh --list --bootstrap-server your-hostname:9093 --security.protocol SSL
  • 生产/消费最小配置：
    security.protocol=SSL
    ssl.truststore.location=/path/to/client.truststore.jks
    ssl.truststore.password=YourPass

    如服务端开启 ssl.client.auth=required，还需配置客户端证书：

    ssl.keystore.location=/path/to/client.keystore.jks
    ssl.keystore.password=YourPass
    ssl.key.password=YourPass

三 启用 SASL 认证

• 选择机制：推荐 SASL/SCRAM-SHA-256（支持动态凭据管理），避免使用 SASL/PLAIN 明文口令。
• 创建 SCRAM 用户（示例用户：admin、producer、consumer）
  • bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config ‘SCRAM-SHA-256=[password=admin-secret]’ --entity-type users --entity-name admin
  • bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config ‘SCRAM-SHA-256=[password=producer-secret]’ --entity-type users --entity-name producer
  • bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config ‘SCRAM-SHA-256=[password=consumer-secret]’ --entity-type users --entity-name consumer
• Broker 端配置 server.properties
  • 监听与协议：
    listeners=SASL_SSL://:9093
    advertised.listeners=SASL_SSL://your-hostname:9093
    security.inter.broker.protocol=SASL_SSL
  • 机制与 JAAS：
    sasl.enabled.mechanisms=SCRAM-SHA-256
    sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256

    方式A（推荐，直接内联 JAAS）：

    sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=“admin” password=“admin-secret”;

    方式B（外部 JAAS 文件）：在启动脚本中注入

    export KAFKA_OPTS=“-Djava.security.auth.login.config=/path/kafka_server_jaas.conf”

• 客户端连接示例
  • 命令行生产：
    kafka-console-producer.sh --broker-list your-hostname:9093 --topic test-topic \ –producer-property security.protocol=SASL_SSL \ –producer-property sasl.mechanism=SCRAM-SHA-256 \ –producer-property sasl.jaas.config=‘org.apache.kafka.common.security.scram.ScramLoginModule required username=“producer” password=“producer-secret”;’
  • 命令行消费：
    kafka-console-consumer.sh --bootstrap-server your-hostname:9093 --topic test-topic \ –consumer-property security.protocol=SASL_SSL \ –consumer-property sasl.mechanism=SCRAM-SHA-256 \ –consumer-property sasl.jaas.config=‘org.apache.kafka.common.security.scram.ScramLoginModule required username=“consumer” password=“consumer-secret”;’ \ –from-beginning

四 授权与 ACL

• 启用 ACL 与超级用户（server.properties）
  • authorizer.class.name=kafka.security.authorizer.AclAuthorizer
  • allow.everyone.if.no.acl.found=false
  • super.users=User:admin
• 常用 ACL 操作（示例）
  • 授予生产者写权限：
    bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \ –add --allow-principal User:producer --operation Write --topic test-topic
  • 授予消费者组读权限（含消费组）：
    bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \ –add --allow-principal User:consumer --operation Read --topic test-topic --group test-consumer-group
  • 查看 ACL：
    bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \ –list --topic test-topic
• 提示：为运维与集群内部通信的账号（如 admin）配置为 super.users，避免被 ACL 拦截导致集群异常。

五 网络与系统加固

• 防火墙与网络隔离
  • 仅开放 9092/9093 等必要端口，并限制来源网段：
    sudo firewall-cmd --permanent --add-port=9093/tcp
    sudo firewall-cmd --permanent --add-rich-rule=‘rule family=“ipv4” source address=“192.168.1.0/24” accept’
    sudo firewall-cmd --reload
• ZooKeeper 安全
  • 为 ZooKeeper 启用 SASL 与 ACL，避免未授权访问元数据；相关 JAAS 与启动脚本配置可参考生产实践。
• 可靠性与监控
  • 设置 min.insync.replicas=2（或更高）提升数据持久性；
  • 关闭 unclean.leader.election.enable（或设为 false）避免数据丢失；
  • 启用 JMX 监控端口、配置日志轮转与磁盘/内存告警。