Broker configurable max_packet_size

Plus tests.

lib/packet_mosq.c

@@ -36,6 +36,7 @@ and the Eclipse Distribution License is available at
 #include "read_handle.h"
 #ifdef WITH_BROKER
 # include "sys_tree.h"
+# include "send_mosq.h"
 #else
 # define G_BYTES_RECEIVED_INC(A)
 # define G_BYTES_SENT_INC(A)
@@ -381,6 +382,17 @@ int packet__read(struct mosquitto *mosq)
  * positive. */
  mosq->in_packet.remaining_count *= -1;
 
+#ifdef WITH_BROKER
+ if(db->config->max_packet_size > 0 && mosq->in_packet.remaining_length+1 > db->config->max_packet_size){
+ log__printf(NULL, MOSQ_LOG_INFO, "Client %s sent too large packet %d, disconnecting.", mosq->id, mosq->in_packet.remaining_length+1);
+ if(mosq->protocol == mosq_p_mqtt5){
+ send__disconnect(mosq, MQTT_RC_PACKET_TOO_LARGE, NULL);
+ }
+ return MOSQ_ERR_OVERSIZE_PACKET;
+ }
+#else
+ // FIXME - client case for incoming message received from broker too large
+#endif
  if(mosq->in_packet.remaining_length > 0){
  mosq->in_packet.payload = mosquitto__malloc(mosq->in_packet.remaining_length*sizeof(uint8_t));
  if(!mosq->in_packet.payload){

man/mosquitto.conf.5.xml

@@ -474,6 +474,28 @@
  <para>Reloaded on reload signal.</para>
  </listitem>
  </varlistentry>
+ <varlistentry>
+ <term><option>max_packet_size</option> <replaceable>value</replaceable></term>
+ <listitem>
+ <para>For MQTT v5 clients, it is possible to have the
+ server send a "maximum packet size" value that will
+ instruct the client it will not accept MQTT packets
+ with size greater than <option>value</option> bytes.
+ This applies to the full MQTT packet, not just the
+ payload. Setting this option to a positive value will
+ set the maximum packet size to that number of bytes. If
+ a client sends a packet which is larger than this
+ value, it will be disconnected. This applies to all
+ clients regardless of the protocol version they are
+ using, but v3.1.1 and earlier clients will of course
+ not have received the maximum packet size information.
+ Defaults to no limit.</para>
+ <para>Setting below 20 bytes is forbidden because it is
+ likely to interfere with normal client operation even
+ with small payloads.</para>
+ <para>Reloaded on reload signal.</para>
+ </listitem>
+ </varlistentry>
  <varlistentry>
  <term><option>max_queued_bytes</option> <replaceable>count</replaceable></term>
  <listitem>

mosquitto.conf

@@ -179,6 +179,21 @@
 # allowable is 65535. Do not set below 10.
 #max_keepalive 65535
 
+
+# For MQTT v5 clients, it is possible to have the server send a "maximum packet
+# size" value that will instruct the client it will not accept MQTT packets
+# with size greater than max_packet_size bytes. This applies to the full MQTT
+# packet, not just the payload. Setting this option to a positive value will
+# set the maximum packet size to that number of bytes. If a client sends a
+# packet which is larger than this value, it will be disconnected. This applies
+# to all clients regardless of the protocol version they are using, but v3.1.1
+# and earlier clients will of course not have received the maximum packet size
+# information. Defaults to no limit. Setting below 20 bytes is forbidden
+# because it is likely to interfere with ordinary client operation, even with
+# very small payloads.
+#max_packet_size 0
+
+
 # =================================================================
 # Default listener
 # =================================================================

src/conf.c

@@ -214,6 +214,7 @@ static void config__init_reload(struct mosquitto_db *db, struct mosquitto__confi
 #endif
  config->log_timestamp = true;
  config->max_keepalive = 65535;
+ config->max_packet_size = 0;
  config->max_inflight_messages = 20;
  config->persistence = false;
  mosquitto__free(config->persistence_location);
@@ -1503,6 +1504,13 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct
  return MOSQ_ERR_INVAL;
  }
  config->max_keepalive = tmp_int;
+ }else if(!strcmp(token, "max_packet_size")){
+ if(conf__parse_int(&token, "max_packet_size", &tmp_int, saveptr)) return MOSQ_ERR_INVAL;
+ if(tmp_int < 20){
+ log__printf(NULL, MOSQ_LOG_ERR, "Error: Invalid max_packet_size value (%d).", tmp_int);
+ return MOSQ_ERR_INVAL;
+ }
+ config->max_packet_size = tmp_int;
  }else if(!strcmp(token, "max_queued_bytes")){
  token = strtok_r(NULL, " ", &saveptr);
  if(token){

src/mosquitto_broker_internal.h

@@ -264,6 +264,7 @@ struct mosquitto__config {
  FILE *log_fptr;
  uint16_t max_inflight_messages;
  uint16_t max_keepalive;
+ uint32_t max_packet_size;
  uint32_t message_size_limit;
  bool persistence;
  char *persistence_location;

src/send_connack.c

@@ -54,6 +54,14 @@ int send__connack(struct mosquitto_db *db, struct mosquitto *context, int ack, i
  return rc;
  }
  }
+ if(db->config->max_packet_size > 0){
+ rc = mosquitto_property_add_int32(&connack_props, MQTT_PROP_MAXIMUM_PACKET_SIZE, db->config->max_packet_size);
+ if(rc){
+ mosquitto_property_free_all(&connack_props);
+ return rc;
+ }
+ }
+
  /* FIXME - disable support until available */
  rc = mosquitto_property_add_byte(&connack_props, MQTT_PROP_SHARED_SUB_AVAILABLE, 0);
  if(rc){

test/broker/12-prop-maximum-packet-size-broker.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+
+# Check whether the broker disconnects a client nicely when they send a too large packet.
+
+from mosq_test_helper import *
+
+def write_config(filename, port):
+ with open(filename, 'w') as f:
+ f.write("port %d\n" % (port))
+ f.write("max_packet_size 30\n")
+
+port = mosq_test.get_port()
+conf_file = os.path.basename(__file__).replace('.py', '.conf')
+write_config(conf_file, port)
+
+rc = 1
+
+keepalive = 10
+connect_packet = mosq_test.gen_connect("test", proto_ver=5, keepalive=keepalive)
+props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_MAXIMUM_PACKET_SIZE, 30)
+connack_packet = mosq_test.gen_connack(rc=0, proto_ver=5, properties=props)
+
+publish_packet = mosq_test.gen_publish("test/topic", qos=0, payload="0123456789012345678901234567890", proto_ver=5)
+disconnect_packet = mosq_test.gen_disconnect(reason_code=149, proto_ver=5)
+
+broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port, use_conf=True)
+
+try:
+ sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
+ mosq_test.do_send_receive(sock, publish_packet, disconnect_packet, "disconnect")
+ rc = 0
+finally:
+ broker.terminate()
+ broker.wait()
+ os.remove(conf_file)
+ (stdo, stde) = broker.communicate()
+ if rc:
+ print(stde)
+
+exit(rc)
+

test/broker/Makefile

@@ -175,5 +175,6 @@ endif
  ./12-prop-server-keepalive.py
  ./12-prop-response-topic.py
  ./12-prop-response-topic-correlation-data.py
+ ./12-prop-maximum-packet-size-broker.py
  ./12-prop-maximum-packet-size-connect.py
  ./12-prop-maximum-packet-size-publish.py

test/broker/test.py

@@ -142,6 +142,7 @@
  (1, './12-prop-server-keepalive.py'),
  (1, './12-prop-response-topic.py'),
  (1, './12-prop-response-topic-correlation-data.py'),
+ (1, './12-prop-maximum-packet-size-broker.py'),
  (1, './12-prop-maximum-packet-size-connect.py'),
  (1, './12-prop-maximum-packet-size-publish.py'),
  ]