afb-stub-ws: Safe handling of deconnections
[src/app-framework-binder.git] / src / afb-stub-ws.c
1 /*
2  * Copyright (C) 2015-2018 "IoT.bzh"
3  * Author José Bollo <jose.bollo@iot.bzh>
4  *
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at
8  *
9  *   http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  */
17
18 #define _GNU_SOURCE
19
20 #include <stdlib.h>
21 #include <string.h>
22 #include <assert.h>
23 #include <fcntl.h>
24 #include <unistd.h>
25 #include <errno.h>
26 #include <endian.h>
27 #include <netdb.h>
28 #include <sys/types.h>
29 #include <sys/socket.h>
30 #include <sys/un.h>
31 #include <pthread.h>
32
33 #include <json-c/json.h>
34
35 #include <afb/afb-event-x2.h>
36
37 #include "afb-session.h"
38 #include "afb-cred.h"
39 #include "afb-api.h"
40 #include "afb-apiset.h"
41 #include "afb-proto-ws.h"
42 #include "afb-stub-ws.h"
43 #include "afb-context.h"
44 #include "afb-evt.h"
45 #include "afb-xreq.h"
46 #include "verbose.h"
47 #include "fdev.h"
48 #include "jobs.h"
49
50 struct afb_stub_ws;
51
52
53 /**
54  * structure for a ws request: requests on server side
55  */
56 struct server_req {
57         struct afb_xreq xreq;           /**< the xreq */
58         struct afb_stub_ws *stubws;     /**< the client of the request */
59         struct afb_proto_ws_call *call; /**< the incoming call */
60 };
61
62 /**
63  * structure for recording events on client side
64  */
65 struct client_event
66 {
67         struct client_event *next;      /**< link to the next */
68         struct afb_event_x2 *event;     /**< the local event */
69         int id;                         /**< the identifier */
70         int refcount;                   /**< a reference count */
71 };
72
73 /**
74  * structure for recording describe requests on the client side
75  */
76 struct client_describe
77 {
78         struct afb_stub_ws *stubws;     /**< the stub */
79         struct jobloop *jobloop;        /**< the jobloop to leave */
80         struct json_object *result;     /**< result */
81 };
82
83 /**
84  * structure for jobs of describing
85  */
86 struct server_describe
87 {
88         struct afb_stub_ws *stubws;
89         struct afb_proto_ws_describe *describe;
90 };
91
92 /**
93  * structure for recording sessions
94  */
95 struct server_session
96 {
97         struct server_session *next;
98         struct afb_session *session;
99 };
100
101 /******************* stub description for client or servers ******************/
102
103 struct afb_stub_ws
104 {
105         /* protocol */
106         struct afb_proto_ws *proto;
107
108         /* apiset */
109         struct afb_apiset *apiset;
110
111         /* on hangup callback */
112         void (*on_hangup)(struct afb_stub_ws *);
113
114         union {
115                 /* server side */
116                 struct {
117                         /* listener for events */
118                         struct afb_evt_listener *listener;
119
120                         /* sessions */
121                         struct server_session *sessions;
122
123                         /* credentials of the client */
124                         struct afb_cred *cred;
125                 };
126
127                 /* client side */
128                 struct {
129                         /* event replica  */
130                         struct client_event *events;
131                 };
132         };
133
134         /* count of references */
135         unsigned refcount;
136
137         /* type of the stub: 0=server, 1=client */
138         uint8_t is_client;
139
140         /* the api name */
141         char apiname[1];
142 };
143
144 /******************* ws request part for server *****************/
145
146 /* decrement the reference count of the request and free/release it on falling to null */
147 static void server_req_destroy_cb(struct afb_xreq *xreq)
148 {
149         struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
150
151         afb_context_disconnect(&wreq->xreq.context);
152         afb_cred_unref(wreq->xreq.cred);
153         json_object_put(wreq->xreq.json);
154         afb_proto_ws_call_unref(wreq->call);
155         afb_stub_ws_unref(wreq->stubws);
156         free(wreq);
157 }
158
159 static void server_req_reply_cb(struct afb_xreq *xreq, struct json_object *obj, const char *error, const char *info)
160 {
161         int rc;
162         struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
163
164         rc = afb_proto_ws_call_reply(wreq->call, obj, error, info);
165         if (rc < 0)
166                 ERROR("error while sending reply");
167         json_object_put(obj);
168 }
169
170 static int server_req_subscribe_cb(struct afb_xreq *xreq, struct afb_event_x2 *event)
171 {
172         int rc;
173         struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
174
175         rc = afb_evt_event_x2_add_watch(wreq->stubws->listener, event);
176         if (rc >= 0)
177                 rc = afb_proto_ws_call_subscribe(wreq->call,  afb_evt_event_x2_fullname(event), afb_evt_event_x2_id(event));
178         if (rc < 0)
179                 ERROR("error while subscribing event");
180         return rc;
181 }
182
183 static int server_req_unsubscribe_cb(struct afb_xreq *xreq, struct afb_event_x2 *event)
184 {
185         int rc, rc2;
186         struct server_req *wreq = CONTAINER_OF_XREQ(struct server_req, xreq);
187
188         rc = afb_proto_ws_call_unsubscribe(wreq->call,  afb_evt_event_x2_fullname(event), afb_evt_event_x2_id(event));
189         rc2 = afb_evt_event_x2_remove_watch(wreq->stubws->listener, event);
190         if (rc >= 0 && rc2 < 0)
191                 rc = rc2;
192         if (rc < 0)
193                 ERROR("error while unsubscribing event");
194         return rc;
195 }
196
197 static const struct afb_xreq_query_itf server_req_xreq_itf = {
198         .reply = server_req_reply_cb,
199         .unref = server_req_destroy_cb,
200         .subscribe = server_req_subscribe_cb,
201         .unsubscribe = server_req_unsubscribe_cb
202 };
203
204 /******************* client part **********************************/
205
206 /* destroy all events */
207 static void client_drop_all_events(struct afb_stub_ws *stubws)
208 {
209         struct client_event *ev, *nxt;
210
211         nxt = __atomic_exchange_n(&stubws->events, NULL, __ATOMIC_RELAXED);
212         while (nxt) {
213                 ev = nxt;
214                 nxt = ev->next;
215                 afb_evt_event_x2_unref(ev->event);
216                 free(ev);
217         }
218 }
219
220 /* search the event */
221 static struct client_event *client_event_search(struct afb_stub_ws *stubws, uint32_t eventid, const char *name)
222 {
223         struct client_event *ev;
224
225         ev = stubws->events;
226         while (ev != NULL && (ev->id != eventid || 0 != strcmp(afb_evt_event_x2_fullname(ev->event), name)))
227                 ev = ev->next;
228
229         return ev;
230 }
231
232 /* on call, propagate it to the ws service */
233 static void client_api_call_cb(void * closure, struct afb_xreq *xreq)
234 {
235         int rc;
236         struct afb_stub_ws *stubws = closure;
237
238         if (stubws->proto == NULL) {
239                 afb_xreq_reply(xreq, NULL, "disconnected", "server hung up");
240                 return;
241         }
242
243         rc = afb_proto_ws_client_call(
244                         stubws->proto,
245                         xreq->request.called_verb,
246                         afb_xreq_json(xreq),
247                         afb_session_uuid(xreq->context.session),
248                         xreq,
249                         xreq_on_behalf_cred_export(xreq));
250         if (rc >= 0)
251                 afb_xreq_unhooked_addref(xreq);
252         else
253                 afb_xreq_reply(xreq, NULL, "internal", "can't send message");
254 }
255
256 static void client_on_description_cb(void *closure, struct json_object *data)
257 {
258         struct client_describe *desc = closure;
259
260         desc->result = data;
261         jobs_leave(desc->jobloop);
262 }
263
264 static void client_send_describe_cb(int signum, void *closure, struct jobloop *jobloop)
265 {
266         struct client_describe *desc = closure;
267
268         if (signum || desc->stubws->proto == NULL)
269                 jobs_leave(jobloop);
270         else {
271                 desc->jobloop = jobloop;
272                 afb_proto_ws_client_describe(desc->stubws->proto, client_on_description_cb, desc);
273         }
274 }
275
276 /* get the description */
277 static struct json_object *client_api_describe_cb(void * closure)
278 {
279         struct client_describe desc;
280
281         /* synchronous job: send the request and wait its result */
282         desc.stubws = closure;
283         desc.result = NULL;
284         jobs_enter(NULL, 0, client_send_describe_cb, &desc);
285         return desc.result;
286 }
287
288 /******************* server part: manage events **********************************/
289
290 static void server_event_add_cb(void *closure, const char *event, int eventid)
291 {
292         struct afb_stub_ws *stubws = closure;
293
294         if (stubws->proto != NULL)
295                 afb_proto_ws_server_event_create(stubws->proto, event, eventid);
296 }
297
298 static void server_event_remove_cb(void *closure, const char *event, int eventid)
299 {
300         struct afb_stub_ws *stubws = closure;
301
302         if (stubws->proto != NULL)
303                 afb_proto_ws_server_event_remove(stubws->proto, event, eventid);
304 }
305
306 static void server_event_push_cb(void *closure, const char *event, int eventid, struct json_object *object)
307 {
308         struct afb_stub_ws *stubws = closure;
309
310         if (stubws->proto != NULL)
311                 afb_proto_ws_server_event_push(stubws->proto, event, eventid, object);
312         json_object_put(object);
313 }
314
315 static void server_event_broadcast_cb(void *closure, const char *event, int eventid, struct json_object *object)
316 {
317         struct afb_stub_ws *stubws = closure;
318
319         if (stubws->proto != NULL)
320                 afb_proto_ws_server_event_broadcast(stubws->proto, event, object);
321         json_object_put(object);
322 }
323
324 /*****************************************************/
325
326 static void client_on_reply_cb(void *closure, void *request, struct json_object *object, const char *error, const char *info)
327 {
328         struct afb_xreq *xreq = request;
329
330         afb_xreq_reply(xreq, object, error, info);
331         afb_xreq_unhooked_unref(xreq);
332 }
333
334 static void client_on_event_create_cb(void *closure, const char *event_name, int event_id)
335 {
336         struct afb_stub_ws *stubws = closure;
337         struct client_event *ev;
338
339         /* check conflicts */
340         ev = client_event_search(stubws, event_id, event_name);
341         if (ev != NULL) {
342                 __atomic_add_fetch(&ev->refcount, 1, __ATOMIC_RELAXED);
343                 return;
344         }
345
346         /* no conflict, try to add it */
347         ev = malloc(sizeof *ev);
348         if (ev != NULL) {
349                 ev->event = afb_evt_event_x2_create(event_name);
350                 if (ev->event != NULL) {
351                         ev->refcount = 1;
352                         ev->id = event_id;
353                         ev->next = stubws->events;
354                         stubws->events = ev;
355                         return;
356                 }
357                 free(ev);
358         }
359         ERROR("can't create event %s, out of memory", event_name);
360 }
361
362 static void client_on_event_remove_cb(void *closure, const char *event_name, int event_id)
363 {
364         struct afb_stub_ws *stubws = closure;
365         struct client_event *ev, **prv;
366
367         /* check conflicts */
368         ev = client_event_search(stubws, event_id, event_name);
369         if (ev == NULL)
370                 return;
371
372         /* decrease the reference count */
373
374         if (__atomic_sub_fetch(&ev->refcount, 1, __ATOMIC_RELAXED))
375                 return;
376
377         /* unlinks the event */
378         prv = &stubws->events;
379         while (*prv != ev)
380                 prv = &(*prv)->next;
381         *prv = ev->next;
382
383         /* destroys the event */
384         afb_evt_event_x2_unref(ev->event);
385         free(ev);
386 }
387
388 static void client_on_event_subscribe_cb(void *closure, void *request, const char *event_name, int event_id)
389 {
390         struct afb_stub_ws *stubws = closure;
391         struct afb_xreq *xreq = request;
392         struct client_event *ev;
393
394         /* check conflicts */
395         ev = client_event_search(stubws, event_id, event_name);
396         if (ev == NULL)
397                 return;
398
399         if (afb_xreq_subscribe(xreq, ev->event) < 0)
400                 ERROR("can't subscribe: %m");
401 }
402
403 static void client_on_event_unsubscribe_cb(void *closure, void *request, const char *event_name, int event_id)
404 {
405         struct afb_stub_ws *stubws = closure;
406         struct afb_xreq *xreq = request;
407         struct client_event *ev;
408
409         /* check conflicts */
410         ev = client_event_search(stubws, event_id, event_name);
411         if (ev == NULL)
412                 return;
413
414         if (afb_xreq_unsubscribe(xreq, ev->event) < 0)
415                 ERROR("can't unsubscribe: %m");
416 }
417
418 static void client_on_event_push_cb(void *closure, const char *event_name, int event_id, struct json_object *data)
419 {
420         struct afb_stub_ws *stubws = closure;
421         struct client_event *ev;
422
423         /* check conflicts */
424         ev = client_event_search(stubws, event_id, event_name);
425         if (ev)
426                 afb_evt_event_x2_push(ev->event, data);
427         else
428                 ERROR("unreadable push event");
429 }
430
431 static void client_on_event_broadcast_cb(void *closure, const char *event_name, struct json_object *data)
432 {
433         afb_evt_broadcast(event_name, data);
434 }
435
436 /*****************************************************/
437
438 static void server_record_session(struct afb_stub_ws *stubws, struct afb_session *session)
439 {
440         struct server_session *s, **prv;
441
442         /* search */
443         prv = &stubws->sessions;
444         while ((s = *prv)) {
445                 if (s->session == session)
446                         return;
447                 if (afb_session_is_closed(s->session)) {
448                         *prv = s->next;
449                         afb_session_unref(s->session);
450                         free(s);
451                 }
452                 else
453                         prv = &s->next;
454         }
455
456         /* create */
457         s = malloc(sizeof *s);
458         if (s) {
459                 s->session = afb_session_addref(session);
460                 s->next = stubws->sessions;
461                 stubws->sessions = s;
462         }
463 }
464
465 static void server_release_all_sessions(struct afb_stub_ws *stubws)
466 {
467         struct server_session *ses, *nses;
468
469         nses = __atomic_exchange_n(&stubws->sessions, NULL, __ATOMIC_RELAXED);
470         while(nses) {
471                 ses = nses;
472                 nses = ses->next;
473                 afb_session_unref(ses->session);
474                 free(ses);
475         }
476 }
477
478 /*****************************************************/
479
480 static void server_on_call_cb(void *closure, struct afb_proto_ws_call *call, const char *verb, struct json_object *args, const char *sessionid, const char *user_creds)
481 {
482         struct afb_stub_ws *stubws = closure;
483         struct server_req *wreq;
484
485         afb_stub_ws_addref(stubws);
486
487         /* create the request */
488         wreq = malloc(sizeof *wreq);
489         if (wreq == NULL)
490                 goto out_of_memory;
491
492         afb_xreq_init(&wreq->xreq, &server_req_xreq_itf);
493         wreq->stubws = stubws;
494         wreq->call = call;
495
496         /* init the context */
497         if (afb_context_connect(&wreq->xreq.context, sessionid, NULL) < 0)
498                 goto unconnected;
499         wreq->xreq.context.validated = 1;
500         server_record_session(stubws, wreq->xreq.context.session);
501         if (wreq->xreq.context.created)
502                 afb_session_set_autoclose(wreq->xreq.context.session, 1);
503
504         /* makes the call */
505         wreq->xreq.cred = afb_cred_mixed_on_behalf_import(stubws->cred, sessionid, user_creds);
506         wreq->xreq.request.called_api = stubws->apiname;
507         wreq->xreq.request.called_verb = verb;
508         wreq->xreq.json = args;
509         afb_xreq_process(&wreq->xreq, stubws->apiset);
510         return;
511
512 unconnected:
513         free(wreq);
514 out_of_memory:
515         json_object_put(args);
516         afb_stub_ws_unref(stubws);
517         afb_proto_ws_call_reply(call, NULL, "internal-error", NULL);
518         afb_proto_ws_call_unref(call);
519 }
520
521 static void server_describe_cb(int signum, void *closure)
522 {
523         struct json_object *obj;
524         struct server_describe *desc = closure;
525
526         /* get the description if possible */
527         obj = !signum ? afb_apiset_describe(desc->stubws->apiset, desc->stubws->apiname) : NULL;
528
529         /* send it */
530         afb_proto_ws_describe_put(desc->describe, obj);
531         json_object_put(obj);
532         afb_stub_ws_unref(desc->stubws);
533 }
534
535 static void server_describe_job(int signum, void *closure)
536 {
537         server_describe_cb(signum, closure);
538         free(closure);
539 }
540
541 static void server_on_describe_cb(void *closure, struct afb_proto_ws_describe *describe)
542 {
543         struct server_describe *desc, sdesc;
544         struct afb_stub_ws *stubws = closure;
545
546         /* allocate (if possible) and init */
547         desc = malloc(sizeof *desc);
548         if (desc == NULL)
549                 desc = &sdesc;
550         desc->stubws = stubws;
551         desc->describe = describe;
552         afb_stub_ws_addref(stubws);
553
554         /* process */
555         if (desc == &sdesc || jobs_queue(NULL, 0, server_describe_job, desc) < 0)
556                 jobs_call(NULL, 0, server_describe_cb, desc);
557 }
558
559 /*****************************************************/
560
561 static const struct afb_proto_ws_client_itf client_itf =
562 {
563         .on_reply = client_on_reply_cb,
564         .on_event_create = client_on_event_create_cb,
565         .on_event_remove = client_on_event_remove_cb,
566         .on_event_subscribe = client_on_event_subscribe_cb,
567         .on_event_unsubscribe = client_on_event_unsubscribe_cb,
568         .on_event_push = client_on_event_push_cb,
569         .on_event_broadcast = client_on_event_broadcast_cb,
570 };
571
572 static struct afb_api_itf client_api_itf = {
573         .call = client_api_call_cb,
574         .describe = client_api_describe_cb
575 };
576
577 static const struct afb_proto_ws_server_itf server_itf =
578 {
579         .on_call = server_on_call_cb,
580         .on_describe = server_on_describe_cb
581 };
582
583 /* the interface for events pushing */
584 static const struct afb_evt_itf server_event_itf = {
585         .broadcast = server_event_broadcast_cb,
586         .push = server_event_push_cb,
587         .add = server_event_add_cb,
588         .remove = server_event_remove_cb
589 };
590
591 /*****************************************************/
592
593 /* disconnect */
594 static void disconnect(struct afb_stub_ws *stubws)
595 {
596         afb_proto_ws_unref(__atomic_exchange_n(&stubws->proto, NULL, __ATOMIC_RELAXED));
597         if (stubws->is_client)
598                 client_drop_all_events(stubws);
599         else {
600                 afb_evt_listener_unref(__atomic_exchange_n(&stubws->listener, NULL, __ATOMIC_RELAXED));
601                 afb_cred_unref(__atomic_exchange_n(&stubws->cred, NULL, __ATOMIC_RELAXED));
602                 server_release_all_sessions(stubws);
603         }
604 }
605
606
607 /* callback when receiving a hangup */
608 static void on_hangup(void *closure)
609 {
610         struct afb_stub_ws *stubws = closure;
611
612         if (stubws->proto) {
613                 afb_stub_ws_addref(stubws);
614                 disconnect(stubws);
615                 if (stubws->on_hangup)
616                         stubws->on_hangup(stubws);
617                 afb_stub_ws_unref(stubws);
618         }
619 }
620
621 static int enqueue_processing(void (*callback)(int signum, void* arg), void *arg)
622 {
623         return jobs_queue(NULL, 0, callback, arg);
624 }
625
626 /*****************************************************/
627
628 static struct afb_proto_ws *afb_stub_ws_create_proto(struct afb_stub_ws *stubws, struct fdev *fdev, uint8_t is_client)
629 {
630         struct afb_proto_ws *proto;
631
632         stubws->proto = proto = is_client
633                   ? afb_proto_ws_create_client(fdev, &client_itf, stubws)
634                   : afb_proto_ws_create_server(fdev, &server_itf, stubws);
635         if (proto) {
636                 afb_proto_ws_on_hangup(proto, on_hangup);
637                 afb_proto_ws_set_queuing(proto, enqueue_processing);
638         }
639
640         return proto;
641 }
642
643 static struct afb_stub_ws *afb_stub_ws_create(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset, uint8_t is_client)
644 {
645         struct afb_stub_ws *stubws;
646
647         stubws = calloc(1, sizeof *stubws + strlen(apiname));
648         if (stubws == NULL)
649                 errno = ENOMEM;
650         else {
651                 if (afb_stub_ws_create_proto(stubws, fdev, is_client)) {
652                         stubws->refcount = 1;
653                         stubws->is_client = is_client;
654                         strcpy(stubws->apiname, apiname);
655                         stubws->apiset = afb_apiset_addref(apiset);
656                         return stubws;
657                 }
658                 free(stubws);
659         }
660         fdev_unref(fdev);
661         return NULL;
662 }
663
664 struct afb_stub_ws *afb_stub_ws_create_client(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset)
665 {
666         return afb_stub_ws_create(fdev, apiname, apiset, 1);
667 }
668
669 struct afb_stub_ws *afb_stub_ws_create_server(struct fdev *fdev, const char *apiname, struct afb_apiset *apiset)
670 {
671         struct afb_stub_ws *stubws;
672
673         stubws = afb_stub_ws_create(fdev, apiname, apiset, 0);
674         if (stubws) {
675                 stubws->cred = afb_cred_create_for_socket(fdev_fd(fdev));
676                 stubws->listener = afb_evt_listener_create(&server_event_itf, stubws);
677                 if (stubws->listener != NULL)
678                         return stubws;
679                 afb_stub_ws_unref(stubws);
680         }
681         return NULL;
682 }
683
684 void afb_stub_ws_unref(struct afb_stub_ws *stubws)
685 {
686         if (stubws && !__atomic_sub_fetch(&stubws->refcount, 1, __ATOMIC_RELAXED)) {
687
688                 disconnect(stubws);
689                 afb_apiset_unref(stubws->apiset);
690                 free(stubws);
691         }
692 }
693
694 void afb_stub_ws_addref(struct afb_stub_ws *stubws)
695 {
696         __atomic_add_fetch(&stubws->refcount, 1, __ATOMIC_RELAXED);
697 }
698
699 void afb_stub_ws_set_on_hangup(struct afb_stub_ws *stubws, void (*on_hangup)(struct afb_stub_ws*))
700 {
701         stubws->on_hangup = on_hangup;
702 }
703
704 const char *afb_stub_ws_name(struct afb_stub_ws *stubws)
705 {
706         return stubws->apiname;
707 }
708
709 struct afb_api_item afb_stub_ws_client_api(struct afb_stub_ws *stubws)
710 {
711         struct afb_api_item api;
712
713         assert(stubws->is_client); /* check client */
714         api.closure = stubws;
715         api.itf = &client_api_itf;
716         api.group = NULL;
717         return api;
718 }
719
720 int afb_stub_ws_client_add(struct afb_stub_ws *stubws, struct afb_apiset *apiset)
721 {
722         return afb_apiset_add(apiset, stubws->apiname, afb_stub_ws_client_api(stubws));
723 }