java - 如何使用 keystore 检查 SSL Kafka 主题是否已授权

Question

我有一个简单的KafkaProducer使用 keystore.jks 生成 SSL Kafka 主题文件

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("security.protocol", "SSL");
props.put("ssl.key.password", "password");
props.put("ssl.keystore.password", "password");
props.put("ssl.keystore.location", "/keystore.jks");

Producer<String, String> producer = new KafkaProducer<>(props);

producer.send(new ProducerRecord<String, String>("my-topic", key, body));

但是自从 KafkaProducer是一个惰性实例，如果主题未经授权，它会在第一条消息到达时创建与 Kafka 集群的连接，从而导致运行时失败
错误:

org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [my-topic]

有没有办法使用 KafkaAdminClient 检查主题授权？还是在启动 KafkaProducer 之前编写自定义逻辑？我确实有以下代码从 keystore 中提取证书和 key 文件但找不到任何后续步骤

KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());  
char[] pwdArray = "password".toCharArray();
keyStore.load(new FileInputStream("/keystore.jks"), pwdArray);

Answer 1

虽然您没有为 trust 配置必要的属性卡夫卡服务器:

ssl.truststore.location=/var/private/ssl/client.truststore.jks
ssl.truststore.password=test1234

可能您的 SSL 设置很好，但正如您所指出的，存在与您的客户端授权相关的某种问题:

org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [my-topic]

请考虑查看 Kafka authorization documentation ，尤其是 part描述了用户名是如何从 CN 派生的您的客户端 SSL 证书:

By default, the SSL user name will be of the form "CN=writeuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown". One can change that by setting ssl.principal.mapping.rules to a customized rule in server.properties. This config allows a list of rules for mapping X.500 distinguished name to short name.

识别用户后，检查是否为他应用了必要的 ACL 和主题 my-topic .请参阅提供的 examples在文档中。
例如添加上述用户为话题的生产者 my-topic尝试类似的东西:

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

您可以验证应用于资源的 ACL， my-topic在您的用例中，例如:

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