res_ari_events: Fix use after free / double-free of JSON message.
[asterisk/asterisk.git] / res / ari / resource_events.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2012 - 2013, Digium, Inc.
5  *
6  * David M. Lee, II <dlee@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 /*! \file
20  *
21  * \brief /api-docs/events.{format} implementation- WebSocket resource
22  *
23  * \author David M. Lee, II <dlee@digium.com>
24  */
25
26 /*** MODULEINFO
27         <depend type="module">res_http_websocket</depend>
28         <support_level>core</support_level>
29  ***/
30
31 #include "asterisk.h"
32
33 #include "resource_events.h"
34 #include "asterisk/astobj2.h"
35 #include "asterisk/http_websocket.h"
36 #include "asterisk/stasis_app.h"
37 #include "asterisk/vector.h"
38
39 /*! Number of buckets for the event session registry. Remember to keep it a prime number! */
40 #define EVENT_SESSION_NUM_BUCKETS 23
41
42 /*! Number of buckets for a websocket apps container. Remember to keep it a prime number! */
43 #define APPS_NUM_BUCKETS 7
44
45 /*! Initial size of a message queue. */
46 #define MESSAGES_INIT_SIZE 23
47
48
49 /*! \brief A wrapper for the /ref ast_ari_websocket_session. */
50 struct event_session {
51         struct ast_ari_websocket_session *ws_session;   /*!< Handle to the websocket session. */
52         struct ao2_container *websocket_apps;           /*!< List of Stasis apps registered to
53                                                              the websocket session. */
54         AST_VECTOR(, struct ast_json *) message_queue;  /*!< Container for holding delayed messages. */
55         char session_id[];                              /*!< The id for the websocket session. */
56 };
57
58 /*! \brief \ref event_session error types. */
59 enum event_session_error_type {
60         ERROR_TYPE_STASIS_REGISTRATION = 1,  /*!< Stasis failed to register the application. */
61         ERROR_TYPE_OOM = 2,                  /*!< Insufficient memory to create the event
62                                                   session. */
63         ERROR_TYPE_MISSING_APP_PARAM = 3,    /*!< HTTP request was missing an [app] parameter. */
64         ERROR_TYPE_INVALID_APP_PARAM = 4,    /*!< HTTP request contained an invalid [app]
65                                                   parameter. */
66 };
67
68 /*! \brief Local registry for created \ref event_session objects. */
69 static struct ao2_container *event_session_registry;
70
71 /*!
72  * \brief Callback handler for Stasis application messages.
73  *
74  * \internal
75  *
76  * \param data      Void pointer to the event session (\ref event_session).
77  * \param app_name  Name of the Stasis application that dispatched the message.
78  * \param message   The dispatched message.
79  */
80 static void stasis_app_message_handler(
81                 void *data, const char *app_name, struct ast_json *message)
82 {
83         struct event_session *session = data;
84         const char *msg_type, *msg_application;
85
86         ast_assert(session != NULL);
87
88         ao2_lock(session);
89
90         msg_type = S_OR(ast_json_string_get(ast_json_object_get(message, "type")), "");
91         msg_application = S_OR(
92                 ast_json_string_get(ast_json_object_get(message, "application")), "");
93
94         /* If we've been replaced, remove the application from our local
95            websocket_apps container */
96         if (strcmp(msg_type, "ApplicationReplaced") == 0 &&
97                 strcmp(msg_application, app_name) == 0) {
98                 ao2_find(session->websocket_apps, msg_application,
99                         OBJ_UNLINK | OBJ_NODATA);
100         }
101
102         /* Now, we need to determine our state to see how we will handle the message */
103         if (ast_json_object_set(message, "application", ast_json_string_create(app_name))) {
104                 /* We failed to add an application element to our json message */
105                 ast_log(LOG_WARNING,
106                         "Failed to dispatch '%s' message from Stasis app '%s'; could not update message\n",
107                         msg_type,
108                         msg_application);
109         } else if (!session->ws_session) {
110                 /* If the websocket is NULL, the message goes to the queue */
111                 if (!AST_VECTOR_APPEND(&session->message_queue, message)) {
112                         ast_json_ref(message);
113                 }
114                 ast_log(LOG_WARNING,
115                                 "Queued '%s' message for Stasis app '%s'; websocket is not ready\n",
116                                 msg_type,
117                                 msg_application);
118         } else {
119                 if (stasis_app_get_debug_by_name(app_name)) {
120                         char *str = ast_json_dump_string_format(message, ast_ari_json_format());
121
122                         ast_verbose("<--- Sending ARI event to %s --->\n%s\n",
123                                 ast_sockaddr_stringify(ast_ari_websocket_session_get_remote_addr(session->ws_session)),
124                                 str);
125                         ast_json_free(str);
126                 }
127
128                 /* We are ready to publish the message */
129                 ast_ari_websocket_session_write(session->ws_session, message);
130         }
131
132         ao2_unlock(session);
133 }
134
135 /*!
136  * \brief AO2 comparison function for \ref event_session objects.
137  *
138  * \internal
139  *
140  * \param obj    Void pointer to the \ref event_session container.
141  * \param arg    Void pointer to the \ref event_session object.
142  * \param flags  The \ref search_flags to use when creating the hash key.
143  *
144  * \retval 0          The objects are not equal.
145  * \retval CMP_MATCH  The objects are equal.
146  */
147 static int event_session_compare(void *obj, void *arg, int flags)
148 {
149         const struct event_session *object_left = obj;
150         const struct event_session *object_right = arg;
151         const char *right_key = arg;
152         int cmp = 0;
153
154         switch (flags & OBJ_SEARCH_MASK) {
155         case OBJ_SEARCH_OBJECT:
156                 right_key = object_right->session_id;
157                 /* Fall through */
158         case OBJ_SEARCH_KEY:
159                 cmp = strcmp(object_left->session_id, right_key);
160                 break;
161         case OBJ_SEARCH_PARTIAL_KEY:
162                 cmp = strncmp(object_left->session_id, right_key, strlen(right_key));
163                 break;
164         default:
165                 break;
166         }
167
168         return cmp ? 0 : CMP_MATCH;
169 }
170
171 /*!
172  * \brief AO2 hash function for \ref event_session objects.
173  *
174  * \details Computes hash value for the given \ref event_session, with respect to the
175  *          provided search flags.
176  *
177  * \internal
178  *
179  * \param obj    Void pointer to the \ref event_session object.
180  * \param flags  The \ref search_flags to use when creating the hash key.
181  *
182  * \retval > 0  on success
183  * \retval   0  on failure
184  */
185 static int event_session_hash(const void *obj, const int flags)
186 {
187         const struct event_session *session;
188         const char *key;
189
190         switch (flags & OBJ_SEARCH_MASK) {
191         case OBJ_SEARCH_KEY:
192                 key = obj;
193                 break;
194         case OBJ_SEARCH_OBJECT:
195                 session = obj;
196                 key = session->session_id;
197                 break;
198         default:
199                 /* Hash can only work on something with a full key. */
200                 ast_assert(0);
201                 return 0;
202         }
203         return ast_str_hash(key);
204 }
205
206 /*!
207  * \brief Explicitly shutdown a session.
208  *
209  * \details An explicit shutdown is necessary, since the \ref stasis_app has a reference
210  *          to this session. We also need to be sure to null out the \c ws_session field,
211  *          since the websocket is about to go away.
212  *
213  * \internal
214  *
215  * \param session  Event session object (\ref event_session).
216  */
217 static void event_session_shutdown(struct event_session *session)
218 {
219         struct ao2_iterator i;
220         char *app;
221         int j;
222         SCOPED_AO2LOCK(lock, session);
223
224         /* Clean up the websocket_apps container */
225         if (session->websocket_apps) {
226                 i = ao2_iterator_init(session->websocket_apps, 0);
227                 while ((app = ao2_iterator_next(&i))) {
228                         stasis_app_unregister(app);
229                         ao2_cleanup(app);
230                 }
231                 ao2_iterator_destroy(&i);
232                 ao2_cleanup(session->websocket_apps);
233                 session->websocket_apps = NULL;
234         }
235
236         /* Clean up the message_queue container */
237         for (j = 0; j < AST_VECTOR_SIZE(&session->message_queue); j++) {
238                 struct ast_json *msg = AST_VECTOR_GET(&session->message_queue, j);
239                 ast_json_unref(msg);
240         }
241         AST_VECTOR_FREE(&session->message_queue);
242
243         /* Remove the handle to the underlying websocket session */
244         session->ws_session = NULL;
245 }
246
247 /*!
248  * \brief Updates the websocket session for an \ref event_session.
249  *
250  * \details The websocket for the given \ref event_session will be updated to the value
251  *          of the \c ws_session argument.
252  *
253  *          If the value of the \c ws_session is not \c NULL and there are messages in the
254  *          event session's \c message_queue, the messages are dispatched and removed from
255  *          the queue.
256  *
257  * \internal
258  *
259  * \param session     The event session object to update (\ref event_session).
260  * \param ws_session  Handle to the underlying websocket session
261  *                    (\ref ast_ari_websocket_session).
262  */
263 static void event_session_update_websocket(
264                 struct event_session *session, struct ast_ari_websocket_session *ws_session)
265 {
266         int i;
267
268         ast_assert(session != NULL);
269
270         ao2_lock(session);
271
272         session->ws_session = ws_session;
273
274         for (i = 0; i < AST_VECTOR_SIZE(&session->message_queue); i++) {
275                 struct ast_json *msg = AST_VECTOR_GET(&session->message_queue, i);
276                 ast_ari_websocket_session_write(session->ws_session, msg);
277                 ast_json_unref(msg);
278         }
279
280         AST_VECTOR_RESET(&session->message_queue, AST_VECTOR_ELEM_CLEANUP_NOOP);
281         ao2_unlock(session);
282 }
283
284 /*!
285  * \brief Processes cleanup actions for a \ref event_session object.
286  *
287  * \internal
288  *
289  * \param session  The event session object to cleanup (\ref event_session).
290  */
291 static void event_session_cleanup(struct event_session *session)
292 {
293         if (!session) {
294                 return;
295         }
296
297         event_session_shutdown(session);
298         if (event_session_registry) {
299                 ao2_unlink(event_session_registry, session);
300         }
301 }
302
303 /*!
304  * \brief Event session object destructor (\ref event_session).
305  *
306  * \internal
307  *
308  * \param obj  Void pointer to the \ref event_session object.
309  */
310 static void event_session_dtor(void *obj)
311 {
312 #ifdef AST_DEVMODE /* Avoid unused variable warning */
313         struct event_session *session = obj;
314
315 #endif
316
317         /* event_session_shutdown should have been called before now */
318         ast_assert(session->ws_session == NULL);
319         ast_assert(session->websocket_apps == NULL);
320         ast_assert(AST_VECTOR_SIZE(&session->message_queue) == 0);
321 }
322
323 /*!
324  * \brief Handles \ref event_session error processing.
325  *
326  * \internal
327  *
328  * \param session  The \ref event_session object.
329  * \param error    The \ref event_session_error_type to handle.
330  * \param ser      HTTP TCP/TLS Server Session (\ref ast_tcptls_session_instance).
331  *
332  * \retval  -1  Always returns -1.
333  */
334 static int event_session_allocation_error_handler(
335                 struct event_session *session, enum event_session_error_type error,
336                 struct ast_tcptls_session_instance *ser)
337 {
338         /* Notify the client */
339         switch (error) {
340         case ERROR_TYPE_STASIS_REGISTRATION:
341                 ast_http_error(ser, 500, "Internal Server Error",
342                         "Stasis registration failed");
343                 break;
344
345         case ERROR_TYPE_OOM:
346                 ast_http_error(ser, 500, "Internal Server Error",
347                         "Allocation failed");
348                 break;
349
350         case ERROR_TYPE_MISSING_APP_PARAM:
351                 ast_http_error(ser, 400, "Bad Request",
352                         "HTTP request is missing param: [app]");
353                 break;
354
355         case ERROR_TYPE_INVALID_APP_PARAM:
356                 ast_http_error(ser, 400, "Bad Request",
357                         "Invalid application provided in param [app].");
358                 break;
359
360         default:
361                 break;
362         }
363
364         /* Cleanup the session */
365         event_session_cleanup(session);
366         return -1;
367 }
368
369 /*!
370  * \brief Creates an \ref event_session object and registers its apps with Stasis.
371  *
372  * \internal
373  *
374  * \param ser         HTTP TCP/TLS Server Session (\ref ast_tcptls_session_instance).
375  * \param args        The Stasis [app] parameters as parsed from the HTTP request
376  *                    (\ref ast_ari_events_event_websocket_args).
377  * \param session_id  The id for the websocket session that will be created for this
378  *                    event session.
379  *
380  * \retval  0  on success
381  * \retval -1  on failure
382  */
383 static int event_session_alloc(struct ast_tcptls_session_instance *ser,
384                 struct ast_ari_events_event_websocket_args *args, const char *session_id)
385 {
386         RAII_VAR(struct event_session *, session, NULL, ao2_cleanup);
387         int (* register_handler)(const char *, stasis_app_cb handler, void *data);
388         size_t size, i;
389
390         /* The request must have at least one [app] parameter */
391         if (args->app_count == 0) {
392                 return event_session_allocation_error_handler(
393                         session, ERROR_TYPE_MISSING_APP_PARAM, ser);
394         }
395
396         size = sizeof(*session) + strlen(session_id) + 1;
397
398         /* Instantiate the event session */
399         session = ao2_alloc(size, event_session_dtor);
400         if (!session) {
401                 return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
402         }
403
404         strncpy(session->session_id, session_id, size - sizeof(*session));
405
406         /* Instantiate the hash table for Stasis apps */
407         session->websocket_apps =
408                 ast_str_container_alloc(APPS_NUM_BUCKETS);
409
410         if (!session->websocket_apps) {
411                 return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
412         }
413
414         /* Instantiate the message queue */
415         if (AST_VECTOR_INIT(&session->message_queue, MESSAGES_INIT_SIZE)) {
416                 return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
417         }
418
419         /* Register the apps with Stasis */
420         if (args->subscribe_all) {
421                 register_handler = &stasis_app_register_all;
422         } else {
423                 register_handler = &stasis_app_register;
424         }
425
426         for (i = 0; i < args->app_count; ++i) {
427                 const char *app = args->app[i];
428
429                 if (ast_strlen_zero(app)) {
430                         return event_session_allocation_error_handler(
431                                 session, ERROR_TYPE_INVALID_APP_PARAM, ser);
432                 }
433
434                 if (ast_str_container_add(session->websocket_apps, app)) {
435                         return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
436                 }
437
438                 if (register_handler(app, stasis_app_message_handler, session)) {
439                         ast_log(LOG_WARNING, "Stasis registration failed for application: '%s'\n", app);
440                         return event_session_allocation_error_handler(
441                                 session, ERROR_TYPE_STASIS_REGISTRATION, ser);                  
442                 }
443         }
444
445         /* Add the event session to the local registry */
446         if (!ao2_link(event_session_registry, session)) {
447                 return event_session_allocation_error_handler(session, ERROR_TYPE_OOM, ser);
448         }
449
450         return 0;
451 }
452
453 static int event_session_shutdown_cb(void *session, void *arg, int flags)
454 {
455         event_session_cleanup(session);
456
457         return 0;
458 }
459
460 void ast_ari_websocket_events_event_websocket_dtor(void)
461 {
462         ao2_callback(event_session_registry, OBJ_MULTIPLE | OBJ_NODATA, event_session_shutdown_cb, NULL);
463
464         ao2_cleanup(event_session_registry);
465         event_session_registry = NULL;
466 }
467
468 int ast_ari_websocket_events_event_websocket_init(void)
469 {
470         /* Try to instantiate the registry */
471         event_session_registry = ao2_container_alloc(EVENT_SESSION_NUM_BUCKETS,
472                                                      event_session_hash,
473                                                      event_session_compare);
474         if (!event_session_registry) {
475                 /* This is bad, bad. */
476                 ast_log(LOG_WARNING,
477                             "Failed to allocate the local registry for websocket applications\n");
478                 return -1;
479         }
480
481         return 0;
482 }
483
484 int ast_ari_websocket_events_event_websocket_attempted(
485                 struct ast_tcptls_session_instance *ser, struct ast_variable *headers,
486                 struct ast_ari_events_event_websocket_args *args, const char *session_id)
487 {
488         ast_debug(3, "/events WebSocket attempted\n");
489
490         /* Create the event session */
491         return event_session_alloc(ser, args, session_id);
492 }
493
494 void ast_ari_websocket_events_event_websocket_established(
495                 struct ast_ari_websocket_session *ws_session, struct ast_variable *headers,
496                 struct ast_ari_events_event_websocket_args *args)
497 {
498         struct event_session *session;
499
500         struct ast_json *msg;
501         const char *session_id;
502
503         ast_debug(3, "/events WebSocket established\n");
504
505         ast_assert(ws_session != NULL);
506
507         session_id = ast_ari_websocket_session_id(ws_session);
508
509         /* Find the event_session and update its websocket  */
510         session = ao2_find(event_session_registry, session_id, OBJ_SEARCH_KEY);
511         if (session) {
512                 ao2_unlink(event_session_registry, session);
513                 event_session_update_websocket(session, ws_session);
514         } else {
515                 ast_log(LOG_WARNING,
516                         "Failed to locate an event session for the provided websocket session\n");
517         }
518
519         /* We don't process any input, but we'll consume it waiting for EOF */
520         while ((msg = ast_ari_websocket_session_read(ws_session))) {
521                 ast_json_unref(msg);
522         }
523
524         event_session_cleanup(session);
525         ao2_ref(session, -1);
526 }
527
528 void ast_ari_events_user_event(struct ast_variable *headers,
529         struct ast_ari_events_user_event_args *args,
530         struct ast_ari_response *response)
531 {
532         enum stasis_app_user_event_res res;
533         struct ast_json *json_variables = NULL;
534
535         if (args->variables) {
536                 ast_ari_events_user_event_parse_body(args->variables, args);
537                 json_variables = ast_json_object_get(args->variables, "variables");
538         }
539
540         if (ast_strlen_zero(args->application)) {
541                 ast_ari_response_error(response, 400, "Bad Request",
542                         "Missing parameter application");
543                 return;
544         }
545
546         res = stasis_app_user_event(args->application,
547                 args->event_name,
548                 args->source, args->source_count,
549                 json_variables);
550
551         switch (res) {
552         case STASIS_APP_USER_OK:
553                 ast_ari_response_no_content(response);
554                 break;
555
556         case STASIS_APP_USER_APP_NOT_FOUND:
557                 ast_ari_response_error(response, 404, "Not Found",
558                         "Application not found");
559                 break;
560
561         case STASIS_APP_USER_EVENT_SOURCE_NOT_FOUND:
562                 ast_ari_response_error(response, 422, "Unprocessable Entity",
563                         "Event source was not found");
564                 break;
565
566         case STASIS_APP_USER_EVENT_SOURCE_BAD_SCHEME:
567                 ast_ari_response_error(response, 400, "Bad Request",
568                         "Invalid event source URI scheme");
569                 break;
570
571         case STASIS_APP_USER_USEREVENT_INVALID:
572                 ast_ari_response_error(response, 400, "Bad Request",
573                         "Invalid userevnet data");
574                 break;
575
576         case STASIS_APP_USER_INTERNAL_ERROR:
577         default:
578                 ast_ari_response_error(response, 500, "Internal Server Error",
579                         "Error processing request");
580         }
581 }