破绽简介
Apache Kafka是一个分布式数据流解决平台，能够实时公布、订阅、存储和解决数据流。Kafka Connect是一种用于在kafka和其余零碎之间可扩大、牢靠的流式传输数据的工具。攻击者能够利用基于SASLJAAS 配置和SASL 协定的任意Kafka客户端，对Kafka Connect worker 创立或批改连接器时，通过结构非凡的配置，进行JNDI 注入来实现近程代码执行。
影响版本
2.4.0<=Apache kafka<=3.3.2
修复计划
更新Apache Kafka至官网最新版本
环境搭建
通过https://github.com/vulhub/vulhub搭建
破绽复现
exp可参考：
https://github.com/projectdiscovery/nuclei-templates/blob/5d90e8275084b0ae9166ec38cacd22e5a5a94fb8/http/vulnerabilities/apache/apache-druid-kafka-connect-rce.yaml
发动攻打申请：
结构payload ,执行新建/tmp/test.txt文件
验证破绽存在，文件新建胜利
开启RASP后发动攻打：
在业务优先模式下，RASP会呈现JNDI注入的告警，拦挡最终的命令执行
堆栈信息为
在防护模式下将间接在JNDI注入处被拦挡
堆栈信息为
破绽剖析
开始org.apache.kafka.clients.producer.KafkaProducer#KafkaProducer(java.util.Properties)
跟进到org.apache.kafka.clients.producer.KafkaProducer#KafkaProducer(java.util.Propertiesorg.apache.kafka.common.serialization.Serializer<K>,org.apache.kafka.common.serialization.Serializer<V>)
调用org.apache.kafka.common.utils.Utils#propsToMap对传入对象进行解决
将map型的对象传入org.apache.kafka.clients.producer.KafkaProducer#KafkaProducer(java.util.Map<java.lang.String,java.lang.Object>org.apache.kafka.common.serialization.Serializer<K>org.apache.kafka.common.serialization.Serializer<V>)
之后调用org.apache.kafka.clients.producer.ProducerConfig#appendSerializerToConfig
将返回的newConfigs传入org.apache.kafka.clients.producer.ProducerConfig#ProducerConfig(java.util.Map<java.lang.String,java.lang.Object>)
将配置参数传入org.apache.kafka.clients.producer.KafkaProducer#KafkaProducer(org.apache.kafka.clients.producer.ProducerConfig,org.apache.kafka.common.serialization.Serializer<K>org.apache.kafka.common.serialization.Serializer<V>org.apache.kafka.clients.producer.internals.ProducerMetadata,org.apache.kafka.clients.KafkaClientorg.apache.kafka.clients.producer.internals.ProducerInterceptors<K,V>org.apache.kafka.common.utils.Time)
赋值后调用org.apache.kafka.clients.producer.KafkaProducer#newSender
调用到org.apache.kafka.clients.ClientUtils#createChannelBuilder
赋值后调用org.apache.kafka.common.network.ChannelBuilders#clientChannelBuilder
这里对值做了一个判断后调用org.apache.kafka.common.network.ChannelBuilders#create
Create办法中失去map型的configs后进行switch，失去SaslChannelBuilder类型channelBuilder的对象，switch完结后调用了org.apache.kafka.common.network.SaslChannelBuilder#configure
org.apache.kafka.common.network.SaslChannelBuilder#configure进入循环后到org.apache.kafka.common.security.authenticator.LoginManager#acquireLoginManager
判断值后到org.apache.kafka.common.security.authenticator.LoginManager#LoginManager
跟进到org.apache.kafka.common.security.authenticator.AbstractLogin#login
调用javax.security.auth.login.LoginContext#login
调用javax.security.auth.login.LoginContext#invokePriv
调用javax.security.auth.login.LoginContext#invoke进行逻辑判断后调用initialize办法
Initialize中失去userProvider
user.provider.url通过jndi提供
调用com.sun.security.auth.module.JndiLoginModule#login
调用com.sun.security.auth.module.JndiLoginModule#attemptAuthentication
通过javax.naming.InitialContext#lookup(java.lang.String)执行userProvider的值
因为RASP对javax.naming.InitialContext.lookup调用做了防护策略检测，所以会在此处拦挡。
Reference
https://github.com/luelueking/Java-CVE-Lists#cve-2023-25194
https://blog.snert.cn/index.php/2023/04/04/cve-2023-25194-kaf...
对于云鲨RASP
悬镜云鲨RASP助力企业构建利用平安爱护体系、搭建利用平安研运闭环，将踊跃防御能力注入业务利用中，实现利用平安自免疫。详情欢送拜访云鲨RASP官网https://rasp.xmirror.cn/