2 * Copyright (C) 2011-2014 Felix Fietkau <nbd@openwrt.org>
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
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.
15 #include <sys/types.h>
17 #include <sys/socket.h>
23 #include <libubox/usock.h>
24 #include <libubox/blob.h>
25 #include <libubox/blobmsg.h>
28 #include "libubus-internal.h"
30 #define STATIC_IOV(_var) { .iov_base = (char *) &(_var), .iov_len = sizeof(_var) }
32 #define UBUS_MSGBUF_REDUCTION_INTERVAL 16
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
},
44 static struct blob_attr
*attrbuf
[UBUS_ATTR_MAX
];
46 __hidden
struct blob_attr
**ubus_parse_msg(struct blob_attr
*msg
, size_t len
)
48 blob_parse_untrusted(msg
, len
, attrbuf
, ubus_policy
, UBUS_ATTR_MAX
);
52 static void wait_data(int fd
, bool write
)
54 struct pollfd pfd
= { .fd
= fd
};
56 pfd
.events
= write
? POLLOUT
: POLLIN
;
60 static int writev_retry(int fd
, struct iovec
*iov
, int iov_len
, int sock_fd
)
62 uint8_t fd_buf
[CMSG_SPACE(sizeof(int))] = { 0 };
63 struct msghdr msghdr
= { 0 };
69 msghdr
.msg_iovlen
= iov_len
,
70 msghdr
.msg_control
= fd_buf
;
71 msghdr
.msg_controllen
= sizeof(fd_buf
);
73 cmsg
= CMSG_FIRSTHDR(&msghdr
);
74 cmsg
->cmsg_type
= SCM_RIGHTS
;
75 cmsg
->cmsg_level
= SOL_SOCKET
;
76 cmsg
->cmsg_len
= CMSG_LEN(sizeof(int));
78 pfd
= (int *) CMSG_DATA(cmsg
);
79 msghdr
.msg_controllen
= cmsg
->cmsg_len
;
85 msghdr
.msg_control
= NULL
;
86 msghdr
.msg_controllen
= 0;
91 cur_len
= sendmsg(fd
, &msghdr
, 0);
109 while (cur_len
>= (ssize_t
) iov
->iov_len
) {
110 cur_len
-= iov
->iov_len
;
116 iov
->iov_base
+= cur_len
;
117 iov
->iov_len
-= cur_len
;
118 msghdr
.msg_iov
= iov
;
119 msghdr
.msg_iovlen
= iov_len
;
122 /* Should never reach here */
126 int __hidden
ubus_send_msg(struct ubus_context
*ctx
, uint32_t seq
,
127 struct blob_attr
*msg
, int cmd
, uint32_t peer
, int fd
)
129 struct ubus_msghdr hdr
;
130 struct iovec iov
[2] = {
137 hdr
.seq
= cpu_to_be16(seq
);
138 hdr
.peer
= cpu_to_be32(peer
);
141 blob_buf_init(&b
, 0);
145 iov
[1].iov_base
= (char *) msg
;
146 iov
[1].iov_len
= blob_raw_len(msg
);
148 ret
= writev_retry(ctx
->sock
.fd
, iov
, ARRAY_SIZE(iov
), fd
);
150 ctx
->sock
.eof
= true;
158 static int recv_retry(struct ubus_context
*ctx
, struct iovec
*iov
, bool wait
, int *recv_fd
)
160 uint8_t fd_buf
[CMSG_SPACE(sizeof(int))] = { 0 };
161 struct msghdr msghdr
= { 0 };
162 struct cmsghdr
*cmsg
;
170 msghdr
.msg_iov
= iov
,
171 msghdr
.msg_iovlen
= 1,
172 msghdr
.msg_control
= fd_buf
;
173 msghdr
.msg_controllen
= sizeof(fd_buf
);
175 cmsg
= CMSG_FIRSTHDR(&msghdr
);
176 cmsg
->cmsg_type
= SCM_RIGHTS
;
177 cmsg
->cmsg_level
= SOL_SOCKET
;
178 cmsg
->cmsg_len
= CMSG_LEN(sizeof(int));
180 pfd
= (int *) CMSG_DATA(cmsg
);
182 while (iov
->iov_len
> 0) {
184 msghdr
.msg_control
= fd_buf
;
185 msghdr
.msg_controllen
= cmsg
->cmsg_len
;
187 msghdr
.msg_control
= NULL
;
188 msghdr
.msg_controllen
= 0;
192 bytes
= recvmsg(fd
, &msghdr
, 0);
213 iov
->iov_len
-= bytes
;
214 iov
->iov_base
+= bytes
;
217 if (iov
->iov_len
> 0)
218 wait_data(fd
, false);
224 bool ubus_validate_hdr(struct ubus_msghdr
*hdr
)
226 struct blob_attr
*data
= (struct blob_attr
*) (hdr
+ 1);
228 if (hdr
->version
!= 0)
231 if (blob_raw_len(data
) < sizeof(*data
))
234 if (blob_pad_len(data
) > UBUS_MAX_MSGLEN
)
240 static bool alloc_msg_buf(struct ubus_context
*ctx
, int len
)
243 int buf_len
= ctx
->msgbuf_data_len
;
246 if (!ctx
->msgbuf
.data
)
249 rem
= (len
% UBUS_MSG_CHUNK_SIZE
);
251 len
+= UBUS_MSG_CHUNK_SIZE
- rem
;
254 ++ctx
->msgbuf_reduction_counter
> UBUS_MSGBUF_REDUCTION_INTERVAL
) {
255 ctx
->msgbuf_reduction_counter
= 0;
262 ptr
= realloc(ctx
->msgbuf
.data
, len
);
266 ctx
->msgbuf
.data
= ptr
;
267 ctx
->msgbuf_data_len
= len
;
271 static bool get_next_msg(struct ubus_context
*ctx
, int *recv_fd
)
274 struct ubus_msghdr hdr
;
275 struct blob_attr data
;
277 struct iovec iov
= STATIC_IOV(hdrbuf
);
281 /* receive header + start attribute */
282 r
= recv_retry(ctx
, &iov
, false, recv_fd
);
285 ctx
->sock
.eof
= true;
290 hdrbuf
.hdr
.seq
= be16_to_cpu(hdrbuf
.hdr
.seq
);
291 hdrbuf
.hdr
.peer
= be32_to_cpu(hdrbuf
.hdr
.peer
);
293 if (!ubus_validate_hdr(&hdrbuf
.hdr
))
296 len
= blob_raw_len(&hdrbuf
.data
);
297 if (!alloc_msg_buf(ctx
, len
))
300 memcpy(&ctx
->msgbuf
.hdr
, &hdrbuf
.hdr
, sizeof(hdrbuf
.hdr
));
301 memcpy(ctx
->msgbuf
.data
, &hdrbuf
.data
, sizeof(hdrbuf
.data
));
303 iov
.iov_base
= (char *)ctx
->msgbuf
.data
+ sizeof(hdrbuf
.data
);
304 iov
.iov_len
= blob_len(ctx
->msgbuf
.data
);
305 if (iov
.iov_len
> 0 &&
306 recv_retry(ctx
, &iov
, true, NULL
) <= 0)
312 void __hidden
ubus_handle_data(struct uloop_fd
*u
, unsigned int events
)
314 struct ubus_context
*ctx
= container_of(u
, struct ubus_context
, sock
);
317 while (get_next_msg(ctx
, &recv_fd
)) {
318 ubus_process_msg(ctx
, &ctx
->msgbuf
, recv_fd
);
319 if (uloop_cancelling() || ctx
->cancel_poll
)
324 ctx
->connection_lost(ctx
);
327 void __hidden
ubus_poll_data(struct ubus_context
*ctx
, int timeout
)
329 struct pollfd pfd
= {
331 .events
= POLLIN
| POLLERR
,
334 ctx
->cancel_poll
= false;
335 poll(&pfd
, 1, timeout
? timeout
: -1);
336 ubus_handle_data(&ctx
->sock
, ULOOP_READ
);
340 ubus_refresh_state(struct ubus_context
*ctx
)
342 struct ubus_object
*obj
, *tmp
;
343 struct ubus_object
**objs
;
346 /* clear all type IDs, they need to be registered again */
347 avl_for_each_element(&ctx
->objects
, obj
, avl
)
351 /* push out all objects again */
352 objs
= alloca(ctx
->objects
.count
* sizeof(*objs
));
353 avl_remove_all_elements(&ctx
->objects
, obj
, avl
, tmp
) {
358 for (n
= i
, i
= 0; i
< n
; i
++)
359 ubus_add_object(ctx
, objs
[i
]);
362 int ubus_reconnect(struct ubus_context
*ctx
, const char *path
)
365 struct ubus_msghdr hdr
;
366 struct blob_attr data
;
368 struct blob_attr
*buf
;
369 int ret
= UBUS_STATUS_UNKNOWN_ERROR
;
372 path
= UBUS_UNIX_SOCKET
;
374 if (ctx
->sock
.fd
>= 0) {
375 if (ctx
->sock
.registered
)
376 uloop_fd_delete(&ctx
->sock
);
381 ctx
->sock
.eof
= false;
382 ctx
->sock
.error
= false;
383 ctx
->sock
.fd
= usock(USOCK_UNIX
, path
, NULL
);
384 if (ctx
->sock
.fd
< 0)
385 return UBUS_STATUS_CONNECTION_FAILED
;
387 if (read(ctx
->sock
.fd
, &hdr
, sizeof(hdr
)) != sizeof(hdr
))
390 if (!ubus_validate_hdr(&hdr
.hdr
))
393 if (hdr
.hdr
.type
!= UBUS_MSG_HELLO
)
396 buf
= calloc(1, blob_raw_len(&hdr
.data
));
400 memcpy(buf
, &hdr
.data
, sizeof(hdr
.data
));
401 if (read(ctx
->sock
.fd
, blob_data(buf
), blob_len(buf
)) != (ssize_t
) blob_len(buf
))
404 ctx
->local_id
= hdr
.hdr
.peer
;
408 ret
= UBUS_STATUS_OK
;
409 fcntl(ctx
->sock
.fd
, F_SETFL
, fcntl(ctx
->sock
.fd
, F_GETFL
) | O_NONBLOCK
| O_CLOEXEC
);
411 ubus_refresh_state(ctx
);