src/afb-ws.c

   1 /*
   2  * Copyright (C) 2015-2020 "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 #include <stdlib.h>
  20 #include <unistd.h>
  21 #include <stdint.h>
  22 #include <assert.h>
  23 #include <errno.h>
  24 #include <sys/uio.h>
  25 #include <string.h>
  26 #include <stdarg.h>
  27 #include <poll.h>
  28
  29 #include "websock.h"
  30 #include "afb-ws.h"
  31 #include "fdev.h"
  32
  33 /*
  34  * declaration of the websock interface for afb-ws
  35  */
  36 static ssize_t aws_writev(struct afb_ws *ws, const struct iovec *iov, int iovcnt);
  37 static ssize_t aws_readv(struct afb_ws *ws, const struct iovec *iov, int iovcnt);
  38 static void aws_on_close(struct afb_ws *ws, uint16_t code, size_t size);
  39 static void aws_on_text(struct afb_ws *ws, int last, size_t size);
  40 static void aws_on_binary(struct afb_ws *ws, int last, size_t size);
  41 static void aws_on_continue(struct afb_ws *ws, int last, size_t size);
  42 static void aws_on_readable(struct afb_ws *ws);
  43 static void aws_on_error(struct afb_ws *ws, uint16_t code, const void *data, size_t size);
  44
  45 static struct websock_itf aws_itf = {
  46         .writev = (void*)aws_writev,
  47         .readv = (void*)aws_readv,
  48
  49         .on_ping = NULL,
  50         .on_pong = NULL,
  51         .on_close = (void*)aws_on_close,
  52         .on_text = (void*)aws_on_text,
  53         .on_binary = (void*)aws_on_binary,
  54         .on_continue = (void*)aws_on_continue,
  55         .on_extension = NULL,
  56
  57         .on_error = (void*)aws_on_error
  58 };
  59
  60 /*
  61  * a common scheme of buffer handling
  62  */
  63 struct buf
  64 {
  65         char *buffer;
  66         size_t size;
  67 };
  68
  69 /*
  70  * the state
  71  */
  72 enum state
  73 {
  74         waiting,
  75         reading_text,
  76         reading_binary
  77 };
  78
  79 /*
  80  * the afb_ws structure
  81  */
  82 struct afb_ws
  83 {
  84         int fd;                 /* the socket file descriptor */
  85         enum state state;       /* current state */
  86         const struct afb_ws_itf *itf; /* the callback interface */
  87         void *closure;          /* closure when calling the callbacks */
  88         struct websock *ws;     /* the websock handler */
  89         struct fdev *fdev;      /* the fdev for the socket */
  90         struct buf buffer;      /* the last read fragment */
  91 };
  92
  93 /*
  94  * Returns the current buffer of 'ws' that is reset.
  95  */
  96 static inline struct buf aws_pick_buffer(struct afb_ws *ws)
  97 {
  98         struct buf result = ws->buffer;
  99         if (result.buffer)
 100                 result.buffer[result.size] = 0;
 101         ws->buffer.buffer = NULL;
 102         ws->buffer.size = 0;
 103         return result;
 104 }
 105
 106 /*
 107  * Clear the current buffer
 108  */
 109 static inline void aws_clear_buffer(struct afb_ws *ws)
 110 {
 111         ws->buffer.size = 0;
 112 }
 113
 114 /*
 115  * Disconnect the websocket 'ws' and calls on_hangup if
 116  * 'call_on_hangup' is not null.
 117  */
 118 static void aws_disconnect(struct afb_ws *ws, int call_on_hangup)
 119 {
 120         struct websock *wsi = ws->ws;
 121         if (wsi != NULL) {
 122                 ws->ws = NULL;
 123                 fdev_set_callback(ws->fdev, NULL, 0);
 124                 fdev_unref(ws->fdev);
 125                 websock_destroy(wsi);
 126                 free(ws->buffer.buffer);
 127                 ws->state = waiting;
 128                 if (call_on_hangup && ws->itf->on_hangup)
 129                         ws->itf->on_hangup(ws->closure);
 130         }
 131 }
 132
 133 static void fdevcb(void *ws, uint32_t revents, struct fdev *fdev)
 134 {
 135         if ((revents & EPOLLHUP) != 0)
 136                 afb_ws_hangup(ws);
 137         else if ((revents & EPOLLIN) != 0)
 138                 aws_on_readable(ws);
 139 }
 140
 141 /*
 142  * Creates the afb_ws structure for the file descritor
 143  * 'fd' and the callbacks described by the interface 'itf'
 144  * and its 'closure'.
 145  * When the creation is a success, the systemd event loop 'eloop' is
 146  * used for handling event for 'fd'.
 147  *
 148  * Returns the handle for the afb_ws created or NULL on error.
 149  */
 150 struct afb_ws *afb_ws_create(struct fdev *fdev, const struct afb_ws_itf *itf, void *closure)
 151 {
 152         struct afb_ws *result;
 153
 154         assert(fdev);
 155
 156         /* allocation */
 157         result = malloc(sizeof * result);
 158         if (result == NULL)
 159                 goto error;
 160
 161         /* init */
 162         result->fdev = fdev;
 163         result->fd = fdev_fd(fdev);
 164         result->state = waiting;
 165         result->itf = itf;
 166         result->closure = closure;
 167         result->buffer.buffer = NULL;
 168         result->buffer.size = 0;
 169
 170         /* creates the websocket */
 171         result->ws = websock_create_v13(&aws_itf, result);
 172         if (result->ws == NULL)
 173                 goto error2;
 174
 175         /* finalize */
 176         fdev_set_events(fdev, EPOLLIN);
 177         fdev_set_callback(fdev, fdevcb, result);
 178         return result;
 179
 180 error2:
 181         free(result);
 182 error:
 183         fdev_unref(fdev);
 184         return NULL;
 185 }
 186
 187 /*
 188  * Destroys the websocket 'ws'
 189  * It first hangup (but without calling on_hangup for safety reasons)
 190  * if needed.
 191  */
 192 void afb_ws_destroy(struct afb_ws *ws)
 193 {
 194         aws_disconnect(ws, 0);
 195         free(ws);
 196 }
 197
 198 /*
 199  * Hangup the websocket 'ws'
 200  */
 201 void afb_ws_hangup(struct afb_ws *ws)
 202 {
 203         aws_disconnect(ws, 1);
 204 }
 205
 206 /*
 207  * Is the websocket 'ws' still connected ?
 208  */
 209 int afb_ws_is_connected(struct afb_ws *ws)
 210 {
 211         return ws->ws != NULL;
 212 }
 213
 214 /*
 215  * Sends a 'close' command to the endpoint of 'ws' with the 'code' and the
 216  * 'reason' (that can be NULL and that else should not be greater than 123
 217  * characters).
 218  * Returns 0 on success or -1 in case of error.
 219  */
 220 int afb_ws_close(struct afb_ws *ws, uint16_t code, const char *reason)
 221 {
 222         if (ws->ws == NULL) {
 223                 /* disconnected */
 224                 errno = EPIPE;
 225                 return -1;
 226         }
 227         return websock_close(ws->ws, code, reason, reason == NULL ? 0 : strlen(reason));
 228 }
 229
 230 /*
 231  * Sends a 'close' command to the endpoint of 'ws' with the 'code' and the
 232  * 'reason' (that can be NULL and that else should not be greater than 123
 233  * characters).
 234  * Raise an error after 'close' command is sent.
 235  * Returns 0 on success or -1 in case of error.
 236  */
 237 int afb_ws_error(struct afb_ws *ws, uint16_t code, const char *reason)
 238 {
 239         if (ws->ws == NULL) {
 240                 /* disconnected */
 241                 errno = EPIPE;
 242                 return -1;
 243         }
 244         return websock_error(ws->ws, code, reason, reason == NULL ? 0 : strlen(reason));
 245 }
 246
 247 /*
 248  * Sends a 'text' of 'length' to the endpoint of 'ws'.
 249  * Returns 0 on success or -1 in case of error.
 250  */
 251 int afb_ws_text(struct afb_ws *ws, const char *text, size_t length)
 252 {
 253         if (ws->ws == NULL) {
 254                 /* disconnected */
 255                 errno = EPIPE;
 256                 return -1;
 257         }
 258         return websock_text(ws->ws, 1, text, length);
 259 }
 260
 261 /*
 262  * Sends a variable list of texts to the endpoint of 'ws'.
 263  * Returns 0 on success or -1 in case of error.
 264  */
 265 int afb_ws_texts(struct afb_ws *ws, ...)
 266 {
 267         va_list args;
 268         struct iovec ios[32];
 269         int count;
 270         const char *s;
 271
 272         if (ws->ws == NULL) {
 273                 /* disconnected */
 274                 errno = EPIPE;
 275                 return -1;
 276         }
 277
 278         count = 0;
 279         va_start(args, ws);
 280         s = va_arg(args, const char *);
 281         while (s != NULL) {
 282                 if (count == 32) {
 283                         errno = EINVAL;
 284                         return -1;
 285                 }
 286                 ios[count].iov_base = (void*)s;
 287                 ios[count].iov_len = strlen(s);
 288                 count++;
 289                 s = va_arg(args, const char *);
 290         }
 291         va_end(args);
 292         return websock_text_v(ws->ws, 1, ios, count);
 293 }
 294
 295 /*
 296  * Sends a text data described in the 'count' 'iovec' to the endpoint of 'ws'.
 297  * Returns 0 on success or -1 in case of error.
 298  */
 299 int afb_ws_text_v(struct afb_ws *ws, const struct iovec *iovec, int count)
 300 {
 301         if (ws->ws == NULL) {
 302                 /* disconnected */
 303                 errno = EPIPE;
 304                 return -1;
 305         }
 306         return websock_text_v(ws->ws, 1, iovec, count);
 307 }
 308
 309 /*
 310  * Sends a binary 'data' of 'length' to the endpoint of 'ws'.
 311  * Returns 0 on success or -1 in case of error.
 312  */
 313 int afb_ws_binary(struct afb_ws *ws, const void *data, size_t length)
 314 {
 315         if (ws->ws == NULL) {
 316                 /* disconnected */
 317                 errno = EPIPE;
 318                 return -1;
 319         }
 320         return websock_binary(ws->ws, 1, data, length);
 321 }
 322
 323 /*
 324  * Sends a binary data described in the 'count' 'iovec' to the endpoint of 'ws'.
 325  * Returns 0 on success or -1 in case of error.
 326  */
 327 int afb_ws_binary_v(struct afb_ws *ws, const struct iovec *iovec, int count)
 328 {
 329         if (ws->ws == NULL) {
 330                 /* disconnected */
 331                 errno = EPIPE;
 332                 return -1;
 333         }
 334         return websock_binary_v(ws->ws, 1, iovec, count);
 335 }
 336
 337 /*
 338  * callback for writing data
 339  */
 340 static ssize_t aws_writev(struct afb_ws *ws, const struct iovec *iov, int iovcnt)
 341 {
 342         int i;
 343         ssize_t rc, sz, dsz;
 344         struct iovec *iov2;
 345         struct pollfd pfd;
 346
 347         /* compute the size */
 348         dsz = 0;
 349         i = 0;
 350         while (i < iovcnt) {
 351                 dsz += iov[i++].iov_len;
 352                 if (dsz < 0) {
 353                         errno = EINVAL;
 354                         return -1;
 355                 }
 356         }
 357         if (dsz == 0)
 358                 return 0;
 359
 360         /* write the data */
 361         iov2 = (struct iovec*)iov;
 362         sz = dsz;
 363         for (;;) {
 364                 rc = writev(ws->fd, iov2, iovcnt);
 365                 if (rc < 0) {
 366                         if (errno == EINTR)
 367                                 continue;
 368                         if (errno != EAGAIN)
 369                                 return -1;
 370                 } else {
 371                         dsz -= rc;
 372                         if (dsz == 0)
 373                                 return sz;
 374
 375                         i = 0;
 376                         while (rc >= (ssize_t)iov2[i].iov_len)
 377                                 rc -= (ssize_t)iov2[i++].iov_len;
 378
 379                         iovcnt -= i;
 380                         if (iov2 != iov)
 381                                 iov2 += i;
 382                         else {
 383                                 iov += i;
 384                                 iov2 = alloca(iovcnt * sizeof *iov2);
 385                                 for (i = 0 ; i < iovcnt ; i++)
 386                                         iov2[i] = iov[i];
 387                         }
 388                         iov2->iov_base += rc;
 389                         iov2->iov_len -= rc;
 390                 }
 391                 pfd.fd = ws->fd;
 392                 pfd.events = POLLOUT;
 393                 poll(&pfd, 1, 10);
 394         }
 395 }
 396
 397 /*
 398  * callback for reading data
 399  */
 400 static ssize_t aws_readv(struct afb_ws *ws, const struct iovec *iov, int iovcnt)
 401 {
 402         ssize_t rc;
 403         do {
 404                 rc = readv(ws->fd, iov, iovcnt);
 405         } while(rc == -1 && errno == EINTR);
 406         if (rc == 0) {
 407                 errno = EPIPE;
 408                 rc = -1;
 409         }
 410         return rc;
 411 }
 412
 413 /*
 414  * callback on incoming data
 415  */
 416 static void aws_on_readable(struct afb_ws *ws)
 417 {
 418         int rc;
 419
 420         assert(ws->ws != NULL);
 421         rc = websock_dispatch(ws->ws, 0);
 422         if (rc < 0 && errno == EPIPE)
 423                 afb_ws_hangup(ws);
 424 }
 425
 426 /*
 427  * Reads from the websocket handled by 'ws' data of length 'size'
 428  * and append it to the current buffer of 'ws'.
 429  * Returns 0 in case of error or 1 in case of success.
 430  */
 431 static int aws_read(struct afb_ws *ws, size_t size)
 432 {
 433         struct pollfd pfd;
 434         ssize_t sz;
 435         char *buffer;
 436
 437         if (size != 0 || ws->buffer.buffer == NULL) {
 438                 buffer = realloc(ws->buffer.buffer, ws->buffer.size + size + 1);
 439                 if (buffer == NULL)
 440                         return 0;
 441                 ws->buffer.buffer = buffer;
 442                 while (size != 0) {
 443                         sz = websock_read(ws->ws, &buffer[ws->buffer.size], size);
 444                         if (sz < 0) {
 445                                 if (errno != EAGAIN)
 446                                         return 0;
 447                                 pfd.fd = ws->fd;
 448                                 pfd.events = POLLIN;
 449                                 poll(&pfd, 1, 10); /* TODO: make fully asynchronous websockets */
 450                         } else {
 451                                 ws->buffer.size += (size_t)sz;
 452                                 size -= (size_t)sz;
 453                         }
 454                 }
 455         }
 456         return 1;
 457 }
 458
 459 /*
 460  * Callback when 'close' command received from 'ws' with 'code' and 'size'.
 461  */
 462 static void aws_on_close(struct afb_ws *ws, uint16_t code, size_t size)
 463 {
 464         struct buf b;
 465
 466         ws->state = waiting;
 467         aws_clear_buffer(ws);
 468         if (ws->itf->on_close == NULL) {
 469                 websock_drop(ws->ws);
 470                 afb_ws_hangup(ws);
 471         } else if (!aws_read(ws, size))
 472                 ws->itf->on_close(ws->closure, code, NULL, 0);
 473         else {
 474                 b = aws_pick_buffer(ws);
 475                 ws->itf->on_close(ws->closure, code, b.buffer, b.size);
 476         }
 477 }
 478
 479 /*
 480  * Drops any incoming data and send an error of 'code'
 481  */
 482 static void aws_drop_error(struct afb_ws *ws, uint16_t code)
 483 {
 484         ws->state = waiting;
 485         aws_clear_buffer(ws);
 486         websock_drop(ws->ws);
 487         websock_error(ws->ws, code, NULL, 0);
 488 }
 489
 490 /*
 491  * Reads either text or binary data of 'size' from 'ws' eventually 'last'.
 492  */
 493 static void aws_continue(struct afb_ws *ws, int last, size_t size)
 494 {
 495         struct buf b;
 496         int istxt;
 497
 498         if (!aws_read(ws, size))
 499                 aws_drop_error(ws, WEBSOCKET_CODE_ABNORMAL);
 500         else if (last) {
 501                 istxt = ws->state == reading_text;
 502                 ws->state = waiting;
 503                 b = aws_pick_buffer(ws);
 504                 (istxt ? ws->itf->on_text : ws->itf->on_binary)(ws->closure, b.buffer, b.size);
 505         }
 506 }
 507
 508 /*
 509  * Callback when 'text' message received from 'ws' with 'size' and possibly 'last'.
 510  */
 511 static void aws_on_text(struct afb_ws *ws, int last, size_t size)
 512 {
 513         if (ws->state != waiting)
 514                 aws_drop_error(ws, WEBSOCKET_CODE_PROTOCOL_ERROR);
 515         else if (ws->itf->on_text == NULL)
 516                 aws_drop_error(ws, WEBSOCKET_CODE_CANT_ACCEPT);
 517         else {
 518                 ws->state = reading_text;
 519                 aws_continue(ws, last, size);
 520         }
 521 }
 522
 523 /*
 524  * Callback when 'binary' message received from 'ws' with 'size' and possibly 'last'.
 525  */
 526 static void aws_on_binary(struct afb_ws *ws, int last, size_t size)
 527 {
 528         if (ws->state != waiting)
 529                 aws_drop_error(ws, WEBSOCKET_CODE_PROTOCOL_ERROR);
 530         else if (ws->itf->on_binary == NULL)
 531                 aws_drop_error(ws, WEBSOCKET_CODE_CANT_ACCEPT);
 532         else {
 533                 ws->state = reading_binary;
 534                 aws_continue(ws, last, size);
 535         }
 536 }
 537
 538 /*
 539  * Callback when 'continue' command received from 'ws' with 'code' and 'size'.
 540  */
 541 static void aws_on_continue(struct afb_ws *ws, int last, size_t size)
 542 {
 543         if (ws->state == waiting)
 544                 aws_drop_error(ws, WEBSOCKET_CODE_PROTOCOL_ERROR);
 545         else
 546                 aws_continue(ws, last, size);
 547 }
 548
 549 /*
 550  * Callback when 'close' command is sent to 'ws' with 'code' and 'size'.
 551  */
 552 static void aws_on_error(struct afb_ws *ws, uint16_t code, const void *data, size_t size)
 553 {
 554         if (ws->itf->on_error != NULL)
 555                 ws->itf->on_error(ws->closure, code, data, size);
 556         else
 557                 afb_ws_hangup(ws);
 558 }
 559
 560