[0.x] SIP events mixed in event handling

[oriol-c]

What version of Janus is this happening on?
0.11.7

Have you tested a more recent version of Janus too?
No, code in relation to janus_sip_sofia_logger has not been changed.

Additional context
When sending about 20cps via SIP, some of the sip-out events sent using the events handlers contain mixed SIP messages (https://pastebin.com/RYx4aW7e) . As you can see, the body starts with an ACK message but is then mixed with the next INVITE message, it actually goes one line (sofia log line) of the ACK and one line of the INVITE.
I could see that the sip plugin redirects the sofia log to this function janus_sip_sofia_logger, but I'm not sure if something can be done in janus to prevent the logs coming from sofia from being mixed.

[lminiero]

No, to my knowledge that's the only way we can intercept outgoing and incoming messages using the NUA stack. I don't remember mixed messages ever being an issue, if not initially before we added locking to the logger function. Not sure if anything changed in Sofia SIP that can cause the problem now, but to be honest I wouldn't know how to fix that, since we're working with raw buffers to reconstruct messages.

[lminiero]

Thinking about it, the logger is shared among all sessions, but we have global properties (e.g., sofia_log and call_id, but skip, started, append too) that we sometimes access without a lock, and the function may be called by different threads.

Since you're using 0.x, can you check if the diff below helps? It uses sessions_mutex as a way to protect the whole function, since we were using it before already to look up call-ids. Notice I haven't tested this, so I don't know if it has other issues, but I'm mostly interested in checking if it helps with the mixup of lines.

diff

diff --git a/plugins/janus_sip.c b/plugins/janus_sip.c
index bbc31d14..bafa8191 100644
--- a/plugins/janus_sip.c
+++ b/plugins/janus_sip.c
@@ -1672,12 +1672,14 @@ static void janus_sip_sofia_logger(void *stream, char const *fmt, va_list ap) {
 #pragma GCC diagnostic pop
 #endif
 #pragma GCC diagnostic pop
+	janus_mutex_lock(&sessions_mutex);
 	if(skip) {
 		/* This is a message we're not interested in: just check when it ends */
 		if(line[3] == '-') {
 			skip = FALSE;
 			append = FALSE;
 		}
+		janus_mutex_unlock(&sessions_mutex);
 		return;
 	}
 	if(append) {
@@ -1694,7 +1696,6 @@ static void janus_sip_sofia_logger(void *stream, char const *fmt, va_list ap) {
 				append = FALSE;
 				/* Look for the session this message belongs to */
 				janus_sip_session *session = NULL;
-				janus_mutex_lock(&sessions_mutex);
 				if(strlen(call_id))
 					session = g_hash_table_lookup(callids, call_id);
 				if(!session) {
@@ -1721,7 +1722,6 @@ static void janus_sip_sofia_logger(void *stream, char const *fmt, va_list ap) {
 				}
 				if(session)
 					janus_refcount_increase(&session->ref);
-				janus_mutex_unlock(&sessions_mutex);
 				if(session) {
 					/* Notify event handlers about the content of the whole outgoing SIP message */
 					json_t *info = json_object();
@@ -1736,6 +1736,7 @@ static void janus_sip_sofia_logger(void *stream, char const *fmt, va_list ap) {
 				sofia_log[0] = '\0';
 				call_id[0] = '\0';
 			}
+			janus_mutex_unlock(&sessions_mutex);
 			return;
 		}
 		if(strlen(line) == 1) {
@@ -1746,6 +1747,7 @@ static void janus_sip_sofia_logger(void *stream, char const *fmt, va_list ap) {
 			char *header = &line[3];
 			if(strstr(header, "OPTIONS") == header) {
 				skip = TRUE;
+				janus_mutex_unlock(&sessions_mutex);
 				return;
 			}
 			/* Is this a Call-ID header? Keep note of it */
@@ -1755,6 +1757,7 @@ static void janus_sip_sofia_logger(void *stream, char const *fmt, va_list ap) {
 			/* Append the line to our buffer, skipping the indent */
 			janus_strlcat(sofia_log, &line[3], sizeof(sofia_log));
 		}
+		janus_mutex_unlock(&sessions_mutex);
 		return;
 	}
 	/* Still waiting to decide if this is a message we need */
@@ -1768,6 +1771,7 @@ static void janus_sip_sofia_logger(void *stream, char const *fmt, va_list ap) {
 		if(strstr(line, "-----"))
 			started = TRUE;
 	}
+	janus_mutex_unlock(&sessions_mutex);
 }
 
 /* Helpers to ref/unref sessions with active calls */

[lminiero]

@oriol-c ping 🙂

[oriol-c]

I can give it a try. I'll come back with the result. But it turns out that in Sofia-SIP 1.13.2 the logging mechanism changed and the callback is not called for each line but once per message. Therefore if using Sofia-SIP higher than 1.13.2 the sofia_log variable can be a local variable inside janus_sip_sofia_logger avoiding the possibility of having mixed messages.

I have a modified version of janus_sip.c that works for the new Sofia-SIP version, but honestly, I'm not that good enough programmer to have a janus_sip.c version that works for both mechanisms and that is a nice piece of code...

This is the a patch to be applied in 0.12.4.

Patch

1657,1659d1656
< char sofia_log[2048];
< char call_id[255];
< gboolean skip = FALSE, started = FALSE, append = FALSE;
1663a1661,1665
> 	char message[16000];
> 	char sofia_log[16000];
> 	char call_id[255];
> 	sofia_log[0] = '\0';
> 	call_id[0] = '\0';
1670c1672
< 	g_vsnprintf(line, sizeof(line), fmt, ap);
---
> 	g_vsnprintf(line, sizeof(message), fmt, ap);
1675,1679c1677,1725
< 	if(skip) {
< 		/* This is a message we're not interested in: just check when it ends */
< 		if(line[3] == '-') {
< 			skip = FALSE;
< 			append = FALSE;
---
> 	gchar **parts = g_strsplit(message, "\n", -1);
> 	if (parts) {
> 		int index = 0;
> 		char *line = NULL, *cr = NULL;
> 		while((line = parts[index]) != NULL) {
> 			cr = strchr(line, '\r');
> 			if(cr != NULL)
> 				*cr = '\0';
> 			if(*line == '\0') {
> 				if(cr != NULL)
> 					*cr = '\r';
> 				index++;
> 				continue;
> 			}
>
> 			/* First line of the log indicates if we send or receive a message. */
> 			/* We are only interested in the sent ones */
> 			if(index==0) {
> 				if(strstr(line, "send ") == line) {
> 					int length = atoi(line+5);
> 					JANUS_LOG(LOG_HUGE, "Intercepting message (%d bytes)\n", length);
> 					index++;
> 					continue;
> 				}
> 				g_strfreev(parts);
> 				return ;
> 			}
>
> 			if(strlen(line) == 1) {
> 				/* Append a carriage and return */
> 				janus_strlcat(sofia_log, "\r\n", sizeof(sofia_log));
> 			} else if(strstr(line, "-") == line ) {
> 				index++;
> 				continue;
> 			} else if(strstr(line, "OPTIONS") == line) {
> 				index++;
> 				continue;
> 			} else {
> 				/* Is this a Call-ID header? Keep note of it */
> 				if(strstr(line, "Call-ID") == line) {
> 					g_snprintf(call_id, sizeof(call_id), "%s", line+9);
> 				}
> 				/* Append the line to our buffer, skipping the indent */
> 				janus_strlcat(sofia_log, line, sizeof(sofia_log));
> 				janus_strlcat(sofia_log, "\r\n", sizeof(sofia_log));
> 			}
> 			if(cr != NULL)
> 				*cr = '\r';
> 			index++;
1680a1727,1728
> 		g_strfreev(parts);
> 	} else {
1683,1719c1731,1753
< 	if(append) {
< 		/* We're copying a message in our buffer: check if this is the end */
< 		if(line[3] == '-') {
< 			if(!started) {
< 				/* Ok, start appending from now on */
< 				started = TRUE;
< 				sofia_log[0] = '\0';
< 				call_id[0] = '\0';
< 			} else {
< 				/* Message ended, handle it */
< 				skip = FALSE;
< 				append = FALSE;
< 				/* Look for the session this message belongs to */
< 				janus_sip_session *session = NULL;
< 				janus_mutex_lock(&sessions_mutex);
< 				if(strlen(call_id))
< 					session = g_hash_table_lookup(callids, call_id);
< 				if(!session) {
< 					/* Couldn't find any SIP session with that Call-ID, check the request */
< 					if(strstr(sofia_log, "REGISTER") == sofia_log || strstr(sofia_log, "SIP/2.0 ") == sofia_log) {
< 						/* FIXME This is a REGISTER or a response code:
< 						 * check the To header and get the identity from there */
< 						char *from = strstr(sofia_log, "To: ");
< 						if(from) {
< 							from = from+4;
< 							char *start = strstr(from, "<");
< 							if(start) {
< 								start++;
< 								char *end = strstr(from, ">");
< 								if(end) {
< 									*end = '\0';
< 									g_snprintf(call_id, sizeof(call_id), "%s", start);
< 									*end = '>';
< 									session = g_hash_table_lookup(identities, call_id);
< 								}
< 							}
< 						}
---
>
> 	/* Look for the session this message belongs to */
> 	janus_sip_session *session = NULL;
> 	janus_mutex_lock(&sessions_mutex);
> 	if(strlen(call_id))
> 		session = g_hash_table_lookup(callids, call_id);
> 	if(!session) {
> 		/* Couldn't find any SIP session with that Call-ID, check the request */
> 		if(strstr(sofia_log, "REGISTER") == sofia_log || strstr(sofia_log, "SIP/2.0 ") == sofia_log) {
> 			/* FIXME This is a REGISTER or a response code:
> 			* check the To header and get the identity from there */
> 			char *from = strstr(sofia_log, "To: ");
> 			if(from) {
> 				from = from+4;
> 				char *start = strstr(from, "<");
> 				if(start) {
> 					start++;
> 					char *end = strstr(from, ">");
> 					if(end) {
> 						*end = '\0';
> 						g_snprintf(call_id, sizeof(call_id), "%s", start);
> 						*end = '>';
> 						session = g_hash_table_lookup(identities, call_id);
1722,1737d1755
< 				if(session)
< 					janus_refcount_increase(&session->ref);
< 				janus_mutex_unlock(&sessions_mutex);
< 				if(session) {
< 					/* Notify event handlers about the content of the whole outgoing SIP message */
< 					json_t *info = json_object();
< 					json_object_set_new(info, "event", json_string("sip-out"));
< 					json_object_set_new(info, "sip", json_string(sofia_log));
< 					gateway->notify_event(&janus_sip_plugin, session->handle, info);
< 					janus_refcount_decrease(&session->ref);
< 				} else {
< 					JANUS_LOG(LOG_WARN, "Couldn't find a session associated to this message, dropping it...\n%s", sofia_log);
< 				}
< 				/* Done, reset the buffers */
< 				sofia_log[0] = '\0';
< 				call_id[0] = '\0';
1739d1756
< 			return;
1741,1758d1757
< 		if(strlen(line) == 1) {
< 			/* Append a carriage and return */
< 			janus_strlcat(sofia_log, "\r\n", sizeof(sofia_log));
< 		} else {
< 			/* If this is an OPTIONS, we don't care: drop it */
< 			char *header = &line[3];
< 			if(strstr(header, "OPTIONS") == header) {
< 				skip = TRUE;
< 				return;
< 			}
< 			/* Is this a Call-ID header? Keep note of it */
< 			if(strstr(header, "Call-ID") == header) {
< 				g_snprintf(call_id, sizeof(call_id), "%s", header+9);
< 			}
< 			/* Append the line to our buffer, skipping the indent */
< 			janus_strlcat(sofia_log, &line[3], sizeof(sofia_log));
< 		}
< 		return;
1760,1769c1759,1770
< 	/* Still waiting to decide if this is a message we need */
< 	if(line[0] == 's' && line[1] == 'e' && line[2] == 'n' && line[3] == 'd' && line[4] == ' ') {
< 		/* An outgoing message is going to be logged, prepare for that */
< 		skip = FALSE;
< 		started = FALSE;
< 		append = TRUE;
< 		int length = atoi(&line[5]);
< 		JANUS_LOG(LOG_HUGE, "Intercepting message (%d bytes)\n", length);
< 		if(strstr(line, "-----"))
< 			started = TRUE;
---
> 	if(session)
> 		janus_refcount_increase(&session->ref);
> 	janus_mutex_unlock(&sessions_mutex);
> 	if(session) {
> 		/* Notify event handlers about the content of the whole outgoing SIP message */
> 		json_t *info = json_object();
> 		json_object_set_new(info, "event", json_string("sip-out"));
> 		json_object_set_new(info, "sip", json_string(sofia_log));
> 		gateway->notify_event(&janus_sip_plugin, session->handle, info);
> 		janus_refcount_decrease(&session->ref);
> 	} else {
> 		JANUS_LOG(LOG_WARN, "Couldn't find a session associated to this message, dropping it...\n%s", sofia_log);
1770a1772
> 	return;

[lminiero]

If you can submit your proposed change as a pull request, that would be better, as a diff shared like that is unfortunately not very readable.

[lminiero]

@oriol-c any update on testing my patch, and on making yours available as a PR?

[oriol-c]

I have just created #3063 . I haven't had the chance to test your patch.