Spring Kafka消息头反序列化漏洞（CVE-2023-34040）

产品介绍

Spring for Apache Kafka (Spring - Kafka)项目将Spring的核心概念应用于基于Kafka的消息传递解决方案的开发。它提供了一个“template”作为发送消息的高级抽象。它还支持带有@KafkaListener注解和“listener container”的消息驱动POJOs 。这些库提倡使用依赖注入和声明式的使用方式。所有这些例子都与Spring框架的JMS支持和Spring AMQP对RabbitMQ的支持有相似之处。

漏洞概述

当Kafka应用服务器允许一个不受信任的源的生产消息的时候，若其能够控制消息的headers，则可能导致一个反序列化漏洞的发生。

受影响版本

2.8.1 to 2.9.10
3.0.0 to 3.0.9

漏洞分析

调试环境

使用idea新建基于maven的java项目，在pom.xml中引入相关依赖包

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.6.14</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.arliya</groupId>
    <artifactId>SpringKafkaTest</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>SpringKafkaTest</name>
    <description>SpringKafkaTest</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
<!--            <version>2.9.11</version>-->
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.28</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>
</project>

这里使用spring-boot-starter-parent版本为2.6.14，其依赖的Spring Kafka版本为2.8.11，为存在漏洞版本；

为了对比，修改Spring Kafka的版本为2.9.11，为不受修复版本。

为了能够进行漏洞复现，我们需要编写一部分代码

测试代码主要包括Kafka的生产者于消费者的配置，一个生产消息的Web接口控制器，一个消费消息的Service，关于Spring Kafka的使用方式，可以参考该文章：

https://juejin.cn/post/7207411012479074364

两外，还需要搭建Apache Kafka服务器，建议使用Docker一键搭建，具体方式可以参考以下文章：

https://juejin.cn/post/6960820341631352868

漏洞原理

在进行漏洞复现前，需要有一些于Apache Kafka相关的基础知识。

Kafka 是一种分布式的，基于发布 / 订阅的消息系统。Kafka包含一些基本概念，如生产者、消费者

生产者：负责消息的产生，并推送到Kafka服务器
消费者：负责消息的消费

Kafka的一条消息有一些更精细的组件，包括Key、Value、Headers。

Key：当一条消息被定义了Key之后，Kafka Server通过消息Key的hash，将相同Key的消息映射到同一个分区中。
Value：数据
Headers：存储消息元数据，headers内容可以自行定义，类似于HTTP协议中的自定义消息头，当然Kafka中也有一些内部的消息头，该漏洞就与消息头有关

关于Kafka的基础概念还有很多，但对于该漏洞来说，能够理解上述概念已经足够了。若需要深入了解Kafka可以参考这篇文章：

https://zhuanlan.zhihu.com/p/74063251

我们还是从官方的通告入手：

https://spring.io/security/cve-2023-34040

漏洞描述中提示了一些关键信息：

存在漏洞的版本为3.x.0 <= version <= 3.0.9 2.x.0 <= 2.9.10，也就是说3.0.10版本是修复版本。
当Spring Kafka Consumer没有配置ErrorHandlingDeserializer 的时候存在漏洞
当checkDeserExWhenKeyNull 或者checkDeserExWhenValueNull 被设置为True时存在漏洞
当攻击者可以生产消息
当上述三个条件同时满足时可能存在漏洞
攻击发生在程序反序列化Record Headers的时候
使用了ErrorHandlingDeserializer 可以避免漏洞，因为它会删除危险的记录头。

首先，我们到官方仓库找到最新发布的漏洞修复版本v3.0.10：

更新日志提示了可能的漏洞修复记录

https://github.com/spring-projects/spring-kafka/pull/2770/files

另外，在ErrorHandlingDeserializer中找到其反序列化过程中会删除的Record headers

由此可知会被删除的header为springDeserializerExceptionKey 与springDeserializerExceptionValue，这也意味着漏洞可能就发生在这两个header的值中。所以，此时，我们需要在生产消息的时候为消息增加这两个消息头。

这里模拟了用户发送HTTP请求设置消息的情况

在前面提到的漏洞存在的条件中，提到需要设置checkDeserExWhenKeyNull 或者checkDeserExWhenValueNull为True，我们首先找到这两个变量在ConsumerProperties类中。顾名思义，这是一个与消费者相关的配置。

这两个配置需要在创建消费者容器工厂的时候配置为True，现在我们的漏洞复现环境才算完整。

我们到上面提到的commit记录中找到一个新增的测试类

这个测试类的功能大概是模拟了一条消息，其携带了Value类型的反序列化异常消息头，即：springDeserializerExceptionValue

然后通过调用SerializationUtils.getExceptionFromHeader方法查看是否能够返回一个Exception对象，若不能则测试通过。同时我们注意到该commit很多地方将ListenerUtils.getExceptionFromHeader方法替换为SerializationUtils.getExceptionFromHeader方法

那么证明SerializationUtils.getExceptionFromHeader方法中存在漏洞的修复，而按照原始的逻辑执行代码会导致漏洞。

那么我们将断点打在ListenerUtils.getExceptionFromHeader处进行调试：

这个时候可以通过IDEA查看以下程序的调用栈

checkDeser:2766, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
invokeOnMessage:2648, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
doInvokeRecordListener:2577, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
doInvokeWithRecords:2457, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
invokeRecordListener:2335, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
invokeListener:2006, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
invokeIfHaveRecords:1375, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
pollAndInvoke:1366, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
run:1257, KafkaMessageListenerContainer$ListenerConsumer (org.springframework.kafka.listener)
call:511, Executors$RunnableAdapter (java.util.concurrent)
run$$$capture:266, FutureTask (java.util.concurrent)
run:-1, FutureTask (java.util.concurrent)

在第8行的代码逻辑中有两个重要的地方，在doPoll方法中会完成消息的反序列化得到records对象，这中间需要保证消息拥有正确的数据格式以避免发生异常导致程序不能运行到第1366行invokeIfHaveRecords方法处，所谓invokeIfHaveRecords即在消费者成功获取到消息并反序列化后调用程序中设置的监听器回调进行跟进一步的处理，即被KafkaListener注解修饰的逻辑的处理逻辑，不过在进行这步操作前包含一些必要的检测，如：记录Headers元数据的检查。

回到ListenerUtils.getExceptionFromHeader的调用逻辑。该方法首先在104行处获取到记录headers能够与headerName 变量值springDeserializerExceptionKey匹配的头信息。然后将该标头的值取出来调用byteArrayToDeserializationException方法

在byteArrayToDeserializationException方法中有典型的反序列化逻辑，不过因为该部分代码重写了resolveClass方法限制了反序列化目标类的全类名必须为org.springframework.kafka.support.serializer.DeserializationException 所以不能进行任意类的反序列化。