kafka在centos上的权限管理如何设置

Kafka 在 CentOS 上的权限管理设置

一核心思路与前置条件

权限管理包含四个层面：
1. 认证 Authentication：确认客户端与 Broker 的身份，常用 SASL/SCRAM-SHA-256（生产推荐）、SASL/PLAIN（仅测试）、或 Kerberos/GSSAPI（企业域环境）。
2. 授权 Authorization：基于 ACL 控制对 Topic/Group/Cluster 等资源的 Read/Write/Describe/Create/Delete 等操作。
3. 加密 Encryption：开启 SSL/TLS 保护传输链路。
4. 网络与系统加固：防火墙、最小暴露面、最小权限运行。
前置条件：
- Java 11+ 运行环境。
- 已部署 Zookeeper（或 KRaft 模式），本文以 Zookeeper 为例。
- 工具脚本位于 Kafka 安装目录的 bin/ 下（如 kafka-configs.sh、kafka-acls.sh）。

二启用认证 SASL SCRAM 与 JAAS 配置

创建 SCRAM 用户（可在不重启 Broker 的情况下动态新增）：
- 建议为 Broker 间通信创建管理员账户（如 admin），并为生产/消费分别创建账户（如 producer/consumer）。
- 示例（SCRAM-SHA-256，迭代次数建议 ≥ 8192）：
  - kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type users --entity-name admin --add-config ‘SCRAM-SHA-256=[iterations=8192,password=AdminPass!]’
  - kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type users --entity-name producer --add-config ‘SCRAM-SHA-256=[iterations=8192,password=ProdPass!]’
  - kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type users --entity-name consumer --add-config ‘SCRAM-SHA-256=[iterations=8192,password=ConsPass!]’
- 说明：SCRAM 支持运行时增删改用户；而 SASL/PLAIN 的用户需在 JAAS 中静态配置，新增用户需重启，生产不推荐。
配置 JAAS（/etc/kafka/kafka_server_jaas.conf）：
- SCRAM 示例：
  - KafkaServer {
    - org.apache.kafka.common.security.scram.ScramLoginModule required
    - username=“admin”
    - password=“AdminPass!”;
  - };
在 server.properties 启用 SASL：
- listeners=SASL_SSL://0.0.0.0:9093（生产建议 SASL_SSL；测试可用 SASL_PLAINTEXT://:9092）
- security.inter.broker.protocol=SASL_SSL
- sasl.enabled.mechanisms=SCRAM-SHA-256
- sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
- 方式一（推荐）：sasl.jaas.config=file:/etc/kafka/kafka_server_jaas.conf
- 方式二：在启动脚本中注入 JVM 参数（kafka-server-start.sh）：
  - export KAFKA_OPTS=“-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf”。

三配置授权 ACL 与常用命令

启用 ACL（server.properties）：
- authorizer.class.name=kafka.security.authorizer.AclAuthorizer
- allow.everyone.if.no.acl.found=false
- super.users=User:admin（授予管理员绕过 ACL 的能力，谨慎使用）
常用 ACL 命令（示例）：
- 为用户 alice 授予 topic1 的读写：
  - kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice --operation Read --operation Write --topic topic1
- 为消费组 group1 授予 topic2 的消费权限（含 Describe 与 Read Group）：
  - kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal Group:group1 --operation Read --operation Describe --group group1 --topic topic2
- 列出 ACL：
  - kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --list --topic topic1
- 移除 ACL：
  - kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --remove --allow-principal User:alice --operation Read --topic topic1
提示：ACL 资源类型包括 Topic/Group/Cluster/TransactionalId；操作包括 Read/Write/Describe/Create/Delete/Alter/DescribeConfigs/AlterConfigs。生产建议按“最小权限”原则分配。

四加密与网络隔离

启用 SSL/TLS（Broker 端）：
- 生成密钥库与信任库（JKS）：
  - keytool -genkey -alias kafka-server -keystore /etc/kafka/kafka.server.keystore.jks -storepass KsPass! -keypass KsPass! -validity 365 -keyalg RSA -keysize 2048 -dname “CN=kafka-server”
  - keytool -export -alias kafka-server -file /etc/kafka/kafka.server.crt -keystore /etc/kafka/kafka.server.keystore.jks -storepass KsPass!
  - keytool -import -alias kafka-server -file /etc/kafka/kafka.server.crt -keystore /etc/kafka/kafka.server.truststore.jks -storepass TsPass! -noprompt
- server.properties（与 SASL 组合为 SASL_SSL 时，SSL 参数仍需配置在监听器上）：
  - listeners=SASL_SSL://0.0.0.0:9093
  - advertised.listeners=SASL_SSL://:9093
  - ssl.keystore.location=/etc/kafka/kafka.server.keystore.jks
  - ssl.keystore.password=KsPass!
  - ssl.key.password=KsPass!
  - ssl.truststore.location=/etc/kafka/kafka.server.truststore.jks
  - ssl.truststore.password=TsPass!
  - ssl.enabled.protocols=TLSv1.2,TLSv1.3
防火墙与网络：
- 仅开放必要端口（如 9093）：firewall-cmd --permanent --add-port=9093/tcp && firewall-cmd --reload
- listeners 绑定内网或受控 IP，避免暴露在公网；必要时使用 VPC/安全组 做白名单。

五客户端配置示例与验证

生产者（producer.properties）：
- bootstrap.servers=:9093
- security.protocol=SASL_SSL
- sasl.mechanism=SCRAM-SHA-256
- sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=“producer” password=“ProdPass!”;
- ssl.truststore.location=/etc/kafka/kafka.client.truststore.jks
- ssl.truststore.password=TsPass!
消费者（consumer.properties）：
- bootstrap.servers=:9093
- security.protocol=SASL_SSL
- sasl.mechanism=SCRAM-SHA-256
- sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=“consumer” password=“ConsPass!”;
- group.id=group1
- ssl.truststore.location=/etc/kafka/kafka.client.truststore.jks
- ssl.truststore.password=TsPass!
验证步骤：
- 使用未授权的用户或缺少 ACL 的账户尝试生产/消费，应被拒绝。
- 使用已授权账户执行操作，确认成功。
- 运行 kafka-acls.sh --list 定期审计 ACL 规则，移除不必要授权。

最新问答

相关标签