环境搭建：
下载3.9.0漏洞版本：https://downloads.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz.asc
解压后先启动zookeeper和kafka
./bin/zookeeper-server-start.sh config/zookeeper.properties
./bin/kafka-server-start.sh config/server.properties
在/bin/kafka-run-class.sh开头添加 export KAFKA_DEBUG=2 开启远程debug
在/config/connect-standalone.properties中将plugin.path修改为指向当前/libs(需要,结尾)
启动kafka connect
./bin/connect-standalone.sh ./config/connect-standalone.properties ./config/connect-file-source.properties
漏洞分析：
查看漏洞补丁测试：增加了新的系统属性org.apache.kafka.sasl.oauthbearer.allowed.urls来判断传入sasl.oauthbearer.token.endpoint.url和sasl.oauthbearer.jwks.endpoint.url的值是否在允许范围内
https://github.com/apache/kafka/commit/15c5c075c17f1fa8a8b958bf299e80a1b1f2ae11?diff=split
其中的测试方法也提示漏洞的触发方式
根据补丁改动的地方下入断点
org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenRetriever
进入URL tokenEndpointUrl = cu.validateUrl(“sasl.oauthbearer.token.endpoint.url”);可以看到没有读取文件的地方
org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils#validateUrl
继续跟进会把sasl.oauthbearer.token.endpoint.url的值传给accesstokenfile
org.apache.kafka.common.security.oauthbearer.internals.secured.FileTokenRetriever#FileTokenRetriever
继续跟进到相关方法的初始化中，会调用Utils.readFileAsString读取accesstokenfile的值
org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler#init
org.apache.kafka.common.security.oauthbearer.internals.secured.FileTokenRetriever#init
继续跟进到登录验证流程
org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule#identifyToken
最后在验证accesstokenfile内容不满足jwt格式从而报错带出
漏洞复现：
