Kafka反序列化RCE漏洞（CVE-2023-34040） - FreeBuf网络安全行业门户

官方公众号企业安全新浪微博

FreeBuf.COM网络安全行业门户，每日发布专业的安全资讯、技术剖析。

FreeBuf+小程序把安全装进口袋

漏洞

Kafka反序列化RCE漏洞（CVE-2023-34040）

蚁景科技 2023-11-03 16:21:55 216733

本文由蚁景科技创作，已纳入「FreeBuf原创奖励计划」，未授权禁止转载

漏洞描述

Spring Kafka 是 Spring Framework 生态系统中的一个模块，用于简化在 Spring 应用程序中集成 Apache Kafka 的过程，记录 (record) 指 Kafka 消息中的一条记录。

受影响版本中默认未对记录配置 ErrorHandlingDeserializer，当用户将容器属性 checkDeserExWhenKeyNull或 checkDeserExWhenValueNull设置为 true(默认为 false)，并且允许不受信任的源发布到 Kafka 主题中时，攻击者可将恶意 payload 注入到 Kafka 主题中，当反序列化记录头时远程执行任意代码。

影响版本

2.8.1 <= Spring-Kafka <= 2.9.103.0.0 <= Spring-Kafka <= 3.0.9

漏洞复现

这一个漏洞所影响的组件其实是 Spring-Kafka，严格意义上来说并不算是 kafka 的漏洞，应该算是 Spring 的漏洞。

漏洞前置知识

先来看一看 SpringBoot 和 Kafka 是怎么完成通讯/消费的

工作流程如下

1、生产者将消息发送到 Kafka 集群中的某个 Broker（也可以是多个）2、Kafka 集群将消息存储在一个或多个分区中，并为每个分区维护一个偏移量3、消费者订阅一个或多个主题，并从 Kafka 集群中读取消息。4、消费者按顺序读取每个分区中的消息，并跟踪每个分区的偏移量。

ErrorHandlingDeserializer：是 Kafka中的一种反序列化器（Deserializer），它可以在反序列化过程中处理异常和错误。
checkDeserExWhenKeyNull && checkDeserExWhenValueNull：是 Kafka 中的一种序列化器（Serializer），它可以在序列化过程中检查键（key/value）是否为 null，并在发现值为 null 时抛出异常。

再简单整理一下漏洞条件

在受到影响的版本中，默认未对记录配置 ErrorHandlingDeserializer容器属性 checkDeserExWhenKeyNull或 checkDeserExWhenValueNull设置为 true

环境搭建

其中需要我们起一个 Kafka 的服务，用来接收消息，本机上起比较麻烦，可以在 vps 上用 docker 迅速搭建，且需注意，Kafka 要能够接受外连，docker-compose.yml如下

version: '2'

services:
zookeeper:
image: zookeeper
restart: always
ports:
- "2181:2181"
container_name: zookeeper

kafka:
image: wurstmeister/kafka
restart: always
ports:
- "9092:9092"
- "9094:9094"
depends_on:
- zookeeper
environment:
KAFKA_ADVERTISED_HOST_NAME: 124.222.21.138
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9094
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://124.222.21.138:9092,SSL://124.222.21.138:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,SSL:SSL
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
container_name: kafka

Spring Kafka 的生产者和消费者可以通过使用 Spring Kafka 提供的 KafkaTemplate和 `@KafkaListener注解来编写。

【----帮助网安学习，以下所有学习资料免费领！加vx：yj009991，备注“freebuf”获取！】

① 网安学习成长路径思维导图
② 60+网安经典常用工具包
③ 100+SRC漏洞分析报告
④ 150+网安攻防实战技术电子书
⑤ 最权威CISSP 认证考试指南+题库
⑥ 超1800页CTF实战技巧手册
⑦ 最新网安大厂面试题合集（含答案）
⑧ APP客户端安全检测指南（安卓+IOS）

生产者可以使用 KafkaTemplate来发送消息到 Kafka 集群：

package com.drunkbaby.springkafkatest.controller;  

import com.drunkbaby.springkafkatest.common.KafkaInfo;  
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.kafka.core.KafkaTemplate;  
import org.springframework.kafka.support.SendResult;  
import org.springframework.util.concurrent.ListenableFuture;  
import org.springframework.web.bind.annotation.PostMapping;  
import org.springframework.web.bind.annotation.RequestMapping;  
import org.springframework.web.bind.annotation.RestController;  

import java.time.LocalDateTime;  
import java.util.concurrent.ExecutionException;  

@RestController  
@RequestMapping("/producer")  
public class ProducerController {  
@Autowired  
private KafkaTemplate<String,String> kafkaTemplate;  

@PostMapping("/fireAndForget")  
public String fireAndForget() {  
kafkaTemplate.send(KafkaInfo.TOPIC_WELCOME, "fireAndForget:" + LocalDateTime.now());  
return "success";  
}  
}

消费者可以使用 @KafkaListener注解来监听 Kafka 集群中的消息：

package com.drunkbaby.springkafkatest.consumer;  

import com.drunkbaby.springkafkatest.common.KafkaInfo;  
import org.springframework.kafka.annotation.KafkaListener;  
import org.springframework.messaging.MessageHeaders;  
import org.springframework.messaging.handler.annotation.Headers;  
import org.springframework.messaging.handler.annotation.Payload;  
import org.springframework.stereotype.Component;  

@Component  
public class Consumer {  
@KafkaListener(topics = KafkaInfo.TOPIC_WELCOME)  
public String consumer2(@Payload String message, @Headers MessageHeaders headers) {  
System.out.println("消费者(注解方式)：收到消息==> ");  
System.out.println("  message：" + message);  
System.out.println("  headers:");  
headers.keySet().forEach(key -> System.out.println("    " + key + ":" + headers.get(key)));  
return "success";  
}

连接成功

访问 http://localhost:8083/producer/sync发送一条记录

post

构造 payload

实际影响到的是 Consumer，且 Consumer 要设置 checkDeserExWhenKeyNull或 checkDeserExWhenValueNull为 true

ConcurrentKafkaListenerContainerFactory<String, Greeting> factory = new ConcurrentKafkaListenerContainerFactory<>();  
factory.getContainerProperties().setCheckDeserExWhenValueNull(true);  
factory.getContainerProperties().setCheckDeserExWhenKeyNull(true);

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