summaryrefslogtreecommitdiff
path: root/res
diff options
context:
space:
mode:
Diffstat (limited to 'res')
-rw-r--r--res/res_http_websocket.c23
-rw-r--r--res/res_pjsip.c92
-rw-r--r--res/res_pjsip/pjsip_distributor.c105
-rw-r--r--res/res_pjsip_dialog_info_body_generator.c9
-rw-r--r--res/res_pjsip_mwi.c45
-rw-r--r--res/res_pjsip_pidf_body_generator.c11
-rw-r--r--res/res_pjsip_pubsub.c2
-rw-r--r--res/res_pjsip_session.c5
-rw-r--r--res/res_pjsip_t38.c22
-rw-r--r--res/res_pjsip_xpidf_body_generator.c9
-rw-r--r--res/res_rtp_asterisk.c207
11 files changed, 393 insertions, 137 deletions
diff --git a/res/res_http_websocket.c b/res/res_http_websocket.c
index 9e8b680a9..1f1f77ce5 100644
--- a/res/res_http_websocket.c
+++ b/res/res_http_websocket.c
@@ -781,13 +781,6 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
return 0;
}
- fprintf(ser->f, "HTTP/1.1 101 Switching Protocols\r\n"
- "Upgrade: %s\r\n"
- "Connection: Upgrade\r\n"
- "Sec-WebSocket-Accept: %s\r\n",
- upgrade,
- websocket_combine_key(key, base64, sizeof(base64)));
-
/* RFC 6455, Section 4.1:
*
* 6. If the response includes a |Sec-WebSocket-Protocol| header
@@ -798,11 +791,23 @@ int AST_OPTIONAL_API_NAME(ast_websocket_uri_cb)(struct ast_tcptls_session_instan
* Connection_.
*/
if (protocol) {
- fprintf(ser->f, "Sec-WebSocket-Protocol: %s\r\n",
+ fprintf(ser->f, "HTTP/1.1 101 Switching Protocols\r\n"
+ "Upgrade: %s\r\n"
+ "Connection: Upgrade\r\n"
+ "Sec-WebSocket-Accept: %s\r\n"
+ "Sec-WebSocket-Protocol: %s\r\n\r\n",
+ upgrade,
+ websocket_combine_key(key, base64, sizeof(base64)),
protocol);
+ } else {
+ fprintf(ser->f, "HTTP/1.1 101 Switching Protocols\r\n"
+ "Upgrade: %s\r\n"
+ "Connection: Upgrade\r\n"
+ "Sec-WebSocket-Accept: %s\r\n\r\n",
+ upgrade,
+ websocket_combine_key(key, base64, sizeof(base64)));
}
- fprintf(ser->f, "\r\n");
fflush(ser->f);
} else {
diff --git a/res/res_pjsip.c b/res/res_pjsip.c
index e92de51bb..658a55e88 100644
--- a/res/res_pjsip.c
+++ b/res/res_pjsip.c
@@ -1867,6 +1867,15 @@
#define MOD_DATA_CONTACT "contact"
+/*! Number of serializers in pool if one not supplied. */
+#define SERIALIZER_POOL_SIZE 8
+
+/*! Next serializer pool index to use. */
+static int serializer_pool_pos;
+
+/*! Pool of serializers to use if not supplied. */
+static struct ast_taskprocessor *serializer_pool[SERIALIZER_POOL_SIZE];
+
static pjsip_endpoint *ast_pjsip_endpoint;
static struct ast_threadpool *sip_threadpool;
@@ -3341,8 +3350,62 @@ struct ast_taskprocessor *ast_sip_create_serializer(void)
return ast_sip_create_serializer_group(NULL);
}
+/*!
+ * \internal
+ * \brief Shutdown the serializers in the default pool.
+ * \since 14.0.0
+ *
+ * \return Nothing
+ */
+static void serializer_pool_shutdown(void)
+{
+ int idx;
+
+ for (idx = 0; idx < SERIALIZER_POOL_SIZE; ++idx) {
+ ast_taskprocessor_unreference(serializer_pool[idx]);
+ serializer_pool[idx] = NULL;
+ }
+}
+
+/*!
+ * \internal
+ * \brief Setup the serializers in the default pool.
+ * \since 14.0.0
+ *
+ * \retval 0 on success.
+ * \retval -1 on error.
+ */
+static int serializer_pool_setup(void)
+{
+ int idx;
+
+ for (idx = 0; idx < SERIALIZER_POOL_SIZE; ++idx) {
+ serializer_pool[idx] = ast_sip_create_serializer();
+ if (!serializer_pool[idx]) {
+ serializer_pool_shutdown();
+ return -1;
+ }
+ }
+ return 0;
+}
+
int ast_sip_push_task(struct ast_taskprocessor *serializer, int (*sip_task)(void *), void *task_data)
{
+ if (!serializer) {
+ unsigned int pos;
+
+ /*
+ * Pick a serializer to use from the pool.
+ *
+ * Note: We don't care about any reentrancy behavior
+ * when incrementing serializer_pool_pos. If it gets
+ * incorrectly incremented it doesn't matter.
+ */
+ pos = serializer_pool_pos++;
+ pos %= SERIALIZER_POOL_SIZE;
+ serializer = serializer_pool[pos];
+ }
+
if (serializer) {
return ast_taskprocessor_push(serializer, sip_task, task_data);
} else {
@@ -3395,18 +3458,10 @@ int ast_sip_push_task_synchronous(struct ast_taskprocessor *serializer, int (*si
std.task = sip_task;
std.task_data = task_data;
- if (serializer) {
- if (ast_taskprocessor_push(serializer, sync_task, &std)) {
- ast_mutex_destroy(&std.lock);
- ast_cond_destroy(&std.cond);
- return -1;
- }
- } else {
- if (ast_threadpool_push(sip_threadpool, sync_task, &std)) {
- ast_mutex_destroy(&std.lock);
- ast_cond_destroy(&std.cond);
- return -1;
- }
+ if (ast_sip_push_task(serializer, sync_task, &std)) {
+ ast_mutex_destroy(&std.lock);
+ ast_cond_destroy(&std.cond);
+ return -1;
}
ast_mutex_lock(&std.lock);
@@ -3697,6 +3752,18 @@ static int load_module(void)
return AST_MODULE_LOAD_DECLINE;
}
+ if (serializer_pool_setup()) {
+ ast_log(LOG_ERROR, "Failed to create SIP serializer pool. Aborting load\n");
+ ast_threadpool_shutdown(sip_threadpool);
+ ast_sip_destroy_system();
+ pj_pool_release(memory_pool);
+ memory_pool = NULL;
+ pjsip_endpt_destroy(ast_pjsip_endpoint);
+ ast_pjsip_endpoint = NULL;
+ pj_caching_pool_destroy(&caching_pool);
+ return AST_MODULE_LOAD_DECLINE;
+ }
+
ast_sip_initialize_dns();
pjsip_tsx_layer_init_module(ast_pjsip_endpoint);
@@ -3826,6 +3893,7 @@ static int unload_module(void)
*/
ast_sip_push_task_synchronous(NULL, unload_pjsip, NULL);
+ serializer_pool_shutdown();
ast_threadpool_shutdown(sip_threadpool);
ast_sip_destroy_cli();
diff --git a/res/res_pjsip/pjsip_distributor.c b/res/res_pjsip/pjsip_distributor.c
index e32f02833..9b052603a 100644
--- a/res/res_pjsip/pjsip_distributor.c
+++ b/res/res_pjsip/pjsip_distributor.c
@@ -22,22 +22,106 @@
#include "asterisk/res_pjsip.h"
#include "include/res_pjsip_private.h"
+#include "asterisk/taskprocessor.h"
+#include "asterisk/threadpool.h"
static int distribute(void *data);
static pj_bool_t distributor(pjsip_rx_data *rdata);
+static pj_status_t record_serializer(pjsip_tx_data *tdata);
static pjsip_module distributor_mod = {
.name = {"Request Distributor", 19},
.priority = PJSIP_MOD_PRIORITY_TSX_LAYER - 6,
+ .on_tx_request = record_serializer,
.on_rx_request = distributor,
.on_rx_response = distributor,
};
+/*!
+ * \internal
+ * \brief Record the task's serializer name on the tdata structure.
+ * \since 14.0.0
+ *
+ * \param tdata The outgoing message.
+ *
+ * \retval PJ_SUCCESS.
+ */
+static pj_status_t record_serializer(pjsip_tx_data *tdata)
+{
+ struct ast_taskprocessor *serializer;
+
+ serializer = ast_threadpool_serializer_get_current();
+ if (serializer) {
+ const char *name;
+
+ name = ast_taskprocessor_name(serializer);
+ if (!ast_strlen_zero(name)
+ && (!tdata->mod_data[distributor_mod.id]
+ || strcmp(tdata->mod_data[distributor_mod.id], name))) {
+ char *tdata_name;
+
+ /* The serializer in use changed. */
+ tdata_name = pj_pool_alloc(tdata->pool, strlen(name) + 1);
+ strcpy(tdata_name, name);/* Safe */
+
+ tdata->mod_data[distributor_mod.id] = tdata_name;
+ }
+ }
+
+ return PJ_SUCCESS;
+}
+
+/*!
+ * \internal
+ * \brief Find the request tdata to get the serializer it used.
+ * \since 14.0.0
+ *
+ * \param rdata The incoming message.
+ *
+ * \retval serializer on success.
+ * \retval NULL on error or could not find the serializer.
+ */
+static struct ast_taskprocessor *find_request_serializer(pjsip_rx_data *rdata)
+{
+ struct ast_taskprocessor *serializer = NULL;
+ pj_str_t tsx_key;
+ pjsip_transaction *tsx;
+
+ pjsip_tsx_create_key(rdata->tp_info.pool, &tsx_key, PJSIP_ROLE_UAC,
+ &rdata->msg_info.cseq->method, rdata);
+
+ tsx = pjsip_tsx_layer_find_tsx(&tsx_key, PJ_TRUE);
+ if (!tsx) {
+ ast_debug(1, "Could not find %.*s transaction for %d response.\n",
+ (int) pj_strlen(&rdata->msg_info.cseq->method.name),
+ pj_strbuf(&rdata->msg_info.cseq->method.name),
+ rdata->msg_info.msg->line.status.code);
+ return NULL;
+ }
+
+ if (tsx->last_tx) {
+ const char *serializer_name;
+
+ serializer_name = tsx->last_tx->mod_data[distributor_mod.id];
+ if (!ast_strlen_zero(serializer_name)) {
+ serializer = ast_taskprocessor_get(serializer_name, TPS_REF_IF_EXISTS);
+ }
+ }
+
+#ifdef HAVE_PJ_TRANSACTION_GRP_LOCK
+ pj_grp_lock_release(tsx->grp_lock);
+#else
+ pj_mutex_unlock(tsx->mutex);
+#endif
+
+ return serializer;
+}
+
/*! Dialog-specific information the distributor uses */
struct distributor_dialog_data {
- /* Serializer to distribute tasks to for this dialog */
+ /*! Serializer to distribute tasks to for this dialog */
struct ast_taskprocessor *serializer;
- /* Endpoint associated with this dialog */
+ /*! Endpoint associated with this dialog */
struct ast_sip_endpoint *endpoint;
};
@@ -167,6 +251,7 @@ static pj_bool_t distributor(pjsip_rx_data *rdata)
pjsip_dialog *dlg = find_dialog(rdata);
struct distributor_dialog_data *dist = NULL;
struct ast_taskprocessor *serializer = NULL;
+ struct ast_taskprocessor *req_serializer = NULL;
pjsip_rx_data *clone;
if (dlg) {
@@ -176,11 +261,16 @@ static pj_bool_t distributor(pjsip_rx_data *rdata)
}
}
- if (rdata->msg_info.msg->type == PJSIP_REQUEST_MSG && (
- !pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_cancel_method) ||
- !pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_bye_method)) &&
- !serializer) {
- pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 481, NULL, NULL, NULL);
+ if (serializer) {
+ /* We have a serializer so we know where to send the message. */
+ } else if (rdata->msg_info.msg->type == PJSIP_RESPONSE_MSG) {
+ req_serializer = find_request_serializer(rdata);
+ serializer = req_serializer;
+ } else if (!pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_cancel_method)
+ || !pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_bye_method)) {
+ /* We have a BYE or CANCEL request without a serializer. */
+ pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata,
+ PJSIP_SC_CALL_TSX_DOES_NOT_EXIST, NULL, NULL, NULL);
goto end;
}
@@ -196,6 +286,7 @@ end:
if (dlg) {
pjsip_dlg_dec_lock(dlg);
}
+ ast_taskprocessor_unreference(req_serializer);
return PJ_TRUE;
}
diff --git a/res/res_pjsip_dialog_info_body_generator.c b/res/res_pjsip_dialog_info_body_generator.c
index d9725f4c5..48ac60f98 100644
--- a/res/res_pjsip_dialog_info_body_generator.c
+++ b/res/res_pjsip_dialog_info_body_generator.c
@@ -163,14 +163,13 @@ static void dialog_info_to_string(void *body, struct ast_str **str)
int size;
do {
- size = pj_xml_print(dialog_info, ast_str_buffer(*str), ast_str_size(*str), PJ_TRUE);
- if (size == AST_PJSIP_XML_PROLOG_LEN) {
+ size = pj_xml_print(dialog_info, ast_str_buffer(*str), ast_str_size(*str) - 1, PJ_TRUE);
+ if (size <= AST_PJSIP_XML_PROLOG_LEN) {
ast_str_make_space(str, ast_str_size(*str) * 2);
++growths;
}
- } while (size == AST_PJSIP_XML_PROLOG_LEN && growths < MAX_STRING_GROWTHS);
-
- if (size == AST_PJSIP_XML_PROLOG_LEN) {
+ } while (size <= AST_PJSIP_XML_PROLOG_LEN && growths < MAX_STRING_GROWTHS);
+ if (size <= AST_PJSIP_XML_PROLOG_LEN) {
ast_log(LOG_WARNING, "dialog-info+xml body text too large\n");
return;
}
diff --git a/res/res_pjsip_mwi.c b/res/res_pjsip_mwi.c
index 6ca6cced0..06587daf7 100644
--- a/res/res_pjsip_mwi.c
+++ b/res/res_pjsip_mwi.c
@@ -138,9 +138,17 @@ static struct mwi_stasis_subscription *mwi_stasis_subscription_alloc(const char
/* Safe strcpy */
strcpy(mwi_stasis_sub->mailbox, mailbox);
+
+ ast_debug(3, "Creating stasis MWI subscription to mailbox %s for endpoint %s\n",
+ mailbox, mwi_sub->id);
ao2_ref(mwi_sub, +1);
- ast_debug(3, "Creating stasis MWI subscription to mailbox %s for endpoint %s\n", mailbox, mwi_sub->id);
mwi_stasis_sub->stasis_sub = stasis_subscribe_pool(topic, mwi_stasis_cb, mwi_sub);
+ if (!mwi_stasis_sub->stasis_sub) {
+ /* Failed to subscribe. */
+ ao2_ref(mwi_stasis_sub, -1);
+ ao2_ref(mwi_sub, -1);
+ mwi_stasis_sub = NULL;
+ }
return mwi_stasis_sub;
}
@@ -491,25 +499,41 @@ static void mwi_subscription_shutdown(struct ast_sip_subscription *sub)
mwi_sub = mwi_datastore->data;
ao2_callback(mwi_sub->stasis_subs, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE, unsubscribe_stasis, NULL);
+ ast_sip_subscription_remove_datastore(sub, MWI_DATASTORE);
ao2_ref(mwi_datastore, -1);
}
-static struct ast_datastore_info mwi_ds_info = { };
+static void mwi_ds_destroy(void *data)
+{
+ struct mwi_subscription *sub = data;
+
+ ao2_ref(sub, -1);
+}
+
+static struct ast_datastore_info mwi_ds_info = {
+ .destroy = mwi_ds_destroy,
+};
static int add_mwi_datastore(struct mwi_subscription *sub)
{
struct ast_datastore *mwi_datastore;
+ int res;
mwi_datastore = ast_sip_subscription_alloc_datastore(&mwi_ds_info, MWI_DATASTORE);
if (!mwi_datastore) {
return -1;
}
+ ao2_ref(sub, +1);
mwi_datastore->data = sub;
- ast_sip_subscription_add_datastore(sub->sip_sub, mwi_datastore);
+ /*
+ * NOTE: Adding the datastore to the subscription creates a ref loop
+ * that must be manually broken.
+ */
+ res = ast_sip_subscription_add_datastore(sub->sip_sub, mwi_datastore);
ao2_ref(mwi_datastore, -1);
- return 0;
+ return res;
}
/*!
@@ -621,8 +645,8 @@ static struct mwi_subscription *mwi_create_subscription(
}
if (add_mwi_datastore(sub)) {
- ast_log(LOG_WARNING, "Unable to allocate datastore on MWI "
- "subscription from %s\n", sub->id);
+ ast_log(LOG_WARNING, "Unable to add datastore for MWI subscription to %s\n",
+ sub->id);
ao2_ref(sub, -1);
return NULL;
}
@@ -714,12 +738,19 @@ static int mwi_subscription_established(struct ast_sip_subscription *sip_sub)
} else {
sub = mwi_subscribe_single(endpoint, sip_sub, resource);
}
-
if (!sub) {
ao2_cleanup(endpoint);
return -1;
}
+ if (!ao2_container_count(sub->stasis_subs)) {
+ /*
+ * We setup no MWI subscriptions so remove the MWI datastore
+ * to break the ref loop.
+ */
+ ast_sip_subscription_remove_datastore(sip_sub, MWI_DATASTORE);
+ }
+
ao2_cleanup(sub);
ao2_cleanup(endpoint);
return 0;
diff --git a/res/res_pjsip_pidf_body_generator.c b/res/res_pjsip_pidf_body_generator.c
index ef0cce599..d3be8c131 100644
--- a/res/res_pjsip_pidf_body_generator.c
+++ b/res/res_pjsip_pidf_body_generator.c
@@ -84,19 +84,18 @@ static int pidf_generate_body_content(void *body, void *data)
static void pidf_to_string(void *body, struct ast_str **str)
{
- int size;
- int growths = 0;
pjpidf_pres *pres = body;
+ int growths = 0;
+ int size;
do {
size = pjpidf_print(pres, ast_str_buffer(*str), ast_str_size(*str) - 1);
- if (size == AST_PJSIP_XML_PROLOG_LEN) {
+ if (size <= AST_PJSIP_XML_PROLOG_LEN) {
ast_str_make_space(str, ast_str_size(*str) * 2);
++growths;
}
- } while (size == AST_PJSIP_XML_PROLOG_LEN && growths < MAX_STRING_GROWTHS);
-
- if (size == AST_PJSIP_XML_PROLOG_LEN) {
+ } while (size <= AST_PJSIP_XML_PROLOG_LEN && growths < MAX_STRING_GROWTHS);
+ if (size <= AST_PJSIP_XML_PROLOG_LEN) {
ast_log(LOG_WARNING, "PIDF body text too large\n");
return;
}
diff --git a/res/res_pjsip_pubsub.c b/res/res_pjsip_pubsub.c
index c00bc76ee..650f5c5c8 100644
--- a/res/res_pjsip_pubsub.c
+++ b/res/res_pjsip_pubsub.c
@@ -1769,7 +1769,7 @@ static int rlmi_print_body(struct pjsip_msg_body *msg_body, char *buf, pj_size_t
pj_xml_node *rlmi = msg_body->data;
num_printed = pj_xml_print(rlmi, buf, size, PJ_TRUE);
- if (num_printed == AST_PJSIP_XML_PROLOG_LEN) {
+ if (num_printed <= AST_PJSIP_XML_PROLOG_LEN) {
return -1;
}
diff --git a/res/res_pjsip_session.c b/res/res_pjsip_session.c
index a8f8cba85..8c04a7cfb 100644
--- a/res/res_pjsip_session.c
+++ b/res/res_pjsip_session.c
@@ -1039,7 +1039,10 @@ void ast_sip_session_resume_reinvite(struct ast_sip_session *session)
return;
}
- pjsip_endpt_process_rx_data(ast_sip_get_pjsip_endpoint(), session->deferred_reinvite, NULL, NULL);
+ if (session->channel) {
+ pjsip_endpt_process_rx_data(ast_sip_get_pjsip_endpoint(),
+ session->deferred_reinvite, NULL, NULL);
+ }
pjsip_rx_data_free_cloned(session->deferred_reinvite);
session->deferred_reinvite = NULL;
}
diff --git a/res/res_pjsip_t38.c b/res/res_pjsip_t38.c
index 06a73cc11..5a2357dfc 100644
--- a/res/res_pjsip_t38.c
+++ b/res/res_pjsip_t38.c
@@ -135,10 +135,13 @@ static void t38_change_state(struct ast_sip_session *session, struct ast_sip_ses
}
session->t38state = new_state;
- ast_debug(2, "T.38 state changed to '%u' from '%u' on channel '%s'\n", new_state, old_state, ast_channel_name(session->channel));
+ ast_debug(2, "T.38 state changed to '%u' from '%u' on channel '%s'\n",
+ new_state, old_state,
+ session->channel ? ast_channel_name(session->channel) : "<gone>");
if (pj_timer_heap_cancel(pjsip_endpt_get_timer_heap(ast_sip_get_pjsip_endpoint()), &state->timer)) {
- ast_debug(2, "Automatic T.38 rejection on channel '%s' terminated\n", ast_channel_name(session->channel));
+ ast_debug(2, "Automatic T.38 rejection on channel '%s' terminated\n",
+ session->channel ? ast_channel_name(session->channel) : "<gone>");
ao2_ref(session, -1);
}
@@ -198,7 +201,8 @@ static int t38_automatic_reject(void *obj)
return 0;
}
- ast_debug(2, "Automatically rejecting T.38 request on channel '%s'\n", ast_channel_name(session->channel));
+ ast_debug(2, "Automatically rejecting T.38 request on channel '%s'\n",
+ session->channel ? ast_channel_name(session->channel) : "<gone>");
t38_change_state(session, session_media, datastore->data, T38_REJECTED);
ast_sip_session_resume_reinvite(session);
@@ -227,9 +231,9 @@ static struct t38_state *t38_state_get_or_alloc(struct ast_sip_session *session)
return datastore->data;
}
- if (!(datastore = ast_sip_session_alloc_datastore(&t38_datastore, "t38")) ||
- !(datastore->data = ast_calloc(1, sizeof(struct t38_state))) ||
- ast_sip_session_add_datastore(session, datastore)) {
+ if (!(datastore = ast_sip_session_alloc_datastore(&t38_datastore, "t38"))
+ || !(datastore->data = ast_calloc(1, sizeof(struct t38_state)))
+ || ast_sip_session_add_datastore(session, datastore)) {
return NULL;
}
@@ -324,9 +328,13 @@ static int t38_interpret_parameters(void *obj)
case AST_T38_REQUEST_NEGOTIATE: /* Request T38 */
/* Negotiation can not take place without a valid max_ifp value. */
if (!parameters->max_ifp) {
- t38_change_state(data->session, session_media, state, T38_REJECTED);
if (data->session->t38state == T38_PEER_REINVITE) {
+ t38_change_state(data->session, session_media, state, T38_REJECTED);
ast_sip_session_resume_reinvite(data->session);
+ } else if (data->session->t38state == T38_ENABLED) {
+ t38_change_state(data->session, session_media, state, T38_DISABLED);
+ ast_sip_session_refresh(data->session, NULL, NULL, NULL,
+ AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1);
}
break;
} else if (data->session->t38state == T38_PEER_REINVITE) {
diff --git a/res/res_pjsip_xpidf_body_generator.c b/res/res_pjsip_xpidf_body_generator.c
index 43cb1e78b..298235cbc 100644
--- a/res/res_pjsip_xpidf_body_generator.c
+++ b/res/res_pjsip_xpidf_body_generator.c
@@ -106,14 +106,13 @@ static void xpidf_to_string(void *body, struct ast_str **str)
int size;
do {
- size = pjxpidf_print(pres, ast_str_buffer(*str), ast_str_size(*str));
- if (size == AST_PJSIP_XML_PROLOG_LEN) {
+ size = pjxpidf_print(pres, ast_str_buffer(*str), ast_str_size(*str) - 1);
+ if (size <= AST_PJSIP_XML_PROLOG_LEN) {
ast_str_make_space(str, ast_str_size(*str) * 2);
++growths;
}
- } while (size == AST_PJSIP_XML_PROLOG_LEN && growths < MAX_STRING_GROWTHS);
-
- if (size == AST_PJSIP_XML_PROLOG_LEN) {
+ } while (size <= AST_PJSIP_XML_PROLOG_LEN && growths < MAX_STRING_GROWTHS);
+ if (size <= AST_PJSIP_XML_PROLOG_LEN) {
ast_log(LOG_WARNING, "XPIDF body text too large\n");
return;
}
diff --git a/res/res_rtp_asterisk.c b/res/res_rtp_asterisk.c
index b78f6e24e..f0e2f8379 100644
--- a/res/res_rtp_asterisk.c
+++ b/res/res_rtp_asterisk.c
@@ -204,11 +204,13 @@ struct rtp_learning_info {
#ifdef HAVE_OPENSSL_SRTP
struct dtls_details {
+ ast_mutex_t lock; /*!< Lock for timeout timer synchronization */
SSL *ssl; /*!< SSL session */
BIO *read_bio; /*!< Memory buffer for reading */
BIO *write_bio; /*!< Memory buffer for writing */
enum ast_rtp_dtls_setup dtls_setup; /*!< Current setup state */
enum ast_rtp_dtls_connection connection; /*!< Whether this is a new or existing connection */
+ int timeout_timer; /*!< Scheduler id for timeout timer */
};
#endif
@@ -317,7 +319,6 @@ struct ast_rtp {
#ifdef HAVE_OPENSSL_SRTP
SSL_CTX *ssl_ctx; /*!< SSL context */
- ast_mutex_t dtls_timer_lock; /*!< Lock for synchronization purposes */
enum ast_rtp_dtls_verify dtls_verify; /*!< What to verify */
enum ast_srtp_suite suite; /*!< SRTP crypto suite */
enum ast_rtp_dtls_hash local_hash; /*!< Local hash used for the fingerprint */
@@ -326,7 +327,6 @@ struct ast_rtp {
unsigned char remote_fingerprint[EVP_MAX_MD_SIZE]; /*!< Fingerprint of the peer certificate */
unsigned int rekey; /*!< Interval at which to renegotiate and rekey */
int rekeyid; /*!< Scheduled item id for rekeying */
- int dtlstimerid; /*!< Scheduled item id for DTLS retransmission for RTP */
struct dtls_details dtls; /*!< DTLS state information */
#endif
};
@@ -444,6 +444,8 @@ static int ast_rtp_sendcng(struct ast_rtp_instance *instance, int level);
#ifdef HAVE_OPENSSL_SRTP
static int ast_rtp_activate(struct ast_rtp_instance *instance);
static void dtls_srtp_check_pending(struct ast_rtp_instance *instance, struct ast_rtp *rtp, int rtcp);
+static void dtls_srtp_start_timeout_timer(struct ast_rtp_instance *instance, struct ast_rtp *rtp, int rtcp);
+static void dtls_srtp_stop_timeout_timer(struct ast_rtp_instance *instance, struct ast_rtp *rtp, int rtcp);
#endif
static int __rtp_sendto(struct ast_rtp_instance *instance, void *buf, size_t size, int flags, struct ast_sockaddr *sa, int rtcp, int *ice, int use_srtp);
@@ -1229,6 +1231,9 @@ static int dtls_details_initialize(struct dtls_details *dtls, SSL_CTX *ssl_ctx,
}
dtls->connection = AST_RTP_DTLS_CONNECTION_NEW;
+ ast_mutex_init(&dtls->lock);
+ dtls->timeout_timer = -1;
+
return 0;
error:
@@ -1397,6 +1402,8 @@ static void ast_rtp_dtls_stop(struct ast_rtp_instance *instance)
{
struct ast_rtp *rtp = ast_rtp_instance_get_data(instance);
+ dtls_srtp_stop_timeout_timer(instance, rtp, 0);
+
if (rtp->ssl_ctx) {
SSL_CTX_free(rtp->ssl_ctx);
rtp->ssl_ctx = NULL;
@@ -1405,11 +1412,17 @@ static void ast_rtp_dtls_stop(struct ast_rtp_instance *instance)
if (rtp->dtls.ssl) {
SSL_free(rtp->dtls.ssl);
rtp->dtls.ssl = NULL;
+ ast_mutex_destroy(&rtp->dtls.lock);
}
- if (rtp->rtcp && rtp->rtcp->dtls.ssl) {
- SSL_free(rtp->rtcp->dtls.ssl);
- rtp->rtcp->dtls.ssl = NULL;
+ if (rtp->rtcp) {
+ dtls_srtp_stop_timeout_timer(instance, rtp, 1);
+
+ if (rtp->rtcp->dtls.ssl) {
+ SSL_free(rtp->rtcp->dtls.ssl);
+ rtp->rtcp->dtls.ssl = NULL;
+ ast_mutex_destroy(&rtp->rtcp->dtls.lock);
+ }
}
}
@@ -1586,21 +1599,25 @@ static void dtls_perform_handshake(struct ast_rtp_instance *instance, struct dtl
{
struct ast_rtp *rtp = ast_rtp_instance_get_data(instance);
- if (!dtls->ssl) {
+ /* If we are not acting as a client connecting to the remote side then
+ * don't start the handshake as it will accomplish nothing and would conflict
+ * with the handshake we receive from the remote side.
+ */
+ if (!dtls->ssl || (dtls->dtls_setup != AST_RTP_DTLS_SETUP_ACTIVE)) {
return;
}
- if (SSL_is_init_finished(dtls->ssl)) {
- SSL_clear(dtls->ssl);
- if (dtls->dtls_setup == AST_RTP_DTLS_SETUP_PASSIVE) {
- SSL_set_accept_state(dtls->ssl);
- } else {
- SSL_set_connect_state(dtls->ssl);
- }
- dtls->connection = AST_RTP_DTLS_CONNECTION_NEW;
- }
SSL_do_handshake(dtls->ssl);
+
+ /* Since the handshake is started in a thread outside of the channel thread it's possible
+ * for the response to be handled in the channel thread before we start the timeout timer.
+ * To ensure this doesn't actually happen we hold the DTLS lock. The channel thread will
+ * block until we're done at which point the timeout timer will be immediately stopped.
+ */
+ ast_mutex_lock(&dtls->lock);
dtls_srtp_check_pending(instance, rtp, rtcp);
+ dtls_srtp_start_timeout_timer(instance, rtp, rtcp);
+ ast_mutex_unlock(&dtls->lock);
}
#endif
@@ -1754,48 +1771,83 @@ static inline int rtcp_debug_test_addr(struct ast_sockaddr *addr)
}
#ifdef HAVE_OPENSSL_SRTP
-
-static int dtls_srtp_handle_timeout(const void *data)
+static int dtls_srtp_handle_timeout(struct ast_rtp_instance *instance, int rtcp)
{
- struct ast_rtp_instance *instance = (struct ast_rtp_instance *)data;
struct ast_rtp *rtp = ast_rtp_instance_get_data(instance);
+ struct dtls_details *dtls = !rtcp ? &rtp->dtls : &rtp->rtcp->dtls;
+ struct timeval dtls_timeout;
+
+ DTLSv1_handle_timeout(dtls->ssl);
+ dtls_srtp_check_pending(instance, rtp, rtcp);
- if (!rtp)
- {
+ /* If a timeout can't be retrieved then this recurring scheduled item must stop */
+ if (!DTLSv1_get_timeout(dtls->ssl, &dtls_timeout)) {
+ dtls->timeout_timer = -1;
return 0;
}
- ast_mutex_lock(&rtp->dtls_timer_lock);
- if (rtp->dtlstimerid == -1)
- {
- ast_mutex_unlock(&rtp->dtls_timer_lock);
+ return dtls_timeout.tv_sec * 1000 + dtls_timeout.tv_usec / 1000;
+}
+
+static int dtls_srtp_handle_rtp_timeout(const void *data)
+{
+ struct ast_rtp_instance *instance = (struct ast_rtp_instance *)data;
+ int reschedule;
+
+ reschedule = dtls_srtp_handle_timeout(instance, 0);
+
+ if (!reschedule) {
ao2_ref(instance, -1);
- return 0;
}
- rtp->dtlstimerid = -1;
- ast_mutex_unlock(&rtp->dtls_timer_lock);
+ return reschedule;
+}
+
+static int dtls_srtp_handle_rtcp_timeout(const void *data)
+{
+ struct ast_rtp_instance *instance = (struct ast_rtp_instance *)data;
+ int reschedule;
+
+ reschedule = dtls_srtp_handle_timeout(instance, 1);
- if (rtp->dtls.ssl && !SSL_is_init_finished(rtp->dtls.ssl)) {
- DTLSv1_handle_timeout(rtp->dtls.ssl);
+ if (!reschedule) {
+ ao2_ref(instance, -1);
}
- dtls_srtp_check_pending(instance, rtp, 0);
- if (rtp->rtcp && rtp->rtcp->dtls.ssl && !SSL_is_init_finished(rtp->rtcp->dtls.ssl)) {
- DTLSv1_handle_timeout(rtp->rtcp->dtls.ssl);
+ return reschedule;
+}
+
+static void dtls_srtp_start_timeout_timer(struct ast_rtp_instance *instance, struct ast_rtp *rtp, int rtcp)
+{
+ struct dtls_details *dtls = !rtcp ? &rtp->dtls : &rtp->rtcp->dtls;
+ struct timeval dtls_timeout;
+
+ if (DTLSv1_get_timeout(dtls->ssl, &dtls_timeout)) {
+ int timeout = dtls_timeout.tv_sec * 1000 + dtls_timeout.tv_usec / 1000;
+
+ ast_assert(dtls->timeout_timer == -1);
+
+ ao2_ref(instance, +1);
+ if ((dtls->timeout_timer = ast_sched_add(rtp->sched, timeout,
+ !rtcp ? dtls_srtp_handle_rtp_timeout : dtls_srtp_handle_rtcp_timeout, instance)) < 0) {
+ ao2_ref(instance, -1);
+ ast_log(LOG_WARNING, "Scheduling '%s' DTLS retransmission for RTP instance [%p] failed.\n",
+ !rtcp ? "RTP" : "RTCP", instance);
+ }
}
- dtls_srtp_check_pending(instance, rtp, 1);
+}
- ao2_ref(instance, -1);
+static void dtls_srtp_stop_timeout_timer(struct ast_rtp_instance *instance, struct ast_rtp *rtp, int rtcp)
+{
+ struct dtls_details *dtls = !rtcp ? &rtp->dtls : &rtp->rtcp->dtls;
- return 0;
+ AST_SCHED_DEL_UNREF(rtp->sched, dtls->timeout_timer, ao2_ref(instance, -1));
}
static void dtls_srtp_check_pending(struct ast_rtp_instance *instance, struct ast_rtp *rtp, int rtcp)
{
struct dtls_details *dtls = !rtcp ? &rtp->dtls : &rtp->rtcp->dtls;
size_t pending;
- struct timeval dtls_timeout; /* timeout on DTLS */
if (!dtls->ssl || !dtls->write_bio) {
return;
@@ -1821,24 +1873,6 @@ static void dtls_srtp_check_pending(struct ast_rtp_instance *instance, struct as
}
out = BIO_read(dtls->write_bio, outgoing, sizeof(outgoing));
-
- /* Stop existing DTLS timer if running */
- ast_mutex_lock(&rtp->dtls_timer_lock);
- if (rtp->dtlstimerid > -1) {
- AST_SCHED_DEL_UNREF(rtp->sched, rtp->dtlstimerid, ao2_ref(instance, -1));
- rtp->dtlstimerid = -1;
- }
-
- if (DTLSv1_get_timeout(dtls->ssl, &dtls_timeout)) {
- int timeout = dtls_timeout.tv_sec * 1000 + dtls_timeout.tv_usec / 1000;
- ao2_ref(instance, +1);
- if ((rtp->dtlstimerid = ast_sched_add(rtp->sched, timeout, dtls_srtp_handle_timeout, instance)) < 0) {
- ao2_ref(instance, -1);
- ast_log(LOG_WARNING, "scheduling DTLS retransmission for RTP instance [%p] failed.\n", instance);
- }
- }
- ast_mutex_unlock(&rtp->dtls_timer_lock);
-
__rtp_sendto(instance, outgoing, out, 0, &remote_address, rtcp, &ice, 0);
}
}
@@ -2014,8 +2048,6 @@ static int __rtp_recvfrom(struct ast_rtp_instance *instance, void *buf, size_t s
}
#ifdef HAVE_OPENSSL_SRTP
- dtls_srtp_check_pending(instance, rtp, rtcp);
-
/* If this is an SSL packet pass it to OpenSSL for processing. RFC section for first byte value:
* https://tools.ietf.org/html/rfc5764#section-5.1.2 */
if ((*in >= 20) && (*in <= 63)) {
@@ -2029,6 +2061,15 @@ static int __rtp_recvfrom(struct ast_rtp_instance *instance, void *buf, size_t s
return -1;
}
+ /* This mutex is locked so that this thread blocks until the dtls_perform_handshake function
+ * completes.
+ */
+ ast_mutex_lock(&dtls->lock);
+ ast_mutex_unlock(&dtls->lock);
+
+ /* Before we feed data into OpenSSL ensure that the timeout timer is either stopped or completed */
+ dtls_srtp_stop_timeout_timer(instance, rtp, rtcp);
+
/* If we don't yet know if we are active or passive and we receive a packet... we are obviously passive */
if (dtls->dtls_setup == AST_RTP_DTLS_SETUP_ACTPASS) {
dtls->dtls_setup = AST_RTP_DTLS_SETUP_PASSIVE;
@@ -2057,6 +2098,9 @@ static int __rtp_recvfrom(struct ast_rtp_instance *instance, void *buf, size_t s
/* Use the keying material to set up key/salt information */
res = dtls_srtp_setup(rtp, srtp, instance);
}
+ } else {
+ /* Since we've sent additional traffic start the timeout timer for retransmission */
+ dtls_srtp_start_timeout_timer(instance, rtp, rtcp);
}
return res;
@@ -2479,7 +2523,6 @@ static int ast_rtp_new(struct ast_rtp_instance *instance,
#ifdef HAVE_OPENSSL_SRTP
rtp->rekeyid = -1;
- rtp->dtlstimerid = -1;
#endif
rtp->f.subclass.format = ao2_bump(ast_format_none);
@@ -2497,6 +2540,10 @@ static int ast_rtp_destroy(struct ast_rtp_instance *instance)
struct timespec ts = { .tv_sec = wait.tv_sec, .tv_nsec = wait.tv_usec * 1000, };
#endif
+#ifdef HAVE_OPENSSL_SRTP
+ ast_rtp_dtls_stop(instance);
+#endif
+
/* Destroy the smoother that was smoothing out audio if present */
if (rtp->smoother) {
ast_smoother_free(rtp->smoother);
@@ -2515,11 +2562,6 @@ static int ast_rtp_destroy(struct ast_rtp_instance *instance)
* RTP instance while it's active.
*/
close(rtp->rtcp->s);
-#ifdef HAVE_OPENSSL_SRTP
- if (rtp->rtcp->dtls.ssl) {
- SSL_free(rtp->rtcp->dtls.ssl);
- }
-#endif
ast_free(rtp->rtcp);
}
@@ -2571,18 +2613,6 @@ static int ast_rtp_destroy(struct ast_rtp_instance *instance)
}
#endif
-#ifdef HAVE_OPENSSL_SRTP
- /* Destroy the SSL context if present */
- if (rtp->ssl_ctx) {
- SSL_CTX_free(rtp->ssl_ctx);
- }
-
- /* Destroy the SSL session if present */
- if (rtp->dtls.ssl) {
- SSL_free(rtp->dtls.ssl);
- }
-#endif
-
ao2_cleanup(rtp->lasttxformat);
ao2_cleanup(rtp->lastrxformat);
ao2_cleanup(rtp->f.subclass.format);
@@ -4909,9 +4939,11 @@ static void ast_rtp_stop(struct ast_rtp_instance *instance)
#ifdef HAVE_OPENSSL_SRTP
AST_SCHED_DEL_UNREF(rtp->sched, rtp->rekeyid, ao2_ref(instance, -1));
- ast_mutex_lock(&rtp->dtls_timer_lock);
- AST_SCHED_DEL_UNREF(rtp->sched, rtp->dtlstimerid, ao2_ref(instance, -1));
- ast_mutex_unlock(&rtp->dtls_timer_lock);
+
+ dtls_srtp_stop_timeout_timer(instance, rtp, 0);
+ if (rtp->rtcp) {
+ dtls_srtp_stop_timeout_timer(instance, rtp, 1);
+ }
#endif
if (rtp->rtcp && rtp->rtcp->schedid > 0) {
@@ -4993,10 +5025,31 @@ static int ast_rtp_sendcng(struct ast_rtp_instance *instance, int level)
}
#ifdef HAVE_OPENSSL_SRTP
+static void dtls_perform_setup(struct dtls_details *dtls)
+{
+ if (!dtls->ssl || !SSL_is_init_finished(dtls->ssl)) {
+ return;
+ }
+
+ SSL_clear(dtls->ssl);
+ if (dtls->dtls_setup == AST_RTP_DTLS_SETUP_PASSIVE) {
+ SSL_set_accept_state(dtls->ssl);
+ } else {
+ SSL_set_connect_state(dtls->ssl);
+ }
+ dtls->connection = AST_RTP_DTLS_CONNECTION_NEW;
+}
+
static int ast_rtp_activate(struct ast_rtp_instance *instance)
{
struct ast_rtp *rtp = ast_rtp_instance_get_data(instance);
+ dtls_perform_setup(&rtp->dtls);
+
+ if (rtp->rtcp) {
+ dtls_perform_setup(&rtp->rtcp->dtls);
+ }
+
/* If ICE negotiation is enabled the DTLS Handshake will be performed upon completion of it */
#ifdef HAVE_PJPROJECT
if (rtp->ice) {
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-01 22:25:18 +0000