物聯(lián)網(wǎng)防火墻himqtt源碼之MQTT協(xié)議分析

himqtt是首款完整源碼的高性能MQTT物聯(lián)網(wǎng)防火墻 - MQTT Application FireWall，C語言編寫，采用epoll模式支持IoT數(shù)十萬的高并發(fā)連接，并且兼容ModSecurity部分規(guī)則。 代碼非常優(yōu)秀，非常值得收藏和學(xué)習(xí)，今天筆者就從結(jié)合himqtt的源碼來進(jìn)行MQTT協(xié)議分析。

一、MQTT協(xié)議指令匯總

MQTT協(xié)議一共有14個(gè)指令，如下表所示：其中有9個(gè)報(bào)文都是固定的2~4個(gè)字節(jié)，非常簡單適合小型物聯(lián)網(wǎng)設(shè)備。

名字值固定報(bào)文描述

CONNECT1否客戶端請求與服務(wù)端建立連接

CONNACK2是服務(wù)端確認(rèn)連接建立

PUBLISH3否發(fā)布消息

PUBACK4是收到發(fā)布消息確認(rèn)

PUBREC5是發(fā)布消息收到

PUBREL6是發(fā)布消息釋放

PUBCOMP7是發(fā)布消息完成

SUBSCRIBE8否訂閱請求

SUBACK9否訂閱確認(rèn)

UNSUBSCRIBE10否取消訂閱

UNSUBACK11是取消訂閱確認(rèn)

PING12是客戶端發(fā)送PING(連接?；?命令

PINGRSP13是PING命令回復(fù)

DISCONNECT14是斷開連接

MQTT協(xié)議由指令號(hào)(1字節(jié))+長度(1-4字節(jié)不定)+內(nèi)容組成，比如下面第一個(gè)字節(jié)0x30表示publish發(fā)布消息指令，0x26表示后面的內(nèi)容長度就是38個(gè)字節(jié)。

---------------MQTT PUBLISH- ------40bytes-------------------------------------------

| 30 26 00 14 68 6f 6d 65 2f 67 61 72 64 65 6e 2f |0&..home/garden/|

| 66 6f 75 6e 74 61 69 6e 31 32 33 34 35 36 37 38 |fountain12345678|

| 39 30 61 62 63 64 65 66 |90abcdef

先到github上下載himqtt最新源碼，https://github.com/qq4108863/himqtt/ ，打開src/waf/mqtt.c文件。

特別注意的是：長度占用的字節(jié)數(shù)是可變的(1-4字節(jié))，具體的計(jì)算方法在process_mqtt_msg這個(gè)函數(shù)里面，理論上這種算法后續(xù)消息內(nèi)容是最大長度是268435455字節(jié)(約255M)。

static void process_mqtt_msg(mqtt_waf_msg *req)

{

......

/* decode mqtt variable length */

len = len_len = 0;

p = req->buf + 1;

eop = &req->buf[req->pos];

while (p < eop) {

lc = *((const unsigned char *) p++);

len += (lc & 0x7f) << 7 * len_len;

len_len++;

if (!(lc & 0x80)) break;

if (len_len > 4){

req->msg_state = MQTT_MSG_ERROR;

return;

}

}

.....

}

......

長度和協(xié)議校驗(yàn)正確后，根據(jù)收到的消息類型，以此對(duì)不同的指令進(jìn)行處理，代碼邏輯非常清晰：

switch (mqtt_msg_type)

{

case MQTT_CONNECT:

req->msg_state = mqtt_connect(req,p,end,&mm);

break;

case MQTT_CONNACK:

break;

case MQTT_PUBLISH:

req->msg_state = mqtt_publish(req,p,end,&mm);

break;

case MQTT_SUBSCRIBE:

req->msg_state = mqtt_subscribe(req,p,end,&mm);

break;

case MQTT_UNSUBSCRIBE:

req->msg_state = mqtt_unsubscribe(req,p,end,&mm);

......

下面我們主要CONNECT、PUBLISH、SUBSCRIBE、UNSUBSCRIBE這幾個(gè)復(fù)雜一點(diǎn)的報(bào)文協(xié)議內(nèi)容。

二、HiMQTT協(xié)議分析

1、CONNECT連接服務(wù)端

CONNECT是客戶端到服務(wù)端的網(wǎng)絡(luò)連接建立后，客戶端發(fā)送給服務(wù)端的第一個(gè)報(bào)文必須是CONNECT報(bào)文，其中登錄的身份認(rèn)證如用戶名、密碼就在這個(gè)指令里面。報(bào)文協(xié)議如下：

--------------MQTT CONNECT-----105bytes-----------------------------

| 10 67 00 04 4d 51 54 54 04 c2 00 3c 00 19 4d 51 |.g..MQTT...<..MQ|

| 54 54 5f 46 58 5f 43 6c 69 65 6e 74 5f 39 69 75 |TT_FX_Client_9iu|

| 79 38 37 36 35 35 35 00 12 69 6f 74 66 72 65 65 |y876555..iotfree|

| 74 65 73 74 2f 74 68 69 6e 67 30 00 2c 59 55 37 |test/thing0.,YU7|

| 54 6f 76 38 7a 46 57 2b 57 75 61 4c 78 39 73 39 |Tov8zFW+WuaLx9s9|

| 49 33 4d 4b 79 63 6c 69 65 39 53 47 44 75 75 4e |I3MKyclie9SGDuuN|

| 6b 6c 36 6f 39 4c 58 6f 3d |kl6o9LXo=

10 //CONNECT指令號(hào)

67 //長度103字節(jié)

00 04 //MQTT協(xié)議長度為4字節(jié)

4d 51 54 54 //MQTT固定字符串

04 //版本3.1.1

c2 //連接標(biāo)記，是否由用戶名/密碼等

00 3c //心跳間隔時(shí)間60秒

00 19 //用戶名長度25字節(jié)，后面是用戶名

4d 51 54 54 5f 46 58 5f 43 6c 69 65 6e 74 5f 39 69 75 79 38 37 36 35 35 35

00 12 //密碼18字節(jié)

69 6f 74 66 72 65 65 74 65 73 74 2f 74 68 69 6e 67 30//密碼

00 2c //will message長度。

59 55 37 ......6f 3d//will message內(nèi)容

2、PUBLISH發(fā)布消息

PUBLISH 是從客戶端向服務(wù)端或者服務(wù)端向客戶端傳輸一個(gè)應(yīng)用消息，這是通信的最重點(diǎn)，就像HTTP協(xié)議的GET一樣。報(bào)文協(xié)議分析如下：

/*

---------------MQTT PUBLISH-------40bytes-----------------------------

| 30 26 00 14 68 6f 6d 65 2f 67 61 72 64 65 6e 2f |0&..home/garden/|

| 66 6f 75 6e 74 61 69 6e 31 32 33 34 35 36 37 38 |fountain12345678|

| 39 30 61 62 63 64 65 66 |90abcdef

*/

30 //PUBLISH指令號(hào)

26 //長度39字節(jié)

00 14 //TOPIC長度20字節(jié)

68 6f 6d 65 2f 67 61 72 64 65 6e 2f 66 6f 75 6e //TOPIC

31 32 33 34 35 36 37 38 39 30 61 62 63 64 65 66 //發(fā)布的消息

在實(shí)際編程中，這個(gè)地方大多是json格式提交給服務(wù)器，SQL注入/XSS攻擊很可能從這里對(duì)物聯(lián)網(wǎng)設(shè)備發(fā)起攻擊，所以一定要做攻擊檢查，himqtt不知道什么原因把ngx_http_dummy_json_parse解析json格式的函數(shù)注釋掉了。

3、SUBSCRIBE訂閱消息

SUBSCRIBE是由客戶端向服務(wù)端發(fā)送的，用于創(chuàng)建一個(gè)或多個(gè)訂閱。每個(gè)訂閱是該客戶端關(guān)注的一個(gè)或多個(gè)主題。服務(wù)端依據(jù)客戶端的訂閱來匹配主題，然后將對(duì)應(yīng)的PUBLISH報(bào)文發(fā)送給客戶端。報(bào)文協(xié)議分析如下：

---------------MQTT SUBSCRIBE------33bytes-----------------------------

| 82 1f 00 01 00 1a 68 6f 6d 65 2f 67 61 72 64 65 |......home/garde|

| 6e 2f 66 6f 75 6e 74 61 69 6e 64 65 6c 65 74 65 |n/fountaindelete|

| 00

82 //SUBSCRIBE指令

1f //長度31字節(jié)

00 01 //Message Identifier

00 1a //TOPIC長度26

68 6f 6d 65 2f 67 61 72 64 65 6e 2f 66 6f 75 6e 74 61 69 6e 64 65 6c 65 74 65 //TOPIC

00 //request QOS

4、UNSUBSCRIBE取消訂閱消息

UNSUBSCRIBE是客戶端發(fā)送本報(bào)文給服務(wù)端，用于取消訂閱主題。報(bào)文協(xié)議分析如下：

---------------MQTT UNSUBSCRIBE-------32bytes-----------------------------

| a2 1e 00 02 00 1a 68 6f 6d 65 2f 67 61 72 64 65 |......home/garde|

| 6e 2f 66 6f 75 6e 74 61 69 6e 64 65 6c 65 74 65 |n/fountaindelete|

a2 //SUBSCRIBE指令

1e //長度30字節(jié)

00 02 //Message Identifier

00 1a //TOPIC長度26

68 6f 6d 65 2f 67 61 72 64 65 6e 2f 66 6f 75 6e 74 61 69 6e 64 65 6c 65 74 65 //TOPIC

總體來說，MQTT協(xié)議比HTTP協(xié)議簡單多了，非常適合物聯(lián)網(wǎng)設(shè)備。另外himqtt其實(shí)也是一款功能強(qiáng)大的WEB應(yīng)用防火墻，其他源碼我們在另外的文章中再介紹。

也許未來幾年IPV6普及后，很可能幾百億帶電的物體都會(huì)聯(lián)網(wǎng)哦，期待himqtt這類高并發(fā)的物聯(lián)網(wǎng)防火墻能扛起信息安全的大旗，徹底阻擋黑客攻擊。