Kafka 开启SASL/SCRAM认证 及 ACL授权（二）ACL

由上一篇博文，讲解了SASL认证的安装过程。本文主要围绕ACL设置进行说明。

参考：

kafka authentorization：官方

一、开启ZK ACL（可选，内网环境，用户无机器访问权限时）

给kafka meta都加上zk的acl，默认是world，即任何人可访问。

zk acl详见：https://www.cnblogs.com/dalianpai/p/12748144.html

执行已有kafka meta的迁移

• vim kafka/bin/zookeeper-security-migration.sh

# sasl
cur_dir=`dirname $0`
export KAFKA_OPTS="-Djava.security.auth.login.config=${cur_dir}/../config/kafka_server_jaas.conf"

• 执行迁移

kafka/bin/zookeeper-security-migration.sh --zookeeper.connect 127.0.0.1:2181 --zookeeper.acl secure

• 到zk上查看生效

getAcl /brokers/topics

二、设置kafka acl

上篇文章已经在server.properties中添加了acl的配置。这里执行命令设置acl。

kafka的acl对象，即资源类型分为 cluster, topic, consumer group, transation id, token。实际应用中，主要以topic及consumer group为主进行运维管控。即控制”哪个用户的哪个消费组可以访问哪个topic“。

常用acl操作

• 列出 test 这个 topic 的 ACL 列表

bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9093 --command-config config/adminclient-configs.conf --list --topic "test"

• 对用户 test 授权，以访问test 这个 topic。

先添加scram用户 test

bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9093 --alter --add-config 'SCRAM-SHA-256=[password=test123],SCRAM-SHA-512=[password=test123]' --entity-type users --entity-name test

因为添加了zk的acl，所以需指定ZK的acl jaas账号信息。若不指定ZK的jaas账号信息，会报如下错误：

org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /config/users/test
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:120)
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:54)

加acl认证信息的方式：先调整bin/kafka-run-class.sh，用于zk acl验证。这个文件被大多数的kafka bin目录下的script所引用，如添加topic的脚本等，所以改它最方便，添加一行。

# Generic jvm settings you want to add
KAFKA_OPTS="-Djava.security.auth.login.config=$base_dir/config/kafka_server_jaas.conf"
if [ -z "$KAFKA_OPTS" ]; then
  KAFKA_OPTS=""
fi

再执行一次添加即可，输出如下提示表示添加用户test完成。

Completed Updating config for entity: user-principal 'test'.

• 添加consumer 授权方式1，先加topic权限：

bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9093 --command-config config/adminclient-configs.conf --add --allow-principal User:test --operation Read --operation Write --operation AlterConfigs --operation Describe --operation DescribeConfigs --operation Alter --topic "test"

输出：

Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test, patternType=LITERAL)`:
     (principal=User:test, host=*, operation=ALTER, permissionType=ALLOW)
    (principal=User:test, host=*, operation=READ, permissionType=ALLOW)
......

再添加消费组权限：

bin/kafka-acls.sh --bootstrap-server localhost:9093 --command-config adminclient-configs.conf --add \
--allow-principal User:test --operation Read --group '*'

• 添加consumer 授权方法2，参考文档。一步到位：

bin/kafka-acls.sh --bootstrap-server localhost:9093 --command-config config/adminclient-configs.conf \
--add --allow-principal User:test \
--consumer --topic test-1 --group '*'

至此，SERVER侧的内容说完了，是以CLI的方式进行配置，实际中，做用户及ACL的管理，可以参考KOWL等工具，使用adminclient api进行管理，做一些web工具。