pjsip_transport_events.c: Fix crash using stale transport pointer.
[asterisk/asterisk.git] / res / res_pjsip / pjsip_distributor.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2013, Digium, Inc.
5  *
6  * Mark Michelson <mmichelson@digium.com>
7  *
8  * See http://www.asterisk.org for more information about
9  * the Asterisk project. Please do not directly contact
10  * any of the maintainers of this project for assistance;
11  * the project provides a web site, mailing lists and IRC
12  * channels for your use.
13  *
14  * This program is free software, distributed under the terms of
15  * the GNU General Public License Version 2. See the LICENSE file
16  * at the top of the source tree.
17  */
18
19 #include "asterisk.h"
20
21 #include <pjsip.h>
22
23 #include "asterisk/res_pjsip.h"
24 #include "asterisk/acl.h"
25 #include "include/res_pjsip_private.h"
26 #include "asterisk/taskprocessor.h"
27 #include "asterisk/threadpool.h"
28 #include "asterisk/res_pjsip_cli.h"
29
30 static int distribute(void *data);
31 static pj_bool_t distributor(pjsip_rx_data *rdata);
32 static pj_status_t record_serializer(pjsip_tx_data *tdata);
33
34 static pjsip_module distributor_mod = {
35         .name = {"Request Distributor", 19},
36         .priority = PJSIP_MOD_PRIORITY_TSX_LAYER - 6,
37         .on_tx_request = record_serializer,
38         .on_rx_request = distributor,
39         .on_rx_response = distributor,
40 };
41
42 struct ast_sched_context *prune_context;
43
44 /* From the auth/realm realtime column size */
45 #define MAX_REALM_LENGTH 40
46
47 #define DEFAULT_SUSPECTS_BUCKETS 53
48
49 static struct ao2_container *unidentified_requests;
50 static unsigned int unidentified_count;
51 static unsigned int unidentified_period;
52 static unsigned int unidentified_prune_interval;
53 static int using_auth_username;
54
55 struct unidentified_request{
56         struct timeval first_seen;
57         int count;
58         char src_name[];
59 };
60
61 /*! Number of serializers in pool if one not otherwise known.  (Best if prime number) */
62 #define DISTRIBUTOR_POOL_SIZE           31
63
64 /*! Pool of serializers to use if not supplied. */
65 static struct ast_taskprocessor *distributor_pool[DISTRIBUTOR_POOL_SIZE];
66
67 /*!
68  * \internal
69  * \brief Record the task's serializer name on the tdata structure.
70  * \since 14.0.0
71  *
72  * \param tdata The outgoing message.
73  *
74  * \retval PJ_SUCCESS.
75  */
76 static pj_status_t record_serializer(pjsip_tx_data *tdata)
77 {
78         struct ast_taskprocessor *serializer;
79
80         serializer = ast_threadpool_serializer_get_current();
81         if (serializer) {
82                 const char *name;
83
84                 name = ast_taskprocessor_name(serializer);
85                 if (!ast_strlen_zero(name)
86                         && (!tdata->mod_data[distributor_mod.id]
87                                 || strcmp(tdata->mod_data[distributor_mod.id], name))) {
88                         char *tdata_name;
89
90                         /* The serializer in use changed. */
91                         tdata_name = pj_pool_alloc(tdata->pool, strlen(name) + 1);
92                         strcpy(tdata_name, name);/* Safe */
93
94                         tdata->mod_data[distributor_mod.id] = tdata_name;
95                 }
96         }
97
98         return PJ_SUCCESS;
99 }
100
101 /*!
102  * \internal
103  * \brief Find the request tdata to get the serializer it used.
104  * \since 14.0.0
105  *
106  * \param rdata The incoming message.
107  *
108  * \retval serializer on success.
109  * \retval NULL on error or could not find the serializer.
110  */
111 static struct ast_taskprocessor *find_request_serializer(pjsip_rx_data *rdata)
112 {
113         struct ast_taskprocessor *serializer = NULL;
114         pj_str_t tsx_key;
115         pjsip_transaction *tsx;
116
117         pjsip_tsx_create_key(rdata->tp_info.pool, &tsx_key, PJSIP_ROLE_UAC,
118                 &rdata->msg_info.cseq->method, rdata);
119
120         tsx = pjsip_tsx_layer_find_tsx(&tsx_key, PJ_TRUE);
121         if (!tsx) {
122                 ast_debug(1, "Could not find transaction for %s.\n",
123                         pjsip_rx_data_get_info(rdata));
124                 return NULL;
125         }
126         ast_debug(3, "Found transaction %s for %s.\n",
127                 tsx->obj_name, pjsip_rx_data_get_info(rdata));
128
129         if (tsx->last_tx) {
130                 const char *serializer_name;
131
132                 serializer_name = tsx->last_tx->mod_data[distributor_mod.id];
133                 if (!ast_strlen_zero(serializer_name)) {
134                         serializer = ast_taskprocessor_get(serializer_name, TPS_REF_IF_EXISTS);
135                         if (serializer) {
136                                 ast_debug(3, "Found serializer %s on transaction %s\n",
137                                                 serializer_name, tsx->obj_name);
138                         }
139                 }
140         }
141
142 #ifdef HAVE_PJ_TRANSACTION_GRP_LOCK
143         pj_grp_lock_release(tsx->grp_lock);
144 #else
145         pj_mutex_unlock(tsx->mutex);
146 #endif
147
148         return serializer;
149 }
150
151 /*! Dialog-specific information the distributor uses */
152 struct distributor_dialog_data {
153         /*! dialog_associations ao2 container key */
154         pjsip_dialog *dlg;
155         /*! Serializer to distribute tasks to for this dialog */
156         struct ast_taskprocessor *serializer;
157         /*! Endpoint associated with this dialog */
158         struct ast_sip_endpoint *endpoint;
159 };
160
161 #define DIALOG_ASSOCIATIONS_BUCKETS 251
162
163 static struct ao2_container *dialog_associations;
164
165 /*!
166  * \internal
167  * \brief Compute a hash value on an arbitrary buffer.
168  * \since 13.17.0
169  *
170  * \param[in] pos The buffer to add to the hash
171  * \param[in] len The buffer length to add to the hash
172  * \param[in] hash The hash value to add to
173  *
174  * \details
175  * This version of the function is for when you need to compute a
176  * hash of more than one buffer.
177  *
178  * This famous hash algorithm was written by Dan Bernstein and is
179  * commonly used.
180  *
181  * \sa http://www.cse.yorku.ca/~oz/hash.html
182  */
183 static int buf_hash_add(const char *pos, size_t len, int hash)
184 {
185         while (len--) {
186                 hash = hash * 33 ^ *pos++;
187         }
188
189         return hash;
190 }
191
192 /*!
193  * \internal
194  * \brief Compute a hash value on an arbitrary buffer.
195  * \since 13.17.0
196  *
197  * \param[in] pos The buffer to add to the hash
198  * \param[in] len The buffer length to add to the hash
199  *
200  * \details
201  * This version of the function is for when you need to compute a
202  * hash of more than one buffer.
203  *
204  * This famous hash algorithm was written by Dan Bernstein and is
205  * commonly used.
206  *
207  * \sa http://www.cse.yorku.ca/~oz/hash.html
208  */
209 static int buf_hash(const char *pos, size_t len)
210 {
211         return buf_hash_add(pos, len, 5381);
212 }
213
214 static int dialog_associations_hash(const void *obj, int flags)
215 {
216         const struct distributor_dialog_data *object;
217         union {
218                 const pjsip_dialog *dlg;
219                 const char buf[sizeof(pjsip_dialog *)];
220         } key;
221
222         switch (flags & OBJ_SEARCH_MASK) {
223         case OBJ_SEARCH_KEY:
224                 key.dlg = obj;
225                 break;
226         case OBJ_SEARCH_OBJECT:
227                 object = obj;
228                 key.dlg = object->dlg;
229                 break;
230         default:
231                 /* Hash can only work on something with a full key. */
232                 ast_assert(0);
233                 return 0;
234         }
235         return ast_str_hash_restrict(buf_hash(key.buf, sizeof(key.buf)));
236 }
237
238 static int dialog_associations_cmp(void *obj, void *arg, int flags)
239 {
240         const struct distributor_dialog_data *object_left = obj;
241         const struct distributor_dialog_data *object_right = arg;
242         const pjsip_dialog *right_key = arg;
243         int cmp = 0;
244
245         switch (flags & OBJ_SEARCH_MASK) {
246         case OBJ_SEARCH_OBJECT:
247                 right_key = object_right->dlg;
248                 /* Fall through */
249         case OBJ_SEARCH_KEY:
250                 if (object_left->dlg == right_key) {
251                         cmp = CMP_MATCH;
252                 }
253                 break;
254         case OBJ_SEARCH_PARTIAL_KEY:
255                 /* There is no such thing for this container. */
256                 ast_assert(0);
257                 break;
258         default:
259                 cmp = 0;
260                 break;
261         }
262         return cmp;
263 }
264
265 void ast_sip_dialog_set_serializer(pjsip_dialog *dlg, struct ast_taskprocessor *serializer)
266 {
267         struct distributor_dialog_data *dist;
268
269         ao2_wrlock(dialog_associations);
270         dist = ao2_find(dialog_associations, dlg, OBJ_SEARCH_KEY | OBJ_NOLOCK);
271         if (!dist) {
272                 if (serializer) {
273                         dist = ao2_alloc(sizeof(*dist), NULL);
274                         if (dist) {
275                                 dist->dlg = dlg;
276                                 dist->serializer = serializer;
277                                 ao2_link_flags(dialog_associations, dist, OBJ_NOLOCK);
278                                 ao2_ref(dist, -1);
279                         }
280                 }
281         } else {
282                 ao2_lock(dist);
283                 dist->serializer = serializer;
284                 if (!dist->serializer && !dist->endpoint) {
285                         ao2_unlink_flags(dialog_associations, dist, OBJ_NOLOCK);
286                 }
287                 ao2_unlock(dist);
288                 ao2_ref(dist, -1);
289         }
290         ao2_unlock(dialog_associations);
291 }
292
293 void ast_sip_dialog_set_endpoint(pjsip_dialog *dlg, struct ast_sip_endpoint *endpoint)
294 {
295         struct distributor_dialog_data *dist;
296
297         ao2_wrlock(dialog_associations);
298         dist = ao2_find(dialog_associations, dlg, OBJ_SEARCH_KEY | OBJ_NOLOCK);
299         if (!dist) {
300                 if (endpoint) {
301                         dist = ao2_alloc(sizeof(*dist), NULL);
302                         if (dist) {
303                                 dist->dlg = dlg;
304                                 dist->endpoint = endpoint;
305                                 ao2_link_flags(dialog_associations, dist, OBJ_NOLOCK);
306                                 ao2_ref(dist, -1);
307                         }
308                 }
309         } else {
310                 ao2_lock(dist);
311                 dist->endpoint = endpoint;
312                 if (!dist->serializer && !dist->endpoint) {
313                         ao2_unlink_flags(dialog_associations, dist, OBJ_NOLOCK);
314                 }
315                 ao2_unlock(dist);
316                 ao2_ref(dist, -1);
317         }
318         ao2_unlock(dialog_associations);
319 }
320
321 struct ast_sip_endpoint *ast_sip_dialog_get_endpoint(pjsip_dialog *dlg)
322 {
323         struct distributor_dialog_data *dist;
324         struct ast_sip_endpoint *endpoint;
325
326         dist = ao2_find(dialog_associations, dlg, OBJ_SEARCH_KEY);
327         if (dist) {
328                 ao2_lock(dist);
329                 endpoint = ao2_bump(dist->endpoint);
330                 ao2_unlock(dist);
331                 ao2_ref(dist, -1);
332         } else {
333                 endpoint = NULL;
334         }
335         return endpoint;
336 }
337
338 static pjsip_dialog *find_dialog(pjsip_rx_data *rdata)
339 {
340         pj_str_t tsx_key;
341         pjsip_transaction *tsx;
342         pjsip_dialog *dlg;
343         pj_str_t *local_tag;
344         pj_str_t *remote_tag;
345
346         if (!rdata->msg_info.msg) {
347                 return NULL;
348         }
349
350         if (rdata->msg_info.msg->type == PJSIP_REQUEST_MSG) {
351                 local_tag = &rdata->msg_info.to->tag;
352                 remote_tag = &rdata->msg_info.from->tag;
353         } else {
354                 local_tag = &rdata->msg_info.from->tag;
355                 remote_tag = &rdata->msg_info.to->tag;
356         }
357
358         /* We can only call the convenient method for
359          *  1) responses
360          *  2) non-CANCEL requests
361          *  3) CANCEL requests with a to-tag
362          */
363         if (rdata->msg_info.msg->type == PJSIP_RESPONSE_MSG ||
364                         pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_cancel_method) ||
365                         rdata->msg_info.to->tag.slen != 0) {
366                 dlg = pjsip_ua_find_dialog(&rdata->msg_info.cid->id, local_tag,
367                                 remote_tag, PJ_FALSE);
368                 if (dlg) {
369                         return dlg;
370                 }
371         }
372
373         /*
374          * There may still be a matching dialog if this is
375          * 1) an incoming CANCEL request without a to-tag
376          * 2) an incoming response to a dialog-creating request.
377          */
378         if (rdata->msg_info.msg->type == PJSIP_REQUEST_MSG) {
379                 /* CANCEL requests will need to match the INVITE we initially received. Any
380                  * other request type will either have been matched already or is not in
381                  * dialog
382                  */
383                 pjsip_tsx_create_key(rdata->tp_info.pool, &tsx_key, PJSIP_ROLE_UAS,
384                                 pjsip_get_invite_method(), rdata);
385         } else {
386                 pjsip_tsx_create_key(rdata->tp_info.pool, &tsx_key, PJSIP_ROLE_UAC,
387                                 &rdata->msg_info.cseq->method, rdata);
388         }
389
390         tsx = pjsip_tsx_layer_find_tsx(&tsx_key, PJ_TRUE);
391         if (!tsx) {
392                 ast_debug(3, "Could not find matching transaction for %s\n",
393                         pjsip_rx_data_get_info(rdata));
394                 return NULL;
395         }
396
397         dlg = pjsip_tsx_get_dlg(tsx);
398
399 #ifdef HAVE_PJ_TRANSACTION_GRP_LOCK
400         pj_grp_lock_release(tsx->grp_lock);
401 #else
402         pj_mutex_unlock(tsx->mutex);
403 #endif
404
405         return dlg;
406 }
407
408 /*!
409  * \internal
410  * \brief Compute a hash value on a pjlib string
411  * \since 13.10.0
412  *
413  * \param[in] str The pjlib string to add to the hash
414  * \param[in] hash The hash value to add to
415  *
416  * \details
417  * This version of the function is for when you need to compute a
418  * string hash of more than one string.
419  *
420  * This famous hash algorithm was written by Dan Bernstein and is
421  * commonly used.
422  *
423  * \sa http://www.cse.yorku.ca/~oz/hash.html
424  */
425 static int pjstr_hash_add(pj_str_t *str, int hash)
426 {
427         return buf_hash_add(pj_strbuf(str), pj_strlen(str), hash);
428 }
429
430 /*!
431  * \internal
432  * \brief Compute a hash value on a pjlib string
433  * \since 13.10.0
434  *
435  * \param[in] str The pjlib string to hash
436  *
437  * This famous hash algorithm was written by Dan Bernstein and is
438  * commonly used.
439  *
440  * http://www.cse.yorku.ca/~oz/hash.html
441  */
442 static int pjstr_hash(pj_str_t *str)
443 {
444         return pjstr_hash_add(str, 5381);
445 }
446
447 struct ast_taskprocessor *ast_sip_get_distributor_serializer(pjsip_rx_data *rdata)
448 {
449         int hash;
450         pj_str_t *remote_tag;
451         struct ast_taskprocessor *serializer;
452
453         if (!rdata->msg_info.msg) {
454                 return NULL;
455         }
456
457         if (rdata->msg_info.msg->type == PJSIP_REQUEST_MSG) {
458                 remote_tag = &rdata->msg_info.from->tag;
459         } else {
460                 remote_tag = &rdata->msg_info.to->tag;
461         }
462
463         /* Compute the hash from the SIP message call-id and remote-tag */
464         hash = pjstr_hash(&rdata->msg_info.cid->id);
465         hash = pjstr_hash_add(remote_tag, hash);
466         hash = ast_str_hash_restrict(hash);
467
468         serializer = ao2_bump(distributor_pool[hash % ARRAY_LEN(distributor_pool)]);
469         if (serializer) {
470                 ast_debug(3, "Calculated serializer %s to use for %s\n",
471                         ast_taskprocessor_name(serializer), pjsip_rx_data_get_info(rdata));
472         }
473         return serializer;
474 }
475
476 static pj_bool_t endpoint_lookup(pjsip_rx_data *rdata);
477
478 static pjsip_module endpoint_mod = {
479         .name = {"Endpoint Identifier", 19},
480         .priority = PJSIP_MOD_PRIORITY_TSX_LAYER - 3,
481         .on_rx_request = endpoint_lookup,
482 };
483
484 static pj_bool_t distributor(pjsip_rx_data *rdata)
485 {
486         pjsip_dialog *dlg;
487         struct distributor_dialog_data *dist = NULL;
488         struct ast_taskprocessor *serializer = NULL;
489         pjsip_rx_data *clone;
490
491         if (!ast_test_flag(&ast_options, AST_OPT_FLAG_FULLY_BOOTED)) {
492                 /*
493                  * Ignore everything until we are fully booted.  Let the
494                  * peer retransmit messages until we are ready.
495                  */
496                 return PJ_TRUE;
497         }
498
499         dlg = find_dialog(rdata);
500         if (dlg) {
501                 ast_debug(3, "Searching for serializer associated with dialog %s for %s\n",
502                         dlg->obj_name, pjsip_rx_data_get_info(rdata));
503                 dist = ao2_find(dialog_associations, dlg, OBJ_SEARCH_KEY);
504                 if (dist) {
505                         ao2_lock(dist);
506                         serializer = ao2_bump(dist->serializer);
507                         ao2_unlock(dist);
508                         if (serializer) {
509                                 ast_debug(3, "Found serializer %s associated with dialog %s\n",
510                                         ast_taskprocessor_name(serializer), dlg->obj_name);
511                         }
512                 }
513         }
514
515         if (serializer) {
516                 /* We have a serializer so we know where to send the message. */
517         } else if (rdata->msg_info.msg->type == PJSIP_RESPONSE_MSG) {
518                 ast_debug(3, "No dialog serializer for %s.  Using request transaction as basis.\n",
519                         pjsip_rx_data_get_info(rdata));
520                 serializer = find_request_serializer(rdata);
521                 if (!serializer) {
522                         /*
523                          * Pick a serializer for the unmatched response.
524                          * We couldn't determine what serializer originally
525                          * sent the request or the serializer is gone.
526                          */
527                         serializer = ast_sip_get_distributor_serializer(rdata);
528                 }
529         } else if (!pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_cancel_method)
530                 || !pjsip_method_cmp(&rdata->msg_info.msg->line.req.method, &pjsip_bye_method)) {
531                 /* We have a BYE or CANCEL request without a serializer. */
532                 pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata,
533                         PJSIP_SC_CALL_TSX_DOES_NOT_EXIST, NULL, NULL, NULL);
534                 ao2_cleanup(dist);
535                 return PJ_TRUE;
536         } else {
537                 if (ast_taskprocessor_alert_get()) {
538                         /*
539                          * When taskprocessors get backed up, there is a good chance that
540                          * we are being overloaded and need to defer adding new work to
541                          * the system.  To defer the work we will ignore the request and
542                          * rely on the peer's transport layer to retransmit the message.
543                          * We usually work off the overload within a few seconds.  The
544                          * alternative is to send back a 503 response to these requests
545                          * and be done with it.
546                          */
547                         ast_debug(3, "Taskprocessor overload alert: Ignoring '%s'.\n",
548                                 pjsip_rx_data_get_info(rdata));
549                         ao2_cleanup(dist);
550                         return PJ_TRUE;
551                 }
552
553                 /* Pick a serializer for the out-of-dialog request. */
554                 serializer = ast_sip_get_distributor_serializer(rdata);
555         }
556
557         if (pjsip_rx_data_clone(rdata, 0, &clone) != PJ_SUCCESS) {
558                 ast_taskprocessor_unreference(serializer);
559                 ao2_cleanup(dist);
560                 return PJ_TRUE;
561         }
562
563         if (dist) {
564                 ao2_lock(dist);
565                 clone->endpt_info.mod_data[endpoint_mod.id] = ao2_bump(dist->endpoint);
566                 ao2_unlock(dist);
567                 ao2_cleanup(dist);
568         }
569
570         if (ast_sip_push_task(serializer, distribute, clone)) {
571                 ao2_cleanup(clone->endpt_info.mod_data[endpoint_mod.id]);
572                 pjsip_rx_data_free_cloned(clone);
573         }
574
575         ast_taskprocessor_unreference(serializer);
576
577         return PJ_TRUE;
578 }
579
580 static struct ast_sip_auth *alloc_artificial_auth(char *default_realm)
581 {
582         struct ast_sip_auth *fake_auth;
583
584         fake_auth = ast_sorcery_alloc(ast_sip_get_sorcery(), SIP_SORCERY_AUTH_TYPE,
585                 "artificial");
586         if (!fake_auth) {
587                 return NULL;
588         }
589
590         ast_string_field_set(fake_auth, realm, default_realm);
591         ast_string_field_set(fake_auth, auth_user, "");
592         ast_string_field_set(fake_auth, auth_pass, "");
593         fake_auth->type = AST_SIP_AUTH_TYPE_ARTIFICIAL;
594
595         return fake_auth;
596 }
597
598 static AO2_GLOBAL_OBJ_STATIC(artificial_auth);
599
600 static int create_artificial_auth(void)
601 {
602         char default_realm[MAX_REALM_LENGTH + 1];
603         struct ast_sip_auth *fake_auth;
604
605         ast_sip_get_default_realm(default_realm, sizeof(default_realm));
606         fake_auth = alloc_artificial_auth(default_realm);
607         if (!fake_auth) {
608                 ast_log(LOG_ERROR, "Unable to create artificial auth\n");
609                 return -1;
610         }
611
612         ao2_global_obj_replace_unref(artificial_auth, fake_auth);
613         ao2_ref(fake_auth, -1);
614         return 0;
615 }
616
617 struct ast_sip_auth *ast_sip_get_artificial_auth(void)
618 {
619         return ao2_global_obj_ref(artificial_auth);
620 }
621
622 static struct ast_sip_endpoint *artificial_endpoint = NULL;
623
624 static int create_artificial_endpoint(void)
625 {
626         artificial_endpoint = ast_sorcery_alloc(ast_sip_get_sorcery(), "endpoint", NULL);
627         if (!artificial_endpoint) {
628                 return -1;
629         }
630
631         AST_VECTOR_INIT(&artificial_endpoint->inbound_auths, 1);
632         /* Pushing a bogus value into the vector will ensure that
633          * the proper size of the vector is returned. This value is
634          * not actually used anywhere
635          */
636         AST_VECTOR_APPEND(&artificial_endpoint->inbound_auths, ast_strdup("artificial-auth"));
637         return 0;
638 }
639
640 struct ast_sip_endpoint *ast_sip_get_artificial_endpoint(void)
641 {
642         ao2_ref(artificial_endpoint, +1);
643         return artificial_endpoint;
644 }
645
646 static void log_failed_request(pjsip_rx_data *rdata, char *msg, unsigned int count, unsigned int period)
647 {
648         char from_buf[PJSIP_MAX_URL_SIZE];
649         char callid_buf[PJSIP_MAX_URL_SIZE];
650         char method_buf[PJSIP_MAX_URL_SIZE];
651         pjsip_uri_print(PJSIP_URI_IN_FROMTO_HDR, rdata->msg_info.from->uri, from_buf, PJSIP_MAX_URL_SIZE);
652         ast_copy_pj_str(callid_buf, &rdata->msg_info.cid->id, PJSIP_MAX_URL_SIZE);
653         ast_copy_pj_str(method_buf, &rdata->msg_info.msg->line.req.method.name, PJSIP_MAX_URL_SIZE);
654         if (count) {
655                 ast_log(LOG_NOTICE, "Request '%s' from '%s' failed for '%s:%d' (callid: %s) - %s"
656                         " after %u tries in %.3f ms\n",
657                         method_buf, from_buf, rdata->pkt_info.src_name, rdata->pkt_info.src_port, callid_buf, msg, count, period / 1000.0);
658         } else {
659                 ast_log(LOG_NOTICE, "Request '%s' from '%s' failed for '%s:%d' (callid: %s) - %s\n",
660                         method_buf, from_buf, rdata->pkt_info.src_name, rdata->pkt_info.src_port, callid_buf, msg);
661         }
662 }
663
664 static void check_endpoint(pjsip_rx_data *rdata, struct unidentified_request *unid,
665         const char *name)
666 {
667         int64_t ms = ast_tvdiff_ms(ast_tvnow(), unid->first_seen);
668
669         ao2_wrlock(unid);
670         unid->count++;
671
672         if (ms < (unidentified_period * 1000) && unid->count >= unidentified_count) {
673                 log_failed_request(rdata, "No matching endpoint found", unid->count, ms);
674                 ast_sip_report_invalid_endpoint(name, rdata);
675         }
676         ao2_unlock(unid);
677 }
678
679 static pj_bool_t endpoint_lookup(pjsip_rx_data *rdata)
680 {
681         struct ast_sip_endpoint *endpoint;
682         struct unidentified_request *unid;
683         int is_ack = rdata->msg_info.msg->line.req.method.id == PJSIP_ACK_METHOD;
684
685         endpoint = rdata->endpt_info.mod_data[endpoint_mod.id];
686         if (endpoint) {
687                 /*
688                  * ao2_find with OBJ_UNLINK always write locks the container before even searching
689                  * for the object.  Since the majority case is that the object won't be found, do
690                  * the find without OBJ_UNLINK to prevent the unnecessary write lock, then unlink
691                  * if needed.
692                  */
693                 if ((unid = ao2_find(unidentified_requests, rdata->pkt_info.src_name, OBJ_SEARCH_KEY))) {
694                         ao2_unlink(unidentified_requests, unid);
695                         ao2_ref(unid, -1);
696                 }
697                 return PJ_FALSE;
698         }
699
700         endpoint = ast_sip_identify_endpoint(rdata);
701         if (endpoint) {
702                 if ((unid = ao2_find(unidentified_requests, rdata->pkt_info.src_name, OBJ_SEARCH_KEY))) {
703                         ao2_unlink(unidentified_requests, unid);
704                         ao2_ref(unid, -1);
705                 }
706         }
707
708         if (!endpoint) {
709
710                 /* always use an artificial endpoint - per discussion no reason
711                    to have "alwaysauthreject" as an option.  It is felt using it
712                    was a bug fix and it is not needed since we are not worried about
713                    breaking old stuff and we really don't want to enable the discovery
714                    of SIP accounts */
715                 endpoint = ast_sip_get_artificial_endpoint();
716         }
717
718         rdata->endpt_info.mod_data[endpoint_mod.id] = endpoint;
719
720         if ((endpoint == artificial_endpoint) && !is_ack) {
721                 char name[AST_UUID_STR_LEN] = "";
722                 pjsip_uri *from = rdata->msg_info.from->uri;
723
724                 if (PJSIP_URI_SCHEME_IS_SIP(from) || PJSIP_URI_SCHEME_IS_SIPS(from)) {
725                         pjsip_sip_uri *sip_from = pjsip_uri_get_uri(from);
726                         ast_copy_pj_str(name, &sip_from->user, sizeof(name));
727                 }
728
729                 if ((unid = ao2_find(unidentified_requests, rdata->pkt_info.src_name, OBJ_SEARCH_KEY))) {
730                         check_endpoint(rdata, unid, name);
731                         ao2_ref(unid, -1);
732                 } else if (using_auth_username) {
733                         ao2_wrlock(unidentified_requests);
734                         /* The check again with the write lock held allows us to eliminate the DUPS_REPLACE and sort_fn */
735                         if ((unid = ao2_find(unidentified_requests, rdata->pkt_info.src_name, OBJ_SEARCH_KEY | OBJ_NOLOCK))) {
736                                 check_endpoint(rdata, unid, name);
737                         } else {
738                                 unid = ao2_alloc_options(sizeof(*unid) + strlen(rdata->pkt_info.src_name) + 1, NULL,
739                                         AO2_ALLOC_OPT_LOCK_RWLOCK);
740                                 if (!unid) {
741                                         ao2_unlock(unidentified_requests);
742                                         return PJ_TRUE;
743                                 }
744                                 strcpy(unid->src_name, rdata->pkt_info.src_name); /* Safe */
745                                 unid->first_seen = ast_tvnow();
746                                 unid->count = 1;
747                                 ao2_link_flags(unidentified_requests, unid, OBJ_NOLOCK);
748                         }
749                         ao2_ref(unid, -1);
750                         ao2_unlock(unidentified_requests);
751                 } else {
752                         log_failed_request(rdata, "No matching endpoint found", 0, 0);
753                         ast_sip_report_invalid_endpoint(name, rdata);
754                 }
755         }
756         return PJ_FALSE;
757 }
758
759 static int apply_endpoint_acl(pjsip_rx_data *rdata, struct ast_sip_endpoint *endpoint)
760 {
761         struct ast_sockaddr addr;
762
763         if (ast_acl_list_is_empty(endpoint->acl)) {
764                 return 0;
765         }
766
767         memset(&addr, 0, sizeof(addr));
768         ast_sockaddr_parse(&addr, rdata->pkt_info.src_name, PARSE_PORT_FORBID);
769         ast_sockaddr_set_port(&addr, rdata->pkt_info.src_port);
770
771         if (ast_apply_acl(endpoint->acl, &addr, "SIP ACL: ") != AST_SENSE_ALLOW) {
772                 log_failed_request(rdata, "Not match Endpoint ACL", 0, 0);
773                 ast_sip_report_failed_acl(endpoint, rdata, "not_match_endpoint_acl");
774                 return 1;
775         }
776         return 0;
777 }
778
779 static int extract_contact_addr(pjsip_contact_hdr *contact, struct ast_sockaddr **addrs)
780 {
781         pjsip_sip_uri *sip_uri;
782         char host[256];
783
784         if (!contact || contact->star) {
785                 *addrs = NULL;
786                 return 0;
787         }
788         if (!PJSIP_URI_SCHEME_IS_SIP(contact->uri) && !PJSIP_URI_SCHEME_IS_SIPS(contact->uri)) {
789                 *addrs = NULL;
790                 return 0;
791         }
792         sip_uri = pjsip_uri_get_uri(contact->uri);
793         ast_copy_pj_str(host, &sip_uri->host, sizeof(host));
794         return ast_sockaddr_resolve(addrs, host, PARSE_PORT_FORBID, AST_AF_UNSPEC);
795 }
796
797 static int apply_endpoint_contact_acl(pjsip_rx_data *rdata, struct ast_sip_endpoint *endpoint)
798 {
799         int num_contact_addrs;
800         int forbidden = 0;
801         struct ast_sockaddr *contact_addrs;
802         int i;
803         pjsip_contact_hdr *contact = (pjsip_contact_hdr *)&rdata->msg_info.msg->hdr;
804
805         if (ast_acl_list_is_empty(endpoint->contact_acl)) {
806                 return 0;
807         }
808
809         while ((contact = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_CONTACT, contact->next))) {
810                 num_contact_addrs = extract_contact_addr(contact, &contact_addrs);
811                 if (num_contact_addrs <= 0) {
812                         continue;
813                 }
814                 for (i = 0; i < num_contact_addrs; ++i) {
815                         if (ast_apply_acl(endpoint->contact_acl, &contact_addrs[i], "SIP Contact ACL: ") != AST_SENSE_ALLOW) {
816                                 log_failed_request(rdata, "Not match Endpoint Contact ACL", 0, 0);
817                                 ast_sip_report_failed_acl(endpoint, rdata, "not_match_endpoint_contact_acl");
818                                 forbidden = 1;
819                                 break;
820                         }
821                 }
822                 ast_free(contact_addrs);
823                 if (forbidden) {
824                         /* No use checking other contacts if we already have failed ACL check */
825                         break;
826                 }
827         }
828
829         return forbidden;
830 }
831
832 static pj_bool_t authenticate(pjsip_rx_data *rdata)
833 {
834         RAII_VAR(struct ast_sip_endpoint *, endpoint, ast_pjsip_rdata_get_endpoint(rdata), ao2_cleanup);
835         int is_ack = rdata->msg_info.msg->line.req.method.id == PJSIP_ACK_METHOD;
836
837         ast_assert(endpoint != NULL);
838
839         if (endpoint!=artificial_endpoint) {
840                 if (apply_endpoint_acl(rdata, endpoint) || apply_endpoint_contact_acl(rdata, endpoint)) {
841                         if (!is_ack) {
842                                 pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 403, NULL, NULL, NULL);
843                         }
844                         return PJ_TRUE;
845                 }
846         }
847
848         if (!is_ack && ast_sip_requires_authentication(endpoint, rdata)) {
849                 pjsip_tx_data *tdata;
850                 struct unidentified_request *unid;
851
852                 pjsip_endpt_create_response(ast_sip_get_pjsip_endpoint(), rdata, 401, NULL, &tdata);
853                 switch (ast_sip_check_authentication(endpoint, rdata, tdata)) {
854                 case AST_SIP_AUTHENTICATION_CHALLENGE:
855                         /* Send the 401 we created for them */
856                         ast_sip_report_auth_challenge_sent(endpoint, rdata, tdata);
857                         if (pjsip_endpt_send_response2(ast_sip_get_pjsip_endpoint(), rdata, tdata, NULL, NULL) != PJ_SUCCESS) {
858                                 pjsip_tx_data_dec_ref(tdata);
859                         }
860                         return PJ_TRUE;
861                 case AST_SIP_AUTHENTICATION_SUCCESS:
862                         /* See note in endpoint_lookup about not holding an unnecessary write lock */
863                         if ((unid = ao2_find(unidentified_requests, rdata->pkt_info.src_name, OBJ_SEARCH_KEY))) {
864                                 ao2_unlink(unidentified_requests, unid);
865                                 ao2_ref(unid, -1);
866                         }
867                         ast_sip_report_auth_success(endpoint, rdata);
868                         break;
869                 case AST_SIP_AUTHENTICATION_FAILED:
870                         log_failed_request(rdata, "Failed to authenticate", 0, 0);
871                         ast_sip_report_auth_failed_challenge_response(endpoint, rdata);
872                         if (pjsip_endpt_send_response2(ast_sip_get_pjsip_endpoint(), rdata, tdata, NULL, NULL) != PJ_SUCCESS) {
873                                 pjsip_tx_data_dec_ref(tdata);
874                         }
875                         return PJ_TRUE;
876                 case AST_SIP_AUTHENTICATION_ERROR:
877                         log_failed_request(rdata, "Error to authenticate", 0, 0);
878                         ast_sip_report_auth_failed_challenge_response(endpoint, rdata);
879                         pjsip_tx_data_dec_ref(tdata);
880                         pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 500, NULL, NULL, NULL);
881                         return PJ_TRUE;
882                 }
883                 pjsip_tx_data_dec_ref(tdata);
884         }
885
886         return PJ_FALSE;
887 }
888
889 static pjsip_module auth_mod = {
890         .name = {"Request Authenticator", 21},
891         .priority = PJSIP_MOD_PRIORITY_APPLICATION - 2,
892         .on_rx_request = authenticate,
893 };
894
895 static int distribute(void *data)
896 {
897         static pjsip_process_rdata_param param = {
898                 .start_mod = &distributor_mod,
899                 .idx_after_start = 1,
900         };
901         pj_bool_t handled = PJ_FALSE;
902         pjsip_rx_data *rdata = data;
903         int is_request = rdata->msg_info.msg->type == PJSIP_REQUEST_MSG;
904         int is_ack = is_request ? rdata->msg_info.msg->line.req.method.id == PJSIP_ACK_METHOD : 0;
905         struct ast_sip_endpoint *endpoint;
906
907         pjsip_endpt_process_rx_data(ast_sip_get_pjsip_endpoint(), rdata, &param, &handled);
908         if (!handled && is_request && !is_ack) {
909                 pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 501, NULL, NULL, NULL);
910         }
911
912         /* The endpoint_mod stores an endpoint reference in the mod_data of rdata. This
913          * is the only appropriate spot to actually decrement the reference.
914          */
915         endpoint = rdata->endpt_info.mod_data[endpoint_mod.id];
916         ao2_cleanup(endpoint);
917         pjsip_rx_data_free_cloned(rdata);
918         return 0;
919 }
920
921 struct ast_sip_endpoint *ast_pjsip_rdata_get_endpoint(pjsip_rx_data *rdata)
922 {
923         struct ast_sip_endpoint *endpoint = rdata->endpt_info.mod_data[endpoint_mod.id];
924         if (endpoint) {
925                 ao2_ref(endpoint, +1);
926         }
927         return endpoint;
928 }
929
930 static int suspects_sort(const void *obj, const void *arg, int flags)
931 {
932         const struct unidentified_request *object_left = obj;
933         const struct unidentified_request *object_right = arg;
934         const char *right_key = arg;
935         int cmp;
936
937         switch (flags & OBJ_SEARCH_MASK) {
938         case OBJ_SEARCH_OBJECT:
939                 right_key = object_right->src_name;
940                 /* Fall through */
941         case OBJ_SEARCH_KEY:
942                 cmp = strcmp(object_left->src_name, right_key);
943                 break;
944         case OBJ_SEARCH_PARTIAL_KEY:
945                 cmp = strncmp(object_left->src_name, right_key, strlen(right_key));
946                 break;
947         default:
948                 cmp = 0;
949                 break;
950         }
951         return cmp;
952 }
953
954 static int suspects_compare(void *obj, void *arg, int flags)
955 {
956         const struct unidentified_request *object_left = obj;
957         const struct unidentified_request *object_right = arg;
958         const char *right_key = arg;
959         int cmp = 0;
960
961         switch (flags & OBJ_SEARCH_MASK) {
962         case OBJ_SEARCH_OBJECT:
963                 right_key = object_right->src_name;
964                 /* Fall through */
965         case OBJ_SEARCH_KEY:
966                 if (strcmp(object_left->src_name, right_key) == 0) {
967                         cmp = CMP_MATCH;
968                 }
969                 break;
970         case OBJ_SEARCH_PARTIAL_KEY:
971                 if (strncmp(object_left->src_name, right_key, strlen(right_key)) == 0) {
972                         cmp = CMP_MATCH;
973                 }
974                 break;
975         default:
976                 cmp = 0;
977                 break;
978         }
979         return cmp;
980 }
981
982 static int suspects_hash(const void *obj, int flags)
983 {
984         const struct unidentified_request *object;
985         const char *key;
986
987         switch (flags & OBJ_SEARCH_MASK) {
988         case OBJ_SEARCH_KEY:
989                 key = obj;
990                 break;
991         case OBJ_SEARCH_OBJECT:
992                 object = obj;
993                 key = object->src_name;
994                 break;
995         default:
996                 /* Hash can only work on something with a full key. */
997                 ast_assert(0);
998                 return 0;
999         }
1000         return ast_str_hash(key);
1001 }
1002
1003 static struct ao2_container *cli_unid_get_container(const char *regex)
1004 {
1005         struct ao2_container *s_container;
1006
1007         s_container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0,
1008                 suspects_sort, suspects_compare);
1009         if (!s_container) {
1010                 return NULL;
1011         }
1012
1013         if (ao2_container_dup(s_container, unidentified_requests, 0)) {
1014                 ao2_ref(s_container, -1);
1015                 return NULL;
1016         }
1017
1018         return s_container;
1019 }
1020
1021 static int cli_unid_iterate(void *container, ao2_callback_fn callback, void *args)
1022 {
1023         ao2_callback(container, 0, callback, args);
1024
1025         return 0;
1026 }
1027
1028 static void *cli_unid_retrieve_by_id(const char *id)
1029 {
1030         return ao2_find(unidentified_requests, id, OBJ_SEARCH_KEY);
1031 }
1032
1033 static const char *cli_unid_get_id(const void *obj)
1034 {
1035         const struct unidentified_request *unid = obj;
1036
1037         return unid->src_name;
1038 }
1039
1040 static int cli_unid_print_header(void *obj, void *arg, int flags)
1041 {
1042         struct ast_sip_cli_context *context = arg;
1043         RAII_VAR(struct ast_sip_cli_formatter_entry *, formatter_entry, NULL, ao2_cleanup);
1044
1045         int indent = CLI_INDENT_TO_SPACES(context->indent_level);
1046         int filler = CLI_LAST_TABSTOP - indent - 7;
1047
1048         ast_assert(context->output_buffer != NULL);
1049
1050         ast_str_append(&context->output_buffer, 0,
1051                 "%*s:  <IP Address%*.*s>  <Count> <Age(sec)>\n",
1052                 indent, "Request", filler, filler, CLI_HEADER_FILLER);
1053
1054         return 0;
1055 }
1056
1057 static int cli_unid_print_body(void *obj, void *arg, int flags)
1058 {
1059         struct unidentified_request *unid = obj;
1060         struct ast_sip_cli_context *context = arg;
1061         int indent;
1062         int flexwidth;
1063         int64_t ms = ast_tvdiff_ms(ast_tvnow(), unid->first_seen);
1064
1065         ast_assert(context->output_buffer != NULL);
1066
1067         indent = CLI_INDENT_TO_SPACES(context->indent_level);
1068         flexwidth = CLI_LAST_TABSTOP - 4;
1069
1070         ast_str_append(&context->output_buffer, 0, "%*s:  %-*.*s  %7d %10.3f\n",
1071                 indent,
1072                 "Request",
1073                 flexwidth, flexwidth,
1074                 unid->src_name, unid->count,  ms / 1000.0);
1075
1076         return 0;
1077 }
1078
1079 static struct ast_cli_entry cli_commands[] = {
1080         AST_CLI_DEFINE(ast_sip_cli_traverse_objects, "Show PJSIP Unidentified Requests",
1081                 .command = "pjsip show unidentified_requests",
1082                 .usage = "Usage: pjsip show unidentified_requests\n"
1083                                 "       Show the PJSIP Unidentified Requests\n"),
1084 };
1085
1086 struct ast_sip_cli_formatter_entry *unid_formatter;
1087
1088 static int expire_requests(void *object, void *arg, int flags)
1089 {
1090         struct unidentified_request *unid = object;
1091         int *maxage = arg;
1092         int64_t ms = ast_tvdiff_ms(ast_tvnow(), unid->first_seen);
1093
1094         if (ms > (*maxage) * 2 * 1000) {
1095                 return CMP_MATCH;
1096         }
1097
1098         return 0;
1099 }
1100
1101 static int prune_task(const void *data)
1102 {
1103         unsigned int maxage;
1104
1105         ast_sip_get_unidentified_request_thresholds(&unidentified_count, &unidentified_period, &unidentified_prune_interval);
1106         maxage = unidentified_period * 2;
1107         ao2_callback(unidentified_requests, OBJ_MULTIPLE | OBJ_NODATA | OBJ_UNLINK, expire_requests, &maxage);
1108
1109         return unidentified_prune_interval * 1000;
1110 }
1111
1112 static int clean_task(const void *data)
1113 {
1114         return 0;
1115 }
1116
1117 static void global_loaded(const char *object_type)
1118 {
1119         char default_realm[MAX_REALM_LENGTH + 1];
1120         struct ast_sip_auth *fake_auth;
1121         char *identifier_order;
1122
1123         /* Update using_auth_username */
1124         identifier_order = ast_sip_get_endpoint_identifier_order();
1125         if (identifier_order) {
1126                 char *identify_method;
1127                 char *io_copy = ast_strdupa(identifier_order);
1128                 int new_using = 0;
1129
1130                 ast_free(identifier_order);
1131                 while ((identify_method = ast_strip(strsep(&io_copy, ",")))) {
1132                         if (!strcmp(identify_method, "auth_username")) {
1133                                 new_using = 1;
1134                                 break;
1135                         }
1136                 }
1137                 using_auth_username = new_using;
1138         }
1139
1140         /* Update default_realm of artificial_auth */
1141         ast_sip_get_default_realm(default_realm, sizeof(default_realm));
1142         fake_auth = ast_sip_get_artificial_auth();
1143         if (!fake_auth || strcmp(fake_auth->realm, default_realm)) {
1144                 ao2_cleanup(fake_auth);
1145
1146                 fake_auth = alloc_artificial_auth(default_realm);
1147                 if (fake_auth) {
1148                         ao2_global_obj_replace_unref(artificial_auth, fake_auth);
1149                 }
1150         }
1151         ao2_cleanup(fake_auth);
1152
1153         ast_sip_get_unidentified_request_thresholds(&unidentified_count, &unidentified_period, &unidentified_prune_interval);
1154
1155         /* Clean out the old task, if any */
1156         ast_sched_clean_by_callback(prune_context, prune_task, clean_task);
1157         /* Have to do something with the return value to shut up the stupid compiler. */
1158         if (ast_sched_add_variable(prune_context, unidentified_prune_interval * 1000, prune_task, NULL, 1) < 0) {
1159                 return;
1160         }
1161 }
1162
1163 /*! \brief Observer which is used to update our interval and default_realm when the global setting changes */
1164 static struct ast_sorcery_observer global_observer = {
1165         .loaded = global_loaded,
1166 };
1167
1168 /*!
1169  * \internal
1170  * \brief Shutdown the serializers in the distributor pool.
1171  * \since 13.10.0
1172  *
1173  * \return Nothing
1174  */
1175 static void distributor_pool_shutdown(void)
1176 {
1177         int idx;
1178
1179         for (idx = 0; idx < ARRAY_LEN(distributor_pool); ++idx) {
1180                 ast_taskprocessor_unreference(distributor_pool[idx]);
1181                 distributor_pool[idx] = NULL;
1182         }
1183 }
1184
1185 /*!
1186  * \internal
1187  * \brief Setup the serializers in the distributor pool.
1188  * \since 13.10.0
1189  *
1190  * \retval 0 on success.
1191  * \retval -1 on error.
1192  */
1193 static int distributor_pool_setup(void)
1194 {
1195         char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1];
1196         int idx;
1197
1198         for (idx = 0; idx < ARRAY_LEN(distributor_pool); ++idx) {
1199                 /* Create name with seq number appended. */
1200                 ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "pjsip/distributor");
1201
1202                 distributor_pool[idx] = ast_sip_create_serializer(tps_name);
1203                 if (!distributor_pool[idx]) {
1204                         return -1;
1205                 }
1206         }
1207         return 0;
1208 }
1209
1210 int ast_sip_initialize_distributor(void)
1211 {
1212         unidentified_requests = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
1213                 DEFAULT_SUSPECTS_BUCKETS, suspects_hash, NULL, suspects_compare);
1214         if (!unidentified_requests) {
1215                 return -1;
1216         }
1217
1218         dialog_associations = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
1219                 DIALOG_ASSOCIATIONS_BUCKETS, dialog_associations_hash, NULL,
1220                 dialog_associations_cmp);
1221         if (!dialog_associations) {
1222                 ast_sip_destroy_distributor();
1223                 return -1;
1224         }
1225
1226         if (distributor_pool_setup()) {
1227                 ast_sip_destroy_distributor();
1228                 return -1;
1229         }
1230
1231         prune_context = ast_sched_context_create();
1232         if (!prune_context) {
1233                 ast_sip_destroy_distributor();
1234                 return -1;
1235         }
1236
1237         if (ast_sched_start_thread(prune_context)) {
1238                 ast_sip_destroy_distributor();
1239                 return -1;
1240         }
1241
1242         ast_sorcery_observer_add(ast_sip_get_sorcery(), "global", &global_observer);
1243         ast_sorcery_reload_object(ast_sip_get_sorcery(), "global");
1244
1245         if (create_artificial_endpoint() || create_artificial_auth()) {
1246                 ast_sip_destroy_distributor();
1247                 return -1;
1248         }
1249
1250         if (ast_sip_register_service(&distributor_mod)) {
1251                 ast_sip_destroy_distributor();
1252                 return -1;
1253         }
1254         if (ast_sip_register_service(&endpoint_mod)) {
1255                 ast_sip_destroy_distributor();
1256                 return -1;
1257         }
1258         if (ast_sip_register_service(&auth_mod)) {
1259                 ast_sip_destroy_distributor();
1260                 return -1;
1261         }
1262
1263         unid_formatter = ao2_alloc_options(sizeof(struct ast_sip_cli_formatter_entry), NULL,
1264                 AO2_ALLOC_OPT_LOCK_NOLOCK);
1265         if (!unid_formatter) {
1266                 ast_sip_destroy_distributor();
1267                 ast_log(LOG_ERROR, "Unable to allocate memory for unid_formatter\n");
1268                 return -1;
1269         }
1270         unid_formatter->name = "unidentified_request";
1271         unid_formatter->print_header = cli_unid_print_header;
1272         unid_formatter->print_body = cli_unid_print_body;
1273         unid_formatter->get_container = cli_unid_get_container;
1274         unid_formatter->iterate = cli_unid_iterate;
1275         unid_formatter->get_id = cli_unid_get_id;
1276         unid_formatter->retrieve_by_id = cli_unid_retrieve_by_id;
1277         ast_sip_register_cli_formatter(unid_formatter);
1278
1279         ast_cli_register_multiple(cli_commands, ARRAY_LEN(cli_commands));
1280
1281         return 0;
1282 }
1283
1284 void ast_sip_destroy_distributor(void)
1285 {
1286         ast_cli_unregister_multiple(cli_commands, ARRAY_LEN(cli_commands));
1287         ast_sip_unregister_cli_formatter(unid_formatter);
1288
1289         ast_sip_unregister_service(&auth_mod);
1290         ast_sip_unregister_service(&endpoint_mod);
1291         ast_sip_unregister_service(&distributor_mod);
1292
1293         ao2_global_obj_release(artificial_auth);
1294         ao2_cleanup(artificial_endpoint);
1295
1296         ast_sorcery_observer_remove(ast_sip_get_sorcery(), "global", &global_observer);
1297
1298         if (prune_context) {
1299                 ast_sched_context_destroy(prune_context);
1300         }
1301
1302         distributor_pool_shutdown();
1303
1304         ao2_cleanup(dialog_associations);
1305         ao2_cleanup(unidentified_requests);
1306 }