Send DISCONNECT with malformed-packet reason code on bad SUBSCRIBEs.

ChangeLog.txt

@@ -16,8 +16,8 @@ Broker:
   publish messages.
 - Add `mosquitto_client_protocol_version()` function which can be used by
   plugins to determine which version of MQTT a client has connected with.
-- Send DISCONNECT with `malformed-packet` reason code on invalid UNSUBSCRIBE
-  packets.
+- Send DISCONNECT with `malformed-packet` reason code on invalid SUBSCRIBE,
+  UNSUBSCRIBE packets.
 
 Client library:
 - Client no longer generates random client ids for v3.1.1 clients, these are

src/handle_subscribe.c

@@ -54,15 +54,24 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 
 	if(context->protocol != mosq_p_mqtt31){
 		if((context->in_packet.command&0x0F) != 0x02){
-			return MOSQ_ERR_PROTOCOL;
+			return MOSQ_ERR_MALFORMED_PACKET;
 		}
 	}
-	if(packet__read_uint16(&context->in_packet, &mid)) return 1;
-	if(mid == 0) return MOSQ_ERR_PROTOCOL;
+	if(packet__read_uint16(&context->in_packet, &mid)) return MOSQ_ERR_MALFORMED_PACKET;
+	if(mid == 0) return MOSQ_ERR_MALFORMED_PACKET;
 
 	if(context->protocol == mosq_p_mqtt5){
 		rc = property__read_all(CMD_SUBSCRIBE, &context->in_packet, &properties);
-		if(rc) return rc;
+		if(rc){
+			/* FIXME - it would be better if property__read_all() returned
+			 * MOSQ_ERR_MALFORMED_PACKET, but this is would change the library
+			 * return codes so needs doc changes as well. */
+			if(rc == MOSQ_ERR_PROTOCOL){
+				return MOSQ_ERR_MALFORMED_PACKET;
+			}else{
+				return rc;
+			}
+		}
 
 		if(mosquitto_property_read_varint(properties, MQTT_PROP_SUBSCRIPTION_IDENTIFIER,
 					&subscription_identifier, false)){
@@ -70,7 +79,7 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 			/* If the identifier was force set to 0, this is an error */
 			if(subscription_identifier == 0){
 				mosquitto_property_free_all(&properties);
-				return MOSQ_ERR_PROTOCOL;
+				return MOSQ_ERR_MALFORMED_PACKET;
 			}
 		}
 
@@ -82,7 +91,7 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 		sub = NULL;
 		if(packet__read_string(&context->in_packet, &sub, &slen)){
 			mosquitto__free(payload);
-			return 1;
+			return MOSQ_ERR_MALFORMED_PACKET;
 		}
 
 		if(sub){
@@ -92,7 +101,7 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 						context->address);
 				mosquitto__free(sub);
 				mosquitto__free(payload);
-				return 1;
+				return MOSQ_ERR_MALFORMED_PACKET;
 			}
 			if(mosquitto_sub_topic_check(sub)){
 				log__printf(NULL, MOSQ_LOG_INFO,
@@ -100,13 +109,13 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 						context->address);
 				mosquitto__free(sub);
 				mosquitto__free(payload);
-				return 1;
+				return MOSQ_ERR_MALFORMED_PACKET;
 			}
 
 			if(packet__read_byte(&context->in_packet, &subscription_options)){
 				mosquitto__free(sub);
 				mosquitto__free(payload);
-				return 1;
+				return MOSQ_ERR_MALFORMED_PACKET;
 			}
 			if(context->protocol == mosq_p_mqtt31 || context->protocol == mosq_p_mqtt311){
 				qos = subscription_options;
@@ -119,7 +128,7 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 
 				retain_handling = (subscription_options & 0x30);
 				if(retain_handling == 0x30 || (subscription_options & 0xC0) != 0){
-					return MOSQ_ERR_PROTOCOL;
+					return MOSQ_ERR_MALFORMED_PACKET;
 				}
 			}
 			if(qos > 2){
@@ -128,7 +137,7 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 						context->address);
 				mosquitto__free(sub);
 				mosquitto__free(payload);
-				return 1;
+				return MOSQ_ERR_MALFORMED_PACKET;
 			}
 
 
@@ -201,7 +210,7 @@ int handle__subscribe(struct mosquitto_db *db, struct mosquitto *context)
 	if(context->protocol != mosq_p_mqtt31){
 		if(payloadlen == 0){
 			/* No subscriptions specified, protocol error. */
-			return MOSQ_ERR_PROTOCOL;
+			return MOSQ_ERR_MALFORMED_PACKET;
 		}
 	}
 	if(send__suback(context, mid, payloadlen, payload)) rc = 1;

src/handle_unsubscribe.c

@@ -55,6 +55,9 @@ int handle__unsubscribe(struct mosquitto_db *db, struct mosquitto *context)
 	if(context->protocol == mosq_p_mqtt5){
 		rc = property__read_all(CMD_UNSUBSCRIBE, &context->in_packet, &properties);
 		if(rc){
+			/* FIXME - it would be better if property__read_all() returned
+			 * MOSQ_ERR_MALFORMED_PACKET, but this is would change the library
+			 * return codes so needs doc changes as well. */
 			if(rc == MOSQ_ERR_PROTOCOL){
 				return MOSQ_ERR_MALFORMED_PACKET;
 			}else{

src/read_handle.c

@@ -56,7 +56,8 @@ int handle__packet(struct mosquitto_db *db, struct mosquitto *context)
 		case CMD_DISCONNECT:
 			return handle__disconnect(db, context);
 		case CMD_SUBSCRIBE:
-			return handle__subscribe(db, context);
+			rc = handle__subscribe(db, context);
+			break;
 		case CMD_UNSUBSCRIBE:
 			rc = handle__unsubscribe(db, context);
 			break;

test/broker/02-subscribe-invalid-utf8.py

@@ -25,7 +25,11 @@ def do_test(proto_ver):
         time.sleep(0.5)
 
         sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
+        if proto_ver == 4:
             mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+        else:
+            disconnect_packet = mosq_test.gen_disconnect(proto_ver=5, reason_code = mqtt5_rc.MQTT_RC_MALFORMED_PACKET)
+            mosq_test.do_send_receive(sock, subscribe_packet, disconnect_packet, "suback")
 
         rc = 0
 

test/broker/02-subscribe-long-topic.py

@@ -21,7 +21,11 @@ def do_test(proto_ver):
 
     try:
         sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
+        if proto_ver == 4:
             mosq_test.do_send_receive(sock, subscribe_packet, b"", "suback")
+        else:
+            disconnect_packet = mosq_test.gen_disconnect(proto_ver=5, reason_code = mqtt5_rc.MQTT_RC_MALFORMED_PACKET)
+            mosq_test.do_send_receive(sock, subscribe_packet, disconnect_packet, "suback")
 
         rc = 0
 

test/broker/13-malformed-subscribe-v5.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+
+# Test whether the broker handles malformed packets correctly - SUBSCRIBE
+# MQTTv5
+
+from mosq_test_helper import *
+
+rc = 1
+
+def do_test(subscribe_packet, reason_code, error_string):
+    global rc
+
+    rc = 1
+
+    keepalive = 10
+    connect_packet = mosq_test.gen_connect("test", proto_ver=5, keepalive=keepalive)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5)
+
+    mid = 0
+    disconnect_packet = mosq_test.gen_disconnect(proto_ver=5, reason_code=reason_code)
+
+    sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
+    mosq_test.do_send_receive(sock, subscribe_packet, disconnect_packet, error_string=error_string)
+    rc = 0
+
+
+port = mosq_test.get_port()
+broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
+
+try:
+    # mid == 0
+    subscribe_packet = mosq_test.gen_subscribe(topic="test/topic", qos=1, mid=0, proto_ver=5)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "mid == 0")
+
+    # qos > 2
+    subscribe_packet = mosq_test.gen_subscribe(topic="test/topic", qos=3, mid=1, proto_ver=5)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "qos > 2")
+
+    # retain handling = 0x30
+    subscribe_packet = mosq_test.gen_subscribe(topic="test/topic", qos=0x30, mid=1, proto_ver=5)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "retain handling = 0x30")
+
+    # subscription options = 0xC0
+    subscribe_packet = mosq_test.gen_subscribe(topic="test/topic", qos=0xC0, mid=1, proto_ver=5)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "subscription options = 0xC0")
+
+    # command flags != 0x02
+    subscribe_packet = mosq_test.gen_subscribe(topic="test/topic", qos=1, mid=1, proto_ver=5, cmd=128)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "command flags != 0x02")
+
+    # Incorrect property
+    props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 0)
+    subscribe_packet = mosq_test.gen_subscribe(topic="test/topic", qos=1, mid=1, proto_ver=5, properties=props)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Incorrect property")
+
+    # Truncated packet, no mid
+    subscribe_packet = struct.pack("!BB", 130, 0)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, no mid")
+
+    # Truncated packet, no properties
+    subscribe_packet = struct.pack("!BBH", 130, 2, 1)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, no properties")
+
+    # Truncated packet, with properties field
+    subscribe_packet = struct.pack("!BBHB", 130, 3, 1, 0)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, with properties field")
+
+    # Truncated packet, with properties field, empty topic
+    subscribe_packet = struct.pack("!BBHBH", 130, 5, 1, 0, 0)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, with properties field, empty topic")
+
+    # Truncated packet, with properties field, empty topic, with qos
+    subscribe_packet = struct.pack("!BBHBHB", 130, 6, 1, 0, 0, 1)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, with properties field, empty topic, with qos")
+
+    # Truncated packet, with properties field, with topic, no qos
+    subscribe_packet = struct.pack("!BBHBH1s", 130, 6, 1, 0, 1, b"a")
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, with properties field, with topic, no qos")
+
+    # Truncated packet, with properties field, with 1st topic and qos ok, second topic ok, no second qos
+    subscribe_packet = struct.pack("!BBHHH1sBH1s", 130, 10, 1, 0, 1, b"a", 0, 1, b"b")
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, with properties field, with 1st topic and qos ok, second topic ok, no second qos")
+
+    # Bad topic
+    subscribe_packet = mosq_test.gen_subscribe(topic="#/test/topic", qos=1, mid=1, proto_ver=5)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Bad topic")
+
+    # Subscription ID set to 0
+    props = mqtt5_props.gen_varint_prop(mqtt5_props.PROP_SUBSCRIPTION_IDENTIFIER, 0)
+    subscribe_packet = mosq_test.gen_subscribe(topic="test/topic", qos=1, mid=1, proto_ver=5, properties=props)
+    do_test(subscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Subscription ID set to 0")
+finally:
+    broker.terminate()
+    broker.wait()
+    (stdo, stde) = broker.communicate()
+    if rc:
+        print(stde.decode('utf-8'))
+        exit(rc)

test/broker/13-malformed-unsubscribe-v5.py

@@ -36,26 +36,26 @@ try:
     unsubscribe_packet = mosq_test.gen_unsubscribe(topic="test/topic", mid=1, proto_ver=5, cmd=160)
     do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "command flags != 0x02")
 
-    # Truncated packet, no mid
-    unsubscribe_packet = struct.pack("!BB", 162, 0)
-    do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, no mid")
-
     # Incorrect property
     props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_SESSION_EXPIRY_INTERVAL, 0)
     unsubscribe_packet = mosq_test.gen_unsubscribe(topic="test/topic", mid=1, proto_ver=5, properties=props)
     do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Incorrect property")
 
-    # Truncated packet, no topic
+    # Truncated packet, no mid
+    unsubscribe_packet = struct.pack("!BB", 162, 0)
+    do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, no mid")
+
+    # Truncated packet, no properties
     unsubscribe_packet = struct.pack("!BBH", 162, 2, 1)
-    do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, no topic")
+    do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, no properties")
 
-    # Truncated packet, empty topic
+    # Truncated packet, with properties field, no topic
     unsubscribe_packet = struct.pack("!BBHH", 162, 4, 1, 0)
-    do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, empty topic")
+    do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, with properties field, no topic")
 
-    # Truncated packet, empty topic, with properties field
+    # Truncated packet, with properties field, empty topic
     unsubscribe_packet = struct.pack("!BBHHH", 162, 5, 1, 0, 0)
-    do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, empty topic, with properties field")
+    do_test(unsubscribe_packet, mqtt5_rc.MQTT_RC_MALFORMED_PACKET, "Truncated packet, with properties field, empty topic")
 
     # Bad topic
     unsubscribe_packet = mosq_test.gen_unsubscribe(topic="#/test/topic", mid=1, proto_ver=5)

test/broker/Makefile

@@ -218,4 +218,5 @@ endif
 	./12-prop-topic-alias-invalid.py
 
 13 :
+	./13-malformed-subscribe-v5.py
 	./13-malformed-unsubscribe-v5.py

test/broker/test.py

@@ -182,6 +182,7 @@ tests = [
     (1, './12-prop-subpub-payload-format.py'),
     (1, './12-prop-topic-alias-invalid.py'),
 
+    (1, './13-malformed-subscribe-v5.py'),
     (1, './13-malformed-unsubscribe-v5.py'),
     ]
 

test/mosq_test.py

@@ -507,9 +507,9 @@ def gen_pubcomp(mid, proto_ver=4, reason_code=-1, properties=None):
     return _gen_command_with_mid(112, mid, proto_ver, reason_code, properties)
 
 
-def gen_subscribe(mid, topic, qos, proto_ver=4, properties=b""):
+def gen_subscribe(mid, topic, qos, cmd=130, proto_ver=4, properties=b""):
     topic = topic.encode("utf-8")
-    packet = struct.pack("!B", 130)
+    packet = struct.pack("!B", cmd)
     if proto_ver == 5:
         if properties == b"":
             packet += pack_remaining_length(2+1+2+len(topic)+1)