一、方案

1. 不接入外部存储，基于kafka原生ACL认证

环境： kafka-2.2.1、zookeeper-3.6.3

​ kafka给我们提供了SASL/SCRAM模式，将SASL、ACL规则信息存储到zookeeper中，并且通过KafkaClientAdmin Api，新增、编辑、删除规则，其特性如下

• 应用发送、消费实现动态身份认证和授权

  基于kafka SASL/SCRAM模式，客户端会在建立连接进行SASL身份验证，在消费发送时，进行ACL鉴权

• 安全认证代码无侵入性、兼容原生kafka api

  有两种配置方式，通过根据-Djava.security.auth.login.config指定jaas配置文件，并且配置producer.properties/consumer.properties 开启SASL配置，兼容原生kafka api

• 账号级别区分：管理账号有最高权限、业务账号限定资源权限

  可通过配置kafka集群超级管理员，通过超级管理员账号登录是不进行ACL鉴权。因此超级管理员可以操作所有的资源权限

• 安全认证支持动态开启、关闭

  kafka提供SASL和ACL管理的api，通过这些api可以新增、修改、查询、删除SASL和ACL策略。通过配置策略，可以实现安全策略开启、关闭

1.1 客户端与kafka集群认证、授权

1.2 安全认证、授权步骤如下

1. 在集群初始化时，配置开启SASL、ACL，包括

   • broker与broker的SASL认证配置
   • broker与zookeeper的SASL认证配置

2. 添加SASL、ACL规则

   • 管理员账号：添加SASL管理员账号，管理员账号新增完成后，broker与broker的认证才会成功。
   • 应用账号：新增主题、消费组时，通过KafkaAdminClient Api新增以alice为用户名，应用scret-1为密码的SASL认证信息。并且给这个用户授予消息写入权限

3. 客户端配置SASL配置，并发送消息

   在properties中配置SASL账号密码。或者指定jaas配置文件

4. 服务端认证、鉴权

   在客户端与服务端建立连接时，对客户端alice进行身份认证（SASL），认证通过后，在发送消息时检查资源（主题）是否授权（ACL）给改用户alice

zookeeper与broker认证、鉴权流程（SASL/SCRAM + ACL）

---

二、实战：kafka SASL+ACL配置步骤

第一步：集群配置

• Zookeeer SASL配置

  修改zoo.cfg配置，开启SASL

  ...
  sessionRequireClientSASLAuth=true
  zookeeper.sasl.client=true
  authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
  

  • sessionRequireClientSASLAuth:默认为false， 当设置为true 时，Zookeeper 服务器将只接受通过 SASL 向服务器进行身份验证来自客户端的连接和请求。使用 SASL 配置但身份验证失败（即使用无效凭据）将无法与服务器建立会话

  • zookeeper.sasl.client: 将值设置为false以禁用SASL身份验证。默认值为true

  • authProvider.1: 自定义ZooKeeper 添加新类型的SASL身份验证方案。指定的类需要实现AuthenticationProvider接口。zookeeper提供实现： org.apache.zookeeper.server.auth.SASLAuthenticationProvider

  新增JAAS配置文件：zoo_jaas.conf，用于定义认证身份信息。我们使用的LoginModule实现为： org.apache.zookeeper.server.auth.DigestLoginModule，它是基于明文用户、密码的形式进行身份认证

  Server {
         org.apache.zookeeper.server.auth.DigestLoginModule required
         username="zookeeper"
         password="zookeepersecret”
         user_kafka="kafkasecret";
  };
  

  指定jaas配置文件

  export SERVER_JVMFLAGS="-Djava.security.auth.login.config=/Users/vhicool/zookeeper/conf/zoo_jaas.conf"
  

  启动zookeeper节点

  ./bin/zkServer.sh start-foreground
  

• Kafka SASL配置

  创建证书

  • 创建证书

    在配置kafka broker之前，我们需要新增broker账号，这一步是必须的，需要在broker重启应用SASL+ACL配置前执行。这个username=admin，password=adminsecret，用了进行broker直接的SASL认证

    bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=adminsecret]' --entity-type users --entity-name admin
    

  • 服务端

    用户认证信息配置 jaas.conf，其中KafkaServer是服务端用来broker与broker直接的SASL认证，Client是kafka broker作为客户端与zookeeper服务端直接的认证。与SASL/PALIN模式不同的是KafkaServer不能指定kafka客户端用户认证账号，而是通过bin/kafka-configs.sh来配置。在Client中的LoginModule同样是明文用户、密码认证，和zookeeper的org.apache.zookeeper.server.auth.DigestLoginModule对应

    //broker与broker
    KafkaServer {
      org.apache.kafka.common.security.scram.ScramLoginModule required
        username="admin"
        password="adminsecret";
    };
    //broker与zookeeper
    Client{
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="kafka"
        password="kafkasecret";
    };
    

    指定kafka环境变量，是SASL生效

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

    server.properties配置，通过指定super.users指定超级用户，超级用户是不需要进行ACL鉴权，可作为管理员账号

    #SASL/SCRAM
    listeners=SASL_PLAINTEXT://host.name:port
    security.inter.broker.protocol=SASL_PLAINTEXT
    sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
    sasl.enabled.mechanisms=SCRAM-SHA-25
    #ACL
    authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
    super.users=User:admin
    

第二部：新增认证、授权规则

• kafka cli工具连接

  除了使用kafka自带cli，同时也可以使用kafka通过的api去维护：KafkaClientAdmin

  • 新增kafka client JAAS认证文件：kafka_client_jaas.conf

    //kafka client与zookeeper SASL
    Client{
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="kafka"
        password="kafkasecret";
    };
    // kafka client与broker SASL
    KafkaClient {
        org.apache.kafka.common.security.scram.ScramLoginModule required
        username="admin"
        password="adminsecret";
    };
    

  • 新增broker启动环境变量，指定JAAS文件

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

  • 通过命令工具新增主题

    验证：kafka client与zookeeper SASL

    bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic-1
    

  • 给用户alice SASL认证

    bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=alice-secret]' --entity-type users --entity-name alice
    

  • 给主题ACL授权

    授予用户alice对主题topic-1生产者权限。涉及验证：kafka client与zookeeper SASL（此处我们并没有开启zookeeper ACL，因此不需要zookeeper ACL权限验证）

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

  • 测试发送消息

    涉及验证： kafka client与broker SASL+ACL

    bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-1 --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=SCRAM-SHA-256
    

第三步：客户端配置

在用户alice配置SASL认证，并且授权主题topic-1发送权限后，客户端通过原生api发送消息

		Properties kafkaProperties = new Properties();
		kafkaProperties.put(org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");
		kafkaProperties.put(org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.ByteArraySerializer");
		kafkaProperties.put("bootstrap.servers", "localhost:9092");
		//SASL配置
		kafkaProperties.put("sasl.jaas.config","org.apache.kafka.common.security.scram.ScramLoginModule required username=\"alice\" password=\"alice-secret\";");
		kafkaProperties.put("security.protocol","SASL_PLAINTEXT");
		kafkaProperties.put("sasl.mechanism","SCRAM-SHA-256");

		KafkaProducer<String, byte[]> producer = new KafkaProducer<>(kafkaProperties);
		ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>("topic-1", null, null, "test message", null);
		producer.send(record).get();

三、如果想自定义SASL、ACL存储形式

1. SASL存储形式

• 本地文件存储

  PLAIN模式

  SASL/PLAIN是一种简单的用户名/密码认证机制，通常与TLS一起用于加密以实现安全认证。他的SASL认证是存储在本地JAAS文件，因此如果需要实现动态身份认证，需要拓展SASL认证接口

• zookeeper存储

  SASL/SCRAM

  它解决了与执行用户名/密码认证的传统机制（如PLAIN和DIGEST-MD5）的安全问题。 RFC 5802中定义了该机制。Kafka支持SCRAM-SHA-256和SCRAM-SHA-512，它们可以与TLS一起使用以执行安全认证。用户名用作配置ACL等的认证主体。Kafka中的默认SCRAM实现在Zookeeper中存储SCRAM凭据，适用于Zookeepher位于专用网络上的Kafka安装

• 外部认证服务器存储

  SASL/Kerberos

kafka默认SASL通过静态JAAS文件配

从Kafka 2.0版开始，可以通过使用配置选项sasl.server.callback.handler.class和sasl.client.callback.handler.class配置自己的回调处理程序来从外部源获取用户名和密码，从而避免在磁盘上存储明文密码。

• sasl.server.callback.handler.class

  实现 AuthenticateCallbackHandler 接口的 SASL 服务器回调处理程序类的完全限定名称。服务器回调处理程序必须以侦听器前缀和小写的 SASL 机制名称为前缀。例如，listener.name.sasl_ssl.plain.sasl.server.callback.handler.class=com.example.CustomPlainCallbackHandler

• sasl.client.callback.handler.class

  实现 AuthenticateCallbackHandler 接口的 SASL 客户端回调处理程序类的完全限定名称。

2. ACL存储形式

​ kafka默认ACL存储在zookeeper

自定义扩展外部存储

通过在客户端配置authorizer.class.name，可自定义实现存储位置。kafka默认是通过zookeeper存储ACL规则，实现类为：kafka.security.auth.SimpleAclAuthorizer。如果我们要扩展ACL存储，需要自定义认证类并实现kafka.security.auth.Authorizer接口，包括实现authorize、addAcls、removeAcls、getAcls等方法

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