7.4 认证和acl

kafka附带一个可插拔的ACL（Access Control List 访问控制列表），它使用zookeeper来存储。通过在server.properties中设置authorizer.class.name来启用：

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

Kafka acls的格式为 "Principal P is [Allowed/Denied] Operation O From Host H On Resource R”，你可以使用Kafka authorizer CLI 来添加，删除或查询所有acl。默认情况下，如果ResourcePatterns与特定的资源R没有匹配，则除了超级用户之外，都不允许访问R。如果要更改该行为，可以在server.properties中包含以下内容。

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

你也可以在server.properties添加超级用户，像这样（注意分隔符是分号，因为SSL的用户名是逗号）。

super.users=User:Bob;User:Alice

默认情况下，SSL用户名的格式为“CN=writeuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown”。可以通过在server.properties中设置自定义的PrincipalBuilder来改变它，如下所示:

principal.builder.class=CustomizedPrincipalBuilderClass

可以通过修改server.properties中的sasl.kerberos.principal.to.local.rules自定义规则。sasl.kerberos.principal.to.local.rules的格式是一个列表，其中每个规则的工作方式与Kerberos 配置文件 (krb5.conf)中的auth_to_local相同。 也支持小写规则，可通过在规则的末尾添加“/L”，强制转移全部结果为小写。每个规则都以RULE开头：并包含一个表达式，格式如下。 有关更多详细信息，请参阅kerberos文档。

RULE:[n:string](regexp)s/pattern/replacement/
RULE:[n:string](regexp)s/pattern/replacement/g
RULE:[n:string](regexp)s/pattern/replacement//L
RULE:[n:string](regexp)s/pattern/replacement/g/L

举个例子，添加规则，将user@MYDOMAIN.COM转换为用户，同时保持默认规则，示例如下：

sasl.kerberos.principal.to.local.rules=RULE:[1:$1@$0](.*@MYDOMAIN.COM)s/@.*//,DEFAULT

命令行界面

Kafka认证管理CLI（和其他的CLI脚本）可以在bin目录中找到。CLI脚本名是kafka-acls.sh。以下列出了所有脚本支持的选项：

选项	描述	默认	类型选择
--add	添加一个acl		Action
--remove	移除一个acl		Action
--list	列出acl		Action
--authorizer	authorizer的完全限定类名	kafka.security.auth.SimpleAclAuthorizer	Configuration
--authorizer-properties	key=val，传给authorizer进行初始化，例如：zookeeper.connect=localhost:2181		Configuration
--cluster	指定集群作为资源。		Resource
--topic [topic-name]	指定topic作为资源。		Resource
--group [group-name]	指定 consumer-group 作为资源。		Resource
-allow-principal	添加到允许访问的ACL中，Principal是PrincipalType:name格式。 你可以指定多个。		Principal
--deny-principal	添加到拒绝访问的ACL中，Principal是PrincipalType:name格式。 你可以指定多个。		Principal
--allow-host	--allow-principal中的principal的IP地址允许访问。	如果--allow-principal指定的默认值是*，则意味着指定“所有主机”	Host
--deny-host	允许或拒绝的操作。 有效值为：读，写，创建，删除，更改，描述，ClusterAction，全部	ALL	Operation
--operation	--deny-principal中的principals的IP地址拒绝访问。	如果 --deny-principal指定的默认值是 * 则意味着指定 "所有主机"	Host
--producer	为producer角色添加/删除acl。生成acl，允许在topic上WRITE, DESCRIBE和CREATE集群。		Convenience
--consumer	为consumer role添加/删除acl，生成acl，允许在topic上READ, DESCRIBE 和 consumer-group上READ。		Convenience
--force	假设所有操作都是yes，规避提示		Convenience

例子

• 添加acl
  假设你要添加一个acl “以允许198.51.100.0和198.51.100.1，Principal为User:Bob和User:Alice对主题是Test-Topic有Read和Write的执行权限” 。可通过以下命令实现：

  bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Bob --allow-principal User:Alice --allow-host 198.51.100.0 --allow-host 198.51.100.1 --operation Read --operation Write --topic Test-topic
  

  默认情况下，所有的principal在没有一个明确的对资源操作访问的acl都是拒绝访问的。在极少的情况下，acl允许访问所有的资源，但一些principal我们可以使用 --deny-principal 和 --deny-host来拒绝访问。例如，如果我们想让所有用户读取Test-topic，只拒绝IP为198.51.100.3的User:BadBob，我们可以使用下面的命令:

  bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:* --allow-host * --deny-principal User:BadBob --deny-host 198.51.100.3 --operation Read --topic Test-topic
  

  需要注意的是--allow-host和deny-host仅支持IP地址（主机名不支持）。上面的例子中通过指定--topic [topic-name]作为资源选项添加ACL到一个topic。同样，用户通过指定--cluster和通过指定--group [group-name]消费者组添加ACL。

• 删除acl
  删除和添加是一样的，--add换成--remove选项，要删除第一个例子中添加的，可以使用下面的命令：

  bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --remove --allow-principal User:Bob --allow-principal User:Alice --allow-host 198.51.100.0 --allow-host 198.51.100.1 --operation Read --operation Write --topic Test-topic
  

• acl列表
  我们可以通过--list选项列出所有资源的ACL。假设要列出Test-topic，我们可以用下面的选项执行CLI所有的ACL：

  bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --list --topic Test-topic
  

• 添加或删除作为生产者或消费者的principal
  acl管理添加/移除一个生产者或消费者principal是最常见的使用情况，所以我们增加更便利的选项处理这些情况。为主题Test-topic添加一个生产者User:Bob，我们可以执行以下命令：

  bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Bob --producer --topic Test-topic
  

  同样，添加Alice作为主题Test-topic的消费者，用消费者组为Group-1，我们只用 --consumer 选项：

  bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:Bob --consumer --topic test-topic --group Group-1
  

  注意，消费者的选择，我们还必须指定消费者组。从生产者或消费者角色删除主体，我们只需要通过--remove选项。