kafka安全认证与授权

一、SASL、SSL、ACL介绍

1．SASL

鉴权协议，主要用来保证客户端登录服务器的时候，传输的鉴权数据的安全性，SASL是对用户名和密码加解密用的

2．SSL

是一种间于传输层（比如TCP/IP）和应用层（比如HTTP）的协议。对传输内容进行加密，如HTTPS

如果使用了SASL但是没有使用SSL，那么服务端可以认证客户端的身份，但是后续传输的数据是明文的。

3．ACL

Access Control List译为访问控制列表，它是通过一系列定义的规则，对访问进行控制，对满足规则的问题予以通过、丢弃或拒绝。

二、kafka安全认证

官方文档：https://kafka.apache.org/21/documentation.html#security_authz

1．kafka的安全特性：0.10版本以上

l 服务器与zookeeper之间连接认证

l 基于SASL认证机制：验证对方身份是否合法（使用用户、密码验证），服务器端与服务器端、服务端与客户端（主题、消费组）、服务器端与管理工具

l 基于ACL授权机制：验证可以访问哪些服务、资源（用户授权可发送主题、可订阅的消费组），服务端与客户端（主题、消费组）

l 基于SSL数据传输加密

2．Kafka 支持以下的四个 SASL 机制:

GSSAPI (Kerberos)

PLAIN

SCRAM-SHA-256

SCRAM-SHA-512

3．broker SASL认证

SASL/PLAIN

在server.properties中配置SASL端口，方法是将SASL_PLAINTEXT或SASL_SSL中的至少一个添加到listeners参数中，该参数包含一个或多个逗号分隔的值：

listeners=SASL_PLAINTEXT://host.name:port

如果只配置一个SASL端口（或者如果希望Kafka代理使用SASL相互验证），请确保为代理间通信设置相同的SASL协议：

security.inter.broker.protocol=SASL_PLAINTEXT (or SASL_SSL)

4．通过SASL PLAIN实现认证

SASL/PLAIN是一种简单的用户名/密码认证机制，通常与TLS一起用于加密以实现安全认证。Kafka支持SASL/PLAIN的默认实现，可以按照这里的描述扩展到生产用途。

用户名用作配置ACL等的身份验证主体

（1）broker 配置

l 在每个Kafka broker的配置目录中添加一个经过适当修改的JAAS文件，类似于下面的文件，在本例中，我们将其称为kafka_server_jaas.conf

KafkaServer {

     org.apache.kafka.common.security.plain.PlainLoginModule required

     username="admin"

     password="admin-secret"

     user_admin="admin-secret"

     user_alice="alice-secret";

 };

此配置定义了两个用户（admin和alice）。KafkaServer中的username、password配置的用户和密码，是用来broker和broker连接认证。在本例中，admin是代理broker间通信的用户。user_userName配置为连接到broker的所有用户定义密码，broker使用这些验证所有客户端连接，包括来自其他broker的连接。

l 将JAAS配置文件位置作为JVM参数传递给每个Kafka broker：export KAFKA_OPTS=

 export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"

l 在server.properties中配置SASL端口和SASL机制，如下所述。例如

security.inter.broker.protocol:用于broker之间通信的安全协议。有效值为：PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL

sasl.mechanism.inter.broker.protocol:用于broker间通信的SASL机制，默认值为GSSAPI

sasl.enabled.mechanisms:Kafka服务器中启用的SASL机制列表。该列表可以包含安全提供程序可用的任何机制。默认情况下仅启用GSSAPI。

listeners=SASL_SSL://host.name:port

security.inter.broker.protocol=SASL_PLAINTEXT

sasl.mechanism.inter.broker.protocol=PLAIN

sasl.enabled.mechanisms=PLAIN

l 启动broker

bin/kafka-server-start.sh config/server-sasl.properties &

（2）客户端配置

l 通过kafka client连接

为producer.properties或consumer.propertie中的每个客户端配置JAAS配置属性。登录模块描述了producer和consumer等客户端如何连接到Kafka Broker。以下是PLAIN机制的客户端配置示例：

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \

        username="alice" \

        password="alice-secret";

客户端使用选项username和password为客户端连接配置用户。在本例中，客户端作为用户alice连接到代理。通过在sasl.jaas.config中指定不同的用户名和密码，JVM中的不同客户端可以作为不同的用户进行连接。

客户机的JAAS配置也可以指定为JVM参数，类似于这里描述的代理。客户端使用名为KafkaClient的登录部分。此选项仅允许一个用户用于JVM中的所有客户端连接。

在producer.properties或consumer.property中添加一下配置

security.protocol=SASL_PLAINTEXT

sasl.mechanism=PLAIN

l 通过命令工具连接

新增JAAS认证文件kafka_client_jaas.conf

KafkaClient {

    org.apache.kafka.common.security.plain.PlainLoginModule required

    username="alice"

    password="alice-secret";

};

设置kafka环境变量KAFKA_OPTS

export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka_2.11-2.2.1/config/kafka_client_jaas.conf"

启动指定配置security.protocol和sasl.mechanism

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-sasl3 --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN

在线上使用SASL/PALIN模式

SASL/PLAIN应仅与SSL一起使用作为传输层，以确保在没有加密的情况下不会在线路上传输清晰的密码。

Kafka中SASL/PLAIN的默认实现在JAAS配置文件中指定用户名和密码，如下所示。从Kafka 2.0版开始，可以通过使用配置选项sasl.server.callback.handler.class和sasl.client.callback.handler.class配置自己的回调处理程序来从外部源获取用户名和密码，从而避免在磁盘上存储明文密码。

在生产系统中，外部认证服务器可以实现密码认证。从Kafka 2.0版开始，可以通过配置sasl.server.callback.handler.class来插入自己的回调处理程序，这些回调处理程序使用外部身份验证服务器进行密码验证。

5．broker授权ACL

Kafka附带了一个可插入的授权器和一个开箱即用的授权器实现，该实现使用zookeeper来存储所有acl。通过在server.properties中设置Authorizer.class.name来配置授权程序。要启用现成的实现，使用：

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

1

Kafka acl的定义格式为“Principal P is[Allowed/Dedenied]Operation O From Host H on any Resource R matching ResourcePattern RP”。可以阅读有关KIP-11中acl结构和KIP-290中资源模式的更多信息。为了添加、删除或列出acl，可以使用Kafka authorizer CLI。默认情况下，如果没有任何ResourcePatterns与特定的Resource R匹配，则R没有关联的acl，因此除了超级用户之外，任何人都不允许访问R。如果要更改该行为，可以在server.properties中包含以下内容

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

还可以在server.property中添加超级用户，如下所示（注意分隔符是分号，因为SSL用户名可能包含逗号）。默认PrincipalType字符串“User”区分大小写

super.users=User:admin

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

principal.builder.class=CustomizedPrincipalBuilderClass

默认情况下，SASL用户名将是Kerberos主体的主要部分。可以通过在server.properties中将sasl.kerberos.principal.to.local.rules设置为自定义规则来更改此设置。sasl.kerberos.pprincipal.to.local-rules的格式是一个列表，其中每个规则的工作方式与kerberos配置文件（krb5.conf）中的auth_to_local相同。这还支持其他小写规则，以强制转换结果全部为小写。这是通过在规则末尾添加“/L”来完成的。检查以下格式的语法。每个规则都以RULE:开头，并包含以下格式的表达式。有关详细信息，请参阅kerberos文档。

l 新增ACL规则

用户alice添加主题test-sasl的资源acl producer权限（WRITE, DESCRIBE , CREATE）

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

基于用户和范围ip的权限控制：用户alice只接受198.51.100.0、198.51.100.1对主题test-sasl进行读写

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-sasl

基于正则匹配，对匹配上的所有资源赋予权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice --producer --topic test- --resource-pattern-type prefixed

l 查询ACL

基于主题资源维度查询

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

所有主题资源ACL

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

删除ACL

  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-sasl

删除基于前缀的ACL

   bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --remove --allow-principal User:Jane --producer --topic Test- --resource-pattern-type Prefixed

通过AdminClient api添加ACL

bin/kafka-acls.sh --bootstrap-server localhost:9092 --command-config /tmp/adminclient-configs.conf --add --allow-principal User:Bob --producer --topic Test-topic

bin/kafka-acls.sh --bootstrap-server localhost:9092 --command-config /tmp/adminclient-configs.conf --add --allow-principal User:Bob --consumer --topic Test-topic --group Group-1

bin/kafka-acls.sh --bootstrap-server localhost:9092 --command-config /tmp/adminclient-configs.conf --list --topic Test-topic