CVE-2023-28366: Missing Release of Memory after Effective Lifetime (CWE-401)
value: The broker in Eclipse Mosquitto 1.3.2 through 2.x before 2.0.16 has a memory leak that can be abused remotely when a client sends many QoS 2 messages with duplicate message IDs, and fails to respond to PUBREC commands. This occurs because of mishandling of EAGAIN from the libc send function.
CWE description: The product does not sufficiently track and release allocated memory after it has been used, which slowly consumes remaining memory.
CVSS v3 base score: 7.5
Published: 2023-09-01T16:15Z
Repository: https://github.com/eclipse/mosquitto
Language: C
Files changed: ChangeLog.txt, lib/packet_mosq.c, src/context.c, src/database.c, src/handle_publish.c, src/mosquitto_broker_internal.h, test/broker/03-publish-qos2-dup.py, test/broker/Makefile, test/broker/test.py, www/pages/security.md
Commit date: 2023-08-16 14:01:21 +0100
Commit message: Fix for CVE-2023-28366

Diff: diff --git a/ChangeLog.txt b/ChangeLog.txt
index 231e4c3a..412d3b4d 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -1,6 +1,9 @@
 Security:
+- CVE-2023-28366: Fix memory leak in broker when clients send multiple QoS 2
+  messages with the same message ID, but then never respond to the PUBREC
+  commands.
 - Broker will now reject Will messages that attempt to publish to $CONTROL/.
 - Broker now validates usernames provided in a TLS certificate or TLS-PSK
   identity are valid UTF-8.
 - Fix potential crash when loading invalid persitence file.
 
diff --git a/lib/packet_mosq.c b/lib/packet_mosq.c
index d737c185..8c06b942 100644
--- a/lib/packet_mosq.c
+++ b/lib/packet_mosq.c
@@ -150,10 +150,25 @@ int packet__queue(struct mosquitto *mosq, struct mosquitto__packet *packet)
 	packet->pos = 0;
 	packet->to_process = packet->packet_length;
 
 	packet->next = NULL;
 	pthread_mutex_lock(&mosq->out_packet_mutex);
+
+#ifdef WITH_BROKER
+	if(mosq->out_packet_count >= db.config->max_queued_messages){
+		mosquitto__free(packet);
+		if(mosq->is_dropping == false){
+			mosq->is_dropping = true;
+			log__printf(NULL, MOSQ_LOG_NOTICE,
+					"Outgoing messages are being dropped for client %s.",
+					mosq->id);
+		}
+		G_MSGS_DROPPED_INC();
+		return MOSQ_ERR_SUCCESS;
+	}
+#endif
+
 	if(mosq->out_packet){
 		mosq->out_packet_last->next = packet;
 	}else{
 		mosq->out_packet = packet;
 	}
diff --git a/src/context.c b/src/context.c
index 923d06f8..3ed49765 100644
--- a/src/context.c
+++ b/src/context.c
@@ -81,13 +81,13 @@ struct mosquitto *context__init(mosq_sock_t sock)
 			mosquitto__free(context);
 			return NULL;
 		}
 	}
 	context->bridge = NULL;
-	context->msgs_in.inflight_maximum = db.config->max_inflight_messages;
+	context->msgs_in.inflight_maximum = 1;
 	context->msgs_out.inflight_maximum = db.config->max_inflight_messages;
-	context->msgs_in.inflight_quota = db.config->max_inflight_messages;
+	context->msgs_in.inflight_quota = 1;
 	context->msgs_out.inflight_quota = db.config->max_inflight_messages;
 	context->max_qos = 2;
 #ifdef WITH_TLS
 	context->ssl = NULL;
 #endif
@@ -96,20 +96,39 @@ struct mosquitto *context__init(mosq_sock_t sock)
 		HASH_ADD(hh_sock, db.contexts_by_sock, sock, sizeof(context->sock), context);
 	}
 	return context;
 }
 
+static void context__cleanup_out_packets(struct mosquitto *context)
+{
+	struct mosquitto__packet *packet;
+
+	if(!context) return;
+
+	if(context->current_out_packet){
+		packet__cleanup(context->current_out_packet);
+		mosquitto__free(context->current_out_packet);
+		context->current_out_packet = NULL;
+	}
+	while(context->out_packet){
+		packet__cleanup(context->out_packet);
+		packet = context->out_packet;
+		context->out_packet = context->out_packet->next;
+		mosquitto__free(packet);
+	}
+	context->out_packet_count = 0;
+}
+
+
 /*
  * This will result in any outgoing packets going unsent. If we're disconnected
  * forcefully then it is usually an error condition and shouldn't be a problem,
  * but it will mean that CONNACK messages will never get sent for bad protocol
  * versions for example.
  */
 void context__cleanup(struct mosquitto *context, bool force_free)
 {
-	struct mosquitto__packet *packet;
-
 	if(!context) return;
 
 	if(force_free){
 		context->clean_start = true;
 	}
@@ -119,10 +138,11 @@ void context__cleanup(struct mosquitto *context, bool force_free)
 		bridge__cleanup(context);
 	}
 #endif
 
 	alias__free_all(context);
+	context__cleanup_out_packets(context);
 
 	mosquitto__free(context->auth_method);
 	context->auth_method = NULL;
 
 	mosquitto__free(context->username);
@@ -146,22 +166,11 @@ void context__cleanup(struct mosquitto *context, bool force_free)
 		context__remove_from_by_id(context);
 		mosquitto__free(context->id);
 		context->id = NULL;
 	}
 	packet__cleanup(&(context->in_packet));
-	if(context->current_out_packet){
-		packet__cleanup(context->current_out_packet);
-		mosquitto__free(context->current_out_packet);
-		context->current_out_packet = NULL;
-	}
-	while(context->out_packet){
-		packet__cleanup(context->out_packet);
-		packet = context->out_packet;
-		context->out_packet = context->out_packet->next;
-		mosquitto__free(packet);
-	}
-	context->out_packet_count = 0;
+	context__cleanup_out_packets(context);
 #if defined(WITH_BROKER) && defined(__GLIBC__) && defined(WITH_ADNS)
 	if(context->adns){
 		gai_cancel(context->adns);
 		mosquitto__free((struct addrinfo *)context->adns->ar_request);
 		mosquitto__free(context->adns);
diff --git a/src/database.c b/src/database.c
index 5e77a281..1a926f50 100644
--- a/src/database.c
+++ b/src/database.c
@@ -553,11 +553,11 @@ int db__message_insert(struct mosquitto *context, uint16_t mid, enum mosquitto_m
 	if(state == mosq_ms_queued){
 		db.persistence_changes++;
 	}
 #endif
 
-	msg = mosquitto__malloc(sizeof(struct mosquitto_client_msg));
+	msg = mosquitto__calloc(1, sizeof(struct mosquitto_client_msg));
 	if(!msg) return MOSQ_ERR_NOMEM;
 	msg->prev = NULL;
 	msg->next = NULL;
 	msg->store = stored;
 	db__msg_store_ref_inc(msg->store);
@@ -611,10 +611,12 @@ int db__message_insert(struct mosquitto *context, uint16_t mid, enum mosquitto_m
 	}
 #endif
 
 	if(dir == mosq_md_out && msg->qos > 0 && state != mosq_ms_queued){
 		util__decrement_send_quota(context);
+	}else if(dir == mosq_md_in && msg->qos > 0 && state != mosq_ms_queued){
+		util__decrement_receive_quota(context);
 	}
 
 	if(dir == mosq_md_out && update){
 		rc = db__message_write_inflight_out_latest(context);
 		if(rc) return rc;
@@ -794,27 +796,28 @@ int db__message_store(const struct mosquitto *source, struct mosquitto_msg_store
 	db__msg_store_add(stored);
 
 	return MOSQ_ERR_SUCCESS;
 }
 
-int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_msg_store **stored)
+int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_client_msg **client_msg)
 {
-	struct mosquitto_client_msg *tail;
+	struct mosquitto_client_msg *cmsg;
+
+	*client_msg = NULL;
 
 	if(!context) return MOSQ_ERR_INVAL;
 
-	*stored = NULL;
-	DL_FOREACH(context->msgs_in.inflight, tail){
-		if(tail->store->source_mid == mid){
-			*stored = tail->store;
+	DL_FOREACH(context->msgs_in.inflight, cmsg){
+		if(cmsg->store->source_mid == mid){
+			*client_msg = cmsg;
 			return MOSQ_ERR_SUCCESS;
 		}
 	}
 
-	DL_FOREACH(context->msgs_in.queued, tail){
-		if(tail->store->source_mid == mid){
-			*stored = tail->store;
+	DL_FOREACH(context->msgs_in.queued, cmsg){
+		if(cmsg->store->source_mid == mid){
+			*client_msg = cmsg;
 			return MOSQ_ERR_SUCCESS;
 		}
 	}
 
 	return 1;
@@ -912,20 +915,22 @@ static int db__message_reconnect_reset_incoming(struct mosquitto *context)
 			 * no harm. */
 			db__message_remove_from_inflight(&context->msgs_in, msg);
 		}else{
 			/* Message state can be preserved here because it should match
 			 * whatever the client has got. */
+			msg->dup = 0;
 		}
 	}
 
 	/* Messages received when the client was disconnected are put
 	 * in the mosq_ms_queued state. If we don't change them to the
 	 * appropriate "publish" state, then the queued messages won't
 	 * get sent until the client next receives a message - and they
 	 * will be sent out of order.
 	 */
 	DL_FOREACH_SAFE(context->msgs_in.queued, msg, tmp){
+		msg->dup = 0;
 		db__msg_add_to_queued_stats(&context->msgs_in, msg);
 		if(db__ready_for_flight(context, mosq_md_in, msg->qos)){
 			switch(msg->qos){
 				case 0:
 					msg->state = mosq_ms_publish_qos0;
diff --git a/src/handle_publish.c b/src/handle_publish.c
index 111f0316..d3aa5ce2 100644
--- a/src/handle_publish.c
+++ b/src/handle_publish.c
@@ -40,10 +40,11 @@ int handle__publish(struct mosquitto *context)
 	int rc = 0;
 	int rc2;
 	uint8_t header = context->in_packet.command;
 	int res = 0;
 	struct mosquitto_msg_store *msg, *stored = NULL;
+	struct mosquitto_client_msg *cmsg_stored = NULL;
 	size_t len;
 	uint16_t slen;
 	char *topic_mount;
 	mosquitto_property *properties = NULL;
 	mosquitto_property *p, *p_prev;
@@ -285,28 +286,28 @@ int handle__publish(struct mosquitto *context)
 			return rc;
 		}
 	}
 
 	if(msg->qos > 0){
-		db__message_store_find(context, msg->source_mid, &stored);
+		db__message_store_find(context, msg->source_mid, &cmsg_stored);
 	}
 
-	if(stored && msg->source_mid != 0 &&
-			(stored->qos != msg->qos
-			 || stored->payloadlen != msg->payloadlen
-			 || strcmp(stored->topic, msg->topic)
-			 || memcmp(stored->payload, msg->payload, msg->payloadlen) )){
+	if(cmsg_stored && cmsg_stored->store && msg->source_mid != 0 &&
+			(cmsg_stored->store->qos != msg->qos
+			 || cmsg_stored->store->payloadlen != msg->payloadlen
+			 || strcmp(cmsg_stored->store->topic, msg->topic)
+			 || memcmp(cmsg_stored->store->payload, msg->payload, msg->payloadlen) )){
 
 		log__printf(NULL, MOSQ_LOG_WARNING, "Reused message ID %u from %s detected. Clearing from storage.", msg->source_mid, context->id);
 		db__message_remove_incoming(context, msg->source_mid);
-		stored = NULL;
+		cmsg_stored = NULL;
 	}
 
-	if(!stored){
+	if(!cmsg_stored){
 		if(msg->qos == 0
 				|| db__ready_for_flight(context, mosq_md_in, msg->qos)
-				|| db__ready_for_queue(context, msg->qos, &context->msgs_in)){
+				){
 
 			dup = 0;
 			rc = db__message_store(context, msg, message_expiry_interval, 0, mosq_mo_client);
 			if(rc) return rc;
 		}else{
@@ -314,14 +315,17 @@ int handle__publish(struct mosquitto *context)
 			reason_code = MQTT_RC_QUOTA_EXCEEDED;
 			goto process_bad_message;
 		}
 		stored = msg;
 		msg = NULL;
+		dup = 0;
 	}else{
 		db__msg_store_free(msg);
 		msg = NULL;
-		dup = 1;
+		stored = cmsg_stored->store;
+		cmsg_stored->dup++;
+		dup = cmsg_stored->dup;
 	}
 
 	switch(stored->qos){
 		case 0:
 			rc2 = sub__messages_queue(context->id, stored->topic, stored->qos, stored->retain, &stored);
@@ -343,15 +347,21 @@ int handle__publish(struct mosquitto *context)
 			if(dup == 0){
 				res = db__message_insert(context, stored->source_mid, mosq_md_in, stored->qos, stored->retain, stored, NULL, false);
 			}else{
 				res = 0;
 			}
+
 			/* db__message_insert() returns 2 to indicate dropped message
 			 * due to queue. This isn't an error so don't disconnect them. */
 			/* FIXME - this is no longer necessary due to failing early above */
 			if(!res){
-				if(send__pubrec(context, stored->source_mid, 0, NULL)) rc = 1;
+				if(dup == 0 || dup == 1){
+					rc2 = send__pubrec(context, stored->source_mid, 0, NULL);
+					if(rc2) rc = rc2;
+				}else{
+					return MOSQ_ERR_PROTOCOL;
+				}
 			}else if(res == 1){
 				rc = 1;
 			}
 			break;
 	}
@@ -372,8 +382,11 @@ process_bad_message:
 				rc = send__pubrec(context, msg->source_mid, reason_code, NULL);
 				break;
 		}
 		db__msg_store_free(msg);
 	}
+	if(context->out_packet_count >= db.config->max_queued_messages){
+		rc = MQTT_RC_QUOTA_EXCEEDED;
+	}
 	return rc;
 }
 
diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h
index c28eaa2a..2cfa9d2b 100644
--- a/src/mosquitto_broker_internal.h
+++ b/src/mosquitto_broker_internal.h
@@ -392,11 +392,11 @@ struct mosquitto_client_msg{
 	uint16_t mid;
 	uint8_t qos;
 	bool retain;
 	enum mosquitto_msg_direction direction;
 	enum mosquitto_msg_state state;
-	bool dup;
+	uint8_t dup;
 };
 
 
 struct mosquitto__unpwd{
 	UT_hash_handle hh;
@@ -649,11 +649,11 @@ int db__message_release_incoming(struct mosquitto *context, uint16_t mid);
 int db__message_update_outgoing(struct mosquitto *context, uint16_t mid, enum mosquitto_msg_state state, int qos);
 void db__message_dequeue_first(struct mosquitto *context, struct mosquitto_msg_data *msg_data);
 int db__messages_delete(struct mosquitto *context, bool force_free);
 int db__messages_easy_queue(struct mosquitto *context, const char *topic, uint8_t qos, uint32_t payloadlen, const void *payload, int retain, uint32_t message_expiry_interval, mosquitto_property **properties);
 int db__message_store(const struct mosquitto *source, struct mosquitto_msg_store *stored, uint32_t message_expiry_interval, dbid_t store_id, enum mosquitto_msg_origin origin);
-int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_msg_store **stored);
+int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_client_msg **client_msg);
 void db__msg_store_add(struct mosquitto_msg_store *store);
 void db__msg_store_remove(struct mosquitto_msg_store *store);
 void db__msg_store_ref_inc(struct mosquitto_msg_store *store);
 void db__msg_store_ref_dec(struct mosquitto_msg_store **store);
 void db__msg_store_clean(void);
diff --git a/test/broker/03-publish-qos2-dup.py b/test/broker/03-publish-qos2-dup.py
new file mode 100755
index 00000000..70834fab
--- /dev/null
+++ b/test/broker/03-publish-qos2-dup.py
@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+
+from mosq_test_helper import *
+
+def do_test(proto_ver):
+    rc = 1
+    connect_packet = mosq_test.gen_connect("03-pub-qos2-dup-test", proto_ver=proto_ver)
+    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)
+
+    mid = 1
+    publish_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="message", proto_ver=proto_ver, dup=1)
+    pubrec_packet = mosq_test.gen_pubrec(mid, proto_ver=proto_ver)
+
+    disconnect_packet = mosq_test.gen_disconnect(reason_code=130, proto_ver=proto_ver)
+
+    port = mosq_test.get_port()
+    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)
+
+    try:
+        sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
+        mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec 1")
+        mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec 2")
+        if proto_ver == 5:
+            mosq_test.do_send_receive(sock, publish_packet, disconnect_packet, "disconnect")
+            rc = 0
+        else:
+            try:
+                mosq_test.do_send_receive(sock, publish_packet, b"", "disconnect1")
+                rc = 0
+            except BrokenPipeError:
+                rc = 0
+
+        sock.close()
+    except Exception as e:
+        print(e)
+    except mosq_test.TestError:
+        pass
+    finally:
+        broker.terminate()
+        broker.wait()
+        (stdo, stde) = broker.communicate()
+        if rc:
+            print(stde.decode('utf-8'))
+            print("proto_ver=%d" % (proto_ver))
+            exit(rc)
+
+
+def all_tests():
+    rc = do_test(proto_ver=4)
+    if rc:
+        return rc;
+    rc = do_test(proto_ver=5)
+    if rc:
+        return rc;
+    return 0
+
+if __name__ == '__main__':
+    all_tests()
diff --git a/test/broker/Makefile b/test/broker/Makefile
index 94cc8026..e66c7ffc 100644
--- a/test/broker/Makefile
+++ b/test/broker/Makefile
@@ -82,10 +82,11 @@ msg_sequence_test:
 	./03-publish-long-topic.py
 	./03-publish-qos1-max-inflight-expire.py
 	./03-publish-qos1-no-subscribers-v5.py
 	./03-publish-qos1-retain-disabled.py
 	./03-publish-qos1.py
+	./03-publish-qos2-dup.py
 	./03-publish-qos2-max-inflight.py
 	./03-publish-qos2.py
 
 04 :
 	./04-retain-check-source-persist-diff-port.py
diff --git a/test/broker/test.py b/test/broker/test.py
index 649746d3..e8956408 100755
--- a/test/broker/test.py
+++ b/test/broker/test.py
@@ -62,10 +62,11 @@ tests = [
     (1, './03-publish-qos1-max-inflight-expire.py'),
     (1, './03-publish-qos1-max-inflight.py'),
     (1, './03-publish-qos1-no-subscribers-v5.py'),
     (1, './03-publish-qos1-retain-disabled.py'),
     (1, './03-publish-qos1.py'),
+    (1, './03-publish-qos2-dup.py'),
     (1, './03-publish-qos2-max-inflight.py'),
     (1, './03-publish-qos2.py'),
 
     (1, './04-retain-check-source-persist.py'),
     (1, './04-retain-check-source.py'),
diff --git a/www/pages/security.md b/www/pages/security.md
index 2f124cca..ddc4423f 100644
--- a/www/pages/security.md
+++ b/www/pages/security.md
@@ -17,10 +17,13 @@ follow the steps on [Eclipse Security] page to report it.
 # Past vulnerabilities
 
 Listed with most recent first. Further information on security related issues
 can be found in the [security category].
 
+* June 2023: [CVE-2023-28366]: Clients sending unacknowledged QoS 2 messages
+  with duplicate message ids cause a memory leak. Affecting versions **1.3.2**
+  to **2.0.15** inclusive, fixed in **2.0.16**.
 * August 2022: Deleting the anonymous group in the dynamic security plugin
   could lead to a crash. Affecting versions **2.0.0** to **2.0.14** inclusive,
   fixed in **2.0.15**.
 * August 2021: [CVE-2021-34434] Affecting versions **2.0.0** to **2.0.11**
   inclusive, fixed in **2.0.12**.

Vulnerable code: 	context->msgs_in.inflight_maximum = db.config->max_inflight_messages;
	context->msgs_in.inflight_quota = db.config->max_inflight_messages;
	struct mosquitto__packet *packet;

	if(context->current_out_packet){
		packet__cleanup(context->current_out_packet);
		mosquitto__free(context->current_out_packet);
		context->current_out_packet = NULL;
	}
	while(context->out_packet){
		packet__cleanup(context->out_packet);
		packet = context->out_packet;
		context->out_packet = context->out_packet->next;
		mosquitto__free(packet);
	}
	context->out_packet_count = 0;
	msg = mosquitto__malloc(sizeof(struct mosquitto_client_msg));
int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_msg_store **stored)
	struct mosquitto_client_msg *tail;
	*stored = NULL;
	DL_FOREACH(context->msgs_in.inflight, tail){
		if(tail->store->source_mid == mid){
			*stored = tail->store;
	DL_FOREACH(context->msgs_in.queued, tail){
		if(tail->store->source_mid == mid){
			*stored = tail->store;
		db__message_store_find(context, msg->source_mid, &stored);
	if(stored && msg->source_mid != 0 &&
			(stored->qos != msg->qos
			 || stored->payloadlen != msg->payloadlen
			 || strcmp(stored->topic, msg->topic)
			 || memcmp(stored->payload, msg->payload, msg->payloadlen) )){
		stored = NULL;
	if(!stored){
				|| db__ready_for_queue(context, msg->qos, &context->msgs_in)){
		dup = 1;
				if(send__pubrec(context, stored->source_mid, 0, NULL)) rc = 1;
	bool dup;
int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_msg_store **stored);
Fixed code: - CVE-2023-28366: Fix memory leak in broker when clients send multiple QoS 2
  messages with the same message ID, but then never respond to the PUBREC
  commands.

#ifdef WITH_BROKER
	if(mosq->out_packet_count >= db.config->max_queued_messages){
		mosquitto__free(packet);
		if(mosq->is_dropping == false){
			mosq->is_dropping = true;
			log__printf(NULL, MOSQ_LOG_NOTICE,
					"Outgoing messages are being dropped for client %s.",
					mosq->id);
		}
		G_MSGS_DROPPED_INC();
		return MOSQ_ERR_SUCCESS;
	}
#endif

	context->msgs_in.inflight_maximum = 1;
	context->msgs_in.inflight_quota = 1;
static void context__cleanup_out_packets(struct mosquitto *context)
{
	struct mosquitto__packet *packet;

	if(!context) return;

	if(context->current_out_packet){
		packet__cleanup(context->current_out_packet);
		mosquitto__free(context->current_out_packet);
		context->current_out_packet = NULL;
	}
	while(context->out_packet){
		packet__cleanup(context->out_packet);
		packet = context->out_packet;
		context->out_packet = context->out_packet->next;
		mosquitto__free(packet);
	}
	context->out_packet_count = 0;
}


	context__cleanup_out_packets(context);
	context__cleanup_out_packets(context);
	msg = mosquitto__calloc(1, sizeof(struct mosquitto_client_msg));
	}else if(dir == mosq_md_in && msg->qos > 0 && state != mosq_ms_queued){
		util__decrement_receive_quota(context);
int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_client_msg **client_msg)
	struct mosquitto_client_msg *cmsg;

	*client_msg = NULL;
	DL_FOREACH(context->msgs_in.inflight, cmsg){
		if(cmsg->store->source_mid == mid){
			*client_msg = cmsg;
	DL_FOREACH(context->msgs_in.queued, cmsg){
		if(cmsg->store->source_mid == mid){
			*client_msg = cmsg;
			msg->dup = 0;
		msg->dup = 0;
	struct mosquitto_client_msg *cmsg_stored = NULL;
		db__message_store_find(context, msg->source_mid, &cmsg_stored);
	if(cmsg_stored && cmsg_stored->store && msg->source_mid != 0 &&
			(cmsg_stored->store->qos != msg->qos
			 || cmsg_stored->store->payloadlen != msg->payloadlen
			 || strcmp(cmsg_stored->store->topic, msg->topic)
			 || memcmp(cmsg_stored->store->payload, msg->payload, msg->payloadlen) )){
		cmsg_stored = NULL;
	if(!cmsg_stored){
				){
		dup = 0;
		stored = cmsg_stored->store;
		cmsg_stored->dup++;
		dup = cmsg_stored->dup;

				if(dup == 0 || dup == 1){
					rc2 = send__pubrec(context, stored->source_mid, 0, NULL);
					if(rc2) rc = rc2;
				}else{
					return MOSQ_ERR_PROTOCOL;
				}
	if(context->out_packet_count >= db.config->max_queued_messages){
		rc = MQTT_RC_QUOTA_EXCEEDED;
	}
	uint8_t dup;
int db__message_store_find(struct mosquitto *context, uint16_t mid, struct mosquitto_client_msg **client_msg);
#!/usr/bin/env python3

from mosq_test_helper import *

def do_test(proto_ver):
    rc = 1
    connect_packet = mosq_test.gen_connect("03-pub-qos2-dup-test", proto_ver=proto_ver)
    connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver)

    mid = 1
    publish_packet = mosq_test.gen_publish("topic", qos=2, mid=mid, payload="message", proto_ver=proto_ver, dup=1)
    pubrec_packet = mosq_test.gen_pubrec(mid, proto_ver=proto_ver)

    disconnect_packet = mosq_test.gen_disconnect(reason_code=130, proto_ver=proto_ver)

    port = mosq_test.get_port()
    broker = mosq_test.start_broker(filename=os.path.basename(__file__), port=port)

    try:
        sock = mosq_test.do_client_connect(connect_packet, connack_packet, port=port)
        mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec 1")
        mosq_test.do_send_receive(sock, publish_packet, pubrec_packet, "pubrec 2")
        if proto_ver == 5:
            mosq_test.do_send_receive(sock, publish_packet, disconnect_packet, "disconnect")
            rc = 0
        else:
            try:
                mosq_test.do_send_receive(sock, publish_packet, b"", "disconnect1")
                rc = 0
            except BrokenPipeError:
                rc = 0

        sock.close()
    except Exception as e:
        print(e)
    except mosq_test.TestError:
        pass
    finally:
        broker.terminate()
        broker.wait()
        (stdo, stde) = broker.communicate()
        if rc:
            print(stde.decode('utf-8'))
            print("proto_ver=%d" % (proto_ver))
            exit(rc)


def all_tests():
    rc = do_test(proto_ver=4)
    if rc:
        return rc;
    rc = do_test(proto_ver=5)
    if rc:
        return rc;
    return 0

if __name__ == '__main__':
    all_tests()
	./03-publish-qos2-dup.py
    (1, './03-publish-qos2-dup.py'),
* June 2023: [CVE-2023-28366]: Clients sending unacknowledged QoS 2 messages
  with duplicate message ids cause a memory leak. Affecting versions **1.3.2**
  to **2.0.15** inclusive, fixed in **2.0.16**.