aaa_diameter: Fix race condition with async dm_send_request()

- Avoid reading the @dmsg after it has been put on the queue, as it
might get freed meanwhile.

* aaa_diameter: Fix race condition on pending async replies

It was possible for the dm_send_request_async_tout() async timeout
function to ran concurrently with a late Diameter server reply, leading
to a use-after-free bug on the @cond struct.

* Add refcounting to the "cond" object

The SHM-stored @cond object is effectively referenced by two separate
processes/threads, which run concurrently:
    - dm_send_request_async_tout(), the reactor async timeout callback
    - dm_receive_msg(), the libfdcore receiver thread(s)

Author: liviuchircu

modules/aaa_diameter/aaa_diameter.c

@@ -472,21 +472,22 @@ struct dm_async_msg {
 	struct dm_cond *cond;
 };
 
-static struct dm_async_msg *dm_get_async_msg(pv_spec_t *rpl_avps_pv, aaa_message *dmsg)
+static struct dm_async_msg *dm_get_async_msg(pv_spec_t *rpl_avps_pv, struct dm_cond *cond)
 {
 	struct dm_async_msg *msg = pkg_malloc(sizeof *msg);
 	if (!msg)
 		return NULL;
 	memset(msg, 0, sizeof *msg);
 	msg->ret = rpl_avps_pv;
-	msg->cond = ((struct dm_message *)(dmsg->avpair))->reply_cond;
+	msg->cond = cond;
+	dm_cond_ref(cond);
 	return msg;
 }
 
 static void dm_free_sync_msg(struct dm_async_msg *amsg)
 {
 	if (amsg->cond)
-		shm_free(amsg->cond);
+		dm_cond_unref(amsg->cond);
 	pkg_free(amsg);
 }
 
@@ -529,12 +530,19 @@ static int dm_send_request_async_reply(int fd,
 static int dm_send_request_async_tout(int fd,
 		struct sip_msg *msg, void *param)
 {
+	int removed;
 	struct dm_async_msg *amsg = (struct dm_async_msg *)param;
 	pv_value_t val = {STR_NULL, 0, PV_VAL_NULL};
 
+	async_status = ASYNC_DONE_CLOSE_FD;
+
 	if (pv_set_value(msg, amsg->ret, 0, &val) != 0)
 		LM_ERR("failed to set output rpl_avps pv to NULL\n");
 
+	removed = dm_drop_pending_reply_cond(amsg->cond);
+	if (removed > 0)
+		dm_cond_unref(amsg->cond);
+
 	dm_free_sync_msg(amsg);
 	return -2;
 }
@@ -546,6 +554,7 @@ static int dm_send_request_async(struct sip_msg *msg, async_ctx *ctx,
 	struct dict_object *req;
 	cJSON *avps;
 	struct dm_async_msg *amsg;
+	struct dm_cond *cond;
 
 	if (fd_dict_search(fd_g_config->cnf_dict, DICT_COMMAND, CMD_BY_CODE_R,
 	      cmd_code, &req, ENOENT) == ENOENT) {
@@ -585,12 +594,12 @@ static int dm_send_request_async(struct sip_msg *msg, async_ctx *ctx,
 		_dm_destroy_message(dmsg);
 		goto error;
 	}
-	if (_dm_send_message_async(NULL, dmsg, &async_status) < 0) {
+	if (_dm_send_message_async(NULL, dmsg, &async_status, &cond) < 0) {
 		LM_ERR("cannot send async message!\n");
 		goto error;
 	}
 
-	amsg = dm_get_async_msg(rpl_avps_pv, dmsg);
+	amsg = dm_get_async_msg(rpl_avps_pv, cond);
 	if (!amsg)
 		goto error;
 	cJSON_Delete(avps);

modules/aaa_diameter/dm_impl.c

@@ -113,6 +113,7 @@ static struct dm_cond *dm_get_cond(int type, diameter_reply_cb *cb, void *param)
 		return NULL;
 	}
 	memset(cond, 0, sizeof *cond);
+	cond->ref = 1;
 	cond->type = type;
 	switch (type) {
 	case DM_TYPE_EVENT:
@@ -274,16 +275,19 @@ static void dm_cond_event_resume(int sender, void *param)
 	} while (ret < 0 && (errno == EINTR || errno == EAGAIN));
 	if (ret < 0)
 		LM_ERR("could not notify resume: %s\n", strerror(errno));
+
+	dm_cond_unref(cond);
 }
 
 static void dm_cond_signal(struct dm_cond *cond)
 {
 	LM_INFO("singalling %p/%d\n", cond, cond->type);
 	switch (cond->type) {
 	case DM_TYPE_EVENT:
+		dm_cond_ref(cond);
 		if (ipc_send_rpc(cond->sync.event.pid, dm_cond_event_resume, cond) < 0) {
 			LM_ERR("could not resume async MI command!\n");
-			shm_free(cond);
+			dm_cond_unref(cond);
 		}
 		break;
 	case DM_TYPE_COND:
@@ -295,7 +299,6 @@ static void dm_cond_signal(struct dm_cond *cond)
 	case DM_TYPE_CB:
 		if (cond->sync.cb.f)
 			cond->sync.cb.f(NULL, &cond->rpl, cond->sync.cb.p);
-		shm_free(cond);
 		break;
 	}
 }
@@ -356,6 +359,7 @@ static int dm_auth_reply(struct msg **_msg, struct avp * avp, struct session * s
 		rpl_cond->rpl.is_error = 0;
 	}
 	dm_cond_signal(rpl_cond);
+	dm_cond_unref(rpl_cond);
 
 out:
 	FD_CHECK(fd_msg_free(msg));
@@ -715,6 +719,7 @@ static int dm_receive_msg(struct msg **_msg, struct avp * avp, struct session *
 		rpl_cond->rpl.is_error = 0;
 	}
 	dm_cond_signal(rpl_cond);
+	dm_cond_unref(rpl_cond);
 
 out:
 	cJSON_InitHooks(NULL);
@@ -830,11 +835,57 @@ int dm_add_pending_reply(const str *callid, struct dm_cond *reply_cond)
 	}
 
 	*cond_holder = reply_cond;
+	dm_cond_ref(reply_cond);
 	hash_unlock(pending_replies, hentry);
 
 	return 0;
 }
 
+struct dm_find_cond_ctx {
+	struct dm_cond *cond;
+	str key;
+	int found;
+};
+
+static int dm_match_pending_cond(void *param, str key, void *value)
+{
+	struct dm_find_cond_ctx *ctx = (struct dm_find_cond_ctx *)param;
+
+	if (value != ctx->cond)
+		return 0;
+
+	ctx->key = key;
+	ctx->found = 1;
+	return 1;
+}
+
+int dm_drop_pending_reply_cond(struct dm_cond *reply_cond)
+{
+	struct dm_find_cond_ctx ctx;
+	unsigned int hentry;
+
+	if (!reply_cond)
+		return 0;
+
+	for (hentry = 0; hentry < hash_size(pending_replies); hentry++) {
+		memset(&ctx, 0, sizeof ctx);
+		ctx.cond = reply_cond;
+
+		hash_lock(pending_replies, hentry);
+		map_for_each(pending_replies->entries[hentry], dm_match_pending_cond, &ctx);
+		if (!ctx.found) {
+			hash_unlock(pending_replies, hentry);
+			continue;
+		}
+
+		hash_remove(pending_replies, hentry, ctx.key);
+		hash_unlock(pending_replies, hentry);
+		return 1;
+	}
+
+	return 0;
+}
+
 
 
 /* all of these AVPs are part of "RADIUS Extension for Digest Auth" RFC 5090 */
@@ -1938,21 +1989,22 @@ static void dm_push_queue(aaa_message *msg, struct dm_cond *cond)
 	pthread_mutex_unlock(msg_send_lk);
 }
 
-int _dm_send_message_async(aaa_conn *_, aaa_message *req, int *fd)
+int _dm_send_message_async(aaa_conn *_, aaa_message *req, int *fd, struct dm_cond **cond)
 {
-	struct dm_cond *cond;
+	struct dm_cond *_cond;
 
 	if (!req)
 		return -1;
 
-	cond = dm_get_cond(DM_TYPE_EVENT, NULL, NULL);
-	if (!cond) {
+	_cond = dm_get_cond(DM_TYPE_EVENT, NULL, NULL);
+	if (!_cond) {
 		LM_ERR("out of memory for cond\n");
 		return -1;
 	}
 
-	*fd = cond->sync.event.fd;
-	dm_push_queue(req, cond);
+	*fd = _cond->sync.event.fd;
+	*cond = _cond;
+	dm_push_queue(req, _cond);
 
 	LM_DBG("message queued for async sending\n");
 
@@ -2191,6 +2243,7 @@ void _dm_destroy_message(aaa_message *msg)
 
 	dm = (struct dm_message *)msg->avpair;
 	dm_free_avps(&dm->avps);
+	dm_cond_unref(dm->reply_cond);
 	shm_free(dm);
 
 	shm_free(msg);

modules/aaa_diameter/dm_impl.h

@@ -22,6 +22,7 @@
 #define AAA_DIAMETER_IMPL
 
 #include "../../aaa/aaa.h"
+#include "../../mem/shm_mem.h"
 #include "diameter_api.h"
 
 #define __FD_CHECK(__call__, __retok__, __retval__) \
@@ -122,6 +123,7 @@ struct dm_avp {
 #define DM_TYPE_CB    (1<<2)
 
 struct dm_cond {
+	volatile int ref;
 	int type;
 	union {
 		struct {
@@ -140,6 +142,23 @@ struct dm_cond {
 
 	diameter_reply rpl;
 };
+
+static inline void dm_cond_ref(struct dm_cond *cond)
+{
+	if (!cond || cond->type == DM_TYPE_COND)
+		return;
+
+	__atomic_add_fetch(&cond->ref, 1, __ATOMIC_SEQ_CST);
+}
+
+static inline void dm_cond_unref(struct dm_cond *cond)
+{
+	if (!cond || cond->type == DM_TYPE_COND)
+		return;
+
+	if (__atomic_sub_fetch(&cond->ref, 1, __ATOMIC_SEQ_CST) == 0)
+		shm_free(cond);
+}
 int init_mutex_cond(pthread_mutex_t *mutex, pthread_cond_t *cond);
 
 extern struct list_head dm_unreplied_req;
@@ -169,7 +188,8 @@ int dm_avp_add(aaa_conn *_, aaa_message *msg, aaa_map *avp, void *val,
 int dm_build_avps(struct list_head *subavps, cJSON *array);
 int dm_send_message(aaa_conn *_, aaa_message *req, aaa_message **__);
 int _dm_send_message(aaa_conn *_, aaa_message *req, struct dm_cond **reply_cond);
-int _dm_send_message_async(aaa_conn *_, aaa_message *req, int *fd);
+int _dm_send_message_async(aaa_conn *_, aaa_message *req, int *fd, struct dm_cond **cond);
+int dm_drop_pending_reply_cond(struct dm_cond *reply_cond);
 int _dm_get_message_response(struct dm_cond *cond, char **rpl_avps);
 void _dm_release_message_response(struct dm_cond *cond, char *rpl_avps);
 int dm_destroy_message(aaa_conn *con, aaa_message *msg);