libubus-io.c

   1 /*
   2  * Copyright (C) 2011-2014 Felix Fietkau <nbd@openwrt.org>
   3  *
   4  * This program is free software; you can redistribute it and/or modify
   5  * it under the terms of the GNU Lesser General Public License version 2.1
   6  * as published by the Free Software Foundation
   7  *
   8  * This program is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11  * GNU General Public License for more details.
  12  */
  13
  14 #define _GNU_SOURCE
  15 #include <sys/types.h>
  16 #include <sys/uio.h>
  17 #include <sys/socket.h>
  18
  19 #include <unistd.h>
  20 #include <fcntl.h>
  21 #include <poll.h>
  22
  23 #include <libubox/usock.h>
  24 #include <libubox/blob.h>
  25 #include <libubox/blobmsg.h>
  26
  27 #include "libubus.h"
  28 #include "libubus-internal.h"
  29
  30 #define STATIC_IOV(_var) { .iov_base = (char *) &(_var), .iov_len = sizeof(_var) }
  31
  32 #define UBUS_MSGBUF_REDUCTION_INTERVAL  16
  33
  34 static const struct blob_attr_info ubus_policy[UBUS_ATTR_MAX] = {
  35         [UBUS_ATTR_STATUS] = { .type = BLOB_ATTR_INT32 },
  36         [UBUS_ATTR_OBJID] = { .type = BLOB_ATTR_INT32 },
  37         [UBUS_ATTR_OBJPATH] = { .type = BLOB_ATTR_STRING },
  38         [UBUS_ATTR_METHOD] = { .type = BLOB_ATTR_STRING },
  39         [UBUS_ATTR_ACTIVE] = { .type = BLOB_ATTR_INT8 },
  40         [UBUS_ATTR_NO_REPLY] = { .type = BLOB_ATTR_INT8 },
  41         [UBUS_ATTR_SUBSCRIBERS] = { .type = BLOB_ATTR_NESTED },
  42 };
  43
  44 static struct blob_attr *attrbuf[UBUS_ATTR_MAX];
  45
  46 __hidden struct blob_attr **ubus_parse_msg(struct blob_attr *msg)
  47 {
  48         blob_parse(msg, attrbuf, ubus_policy, UBUS_ATTR_MAX);
  49         return attrbuf;
  50 }
  51
  52 static void wait_data(int fd, bool write)
  53 {
  54         struct pollfd pfd = { .fd = fd };
  55
  56         pfd.events = write ? POLLOUT : POLLIN;
  57         poll(&pfd, 1, -1);
  58 }
  59
  60 static int writev_retry(int fd, struct iovec *iov, int iov_len, int sock_fd)
  61 {
  62         static struct {
  63                 struct cmsghdr h;
  64                 int fd;
  65         } fd_buf = {
  66                 .h = {
  67                         .cmsg_len = sizeof(fd_buf),
  68                         .cmsg_level = SOL_SOCKET,
  69                         .cmsg_type = SCM_RIGHTS,
  70                 }
  71         };
  72         struct msghdr msghdr = {
  73                 .msg_iov = iov,
  74                 .msg_iovlen = iov_len,
  75                 .msg_control = &fd_buf,
  76                 .msg_controllen = sizeof(fd_buf),
  77         };
  78         int len = 0;
  79
  80         do {
  81                 int cur_len;
  82
  83                 if (sock_fd < 0) {
  84                         msghdr.msg_control = NULL;
  85                         msghdr.msg_controllen = 0;
  86                 } else {
  87                         fd_buf.fd = sock_fd;
  88                 }
  89
  90                 cur_len = sendmsg(fd, &msghdr, 0);
  91                 if (cur_len < 0) {
  92                         switch(errno) {
  93                         case EAGAIN:
  94                                 wait_data(fd, true);
  95                                 break;
  96                         case EINTR:
  97                                 break;
  98                         default:
  99                                 return -1;
 100                         }
 101                         continue;
 102                 }
 103
 104                 if (len > 0)
 105                         sock_fd = -1;
 106
 107                 len += cur_len;
 108                 while (cur_len >= iov->iov_len) {
 109                         cur_len -= iov->iov_len;
 110                         iov_len--;
 111                         iov++;
 112                         if (!iov_len)
 113                                 return len;
 114                 }
 115                 iov->iov_base += cur_len;
 116                 iov->iov_len -= cur_len;
 117                 msghdr.msg_iov = iov;
 118                 msghdr.msg_iovlen = iov_len;
 119         } while (1);
 120
 121         /* Should never reach here */
 122         return -1;
 123 }
 124
 125 int __hidden ubus_send_msg(struct ubus_context *ctx, uint32_t seq,
 126                            struct blob_attr *msg, int cmd, uint32_t peer, int fd)
 127 {
 128         struct ubus_msghdr hdr;
 129         struct iovec iov[2] = {
 130                 STATIC_IOV(hdr)
 131         };
 132         int ret;
 133
 134         hdr.version = 0;
 135         hdr.type = cmd;
 136         hdr.seq = cpu_to_be16(seq);
 137         hdr.peer = cpu_to_be32(peer);
 138
 139         if (!msg) {
 140                 blob_buf_init(&b, 0);
 141                 msg = b.head;
 142         }
 143
 144         iov[1].iov_base = (char *) msg;
 145         iov[1].iov_len = blob_raw_len(msg);
 146
 147         ret = writev_retry(ctx->sock.fd, iov, ARRAY_SIZE(iov), fd);
 148         if (ret < 0)
 149                 ctx->sock.eof = true;
 150
 151         if (fd >= 0)
 152                 close(fd);
 153
 154         return ret;
 155 }
 156
 157 static int recv_retry(int fd, struct iovec *iov, bool wait, int *recv_fd)
 158 {
 159         int bytes, total = 0;
 160         static struct {
 161                 struct cmsghdr h;
 162                 int fd;
 163         } fd_buf = {
 164                 .h = {
 165                         .cmsg_type = SCM_RIGHTS,
 166                         .cmsg_level = SOL_SOCKET,
 167                         .cmsg_len = sizeof(fd_buf),
 168                 },
 169         };
 170         struct msghdr msghdr = {
 171                 .msg_iov = iov,
 172                 .msg_iovlen = 1,
 173         };
 174
 175         while (iov->iov_len > 0) {
 176                 if (wait)
 177                         wait_data(fd, false);
 178
 179                 if (recv_fd) {
 180                         msghdr.msg_control = &fd_buf;
 181                         msghdr.msg_controllen = sizeof(fd_buf);
 182                 } else {
 183                         msghdr.msg_control = NULL;
 184                         msghdr.msg_controllen = 0;
 185                 }
 186
 187                 fd_buf.fd = -1;
 188                 bytes = recvmsg(fd, &msghdr, 0);
 189                 if (!bytes)
 190                         return -1;
 191
 192                 if (bytes < 0) {
 193                         bytes = 0;
 194                         if (uloop_cancelled)
 195                                 return 0;
 196                         if (errno == EINTR)
 197                                 continue;
 198
 199                         if (errno != EAGAIN)
 200                                 return -1;
 201                 }
 202                 if (!wait && !bytes)
 203                         return 0;
 204
 205                 if (recv_fd)
 206                         *recv_fd = fd_buf.fd;
 207
 208                 recv_fd = NULL;
 209
 210                 wait = true;
 211                 iov->iov_len -= bytes;
 212                 iov->iov_base += bytes;
 213                 total += bytes;
 214         }
 215
 216         return total;
 217 }
 218
 219 static bool ubus_validate_hdr(struct ubus_msghdr *hdr)
 220 {
 221         struct blob_attr *data = (struct blob_attr *) (hdr + 1);
 222
 223         if (hdr->version != 0)
 224                 return false;
 225
 226         if (blob_raw_len(data) < sizeof(*data))
 227                 return false;
 228
 229         if (blob_pad_len(data) > UBUS_MAX_MSGLEN)
 230                 return false;
 231
 232         return true;
 233 }
 234
 235 static bool alloc_msg_buf(struct ubus_context *ctx, int len)
 236 {
 237         void *ptr;
 238         int buf_len = ctx->msgbuf_data_len;
 239         int rem;
 240
 241         if (!ctx->msgbuf.data)
 242                 buf_len = 0;
 243
 244         rem = (len % UBUS_MSG_CHUNK_SIZE);
 245         if (rem > 0)
 246                 len += UBUS_MSG_CHUNK_SIZE - rem;
 247
 248         if (len < buf_len &&
 249             ++ctx->msgbuf_reduction_counter > UBUS_MSGBUF_REDUCTION_INTERVAL) {
 250                 ctx->msgbuf_reduction_counter = 0;
 251                 buf_len = 0;
 252         }
 253
 254         if (len <= buf_len)
 255                 return true;
 256
 257         ptr = realloc(ctx->msgbuf.data, len);
 258         if (!ptr)
 259                 return false;
 260
 261         ctx->msgbuf.data = ptr;
 262         ctx->msgbuf_data_len = len;
 263         return true;
 264 }
 265
 266 static bool get_next_msg(struct ubus_context *ctx, int *recv_fd)
 267 {
 268         struct {
 269                 struct ubus_msghdr hdr;
 270                 struct blob_attr data;
 271         } hdrbuf;
 272         struct iovec iov = STATIC_IOV(hdrbuf);
 273         int len;
 274         int r;
 275
 276         /* receive header + start attribute */
 277         r = recv_retry(ctx->sock.fd, &iov, false, recv_fd);
 278         if (r <= 0) {
 279                 if (r < 0)
 280                         ctx->sock.eof = true;
 281
 282                 return false;
 283         }
 284
 285         hdrbuf.hdr.seq = be16_to_cpu(hdrbuf.hdr.seq);
 286         hdrbuf.hdr.peer = be32_to_cpu(hdrbuf.hdr.peer);
 287
 288         if (!ubus_validate_hdr(&hdrbuf.hdr))
 289                 return false;
 290
 291         len = blob_raw_len(&hdrbuf.data);
 292         if (!alloc_msg_buf(ctx, len))
 293                 return false;
 294
 295         memcpy(&ctx->msgbuf.hdr, &hdrbuf.hdr, sizeof(hdrbuf.hdr));
 296         memcpy(ctx->msgbuf.data, &hdrbuf.data, sizeof(hdrbuf.data));
 297
 298         iov.iov_base = (char *)ctx->msgbuf.data + sizeof(hdrbuf.data);
 299         iov.iov_len = blob_len(ctx->msgbuf.data);
 300         if (iov.iov_len > 0 &&
 301             recv_retry(ctx->sock.fd, &iov, true, NULL) <= 0)
 302                 return false;
 303
 304         return true;
 305 }
 306
 307 void __hidden ubus_handle_data(struct uloop_fd *u, unsigned int events)
 308 {
 309         struct ubus_context *ctx = container_of(u, struct ubus_context, sock);
 310         int recv_fd = -1;
 311
 312         while (get_next_msg(ctx, &recv_fd)) {
 313                 ubus_process_msg(ctx, &ctx->msgbuf, recv_fd);
 314                 if (uloop_cancelled)
 315                         break;
 316         }
 317
 318         if (u->eof)
 319                 ctx->connection_lost(ctx);
 320 }
 321
 322 void __hidden ubus_poll_data(struct ubus_context *ctx, int timeout)
 323 {
 324         struct pollfd pfd = {
 325                 .fd = ctx->sock.fd,
 326                 .events = POLLIN | POLLERR,
 327         };
 328
 329         poll(&pfd, 1, timeout ? timeout : -1);
 330         ubus_handle_data(&ctx->sock, ULOOP_READ);
 331 }
 332
 333 static void
 334 ubus_refresh_state(struct ubus_context *ctx)
 335 {
 336         struct ubus_object *obj, *tmp;
 337         struct ubus_object **objs;
 338         int n, i = 0;
 339
 340         /* clear all type IDs, they need to be registered again */
 341         avl_for_each_element(&ctx->objects, obj, avl)
 342                 if (obj->type)
 343                         obj->type->id = 0;
 344
 345         /* push out all objects again */
 346         objs = alloca(ctx->objects.count * sizeof(*objs));
 347         avl_remove_all_elements(&ctx->objects, obj, avl, tmp) {
 348                 objs[i++] = obj;
 349                 obj->id = 0;
 350         }
 351
 352         for (n = i, i = 0; i < n; i++)
 353                 ubus_add_object(ctx, objs[i]);
 354 }
 355
 356 int ubus_reconnect(struct ubus_context *ctx, const char *path)
 357 {
 358         struct {
 359                 struct ubus_msghdr hdr;
 360                 struct blob_attr data;
 361         } hdr;
 362         struct blob_attr *buf;
 363         int ret = UBUS_STATUS_UNKNOWN_ERROR;
 364
 365         if (!path)
 366                 path = UBUS_UNIX_SOCKET;
 367
 368         if (ctx->sock.fd >= 0) {
 369                 if (ctx->sock.registered)
 370                         uloop_fd_delete(&ctx->sock);
 371
 372                 close(ctx->sock.fd);
 373         }
 374
 375         ctx->sock.fd = usock(USOCK_UNIX, path, NULL);
 376         if (ctx->sock.fd < 0)
 377                 return UBUS_STATUS_CONNECTION_FAILED;
 378
 379         if (read(ctx->sock.fd, &hdr, sizeof(hdr)) != sizeof(hdr))
 380                 goto out_close;
 381
 382         if (!ubus_validate_hdr(&hdr.hdr))
 383                 goto out_close;
 384
 385         if (hdr.hdr.type != UBUS_MSG_HELLO)
 386                 goto out_close;
 387
 388         buf = calloc(1, blob_raw_len(&hdr.data));
 389         if (!buf)
 390                 goto out_close;
 391
 392         memcpy(buf, &hdr.data, sizeof(hdr.data));
 393         if (read(ctx->sock.fd, blob_data(buf), blob_len(buf)) != blob_len(buf))
 394                 goto out_free;
 395
 396         ctx->local_id = hdr.hdr.peer;
 397         if (!ctx->local_id)
 398                 goto out_free;
 399
 400         ret = UBUS_STATUS_OK;
 401         fcntl(ctx->sock.fd, F_SETFL, fcntl(ctx->sock.fd, F_GETFL) | O_NONBLOCK | O_CLOEXEC);
 402
 403         ubus_refresh_state(ctx);
 404
 405 out_free:
 406         free(buf);
 407 out_close:
 408         if (ret)
 409                 close(ctx->sock.fd);
 410
 411         return ret;
 412 }