技术储备

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

• SSL
  是一种间于传输层（比如TCP/IP）和应用层（比如HTTP）的协议。对传输内容进行加密，如HTTPS
  如果使用了SASL但是没有使用SSL，那么服务端可以认证客户端的身份，但是后续传输的数据是明文的。

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

---

kafka安全认证

官方文档：Apache Kafka

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

• 服务器与zookeeper之间连接认证
• 基于SASL认证机制：验证对方身份是否合法（使用用户、密码验证），服务器端与服务器端、服务端与客户端（主题、消费组）、服务器端与管理工具
• 基于ACL授权机制：验证可以访问哪些服务、资源（用户授权可发送主题、可订阅的消费组），服务端与客户端（主题、消费组）
• 基于SSL数据传输加密

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

• GSSAPI (Kerberos)
• PLAIN
• SCRAM-SHA-256
• SCRAM-SHA-512

1. 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)

2. 通过SASL PLAIN实现认证

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

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

2.1 broker 配置

• 在每个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的连接。

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

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

• 在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
  

• 启动broker

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

2.2 客户端配置

• 通过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
  

• 通过命令工具连接

  • 新增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来插入自己的回调处理程序，这些回调处理程序使用外部身份验证服务器进行密码验证。

---

2. broker授权ACL

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

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

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文档。

• 新增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
  

• 查询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
  

扩展阅读

实战：kafka、zookeeper SASL+ACL实现动态权限认证、授权