From cd0987a6616bcf186de0423b25fd0430efb75688 Mon Sep 17 00:00:00 2001 From: "Roger A. Light" Date: Thu, 27 Feb 2025 00:02:26 +0000 Subject: [PATCH] Fix potential deadlock in mosquitto_sub if `-W` is used. Closes #3175. Thanks to Audric Schiltknecht --- ChangeLog.txt | 3 +++ lib/connect.c | 26 ++++++++++++++++---------- lib/handle_connack.c | 38 +++++++++++++++++++++++--------------- lib/handle_pubackcomp.c | 26 ++++++++++++++++---------- lib/handle_publish.c | 34 ++++++++++++++++++++-------------- lib/handle_pubrec.c | 9 ++++++--- lib/handle_pubrel.c | 24 ++++++++++++++---------- lib/handle_suback.c | 24 ++++++++++++++---------- lib/handle_unsuback.c | 24 ++++++++++++++---------- lib/logging_mosq.c | 10 ++++++---- lib/loop.c | 25 +++++++++++++++---------- lib/packet_mosq.c | 28 ++++++++++++++++------------ lib/srv_mosq.c | 24 ++++++++++++++---------- lib/util_mosq.c | 24 ++++++++++++++---------- 14 files changed, 191 insertions(+), 128 deletions(-) diff --git a/ChangeLog.txt b/ChangeLog.txt index 197674bd..66de676b 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -9,6 +9,9 @@ Broker: - Fix mismatched wrapped/unwrapped memory alloc/free in properties. Closes #3192. - Fix `allow_anonymous false` not being applied in local only mode. Closes #3198. +Client library: +- Fix potential deadlock in mosquitto_sub if `-W` is used. Closes #3175. + Apps: - mosquitto_ctrl dynsec now also allows `-i` to specify a clientid as well as `-c`. This matches the documentation which states `-i`. Closes #3219. diff --git a/lib/connect.c b/lib/connect.c index 3dd3353a..314d2c52 100644 --- a/lib/connect.c +++ b/lib/connect.c @@ -267,6 +267,9 @@ int mosquitto_disconnect_v5(struct mosquitto *mosq, int reason_code, const mosqu void do_client_disconnect(struct mosquitto *mosq, int reason_code, const mosquitto_property *properties) { + void (*on_disconnect)(struct mosquitto *, void *userdata, int rc); + void (*on_disconnect_v5)(struct mosquitto *, void *userdata, int rc, const mosquitto_property *props); + mosquitto__set_state(mosq, mosq_cs_disconnected); net__socket_close(mosq); @@ -287,17 +290,20 @@ void do_client_disconnect(struct mosquitto *mosq, int reason_code, const mosquit COMPAT_pthread_mutex_unlock(&mosq->msgtime_mutex); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_disconnect){ - mosq->in_callback = true; - mosq->on_disconnect(mosq, mosq->userdata, reason_code); - mosq->in_callback = false; - } - if(mosq->on_disconnect_v5){ - mosq->in_callback = true; - mosq->on_disconnect_v5(mosq, mosq->userdata, reason_code, properties); - mosq->in_callback = false; - } + on_disconnect = mosq->on_disconnect; + on_disconnect_v5 = mosq->on_disconnect_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + + if(on_disconnect){ + mosq->in_callback = true; + on_disconnect(mosq, mosq->userdata, reason_code); + mosq->in_callback = false; + } + if(on_disconnect_v5){ + mosq->in_callback = true; + on_disconnect_v5(mosq, mosq->userdata, reason_code, properties); + mosq->in_callback = false; + } COMPAT_pthread_mutex_unlock(&mosq->current_out_packet_mutex); } diff --git a/lib/handle_connack.c b/lib/handle_connack.c index a3ae5c76..bae41b7b 100644 --- a/lib/handle_connack.c +++ b/lib/handle_connack.c @@ -32,27 +32,35 @@ Contributors: static void connack_callback(struct mosquitto *mosq, uint8_t reason_code, uint8_t connect_flags, const mosquitto_property *properties) { + void (*on_connect)(struct mosquitto *, void *userdata, int rc); + void (*on_connect_with_flags)(struct mosquitto *, void *userdata, int rc, int flags); + void (*on_connect_v5)(struct mosquitto *, void *userdata, int rc, int flags, const mosquitto_property *props); + log__printf(mosq, MOSQ_LOG_DEBUG, "Client %s received CONNACK (%d)", SAFE_PRINT(mosq->id), reason_code); if(reason_code == MQTT_RC_SUCCESS){ mosq->reconnects = 0; } COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_connect){ - mosq->in_callback = true; - mosq->on_connect(mosq, mosq->userdata, reason_code); - mosq->in_callback = false; - } - if(mosq->on_connect_with_flags){ - mosq->in_callback = true; - mosq->on_connect_with_flags(mosq, mosq->userdata, reason_code, connect_flags); - mosq->in_callback = false; - } - if(mosq->on_connect_v5){ - mosq->in_callback = true; - mosq->on_connect_v5(mosq, mosq->userdata, reason_code, connect_flags, properties); - mosq->in_callback = false; - } + on_connect = mosq->on_connect; + on_connect_with_flags = mosq->on_connect_with_flags; + on_connect_v5 = mosq->on_connect_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + + if(on_connect){ + mosq->in_callback = true; + on_connect(mosq, mosq->userdata, reason_code); + mosq->in_callback = false; + } + if(on_connect_with_flags){ + mosq->in_callback = true; + on_connect_with_flags(mosq, mosq->userdata, reason_code, connect_flags); + mosq->in_callback = false; + } + if(on_connect_v5){ + mosq->in_callback = true; + on_connect_v5(mosq, mosq->userdata, reason_code, connect_flags, properties); + mosq->in_callback = false; + } } diff --git a/lib/handle_pubackcomp.c b/lib/handle_pubackcomp.c index 545ccbee..30d91239 100644 --- a/lib/handle_pubackcomp.c +++ b/lib/handle_pubackcomp.c @@ -136,19 +136,25 @@ int handle__pubackcomp(struct mosquitto *mosq, const char *type) rc = message__delete(mosq, mid, mosq_md_out, qos); if(rc == MOSQ_ERR_SUCCESS){ + void (*on_publish)(struct mosquitto *, void *userdata, int mid); + void (*on_publish_v5)(struct mosquitto *, void *userdata, int mid, int reason_code, const mosquitto_property *props); + /* Only inform the client the message has been sent once. */ COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_publish){ - mosq->in_callback = true; - mosq->on_publish(mosq, mosq->userdata, mid); - mosq->in_callback = false; - } - if(mosq->on_publish_v5){ - mosq->in_callback = true; - mosq->on_publish_v5(mosq, mosq->userdata, mid, reason_code, properties); - mosq->in_callback = false; - } + on_publish = mosq->on_publish; + on_publish_v5 = mosq->on_publish_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + + if(on_publish){ + mosq->in_callback = true; + on_publish(mosq, mosq->userdata, mid); + mosq->in_callback = false; + } + if(on_publish_v5){ + mosq->in_callback = true; + on_publish_v5(mosq, mosq->userdata, mid, reason_code, properties); + mosq->in_callback = false; + } mosquitto_property_free_all(&properties); }else if(rc != MOSQ_ERR_NOT_FOUND){ return rc; diff --git a/lib/handle_publish.c b/lib/handle_publish.c index dc5776fe..98bd4240 100644 --- a/lib/handle_publish.c +++ b/lib/handle_publish.c @@ -120,20 +120,24 @@ int handle__publish(struct mosquitto *mosq) (long)message->msg.payloadlen); message->timestamp = mosquitto_time(); + void (*on_message)(struct mosquitto *, void *userdata, const struct mosquitto_message *message); + void (*on_message_v5)(struct mosquitto *, void *userdata, const struct mosquitto_message *message, const mosquitto_property *props); switch(message->msg.qos){ case 0: COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_message){ + on_message = mosq->on_message; + on_message_v5 = mosq->on_message_v5; + COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_message){ mosq->in_callback = true; - mosq->on_message(mosq, mosq->userdata, &message->msg); + on_message(mosq, mosq->userdata, &message->msg); mosq->in_callback = false; } if(mosq->on_message_v5){ mosq->in_callback = true; - mosq->on_message_v5(mosq, mosq->userdata, &message->msg, properties); + on_message_v5(mosq, mosq->userdata, &message->msg, properties); mosq->in_callback = false; } - COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); message__cleanup(&message); mosquitto_property_free_all(&properties); return MOSQ_ERR_SUCCESS; @@ -141,17 +145,19 @@ int handle__publish(struct mosquitto *mosq) util__decrement_receive_quota(mosq); rc = send__puback(mosq, mid, 0, NULL); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_message){ - mosq->in_callback = true; - mosq->on_message(mosq, mosq->userdata, &message->msg); - mosq->in_callback = false; - } - if(mosq->on_message_v5){ - mosq->in_callback = true; - mosq->on_message_v5(mosq, mosq->userdata, &message->msg, properties); - mosq->in_callback = false; - } + on_message = mosq->on_message; + on_message_v5 = mosq->on_message_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_message){ + mosq->in_callback = true; + on_message(mosq, mosq->userdata, &message->msg); + mosq->in_callback = false; + } + if(on_message_v5){ + mosq->in_callback = true; + on_message_v5(mosq, mosq->userdata, &message->msg, properties); + mosq->in_callback = false; + } message__cleanup(&message); mosquitto_property_free_all(&properties); return rc; diff --git a/lib/handle_pubrec.c b/lib/handle_pubrec.c index c46ec73a..13e78128 100644 --- a/lib/handle_pubrec.c +++ b/lib/handle_pubrec.c @@ -107,13 +107,16 @@ int handle__pubrec(struct mosquitto *mosq) }else{ if(!message__delete(mosq, mid, mosq_md_out, 2)){ /* Only inform the client the message has been sent once. */ + void (*on_publish_v5)(struct mosquitto *, void *userdata, int mid, int reason_code, const mosquitto_property *props); + COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_publish_v5){ + on_publish_v5 = mosq->on_publish_v5; + COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_publish_v5){ mosq->in_callback = true; - mosq->on_publish_v5(mosq, mosq->userdata, mid, reason_code, properties); + on_publish_v5(mosq, mosq->userdata, mid, reason_code, properties); mosq->in_callback = false; } - COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); } util__increment_send_quota(mosq); COMPAT_pthread_mutex_lock(&mosq->msgs_out.mutex); diff --git a/lib/handle_pubrel.c b/lib/handle_pubrel.c index 29465a5d..0e5251c6 100644 --- a/lib/handle_pubrel.c +++ b/lib/handle_pubrel.c @@ -116,18 +116,22 @@ int handle__pubrel(struct mosquitto *mosq) if(rc == MOSQ_ERR_SUCCESS){ /* Only pass the message on if we have removed it from the queue - this * prevents multiple callbacks for the same message. */ + void (*on_message)(struct mosquitto *, void *userdata, const struct mosquitto_message *message); + void (*on_message_v5)(struct mosquitto *, void *userdata, const struct mosquitto_message *message, const mosquitto_property *props); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_message){ - mosq->in_callback = true; - mosq->on_message(mosq, mosq->userdata, &message->msg); - mosq->in_callback = false; - } - if(mosq->on_message_v5){ - mosq->in_callback = true; - mosq->on_message_v5(mosq, mosq->userdata, &message->msg, message->properties); - mosq->in_callback = false; - } + on_message = mosq->on_message; + on_message_v5 = mosq->on_message_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_message){ + mosq->in_callback = true; + on_message(mosq, mosq->userdata, &message->msg); + mosq->in_callback = false; + } + if(on_message_v5){ + mosq->in_callback = true; + on_message_v5(mosq, mosq->userdata, &message->msg, message->properties); + mosq->in_callback = false; + } mosquitto_property_free_all(&properties); message__cleanup(&message); }else if(rc == MOSQ_ERR_NOT_FOUND){ diff --git a/lib/handle_suback.c b/lib/handle_suback.c index 50367fc1..100177cc 100644 --- a/lib/handle_suback.c +++ b/lib/handle_suback.c @@ -97,18 +97,22 @@ int handle__suback(struct mosquitto *mosq) /* Immediately free, we don't do anything with Reason String or User Property at the moment */ mosquitto_property_free_all(&properties); #else + void (*on_subscribe)(struct mosquitto *, void *userdata, int mid, int qos_count, const int *granted_qos); + void (*on_subscribe_v5)(struct mosquitto *, void *userdata, int mid, int qos_count, const int *granted_qos, const mosquitto_property *props); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_subscribe){ - mosq->in_callback = true; - mosq->on_subscribe(mosq, mosq->userdata, mid, qos_count, granted_qos); - mosq->in_callback = false; - } - if(mosq->on_subscribe_v5){ - mosq->in_callback = true; - mosq->on_subscribe_v5(mosq, mosq->userdata, mid, qos_count, granted_qos, properties); - mosq->in_callback = false; - } + on_subscribe = mosq->on_subscribe; + on_subscribe_v5 = mosq->on_subscribe_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_subscribe){ + mosq->in_callback = true; + on_subscribe(mosq, mosq->userdata, mid, qos_count, granted_qos); + mosq->in_callback = false; + } + if(on_subscribe_v5){ + mosq->in_callback = true; + on_subscribe_v5(mosq, mosq->userdata, mid, qos_count, granted_qos, properties); + mosq->in_callback = false; + } mosquitto_property_free_all(&properties); #endif mosquitto__free(granted_qos); diff --git a/lib/handle_unsuback.c b/lib/handle_unsuback.c index 4bd7986a..94e45f48 100644 --- a/lib/handle_unsuback.c +++ b/lib/handle_unsuback.c @@ -76,18 +76,22 @@ int handle__unsuback(struct mosquitto *mosq) /* Immediately free, we don't do anything with Reason String or User Property at the moment */ mosquitto_property_free_all(&properties); #else + void (*on_unsubscribe)(struct mosquitto *, void *userdata, int mid); + void (*on_unsubscribe_v5)(struct mosquitto *, void *userdata, int mid, const mosquitto_property *props); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_unsubscribe){ - mosq->in_callback = true; - mosq->on_unsubscribe(mosq, mosq->userdata, mid); - mosq->in_callback = false; - } - if(mosq->on_unsubscribe_v5){ - mosq->in_callback = true; - mosq->on_unsubscribe_v5(mosq, mosq->userdata, mid, properties); - mosq->in_callback = false; - } + on_unsubscribe = mosq->on_unsubscribe; + on_unsubscribe_v5 = mosq->on_unsubscribe_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_unsubscribe){ + mosq->in_callback = true; + on_unsubscribe(mosq, mosq->userdata, mid); + mosq->in_callback = false; + } + if(on_unsubscribe_v5){ + mosq->in_callback = true; + on_unsubscribe_v5(mosq, mosq->userdata, mid, properties); + mosq->in_callback = false; + } mosquitto_property_free_all(&properties); #endif diff --git a/lib/logging_mosq.c b/lib/logging_mosq.c index a71d2514..3f6a0d95 100644 --- a/lib/logging_mosq.c +++ b/lib/logging_mosq.c @@ -33,16 +33,19 @@ int log__printf(struct mosquitto *mosq, unsigned int priority, const char *fmt, va_list va; char *s; size_t len; + void (*on_log)(struct mosquitto *, void *userdata, int level, const char *str); assert(mosq); assert(fmt); COMPAT_pthread_mutex_lock(&mosq->log_callback_mutex); - if(mosq->on_log){ + on_log = mosq->on_log; + COMPAT_pthread_mutex_unlock(&mosq->log_callback_mutex); + + if(on_log){ len = strlen(fmt) + 500; s = mosquitto__malloc(len*sizeof(char)); if(!s){ - COMPAT_pthread_mutex_unlock(&mosq->log_callback_mutex); return MOSQ_ERR_NOMEM; } @@ -51,11 +54,10 @@ int log__printf(struct mosquitto *mosq, unsigned int priority, const char *fmt, va_end(va); s[len-1] = '\0'; /* Ensure string is null terminated. */ - mosq->on_log(mosq, mosq->userdata, (int)priority, s); + on_log(mosq, mosq->userdata, (int)priority, s); mosquitto__free(s); } - COMPAT_pthread_mutex_unlock(&mosq->log_callback_mutex); return MOSQ_ERR_SUCCESS; } diff --git a/lib/loop.c b/lib/loop.c index eb1e6e3f..4905fd99 100644 --- a/lib/loop.c +++ b/lib/loop.c @@ -335,18 +335,23 @@ static int mosquitto__loop_rc_handle(struct mosquitto *mosq, int rc) if(state == mosq_cs_disconnecting || state == mosq_cs_disconnected){ rc = MOSQ_ERR_SUCCESS; } + + void (*on_disconnect)(struct mosquitto *, void *userdata, int rc); + void (*on_disconnect_v5)(struct mosquitto *, void *userdata, int rc, const mosquitto_property *props); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_disconnect){ - mosq->in_callback = true; - mosq->on_disconnect(mosq, mosq->userdata, rc); - mosq->in_callback = false; - } - if(mosq->on_disconnect_v5){ - mosq->in_callback = true; - mosq->on_disconnect_v5(mosq, mosq->userdata, rc, NULL); - mosq->in_callback = false; - } + on_disconnect = mosq->on_disconnect; + on_disconnect_v5 = mosq->on_disconnect_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_disconnect){ + mosq->in_callback = true; + on_disconnect(mosq, mosq->userdata, rc); + mosq->in_callback = false; + } + if(on_disconnect_v5){ + mosq->in_callback = true; + on_disconnect_v5(mosq, mosq->userdata, rc, NULL); + mosq->in_callback = false; + } } return rc; } diff --git a/lib/packet_mosq.c b/lib/packet_mosq.c index 2ea41b0e..4d2f4fe6 100644 --- a/lib/packet_mosq.c +++ b/lib/packet_mosq.c @@ -296,20 +296,24 @@ int packet__write(struct mosquitto *mosq) if(((packet->command)&0xF6) == CMD_PUBLISH){ G_PUB_MSGS_SENT_INC(1); #ifndef WITH_BROKER + void (*on_publish)(struct mosquitto *, void *userdata, int mid); + void (*on_publish_v5)(struct mosquitto *, void *userdata, int mid, int reason_code, const mosquitto_property *props); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_publish){ - /* This is a QoS=0 message */ - mosq->in_callback = true; - mosq->on_publish(mosq, mosq->userdata, packet->mid); - mosq->in_callback = false; - } - if(mosq->on_publish_v5){ - /* This is a QoS=0 message */ - mosq->in_callback = true; - mosq->on_publish_v5(mosq, mosq->userdata, packet->mid, 0, NULL); - mosq->in_callback = false; - } + on_publish = mosq->on_publish; + on_publish_v5 = mosq->on_publish_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_publish){ + /* This is a QoS=0 message */ + mosq->in_callback = true; + on_publish(mosq, mosq->userdata, packet->mid); + mosq->in_callback = false; + } + if(on_publish_v5){ + /* This is a QoS=0 message */ + mosq->in_callback = true; + on_publish_v5(mosq, mosq->userdata, packet->mid, 0, NULL); + mosq->in_callback = false; + } }else if(((packet->command)&0xF0) == CMD_DISCONNECT){ do_client_disconnect(mosq, MOSQ_ERR_SUCCESS, NULL); packet__cleanup(packet); diff --git a/lib/srv_mosq.c b/lib/srv_mosq.c index 2e46745b..963cbd15 100644 --- a/lib/srv_mosq.c +++ b/lib/srv_mosq.c @@ -49,18 +49,22 @@ static void srv_callback(void *arg, int status, int timeouts, unsigned char *abu }else{ log__printf(mosq, MOSQ_LOG_ERR, "Error: SRV lookup failed (%d).", status); /* FIXME - calling on_disconnect here isn't correct. */ + void (*on_disconnect)(struct mosquitto *, void *userdata, int rc); + void (*on_disconnect_v5)(struct mosquitto *, void *userdata, int rc, const mosquitto_property *props); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_disconnect){ - mosq->in_callback = true; - mosq->on_disconnect(mosq, mosq->userdata, MOSQ_ERR_LOOKUP); - mosq->in_callback = false; - } - if(mosq->on_disconnect_v5){ - mosq->in_callback = true; - mosq->on_disconnect_v5(mosq, mosq->userdata, MOSQ_ERR_LOOKUP, NULL); - mosq->in_callback = false; - } + on_disconnect = mosq->on_disconnect; + on_disconnect_v5 = mosq->on_disconnect_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_disconnect){ + mosq->in_callback = true; + on_disconnect(mosq, mosq->userdata, MOSQ_ERR_LOOKUP); + mosq->in_callback = false; + } + if(on_disconnect_v5){ + mosq->in_callback = true; + on_disconnect_v5(mosq, mosq->userdata, MOSQ_ERR_LOOKUP, NULL); + mosq->in_callback = false; + } } } #endif diff --git a/lib/util_mosq.c b/lib/util_mosq.c index 4bebcbd0..134e78fd 100644 --- a/lib/util_mosq.c +++ b/lib/util_mosq.c @@ -118,18 +118,22 @@ int mosquitto__check_keepalive(struct mosquitto *mosq) }else{ rc = MOSQ_ERR_KEEPALIVE; } + void (*on_disconnect)(struct mosquitto *, void *userdata, int rc); + void (*on_disconnect_v5)(struct mosquitto *, void *userdata, int rc, const mosquitto_property *props); COMPAT_pthread_mutex_lock(&mosq->callback_mutex); - if(mosq->on_disconnect){ - mosq->in_callback = true; - mosq->on_disconnect(mosq, mosq->userdata, rc); - mosq->in_callback = false; - } - if(mosq->on_disconnect_v5){ - mosq->in_callback = true; - mosq->on_disconnect_v5(mosq, mosq->userdata, rc, NULL); - mosq->in_callback = false; - } + on_disconnect = mosq->on_disconnect; + on_disconnect_v5 = mosq->on_disconnect_v5; COMPAT_pthread_mutex_unlock(&mosq->callback_mutex); + if(on_disconnect){ + mosq->in_callback = true; + on_disconnect(mosq, mosq->userdata, rc); + mosq->in_callback = false; + } + if(on_disconnect_v5){ + mosq->in_callback = true; + on_disconnect_v5(mosq, mosq->userdata, rc, NULL); + mosq->in_callback = false; + } return rc; #endif