fix(mqtt): close the connection on invalid fixed header flags

Makefile

@@ -393,6 +393,7 @@ unittest: prepare libhv
 	$(CC)  -g -Wall -O0 -std=c99   -I. -Iutil            -o bin/base64            unittest/base64_test.c        util/base64.c
 	$(CC)  -g -Wall -O0 -std=c99   -I. -Iutil            -o bin/md5               unittest/md5_test.c           util/md5.c
 	$(CC)  -g -Wall -O0 -std=c99   -I. -Iutil            -o bin/sha1              unittest/sha1_test.c          util/sha1.c
+	$(CC)  -g -Wall -O0 -std=c99   -I. -Ibase -Imqtt     -o bin/mqtt_protocol_test unittest/mqtt_protocol_test.c mqtt/mqtt_protocol.c
 	$(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Icpputil  -o bin/hstring_test      unittest/hstring_test.cpp     cpputil/hstring.cpp
 	$(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Icpputil  -o bin/hpath_test        unittest/hpath_test.cpp       cpputil/hpath.cpp
 	$(CXX) -g -Wall -O0 -std=c++11 -I. -Ibase -Icpputil  -o bin/hurl_test         unittest/hurl_test.cpp        cpputil/hurl.cpp base/hbase.c

mqtt/mqtt_client.c

@@ -246,7 +246,12 @@ static void on_packet(hio_t* io, void* buf, int len) {
     unsigned char* end = p + len;
     memset(&cli->head, 0, sizeof(mqtt_head_t));
     int headlen = mqtt_head_unpack(&cli->head, p, len);
-    if (headlen <= 0) return;
+    if (headlen < 0 || (headlen > 0 && !mqtt_head_flags_valid(&cli->head))) {
+        hloge("MQTT malformed fixed header!");
+        hio_close(io);
+        return;
+    }
+    if (headlen == 0) return;
     p += headlen;
     switch (cli->head.type) {
     // case MQTT_TYPE_CONNECT:

mqtt/mqtt_protocol.c

@@ -18,5 +18,22 @@ int mqtt_head_unpack(mqtt_head_t* head, const unsigned char* buf, int len) {
     int bytes = len - 1;
     head->length = varint_decode(buf + 1, &bytes);
     if (bytes <= 0) return bytes;
+    // Remaining Length is at most 4 bytes
+    if (bytes > 4) return -1;
     return 1 + bytes;
 }
+
+int mqtt_head_flags_valid(const mqtt_head_t* head) {
+    int flags = (head->dup << 3) | (head->qos << 1) | head->retain;
+    switch (head->type) {
+    case MQTT_TYPE_PUBLISH:
+        // QoS 3 is reserved
+        return head->qos != 3;
+    case MQTT_TYPE_PUBREL:
+    case MQTT_TYPE_SUBSCRIBE:
+    case MQTT_TYPE_UNSUBSCRIBE:
+        return flags == 0x2;
+    default:
+        return flags == 0;
+    }
+}

mqtt/mqtt_protocol.h

@@ -77,6 +77,9 @@ HV_INLINE int mqtt_estimate_length(mqtt_head_t* head) {
 
 HV_EXPORT int mqtt_head_pack(mqtt_head_t* head, unsigned char buf[]);
 HV_EXPORT int mqtt_head_unpack(mqtt_head_t* head, const unsigned char* buf, int len);
+// Check the fixed header flags (dup, qos, retain) required for the packet type.
+// Returns 1 if valid, 0 if the connection should be closed.
+HV_EXPORT int mqtt_head_flags_valid(const mqtt_head_t* head);
 
 END_EXTERN_C
 

scripts/unittest.sh

@@ -18,6 +18,7 @@ bin/hlog_test
 bin/base64
 bin/md5
 bin/sha1
+bin/mqtt_protocol_test
 
 bin/defer_test
 bin/hstring_test

unittest/CMakeLists.txt

@@ -41,6 +41,10 @@ target_include_directories(md5 PRIVATE .. ../util)
 add_executable(sha1 sha1_test.c ../util/sha1.c)
 target_include_directories(sha1 PRIVATE .. ../util)
 
+# ------mqtt------
+add_executable(mqtt_protocol_test mqtt_protocol_test.c ../mqtt/mqtt_protocol.c)
+target_include_directories(mqtt_protocol_test PRIVATE .. ../base ../mqtt)
+
 # ------cpputil------
 add_executable(hstring_test hstring_test.cpp ../cpputil/hstring.cpp)
 target_include_directories(hstring_test PRIVATE .. ../base ../cpputil)

unittest/mqtt_protocol_test.c

@@ -0,0 +1,69 @@
+/*
+ * @build: gcc -o bin/mqtt_protocol_test unittest/mqtt_protocol_test.c mqtt/mqtt_protocol.c -I. -Ibase -Imqtt
+ *
+ */
+
+#include <stdio.h>
+#include <string.h>
+#include <assert.h>
+
+#include "mqtt_protocol.h"
+
+static void test_head_unpack() {
+    mqtt_head_t head;
+    // PUBLISH qos=1 retain=1, remaining length 3
+    const unsigned char publish[] = { 0x33, 0x03, 0x00, 0x01, 0x74 };
+    memset(&head, 0, sizeof(head));
+    assert(mqtt_head_unpack(&head, publish, sizeof(publish)) == 2);
+    assert(head.type == MQTT_TYPE_PUBLISH);
+    assert(head.qos == 1);
+    assert(head.retain == 1);
+    assert(head.length == 3);
+
+    // Remaining Length is at most 4 bytes
+    const unsigned char four[] = { 0x30, 0xFF, 0xFF, 0xFF, 0x7F };
+    memset(&head, 0, sizeof(head));
+    assert(mqtt_head_unpack(&head, four, sizeof(four)) == 5);
+    assert(head.length == 268435455);
+
+    const unsigned char five[] = { 0x30, 0x83, 0x80, 0x80, 0x80, 0x00, 0x00, 0x01, 0x74 };
+    memset(&head, 0, sizeof(head));
+    assert(mqtt_head_unpack(&head, five, sizeof(five)) < 0);
+}
+
+static void test_head_flags_valid() {
+    mqtt_head_t head;
+    int type, qos, dup, retain;
+    for (type = MQTT_TYPE_CONNECT; type <= MQTT_TYPE_AUTH; ++type) {
+        for (dup = 0; dup <= 1; ++dup) {
+            for (qos = 0; qos <= 3; ++qos) {
+                for (retain = 0; retain <= 1; ++retain) {
+                    memset(&head, 0, sizeof(head));
+                    head.type = type;
+                    head.dup = dup;
+                    head.qos = qos;
+                    head.retain = retain;
+                    int flags = (dup << 3) | (qos << 1) | retain;
+                    int expected;
+                    if (type == MQTT_TYPE_PUBLISH) {
+                        expected = qos != 3;
+                    } else if (type == MQTT_TYPE_PUBREL ||
+                               type == MQTT_TYPE_SUBSCRIBE ||
+                               type == MQTT_TYPE_UNSUBSCRIBE) {
+                        expected = flags == 0x2;
+                    } else {
+                        expected = flags == 0;
+                    }
+                    assert(mqtt_head_flags_valid(&head) == expected);
+                }
+            }
+        }
+    }
+}
+
+int main() {
+    test_head_unpack();
+    test_head_flags_valid();
+    printf("mqtt_protocol_test OK\n");
+    return 0;
+}