kafka的访问控制

【概述】

通常情况下，Kafka部署后都是自己的业务进行生产消费，但也有一些情况，比如通过kafka和第三方对接，甚至是多个三方对接；或者是多用户使用同一套kafka集群，各自使用不同的topic。在这种场景下，一般不希望不同的用户能访问彼此的数据，因此需要进行权限控制，这就会用到Kafka中的ACL。

Kafka中的ACL定义为：来自指定主机（Host）的指定用户（User）对任意资源（Resource）的操作（Operation）是否符合指定的资源模式（ResourcePattern）。默认情况下是不开启的，通过在配置文件中增加如下配置项，则启用了kafka的ACL机制。

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

kafka中的ACL模块是以插件化的形式存在的，除了自带默认的实现外，可以较容易的引入外部插件实现权限控制，例如ranger、santry。

【kafka中的资源与对应操作】

上面提到了ACL定义本质上是对指定资源的指定操作的访问控制，在kafka中的资源包括主题（Topic）、消费者组（ConsumerGroup）、集群（Cluster）、事务（ClusterID）以及委派token（DelegationToken）。对我们而言，最常用的则是主题、消费者组和集群。

对于topic涉及的操作包括：

对于消费者组涉及的操作包括：

对于集群涉及的操作包括：

注：kafka自身需要有对cluster的各种访问权限，尤其是ClusterAction，这个涉及分区leader选举、ISR同步等等操作，可以理解为生产消费的基础，否则即便给用户配置了read、write、describe权限，仍旧是无法正常生产消费的。

【ACL的增删查改】

知道了kafka中的资源以及对应的操作后，剩下的就是对ACL的增删查了。这些操作都是通过"kafka-acls.sh"脚本来完成的，命令使用方式为：

kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:xxx --alow-host xxx --operation xxx --topic/cluster/group/delegation-token/transactional-id
# --authorizer-properties 为必填项, 后面的值以K=V形式体现, 当前仅能配置zookeeper.connect指定zk的地址
# --add 添加acl 或者 --remove 删除acl 或者 --list 罗列acl
# --allow-principal 指定用户, 用户格式为User:xxx, 其中xxx为具体的用户名
# --allow-host 指定主机地址
# --operation 指定操作
# --topic 指定topic
# --cluster 指定集群
# --group 指定消费者组

例如，添加允许用户hncscwc（不限定主机）在名为bigdata的topic上进行生产的命令为：

kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:hncscwc --operation Write --operation Describe --topic bigdata

效果如下所示：

进行消息生产的验证：

同样，如果需要允许该用户在指定topic上进行消费，除了需要为topic资源配置ACL以外，还需要为group配置资源。kafka为了简化操作，在kafka-acls.sh脚本中提供了"--consumer"参数，通过该参数会自动对topic与group进行配置，同样还包括"--producer"参数。

注1：使用"--consumer"，需要同时通过"--topic"、"--group"分别指定主题和消费者组

注2：通配符*需要加''

默认情况下，不在指定ACL的非超级用户的操作都是没有权限的，可以通过如下配置修改默认行为，即默认使其具有操作权限。

allow.everyone.if.no.acl.found=true

另外，配置超级用户的方式为：

super.users=User:xxx;User:xxx

【运行机制】

从上面的使用介绍中可以知道，kafka的acl使用还是相对比较简单的，具体的实现逻辑如下图所示：

首先通过"kafka-acls.sh"脚本增加删除ACL的操作，本质是与ACL进行交互，并将信息写入对应的znode中，也就是说kafka的acl信息是存储在zookeeper中的。

具体为指定根节点下，名为kafka-acl的znode，在这个下面分别存储了Topic、Group、Cluster等资源的ACL信息。

注：从这个流程可以看出，为什么一定要指定"--authorizer-properties"参数。

其次，"kafka-acls.sh"脚本向zookeeper写入ACL信息后，还会写入另外一个znode信息："kafka-acl-changes"，通过以版本号递增的形式写入改动的资源项。

而kafka broker则watch该节点，以知晓acl的新增删除等变更情况。一旦知道acl有变更，重新从zookeeper中读取资源的acl信息，并更新内存中的记录信息。此后，客户端进行生产消费时则可以进行权限校验。

【总结】

小结一下，本文简要概述了Kafka中的访问控制，如何配置，如何增删查ACL，以及背后的实现逻辑。而所有这一切的一个前提，也是本文没有提到的：kafka需要开启认证（通常是kerberos认证），这样kafka才能正确感知客户端的用户信息，也就才能正确地进行访问控制。另外，在对资源进行配置时，还可以进行一些表达式的匹配来进行灵活控制，详细可以查看官网文档。