POC来源于Github请在虚拟环境中，进行使用。要求：

• Java 11

• Maven

• Docker ( or Kafka )

用法

首先启动 Dockerized Kafka 实例。这将启动 Kafka 并使其在端口 29092 上可用。

docker-compose up

启动消费者应用程序。

cd spring-kafka-consumermvn clean installmvn spring-boot:run

这将构建并启动消费者应用程序。消费者在关闭之前将等待最多 10 分钟的消息。

cd spring-kafka-producermvn clean installmvn spring-boot:run

这将构建并启动生产者应用程序。生产者将向 kafka 队列发送一条消息然后关闭。

Spring-Kafka-Consumer配置

要使此漏洞成功，需要在使用者上启用以下一个或两个标志。CheckDeserExWhenValueNull CheckDeserExWhenKeyNull 这是在消费者应用程序中完成的KafkaConsumerConfig.greetingKafkaListenerContainerFactory()

有效载荷

有两种可以触发的有效负载，默认情况下使用 RCE 有效负载。

拒绝服务

要启用 DOS 有效负载，请修改方法  KafkaApplication.sendGreetingMessage()更改作为标头添加的有效负载dosPayload并重建并运行生产者。注意：由于 DOS 在读取消息时发生，并且读取永远不会完成，因此消息将保留在队列中，直到手动删除或消息保留时间到期。这增加了 DOS 的效力，因为它使该队列无用，直到发生手动干预或保留时间到期。可能会导致 DOS 消息之后立即发送的消息丢失数据。

远程代码执行

要启用 DOS 有效负载，请修改方法 KafkaApplication.sendGreetingMessage() 更改作为标头添加的有效负载rcePayload并重建并运行生产者。默认情况下，运行的命令是 touch /tmp/newfile查看攻击是否成功，在  /tmp 下查找名为 newfile 的文件。如果在 Windows 上运行，您可以将命令字符串修改为更合适的内容。这个小工具很大程度上只是一个 POC。为了在现实世界中发生 RCE，消费者类路径上需要有一个可用的小工具类。

怎么运行的

拒绝服务不要求消费者类路径上有任何可用的特定小工具类。org.springframework.kafka.support.serializer.DeserializationException它依赖于生成包含对象的类的修改版本 。这使得我们可以轻松地将我们喜欢的任何有效负载添加到序列化对象中。这个修改后的类被称为xrg.springframework.kafka.support.serializer.DeserializationException（注意包名称开头的 x）。一旦有效负载被注入，在这种情况下，就会使用十亿笑式攻击java.util.Set，并且java.lang.Object它会被序列化。然后修改二进制数据以将 x 更改为 o 以满足消费者的期望。

然后将此序列化的异常类作为消息头添加到springDeserializerExceptionValue标头和springDeserializerExceptionKey标头中。如果密钥或消息为空，则消费者将读取这些内容。之后，只需确保密钥或消息为空，消费者就会读取它。

Spring-Kafka 内部有一些反序列化保护。在 ListenerUtils 中。

public static DeserializationException byteArrayToDeserializationException(LogAccessor logger, byte[] value) {    try {        ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(value)) {
            boolean first = true;
            @Override            protected Class<?> resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException {                if (this.first) {                    this.first = false;                    Assert.state(desc.getName().equals(DeserializationException.class.getName()),                            "Header does not contain a DeserializationException");                }                return super.resolveClass(desc);            }

        };        return (DeserializationException) ois.readObject();    }    catch (IOException | ClassNotFoundException | ClassCastException e) {        logger.error(e, "Failed to deserialize a deserialization exception");        return null;    }}

您可以看到进行了检查以确保顶级类是org.springframework.kafka.support.serializer.DeserializationException但请注意，仅检查顶级类，然后仅检查类名称（攻击者可以修改）。因此，低于该级别的任何有效负载都会被反序列化。

项目地址：

https://github.com/Contrast-Security-OSS/Spring-Kafka-POC-CVE-2023-34040