]> git.ipfire.org Git - thirdparty/systemd.git/blob - src/shared/varlink.c
projects / thirdparty / systemd.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
blame | history | raw | HEAD
Merge pull request #27504 from mrc0mmand/fuzz-manager-serialize
[thirdparty/systemd.git] / src / shared / varlink.c
1 /* SPDX-License-Identifier: LGPL-2.1-or-later */
2
3 #include <malloc.h>
4 #include <poll.h>
5
6 #include "alloc-util.h"
7 #include "errno-util.h"
8 #include "fd-util.h"
9 #include "glyph-util.h"
10 #include "hashmap.h"
11 #include "io-util.h"
12 #include "list.h"
13 #include "process-util.h"
14 #include "selinux-util.h"
15 #include "serialize.h"
16 #include "set.h"
17 #include "socket-util.h"
18 #include "string-table.h"
19 #include "string-util.h"
20 #include "strv.h"
21 #include "time-util.h"
22 #include "umask-util.h"
23 #include "user-util.h"
24 #include "varlink.h"
25 #include "varlink-internal.h"
26
27 #define VARLINK_DEFAULT_CONNECTIONS_MAX 4096U
28 #define VARLINK_DEFAULT_CONNECTIONS_PER_UID_MAX 1024U
29
30 #define VARLINK_DEFAULT_TIMEOUT_USEC (45U*USEC_PER_SEC)
31 #define VARLINK_BUFFER_MAX (16U*1024U*1024U)
32 #define VARLINK_READ_SIZE (64U*1024U)
33
34 typedef enum VarlinkState {
35 /* Client side states */
36 VARLINK_IDLE_CLIENT,
37 VARLINK_AWAITING_REPLY,
38 VARLINK_AWAITING_REPLY_MORE,
39 VARLINK_CALLING,
40 VARLINK_CALLED,
41 VARLINK_PROCESSING_REPLY,
42
43 /* Server side states */
44 VARLINK_IDLE_SERVER,
45 VARLINK_PROCESSING_METHOD,
46 VARLINK_PROCESSING_METHOD_MORE,
47 VARLINK_PROCESSING_METHOD_ONEWAY,
48 VARLINK_PROCESSED_METHOD,
49 VARLINK_PENDING_METHOD,
50 VARLINK_PENDING_METHOD_MORE,
51
52 /* Common states (only during shutdown) */
53 VARLINK_PENDING_DISCONNECT,
54 VARLINK_PENDING_TIMEOUT,
55 VARLINK_PROCESSING_DISCONNECT,
56 VARLINK_PROCESSING_TIMEOUT,
57 VARLINK_PROCESSING_FAILURE,
58 VARLINK_DISCONNECTED,
59
60 _VARLINK_STATE_MAX,
61 _VARLINK_STATE_INVALID = -EINVAL,
62 } VarlinkState;
63
64 /* Tests whether we are not yet disconnected. Note that this is true during all states where the connection
65 * is still good for something, and false only when it's dead for good. This means: when we are
66 * asynchronously connecting to a peer and the connect() is still pending, then this will return 'true', as
67 * the connection is still good, and we are likely to be able to properly operate on it soon. */
68 #define VARLINK_STATE_IS_ALIVE(state) \
69 IN_SET(state, \
70 VARLINK_IDLE_CLIENT, \
71 VARLINK_AWAITING_REPLY, \
72 VARLINK_AWAITING_REPLY_MORE, \
73 VARLINK_CALLING, \
74 VARLINK_CALLED, \
75 VARLINK_PROCESSING_REPLY, \
76 VARLINK_IDLE_SERVER, \
77 VARLINK_PROCESSING_METHOD, \
78 VARLINK_PROCESSING_METHOD_MORE, \
79 VARLINK_PROCESSING_METHOD_ONEWAY, \
80 VARLINK_PROCESSED_METHOD, \
81 VARLINK_PENDING_METHOD, \
82 VARLINK_PENDING_METHOD_MORE)
83
84 typedef struct VarlinkJsonQueueItem VarlinkJsonQueueItem;
85
86 /* A queued message we shall write into the socket, along with the file descriptors to send at the same
87 * time. This queue item binds them together so that message/fd boundaries are maintained throughout the
88 * whole pipeline. */
89 struct VarlinkJsonQueueItem {
90 LIST_FIELDS(VarlinkJsonQueueItem, queue);
91 JsonVariant *data;
92 size_t n_fds;
93 int fds[];
94 };
95
96 struct Varlink {
97 unsigned n_ref;
98
99 VarlinkServer *server;
100
101 VarlinkState state;
102 bool connecting; /* This boolean indicates whether the socket fd we are operating on is currently
103 * processing an asynchronous connect(). In that state we watch the socket for
104 * EPOLLOUT, but we refrain from calling read() or write() on the socket as that
105 * will trigger ENOTCONN. Note that this boolean is kept separate from the
106 * VarlinkState above on purpose: while the connect() is still not complete we
107 * already want to allow queuing of messages and similar. Thus it's nice to keep
108 * these two state concepts separate: the VarlinkState encodes what our own view of
109 * the connection is, i.e. whether we think it's a server, a client, and has
110 * something queued already, while 'connecting' tells us a detail about the
111 * transport used below, that should have no effect on how we otherwise accept and
112 * process operations from the user.
113 *
114 * Or to say this differently: VARLINK_STATE_IS_ALIVE(state) tells you whether the
115 * connection is good to use, even if it might not be fully connected
116 * yet. connecting=true then informs you that actually we are still connecting, and
117 * the connection is actually not established yet and thus any requests you enqueue
118 * now will still work fine but will be queued only, not sent yet, but that
119 * shouldn't stop you from using the connection, since eventually whatever you queue
120 * *will* be sent.
121 *
122 * Or to say this even differently: 'state' is a high-level ("application layer"
123 * high, if you so will) state, while 'conecting' is a low-level ("transport layer"
124 * low, if you so will) state, and while they are not entirely unrelated and
125 * sometimes propagate effects to each other they are only asynchronously connected
126 * at most. */
127 unsigned n_pending;
128
129 int fd;
130
131 char *input_buffer; /* valid data starts at input_buffer_index, ends at input_buffer_index+input_buffer_size */
132 size_t input_buffer_index;
133 size_t input_buffer_size;
134 size_t input_buffer_unscanned;
135
136 char *output_buffer; /* valid data starts at output_buffer_index, ends at output_buffer_index+output_buffer_size */
137 size_t output_buffer_index;
138 size_t output_buffer_size;
139
140 int *input_fds; /* file descriptors associated with the data in input_buffer (for fd passing) */
141 size_t n_input_fds;
142
143 int *output_fds; /* file descriptors associated with the data in output_buffer (for fd passing) */
144 size_t n_output_fds;
145
146 /* Further messages to output not yet formatted into text, and thus not included in output_buffer
147 * yet. We keep them separate from output_buffer, to not violate fd message boundaries: we want that
148 * each fd that is sent is associated with its fds, and that fds cannot be accidentally associated
149 * with preceding or following messages. */
150 LIST_HEAD(VarlinkJsonQueueItem, output_queue);
151 VarlinkJsonQueueItem *output_queue_tail;
152
153 /* The fds to associate with the next message that is about to be enqueued. The user first pushes the
154 * fds it intends to send via varlink_push_fd() into this queue, and then once the message data is
155 * submitted we'll combine the fds and the message data into one. */
156 int *pushed_fds;
157 size_t n_pushed_fds;
158
159 VarlinkReply reply_callback;
160
161 JsonVariant *current;
162
163 struct ucred ucred;
164 bool ucred_acquired:1;
165
166 bool write_disconnected:1;
167 bool read_disconnected:1;
168 bool prefer_read_write:1;
169 bool got_pollhup:1;
170
171 bool allow_fd_passing_input:1;
172 bool allow_fd_passing_output:1;
173
174 bool output_buffer_sensitive:1; /* whether to erase the output buffer after writing it to the socket */
175
176 int af; /* address family if socket; AF_UNSPEC if not socket; negative if not known */
177
178 usec_t timestamp;
179 usec_t timeout;
180
181 void *userdata;
182 char *description;
183
184 sd_event *event;
185 sd_event_source *io_event_source;
186 sd_event_source *time_event_source;
187 sd_event_source *quit_event_source;
188 sd_event_source *defer_event_source;
189 };
190
191 typedef struct VarlinkServerSocket VarlinkServerSocket;
192
193 struct VarlinkServerSocket {
194 VarlinkServer *server;
195
196 int fd;
197 char *address;
198
199 sd_event_source *event_source;
200
201 LIST_FIELDS(VarlinkServerSocket, sockets);
202 };
203
204 struct VarlinkServer {
205 unsigned n_ref;
206 VarlinkServerFlags flags;
207
208 LIST_HEAD(VarlinkServerSocket, sockets);
209
210 Hashmap *methods;
211 VarlinkConnect connect_callback;
212 VarlinkDisconnect disconnect_callback;
213
214 sd_event *event;
215 int64_t event_priority;
216
217 unsigned n_connections;
218 Hashmap *by_uid;
219
220 void *userdata;
221 char *description;
222
223 unsigned connections_max;
224 unsigned connections_per_uid_max;
225 };
226
227 static const char* const varlink_state_table[_VARLINK_STATE_MAX] = {
228 [VARLINK_IDLE_CLIENT] = "idle-client",
229 [VARLINK_AWAITING_REPLY] = "awaiting-reply",
230 [VARLINK_AWAITING_REPLY_MORE] = "awaiting-reply-more",
231 [VARLINK_CALLING] = "calling",
232 [VARLINK_CALLED] = "called",
233 [VARLINK_PROCESSING_REPLY] = "processing-reply",
234 [VARLINK_IDLE_SERVER] = "idle-server",
235 [VARLINK_PROCESSING_METHOD] = "processing-method",
236 [VARLINK_PROCESSING_METHOD_MORE] = "processing-method-more",
237 [VARLINK_PROCESSING_METHOD_ONEWAY] = "processing-method-oneway",
238 [VARLINK_PROCESSED_METHOD] = "processed-method",
239 [VARLINK_PENDING_METHOD] = "pending-method",
240 [VARLINK_PENDING_METHOD_MORE] = "pending-method-more",
241 [VARLINK_PENDING_DISCONNECT] = "pending-disconnect",
242 [VARLINK_PENDING_TIMEOUT] = "pending-timeout",
243 [VARLINK_PROCESSING_DISCONNECT] = "processing-disconnect",
244 [VARLINK_PROCESSING_TIMEOUT] = "processing-timeout",
245 [VARLINK_PROCESSING_FAILURE] = "processing-failure",
246 [VARLINK_DISCONNECTED] = "disconnected",
247 };
248
249 DEFINE_PRIVATE_STRING_TABLE_LOOKUP_TO_STRING(varlink_state, VarlinkState);
250
251 #define varlink_log_errno(v, error, fmt, ...) \
252 log_debug_errno(error, "%s: " fmt, varlink_description(v), ##__VA_ARGS__)
253
254 #define varlink_log(v, fmt, ...) \
255 log_debug("%s: " fmt, varlink_description(v), ##__VA_ARGS__)
256
257 #define varlink_server_log_errno(s, error, fmt, ...) \
258 log_debug_errno(error, "%s: " fmt, varlink_server_description(s), ##__VA_ARGS__)
259
260 #define varlink_server_log(s, fmt, ...) \
261 log_debug("%s: " fmt, varlink_server_description(s), ##__VA_ARGS__)
262
263 static int varlink_format_queue(Varlink *v);
264
265 static inline const char *varlink_description(Varlink *v) {
266 return (v ? v->description : NULL) ?: "varlink";
267 }
268
269 static inline const char *varlink_server_description(VarlinkServer *s) {
270 return (s ? s->description : NULL) ?: "varlink";
271 }
272
273 static VarlinkJsonQueueItem *varlink_json_queue_item_free(VarlinkJsonQueueItem *q) {
274 if (!q)
275 return NULL;
276
277 json_variant_unref(q->data);
278 close_many(q->fds, q->n_fds);
279
280 return mfree(q);
281 }
282
283 static VarlinkJsonQueueItem *varlink_json_queue_item_new(JsonVariant *m, const int fds[], size_t n_fds) {
284 VarlinkJsonQueueItem *q;
285
286 assert(m);
287 assert(fds || n_fds == 0);
288
289 q = malloc(offsetof(VarlinkJsonQueueItem, fds) + sizeof(int) * n_fds);
290 if (!q)
291 return NULL;
292
293 *q = (VarlinkJsonQueueItem) {
294 .data = json_variant_ref(m),
295 .n_fds = n_fds,
296 };
297
298 memcpy_safe(q->fds, fds, n_fds * sizeof(int));
299
300 return TAKE_PTR(q);
301 }
302
303 static void varlink_set_state(Varlink *v, VarlinkState state) {
304 assert(v);
305 assert(state >= 0 && state < _VARLINK_STATE_MAX);
306
307 if (v->state < 0)
308 varlink_log(v, "Setting state %s",
309 varlink_state_to_string(state));
310 else
311 varlink_log(v, "Changing state %s %s %s",
312 varlink_state_to_string(v->state),
313 special_glyph(SPECIAL_GLYPH_ARROW_RIGHT),
314 varlink_state_to_string(state));
315
316 v->state = state;
317 }
318
319 static int varlink_new(Varlink **ret) {
320 Varlink *v;
321
322 assert(ret);
323
324 v = new(Varlink, 1);
325 if (!v)
326 return -ENOMEM;
327
328 *v = (Varlink) {
329 .n_ref = 1,
330 .fd = -EBADF,
331
332 .state = _VARLINK_STATE_INVALID,
333
334 .ucred = UCRED_INVALID,
335
336 .timestamp = USEC_INFINITY,
337 .timeout = VARLINK_DEFAULT_TIMEOUT_USEC,
338
339 .af = -1,
340 };
341
342 *ret = v;
343 return 0;
344 }
345
346 int varlink_connect_address(Varlink **ret, const char *address) {
347 _cleanup_(varlink_unrefp) Varlink *v = NULL;
348 union sockaddr_union sockaddr;
349 int r;
350
351 assert_return(ret, -EINVAL);
352 assert_return(address, -EINVAL);
353
354 r = varlink_new(&v);
355 if (r < 0)
356 return log_debug_errno(r, "Failed to create varlink object: %m");
357
358 v->fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
359 if (v->fd < 0)
360 return log_debug_errno(errno, "Failed to create AF_UNIX socket: %m");
361
362 v->fd = fd_move_above_stdio(v->fd);
363 v->af = AF_UNIX;
364
365 r = sockaddr_un_set_path(&sockaddr.un, address);
366 if (r < 0) {
367 if (r != -ENAMETOOLONG)
368 return log_debug_errno(r, "Failed to set socket address '%s': %m", address);
369
370 /* This is a file system path, and too long to fit into sockaddr_un. Let's connect via O_PATH
371 * to this socket. */
372
373 r = connect_unix_path(v->fd, AT_FDCWD, address);
374 } else
375 r = RET_NERRNO(connect(v->fd, &sockaddr.sa, r));
376
377 if (r < 0) {
378 if (!IN_SET(r, -EAGAIN, -EINPROGRESS))
379 return log_debug_errno(r, "Failed to connect to %s: %m", address);
380
381 v->connecting = true; /* We are asynchronously connecting, i.e. the connect() is being
382 * processed in the background. As long as that's the case the socket
383 * is in a special state: it's there, we can poll it for EPOLLOUT, but
384 * if we attempt to write() to it before we see EPOLLOUT we'll get
385 * ENOTCONN (and not EAGAIN, like we would for a normal connected
386 * socket that isn't writable at the moment). Since ENOTCONN on write()
387 * hence can mean two different things (i.e. connection not complete
388 * yet vs. already disconnected again), we store as a boolean whether
389 * we are still in connect(). */
390 }
391
392 varlink_set_state(v, VARLINK_IDLE_CLIENT);
393
394 *ret = TAKE_PTR(v);
395 return 0;
396 }
397
398 int varlink_connect_fd(Varlink **ret, int fd) {
399 Varlink *v;
400 int r;
401
402 assert_return(ret, -EINVAL);
403 assert_return(fd >= 0, -EBADF);
404
405 r = fd_nonblock(fd, true);
406 if (r < 0)
407 return log_debug_errno(r, "Failed to make fd %d nonblocking: %m", fd);
408
409 r = varlink_new(&v);
410 if (r < 0)
411 return log_debug_errno(r, "Failed to create varlink object: %m");
412
413 v->fd = fd;
414 v->af = -1,
415 varlink_set_state(v, VARLINK_IDLE_CLIENT);
416
417 /* Note that if this function is called we assume the passed socket (if it is one) is already
418 * properly connected, i.e. any asynchronous connect() done on it already completed. Because of that
419 * we'll not set the 'connecting' boolean here, i.e. we don't need to avoid write()ing to the socket
420 * until the connection is fully set up. Behaviour here is hence a bit different from
421 * varlink_connect_address() above, as there we do handle asynchronous connections ourselves and
422 * avoid doing write() on it before we saw EPOLLOUT for the first time. */
423
424 *ret = v;
425 return 0;
426 }
427
428 static void varlink_detach_event_sources(Varlink *v) {
429 assert(v);
430
431 v->io_event_source = sd_event_source_disable_unref(v->io_event_source);
432 v->time_event_source = sd_event_source_disable_unref(v->time_event_source);
433 v->quit_event_source = sd_event_source_disable_unref(v->quit_event_source);
434 v->defer_event_source = sd_event_source_disable_unref(v->defer_event_source);
435 }
436
437 static void varlink_clear_current(Varlink *v) {
438 assert(v);
439
440 /* Clears the currently processed incoming message */
441 v->current = json_variant_unref(v->current);
442
443 close_many(v->input_fds, v->n_input_fds);
444 v->input_fds = mfree(v->input_fds);
445 v->n_input_fds = 0;
446 }
447
448 static void varlink_clear(Varlink *v) {
449 assert(v);
450
451 varlink_detach_event_sources(v);
452
453 v->fd = safe_close(v->fd);
454
455 varlink_clear_current(v);
456
457 v->input_buffer = mfree(v->input_buffer);
458 v->output_buffer = v->output_buffer_sensitive ? erase_and_free(v->output_buffer) : mfree(v->output_buffer);
459
460 varlink_clear_current(v);
461
462 close_many(v->output_fds, v->n_output_fds);
463 v->output_fds = mfree(v->output_fds);
464 v->n_output_fds = 0;
465
466 close_many(v->pushed_fds, v->n_pushed_fds);
467 v->pushed_fds = mfree(v->pushed_fds);
468 v->n_pushed_fds = 0;
469
470 while (v->output_queue) {
471 VarlinkJsonQueueItem *q = v->output_queue;
472
473 LIST_REMOVE(queue, v->output_queue, q);
474 varlink_json_queue_item_free(q);
475 }
476 v->output_queue_tail = NULL;
477
478 v->event = sd_event_unref(v->event);
479 }
480
481 static Varlink* varlink_destroy(Varlink *v) {
482 if (!v)
483 return NULL;
484
485 /* If this is called the server object must already been unreffed here. Why that? because when we
486 * linked up the varlink connection with the server object we took one ref in each direction */
487 assert(!v->server);
488
489 varlink_clear(v);
490
491 free(v->description);
492 return mfree(v);
493 }
494
495 DEFINE_TRIVIAL_REF_UNREF_FUNC(Varlink, varlink, varlink_destroy);
496
497 static int varlink_test_disconnect(Varlink *v) {
498 assert(v);
499
500 /* Tests whether we the connection has been terminated. We are careful to not stop processing it
501 * prematurely, since we want to handle half-open connections as well as possible and want to flush
502 * out and read data before we close down if we can. */
503
504 /* Already disconnected? */
505 if (!VARLINK_STATE_IS_ALIVE(v->state))
506 return 0;
507
508 /* Wait until connection setup is complete, i.e. until asynchronous connect() completes */
509 if (v->connecting)
510 return 0;
511
512 /* Still something to write and we can write? Stay around */
513 if (v->output_buffer_size > 0 && !v->write_disconnected)
514 return 0;
515
516 /* Both sides gone already? Then there's no need to stick around */
517 if (v->read_disconnected && v->write_disconnected)
518 goto disconnect;
519
520 /* If we are waiting for incoming data but the read side is shut down, disconnect. */
521 if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_IDLE_SERVER) && v->read_disconnected)
522 goto disconnect;
523
524 /* Similar, if are a client that hasn't written anything yet but the write side is dead, also
525 * disconnect. We also explicitly check for POLLHUP here since we likely won't notice the write side
526 * being down if we never wrote anything. */
527 if (v->state == VARLINK_IDLE_CLIENT && (v->write_disconnected || v->got_pollhup))
528 goto disconnect;
529
530 /* We are on the server side and still want to send out more replies, but we saw POLLHUP already, and
531 * either got no buffered bytes to write anymore or already saw a write error. In that case we should
532 * shut down the varlink link. */
533 if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE) && (v->write_disconnected || v->output_buffer_size == 0) && v->got_pollhup)
534 goto disconnect;
535
536 return 0;
537
538 disconnect:
539 varlink_set_state(v, VARLINK_PENDING_DISCONNECT);
540 return 1;
541 }
542
543 static int varlink_write(Varlink *v) {
544 ssize_t n;
545 int r;
546
547 assert(v);
548
549 if (!VARLINK_STATE_IS_ALIVE(v->state))
550 return 0;
551 if (v->connecting) /* Writing while we are still wait for a non-blocking connect() to complete will
552 * result in ENOTCONN, hence exit early here */
553 return 0;
554 if (v->write_disconnected)
555 return 0;
556
557 /* If needed let's convert some output queue json variants into text form */
558 r = varlink_format_queue(v);
559 if (r < 0)
560 return r;
561
562 if (v->output_buffer_size == 0)
563 return 0;
564
565 assert(v->fd >= 0);
566
567 if (v->n_output_fds > 0) { /* If we shall send fds along, we must use sendmsg() */
568 struct iovec iov = {
569 .iov_base = v->output_buffer + v->output_buffer_index,
570 .iov_len = v->output_buffer_size,
571 };
572 struct msghdr mh = {
573 .msg_iov = &iov,
574 .msg_iovlen = 1,
575 .msg_controllen = CMSG_SPACE(sizeof(int) * v->n_output_fds),
576 };
577
578 mh.msg_control = alloca0(mh.msg_controllen);
579
580 struct cmsghdr *control = CMSG_FIRSTHDR(&mh);
581 control->cmsg_len = CMSG_LEN(sizeof(int) * v->n_output_fds);
582 control->cmsg_level = SOL_SOCKET;
583 control->cmsg_type = SCM_RIGHTS;
584 memcpy(CMSG_DATA(control), v->output_fds, sizeof(int) * v->n_output_fds);
585
586 n = sendmsg(v->fd, &mh, MSG_DONTWAIT|MSG_NOSIGNAL);
587 } else {
588 /* We generally prefer recv()/send() (mostly because of MSG_NOSIGNAL) but also want to be compatible
589 * with non-socket IO, hence fall back automatically.
590 *
591 * Use a local variable to help gcc figure out that we set 'n' in all cases. */
592 bool prefer_write = v->prefer_read_write;
593 if (!prefer_write) {
594 n = send(v->fd, v->output_buffer + v->output_buffer_index, v->output_buffer_size, MSG_DONTWAIT|MSG_NOSIGNAL);
595 if (n < 0 && errno == ENOTSOCK)
596 prefer_write = v->prefer_read_write = true;
597 }
598 if (prefer_write)
599 n = write(v->fd, v->output_buffer + v->output_buffer_index, v->output_buffer_size);
600 }
601 if (n < 0) {
602 if (errno == EAGAIN)
603 return 0;
604
605 if (ERRNO_IS_DISCONNECT(errno)) {
606 /* If we get informed about a disconnect on write, then let's remember that, but not
607 * act on it just yet. Let's wait for read() to report the issue first. */
608 v->write_disconnected = true;
609 return 1;
610 }
611
612 return -errno;
613 }
614
615 if (v->output_buffer_sensitive)
616 explicit_bzero_safe(v->output_buffer + v->output_buffer_index, n);
617
618 v->output_buffer_size -= n;
619
620 if (v->output_buffer_size == 0) {
621 v->output_buffer_index = 0;
622 v->output_buffer_sensitive = false; /* We can reset the sensitive flag once the buffer is empty */
623 } else
624 v->output_buffer_index += n;
625
626 close_many(v->output_fds, v->n_output_fds);
627 v->n_output_fds = 0;
628
629 v->timestamp = now(CLOCK_MONOTONIC);
630 return 1;
631 }
632
633 #define VARLINK_FDS_MAX (16U*1024U)
634
635 static int varlink_read(Varlink *v) {
636 CMSG_BUFFER_TYPE(CMSG_SPACE(sizeof(int) * VARLINK_FDS_MAX)) control;
637 struct iovec iov;
638 struct msghdr mh;
639 size_t rs;
640 ssize_t n;
641 void *p;
642
643 assert(v);
644
645 if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_IDLE_SERVER))
646 return 0;
647 if (v->connecting) /* read() on a socket while we are in connect() will fail with EINVAL, hence exit early here */
648 return 0;
649 if (v->current)
650 return 0;
651 if (v->input_buffer_unscanned > 0)
652 return 0;
653 if (v->read_disconnected)
654 return 0;
655
656 if (v->input_buffer_size >= VARLINK_BUFFER_MAX)
657 return -ENOBUFS;
658
659 assert(v->fd >= 0);
660
661 if (MALLOC_SIZEOF_SAFE(v->input_buffer) <= v->input_buffer_index + v->input_buffer_size) {
662 size_t add;
663
664 add = MIN(VARLINK_BUFFER_MAX - v->input_buffer_size, VARLINK_READ_SIZE);
665
666 if (v->input_buffer_index == 0) {
667
668 if (!GREEDY_REALLOC(v->input_buffer, v->input_buffer_size + add))
669 return -ENOMEM;
670
671 } else {
672 char *b;
673
674 b = new(char, v->input_buffer_size + add);
675 if (!b)
676 return -ENOMEM;
677
678 memcpy(b, v->input_buffer + v->input_buffer_index, v->input_buffer_size);
679
680 free_and_replace(v->input_buffer, b);
681 v->input_buffer_index = 0;
682 }
683 }
684
685 p = v->input_buffer + v->input_buffer_index + v->input_buffer_size;
686 rs = MALLOC_SIZEOF_SAFE(v->input_buffer) - (v->input_buffer_index + v->input_buffer_size);
687
688 if (v->allow_fd_passing_input) {
689 iov = (struct iovec) {
690 .iov_base = p,
691 .iov_len = rs,
692 };
693 mh = (struct msghdr) {
694 .msg_iov = &iov,
695 .msg_iovlen = 1,
696 .msg_control = &control,
697 .msg_controllen = sizeof(control),
698 };
699 n = recvmsg_safe(v->fd, &mh, MSG_DONTWAIT|MSG_CMSG_CLOEXEC);
700 } else {
701 bool prefer_read = v->prefer_read_write;
702 if (!prefer_read) {
703 n = recv(v->fd, p, rs, MSG_DONTWAIT);
704 if (n < 0 && errno == ENOTSOCK)
705 prefer_read = v->prefer_read_write = true;
706 }
707 if (prefer_read)
708 n = read(v->fd, p, rs);
709 }
710 if (n < 0) {
711 if (errno == EAGAIN)
712 return 0;
713
714 if (ERRNO_IS_DISCONNECT(errno)) {
715 v->read_disconnected = true;
716 return 1;
717 }
718
719 return -errno;
720 }
721 if (n == 0) { /* EOF */
722
723 if (v->allow_fd_passing_input)
724 cmsg_close_all(&mh);
725
726 v->read_disconnected = true;
727 return 1;
728 }
729
730 if (v->allow_fd_passing_input) {
731 struct cmsghdr* cmsg;
732
733 cmsg = cmsg_find(&mh, SOL_SOCKET, SCM_RIGHTS, (socklen_t) -1);
734 if (cmsg) {
735 size_t add;
736
737 /* We only allow file descriptors to be passed along with the first byte of a
738 * message. If they are passed with any other byte this is a protocol violation. */
739 if (v->input_buffer_size != 0) {
740 cmsg_close_all(&mh);
741 return -EPROTO;
742 }
743
744 add = (cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int);
745 if (add > INT_MAX - v->n_input_fds) {
746 cmsg_close_all(&mh);
747 return -EBADF;
748 }
749
750 if (!GREEDY_REALLOC(v->input_fds, v->n_input_fds + add)) {
751 cmsg_close_all(&mh);
752 return -ENOMEM;
753 }
754
755 memcpy_safe(v->input_fds + v->n_input_fds, CMSG_TYPED_DATA(cmsg, int), add * sizeof(int));
756 v->n_input_fds += add;
757 }
758 }
759
760 v->input_buffer_size += n;
761 v->input_buffer_unscanned += n;
762
763 return 1;
764 }
765
766 static int varlink_parse_message(Varlink *v) {
767 const char *e, *begin;
768 size_t sz;
769 int r;
770
771 assert(v);
772
773 if (v->current)
774 return 0;
775 if (v->input_buffer_unscanned <= 0)
776 return 0;
777
778 assert(v->input_buffer_unscanned <= v->input_buffer_size);
779 assert(v->input_buffer_index + v->input_buffer_size <= MALLOC_SIZEOF_SAFE(v->input_buffer));
780
781 begin = v->input_buffer + v->input_buffer_index;
782
783 e = memchr(begin + v->input_buffer_size - v->input_buffer_unscanned, 0, v->input_buffer_unscanned);
784 if (!e) {
785 v->input_buffer_unscanned = 0;
786 return 0;
787 }
788
789 sz = e - begin + 1;
790
791 varlink_log(v, "New incoming message: %s", begin); /* FIXME: should we output the whole message here before validation?
792 * This may produce a non-printable journal entry if the message
793 * is invalid. We may also expose privileged information. */
794
795 r = json_parse(begin, 0, &v->current, NULL, NULL);
796 if (r < 0) {
797 /* If we encounter a parse failure flush all data. We cannot possibly recover from this,
798 * hence drop all buffered data now. */
799 v->input_buffer_index = v->input_buffer_size = v->input_buffer_unscanned = 0;
800 return varlink_log_errno(v, r, "Failed to parse JSON: %m");
801 }
802
803 v->input_buffer_size -= sz;
804
805 if (v->input_buffer_size == 0)
806 v->input_buffer_index = 0;
807 else
808 v->input_buffer_index += sz;
809
810 v->input_buffer_unscanned = v->input_buffer_size;
811 return 1;
812 }
813
814 static int varlink_test_timeout(Varlink *v) {
815 assert(v);
816
817 if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING))
818 return 0;
819 if (v->timeout == USEC_INFINITY)
820 return 0;
821
822 if (now(CLOCK_MONOTONIC) < usec_add(v->timestamp, v->timeout))
823 return 0;
824
825 varlink_set_state(v, VARLINK_PENDING_TIMEOUT);
826
827 return 1;
828 }
829
830 static int varlink_dispatch_local_error(Varlink *v, const char *error) {
831 int r;
832
833 assert(v);
834 assert(error);
835
836 if (!v->reply_callback)
837 return 0;
838
839 r = v->reply_callback(v, NULL, error, VARLINK_REPLY_ERROR|VARLINK_REPLY_LOCAL, v->userdata);
840 if (r < 0)
841 log_debug_errno(r, "Reply callback returned error, ignoring: %m");
842
843 return 1;
844 }
845
846 static int varlink_dispatch_timeout(Varlink *v) {
847 assert(v);
848
849 if (v->state != VARLINK_PENDING_TIMEOUT)
850 return 0;
851
852 varlink_set_state(v, VARLINK_PROCESSING_TIMEOUT);
853 varlink_dispatch_local_error(v, VARLINK_ERROR_TIMEOUT);
854 varlink_close(v);
855
856 return 1;
857 }
858
859 static int varlink_dispatch_disconnect(Varlink *v) {
860 assert(v);
861
862 if (v->state != VARLINK_PENDING_DISCONNECT)
863 return 0;
864
865 varlink_set_state(v, VARLINK_PROCESSING_DISCONNECT);
866 varlink_dispatch_local_error(v, VARLINK_ERROR_DISCONNECTED);
867 varlink_close(v);
868
869 return 1;
870 }
871
872 static int varlink_sanitize_parameters(JsonVariant **v) {
873 assert(v);
874
875 /* Varlink always wants a parameters list, hence make one if the caller doesn't want any */
876 if (!*v)
877 return json_variant_new_object(v, NULL, 0);
878 else if (!json_variant_is_object(*v))
879 return -EINVAL;
880
881 return 0;
882 }
883
884 static int varlink_dispatch_reply(Varlink *v) {
885 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
886 VarlinkReplyFlags flags = 0;
887 const char *error = NULL;
888 JsonVariant *e;
889 const char *k;
890 int r;
891
892 assert(v);
893
894 if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING))
895 return 0;
896 if (!v->current)
897 return 0;
898
899 assert(v->n_pending > 0);
900
901 if (!json_variant_is_object(v->current))
902 goto invalid;
903
904 JSON_VARIANT_OBJECT_FOREACH(k, e, v->current) {
905
906 if (streq(k, "error")) {
907 if (error)
908 goto invalid;
909 if (!json_variant_is_string(e))
910 goto invalid;
911
912 error = json_variant_string(e);
913 flags |= VARLINK_REPLY_ERROR;
914
915 } else if (streq(k, "parameters")) {
916 if (parameters)
917 goto invalid;
918 if (!json_variant_is_object(e))
919 goto invalid;
920
921 parameters = json_variant_ref(e);
922
923 } else if (streq(k, "continues")) {
924 if (FLAGS_SET(flags, VARLINK_REPLY_CONTINUES))
925 goto invalid;
926
927 if (!json_variant_is_boolean(e))
928 goto invalid;
929
930 if (json_variant_boolean(e))
931 flags |= VARLINK_REPLY_CONTINUES;
932 } else
933 goto invalid;
934 }
935
936 /* Replies with 'continue' set are only OK if we set 'more' when the method call was initiated */
937 if (v->state != VARLINK_AWAITING_REPLY_MORE && FLAGS_SET(flags, VARLINK_REPLY_CONTINUES))
938 goto invalid;
939
940 /* An error is final */
941 if (error && FLAGS_SET(flags, VARLINK_REPLY_CONTINUES))
942 goto invalid;
943
944 r = varlink_sanitize_parameters(&parameters);
945 if (r < 0)
946 goto invalid;
947
948 if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE)) {
949 varlink_set_state(v, VARLINK_PROCESSING_REPLY);
950
951 if (v->reply_callback) {
952 r = v->reply_callback(v, parameters, error, flags, v->userdata);
953 if (r < 0)
954 log_debug_errno(r, "Reply callback returned error, ignoring: %m");
955 }
956
957 varlink_clear_current(v);
958
959 if (v->state == VARLINK_PROCESSING_REPLY) {
960
961 assert(v->n_pending > 0);
962
963 if (!FLAGS_SET(flags, VARLINK_REPLY_CONTINUES))
964 v->n_pending--;
965
966 varlink_set_state(v,
967 FLAGS_SET(flags, VARLINK_REPLY_CONTINUES) ? VARLINK_AWAITING_REPLY_MORE :
968 v->n_pending == 0 ? VARLINK_IDLE_CLIENT : VARLINK_AWAITING_REPLY);
969 }
970 } else {
971 assert(v->state == VARLINK_CALLING);
972 varlink_set_state(v, VARLINK_CALLED);
973 }
974
975 return 1;
976
977 invalid:
978 varlink_set_state(v, VARLINK_PROCESSING_FAILURE);
979 varlink_dispatch_local_error(v, VARLINK_ERROR_PROTOCOL);
980 varlink_close(v);
981
982 return 1;
983 }
984
985 static int varlink_dispatch_method(Varlink *v) {
986 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
987 VarlinkMethodFlags flags = 0;
988 const char *method = NULL, *error;
989 JsonVariant *e;
990 VarlinkMethod callback;
991 const char *k;
992 int r;
993
994 assert(v);
995
996 if (v->state != VARLINK_IDLE_SERVER)
997 return 0;
998 if (!v->current)
999 return 0;
1000
1001 if (!json_variant_is_object(v->current))
1002 goto invalid;
1003
1004 JSON_VARIANT_OBJECT_FOREACH(k, e, v->current) {
1005
1006 if (streq(k, "method")) {
1007 if (method)
1008 goto invalid;
1009 if (!json_variant_is_string(e))
1010 goto invalid;
1011
1012 method = json_variant_string(e);
1013
1014 } else if (streq(k, "parameters")) {
1015 if (parameters)
1016 goto invalid;
1017 if (!json_variant_is_object(e))
1018 goto invalid;
1019
1020 parameters = json_variant_ref(e);
1021
1022 } else if (streq(k, "oneway")) {
1023
1024 if ((flags & (VARLINK_METHOD_ONEWAY|VARLINK_METHOD_MORE)) != 0)
1025 goto invalid;
1026
1027 if (!json_variant_is_boolean(e))
1028 goto invalid;
1029
1030 if (json_variant_boolean(e))
1031 flags |= VARLINK_METHOD_ONEWAY;
1032
1033 } else if (streq(k, "more")) {
1034
1035 if ((flags & (VARLINK_METHOD_ONEWAY|VARLINK_METHOD_MORE)) != 0)
1036 goto invalid;
1037
1038 if (!json_variant_is_boolean(e))
1039 goto invalid;
1040
1041 if (json_variant_boolean(e))
1042 flags |= VARLINK_METHOD_MORE;
1043
1044 } else
1045 goto invalid;
1046 }
1047
1048 if (!method)
1049 goto invalid;
1050
1051 r = varlink_sanitize_parameters(&parameters);
1052 if (r < 0)
1053 goto fail;
1054
1055 varlink_set_state(v, (flags & VARLINK_METHOD_MORE) ? VARLINK_PROCESSING_METHOD_MORE :
1056 (flags & VARLINK_METHOD_ONEWAY) ? VARLINK_PROCESSING_METHOD_ONEWAY :
1057 VARLINK_PROCESSING_METHOD);
1058
1059 assert(v->server);
1060
1061 if (STR_IN_SET(method, "org.varlink.service.GetInfo", "org.varlink.service.GetInterface")) {
1062 /* For now, we don't implement a single of varlink's own methods */
1063 callback = NULL;
1064 error = VARLINK_ERROR_METHOD_NOT_IMPLEMENTED;
1065 } else if (startswith(method, "org.varlink.service.")) {
1066 callback = NULL;
1067 error = VARLINK_ERROR_METHOD_NOT_FOUND;
1068 } else {
1069 callback = hashmap_get(v->server->methods, method);
1070 error = VARLINK_ERROR_METHOD_NOT_FOUND;
1071 }
1072
1073 if (callback) {
1074 r = callback(v, parameters, flags, v->userdata);
1075 if (r < 0) {
1076 log_debug_errno(r, "Callback for %s returned error: %m", method);
1077
1078 /* We got an error back from the callback. Propagate it to the client if the method call remains unanswered. */
1079 if (!FLAGS_SET(flags, VARLINK_METHOD_ONEWAY)) {
1080 r = varlink_error_errno(v, r);
1081 if (r < 0)
1082 return r;
1083 }
1084 }
1085 } else if (!FLAGS_SET(flags, VARLINK_METHOD_ONEWAY)) {
1086 assert(error);
1087
1088 r = varlink_errorb(v, error, JSON_BUILD_OBJECT(JSON_BUILD_PAIR("method", JSON_BUILD_STRING(method))));
1089 if (r < 0)
1090 return r;
1091 }
1092
1093 switch (v->state) {
1094
1095 case VARLINK_PROCESSED_METHOD: /* Method call is fully processed */
1096 case VARLINK_PROCESSING_METHOD_ONEWAY: /* ditto */
1097 varlink_clear_current(v);
1098 varlink_set_state(v, VARLINK_IDLE_SERVER);
1099 break;
1100
1101 case VARLINK_PROCESSING_METHOD: /* Method call wasn't replied to, will be replied to later */
1102 varlink_set_state(v, VARLINK_PENDING_METHOD);
1103 break;
1104
1105 case VARLINK_PROCESSING_METHOD_MORE: /* No reply for a "more" message was sent, more to come */
1106 varlink_set_state(v, VARLINK_PENDING_METHOD_MORE);
1107 break;
1108
1109 default:
1110 assert_not_reached();
1111
1112 }
1113
1114 return r;
1115
1116 invalid:
1117 r = -EINVAL;
1118
1119 fail:
1120 varlink_set_state(v, VARLINK_PROCESSING_FAILURE);
1121 varlink_dispatch_local_error(v, VARLINK_ERROR_PROTOCOL);
1122 varlink_close(v);
1123
1124 return r;
1125 }
1126
1127 int varlink_process(Varlink *v) {
1128 int r;
1129
1130 assert_return(v, -EINVAL);
1131
1132 if (v->state == VARLINK_DISCONNECTED)
1133 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1134
1135 varlink_ref(v);
1136
1137 r = varlink_write(v);
1138 if (r < 0)
1139 varlink_log_errno(v, r, "Write failed: %m");
1140 if (r != 0)
1141 goto finish;
1142
1143 r = varlink_dispatch_reply(v);
1144 if (r < 0)
1145 varlink_log_errno(v, r, "Reply dispatch failed: %m");
1146 if (r != 0)
1147 goto finish;
1148
1149 r = varlink_dispatch_method(v);
1150 if (r < 0)
1151 varlink_log_errno(v, r, "Method dispatch failed: %m");
1152 if (r != 0)
1153 goto finish;
1154
1155 r = varlink_parse_message(v);
1156 if (r < 0)
1157 varlink_log_errno(v, r, "Message parsing failed: %m");
1158 if (r != 0)
1159 goto finish;
1160
1161 r = varlink_read(v);
1162 if (r < 0)
1163 varlink_log_errno(v, r, "Read failed: %m");
1164 if (r != 0)
1165 goto finish;
1166
1167 r = varlink_test_disconnect(v);
1168 assert(r >= 0);
1169 if (r != 0)
1170 goto finish;
1171
1172 r = varlink_dispatch_disconnect(v);
1173 assert(r >= 0);
1174 if (r != 0)
1175 goto finish;
1176
1177 r = varlink_test_timeout(v);
1178 assert(r >= 0);
1179 if (r != 0)
1180 goto finish;
1181
1182 r = varlink_dispatch_timeout(v);
1183 assert(r >= 0);
1184 if (r != 0)
1185 goto finish;
1186
1187 finish:
1188 if (r >= 0 && v->defer_event_source) {
1189 int q;
1190
1191 /* If we did some processing, make sure we are called again soon */
1192 q = sd_event_source_set_enabled(v->defer_event_source, r > 0 ? SD_EVENT_ON : SD_EVENT_OFF);
1193 if (q < 0)
1194 r = varlink_log_errno(v, q, "Failed to enable deferred event source: %m");
1195 }
1196
1197 if (r < 0) {
1198 if (VARLINK_STATE_IS_ALIVE(v->state))
1199 /* Initiate disconnection */
1200 varlink_set_state(v, VARLINK_PENDING_DISCONNECT);
1201 else
1202 /* We failed while disconnecting, in that case close right away */
1203 varlink_close(v);
1204 }
1205
1206 varlink_unref(v);
1207 return r;
1208 }
1209
1210 static void handle_revents(Varlink *v, int revents) {
1211 assert(v);
1212
1213 if (v->connecting) {
1214 /* If we have seen POLLOUT or POLLHUP on a socket we are asynchronously waiting a connect()
1215 * to complete on, we know we are ready. We don't read the connection error here though,
1216 * we'll get the error on the next read() or write(). */
1217 if ((revents & (POLLOUT|POLLHUP)) == 0)
1218 return;
1219
1220 varlink_log(v, "Asynchronous connection completed.");
1221 v->connecting = false;
1222 } else {
1223 /* Note that we don't care much about POLLIN/POLLOUT here, we'll just try reading and writing
1224 * what we can. However, we do care about POLLHUP to detect connection termination even if we
1225 * momentarily don't want to read nor write anything. */
1226
1227 if (!FLAGS_SET(revents, POLLHUP))
1228 return;
1229
1230 varlink_log(v, "Got POLLHUP from socket.");
1231 v->got_pollhup = true;
1232 }
1233 }
1234
1235 int varlink_wait(Varlink *v, usec_t timeout) {
1236 int r, fd, events;
1237 usec_t t;
1238
1239 assert_return(v, -EINVAL);
1240
1241 if (v->state == VARLINK_DISCONNECTED)
1242 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1243
1244 r = varlink_get_timeout(v, &t);
1245 if (r < 0)
1246 return r;
1247 if (t != USEC_INFINITY) {
1248 usec_t n;
1249
1250 n = now(CLOCK_MONOTONIC);
1251 if (t < n)
1252 t = 0;
1253 else
1254 t = usec_sub_unsigned(t, n);
1255 }
1256
1257 if (timeout != USEC_INFINITY &&
1258 (t == USEC_INFINITY || timeout < t))
1259 t = timeout;
1260
1261 fd = varlink_get_fd(v);
1262 if (fd < 0)
1263 return fd;
1264
1265 events = varlink_get_events(v);
1266 if (events < 0)
1267 return events;
1268
1269 r = fd_wait_for_event(fd, events, t);
1270 if (r < 0 && ERRNO_IS_TRANSIENT(r)) /* Treat EINTR as not a timeout, but also nothing happened, and
1271 * the caller gets a chance to call back into us */
1272 return 1;
1273 if (r <= 0)
1274 return r;
1275
1276 handle_revents(v, r);
1277 return 1;
1278 }
1279
1280 int varlink_get_fd(Varlink *v) {
1281
1282 assert_return(v, -EINVAL);
1283
1284 if (v->state == VARLINK_DISCONNECTED)
1285 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1286 if (v->fd < 0)
1287 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBADF), "No valid fd.");
1288
1289 return v->fd;
1290 }
1291
1292 int varlink_get_events(Varlink *v) {
1293 int ret = 0;
1294
1295 assert_return(v, -EINVAL);
1296
1297 if (v->state == VARLINK_DISCONNECTED)
1298 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1299
1300 if (v->connecting) /* When processing an asynchronous connect(), we only wait for EPOLLOUT, which
1301 * tells us that the connection is now complete. Before that we should neither
1302 * write() or read() from the fd. */
1303 return EPOLLOUT;
1304
1305 if (!v->read_disconnected &&
1306 IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_IDLE_SERVER) &&
1307 !v->current &&
1308 v->input_buffer_unscanned <= 0)
1309 ret |= EPOLLIN;
1310
1311 if (!v->write_disconnected &&
1312 v->output_buffer_size > 0)
1313 ret |= EPOLLOUT;
1314
1315 return ret;
1316 }
1317
1318 int varlink_get_timeout(Varlink *v, usec_t *ret) {
1319 assert_return(v, -EINVAL);
1320
1321 if (v->state == VARLINK_DISCONNECTED)
1322 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1323
1324 if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING) &&
1325 v->timeout != USEC_INFINITY) {
1326 if (ret)
1327 *ret = usec_add(v->timestamp, v->timeout);
1328 return 1;
1329 } else {
1330 if (ret)
1331 *ret = USEC_INFINITY;
1332 return 0;
1333 }
1334 }
1335
1336 int varlink_flush(Varlink *v) {
1337 int ret = 0, r;
1338
1339 assert_return(v, -EINVAL);
1340
1341 if (v->state == VARLINK_DISCONNECTED)
1342 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1343
1344 for (;;) {
1345 if (v->output_buffer_size == 0)
1346 break;
1347 if (v->write_disconnected)
1348 return -ECONNRESET;
1349
1350 r = varlink_write(v);
1351 if (r < 0)
1352 return r;
1353 if (r > 0) {
1354 ret = 1;
1355 continue;
1356 }
1357
1358 r = fd_wait_for_event(v->fd, POLLOUT, USEC_INFINITY);
1359 if (r < 0) {
1360 if (ERRNO_IS_TRANSIENT(r))
1361 continue;
1362
1363 return varlink_log_errno(v, r, "Poll failed on fd: %m");
1364 }
1365
1366 assert(r != 0);
1367
1368 handle_revents(v, r);
1369 }
1370
1371 return ret;
1372 }
1373
1374 static void varlink_detach_server(Varlink *v) {
1375 VarlinkServer *saved_server;
1376 assert(v);
1377
1378 if (!v->server)
1379 return;
1380
1381 if (v->server->by_uid &&
1382 v->ucred_acquired &&
1383 uid_is_valid(v->ucred.uid)) {
1384 unsigned c;
1385
1386 c = PTR_TO_UINT(hashmap_get(v->server->by_uid, UID_TO_PTR(v->ucred.uid)));
1387 assert(c > 0);
1388
1389 if (c == 1)
1390 (void) hashmap_remove(v->server->by_uid, UID_TO_PTR(v->ucred.uid));
1391 else
1392 (void) hashmap_replace(v->server->by_uid, UID_TO_PTR(v->ucred.uid), UINT_TO_PTR(c - 1));
1393 }
1394
1395 assert(v->server->n_connections > 0);
1396 v->server->n_connections--;
1397
1398 /* If this is a connection associated to a server, then let's disconnect the server and the
1399 * connection from each other. This drops the dangling reference that connect_callback() set up. But
1400 * before we release the references, let's call the disconnection callback if it is defined. */
1401
1402 saved_server = TAKE_PTR(v->server);
1403
1404 if (saved_server->disconnect_callback)
1405 saved_server->disconnect_callback(saved_server, v, saved_server->userdata);
1406
1407 varlink_server_unref(saved_server);
1408 varlink_unref(v);
1409 }
1410
1411 int varlink_close(Varlink *v) {
1412 assert_return(v, -EINVAL);
1413
1414 if (v->state == VARLINK_DISCONNECTED)
1415 return 0;
1416
1417 varlink_set_state(v, VARLINK_DISCONNECTED);
1418
1419 /* Let's take a reference first, since varlink_detach_server() might drop the final (dangling) ref
1420 * which would destroy us before we can call varlink_clear() */
1421 varlink_ref(v);
1422 varlink_detach_server(v);
1423 varlink_clear(v);
1424 varlink_unref(v);
1425
1426 return 1;
1427 }
1428
1429 Varlink* varlink_close_unref(Varlink *v) {
1430 if (!v)
1431 return NULL;
1432
1433 (void) varlink_close(v);
1434 return varlink_unref(v);
1435 }
1436
1437 Varlink* varlink_flush_close_unref(Varlink *v) {
1438 if (!v)
1439 return NULL;
1440
1441 (void) varlink_flush(v);
1442 return varlink_close_unref(v);
1443 }
1444
1445 static int varlink_format_json(Varlink *v, JsonVariant *m) {
1446 _cleanup_(erase_and_freep) char *text = NULL;
1447 int r;
1448
1449 assert(v);
1450 assert(m);
1451
1452 r = json_variant_format(m, 0, &text);
1453 if (r < 0)
1454 return r;
1455 assert(text[r] == '\0');
1456
1457 if (v->output_buffer_size + r + 1 > VARLINK_BUFFER_MAX)
1458 return -ENOBUFS;
1459
1460 varlink_log(v, "Sending message: %s", text);
1461
1462 if (v->output_buffer_size == 0) {
1463
1464 free_and_replace(v->output_buffer, text);
1465
1466 v->output_buffer_size = r + 1;
1467 v->output_buffer_index = 0;
1468
1469 } else if (v->output_buffer_index == 0) {
1470
1471 if (!GREEDY_REALLOC(v->output_buffer, v->output_buffer_size + r + 1))
1472 return -ENOMEM;
1473
1474 memcpy(v->output_buffer + v->output_buffer_size, text, r + 1);
1475 v->output_buffer_size += r + 1;
1476 } else {
1477 char *n;
1478 const size_t new_size = v->output_buffer_size + r + 1;
1479
1480 n = new(char, new_size);
1481 if (!n)
1482 return -ENOMEM;
1483
1484 memcpy(mempcpy(n, v->output_buffer + v->output_buffer_index, v->output_buffer_size), text, r + 1);
1485
1486 free_and_replace(v->output_buffer, n);
1487 v->output_buffer_size = new_size;
1488 v->output_buffer_index = 0;
1489 }
1490
1491 if (json_variant_is_sensitive(m))
1492 v->output_buffer_sensitive = true; /* Propagate sensitive flag */
1493 else
1494 text = mfree(text); /* No point in the erase_and_free() destructor declared above */
1495
1496 return 0;
1497 }
1498
1499 static int varlink_enqueue_json(Varlink *v, JsonVariant *m) {
1500 VarlinkJsonQueueItem *q;
1501
1502 assert(v);
1503 assert(m);
1504
1505 /* If there are no file descriptors to be queued and no queue entries yet we can shortcut things and
1506 * append this entry directly to the output buffer */
1507 if (v->n_pushed_fds == 0 && !v->output_queue)
1508 return varlink_format_json(v, m);
1509
1510 /* Otherwise add a queue entry for this */
1511 q = varlink_json_queue_item_new(m, v->pushed_fds, v->n_pushed_fds);
1512 if (!q)
1513 return -ENOMEM;
1514
1515 v->n_pushed_fds = 0; /* fds now belong to the queue entry */
1516
1517 LIST_INSERT_AFTER(queue, v->output_queue, v->output_queue_tail, q);
1518 v->output_queue_tail = q;
1519 return 0;
1520 }
1521
1522 static int varlink_format_queue(Varlink *v) {
1523 int r;
1524
1525 assert(v);
1526
1527 /* Takes entries out of the output queue and formats them into the output buffer. But only if this
1528 * would not corrupt our fd message boundaries */
1529
1530 while (v->output_queue) {
1531 _cleanup_free_ int *array = NULL;
1532 VarlinkJsonQueueItem *q = v->output_queue;
1533
1534 if (v->n_output_fds > 0) /* unwritten fds? if we'd add more we'd corrupt the fd message boundaries, hence wait */
1535 return 0;
1536
1537 if (q->n_fds > 0) {
1538 array = newdup(int, q->fds, q->n_fds);
1539 if (!array)
1540 return -ENOMEM;
1541 }
1542
1543 r = varlink_format_json(v, q->data);
1544 if (r < 0)
1545 return r;
1546
1547 /* Take possession of the queue element's fds */
1548 free(v->output_fds);
1549 v->output_fds = TAKE_PTR(array);
1550 v->n_output_fds = q->n_fds;
1551 q->n_fds = 0;
1552
1553 LIST_REMOVE(queue, v->output_queue, q);
1554 if (!v->output_queue)
1555 v->output_queue_tail = NULL;
1556
1557 varlink_json_queue_item_free(q);
1558 }
1559
1560 return 0;
1561 }
1562
1563 int varlink_send(Varlink *v, const char *method, JsonVariant *parameters) {
1564 _cleanup_(json_variant_unrefp) JsonVariant *m = NULL;
1565 int r;
1566
1567 assert_return(v, -EINVAL);
1568 assert_return(method, -EINVAL);
1569
1570 if (v->state == VARLINK_DISCONNECTED)
1571 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1572
1573 /* We allow enqueuing multiple method calls at once! */
1574 if (!IN_SET(v->state, VARLINK_IDLE_CLIENT, VARLINK_AWAITING_REPLY))
1575 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
1576
1577 r = varlink_sanitize_parameters(&parameters);
1578 if (r < 0)
1579 return varlink_log_errno(v, r, "Failed to sanitize parameters: %m");
1580
1581 r = json_build(&m, JSON_BUILD_OBJECT(
1582 JSON_BUILD_PAIR("method", JSON_BUILD_STRING(method)),
1583 JSON_BUILD_PAIR("parameters", JSON_BUILD_VARIANT(parameters)),
1584 JSON_BUILD_PAIR("oneway", JSON_BUILD_BOOLEAN(true))));
1585 if (r < 0)
1586 return varlink_log_errno(v, r, "Failed to build json message: %m");
1587
1588 r = varlink_enqueue_json(v, m);
1589 if (r < 0)
1590 return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
1591
1592 /* No state change here, this is one-way only after all */
1593 v->timestamp = now(CLOCK_MONOTONIC);
1594 return 0;
1595 }
1596
1597 int varlink_sendb(Varlink *v, const char *method, ...) {
1598 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
1599 va_list ap;
1600 int r;
1601
1602 assert_return(v, -EINVAL);
1603
1604 va_start(ap, method);
1605 r = json_buildv(&parameters, ap);
1606 va_end(ap);
1607
1608 if (r < 0)
1609 return varlink_log_errno(v, r, "Failed to build json message: %m");
1610
1611 return varlink_send(v, method, parameters);
1612 }
1613
1614 int varlink_invoke(Varlink *v, const char *method, JsonVariant *parameters) {
1615 _cleanup_(json_variant_unrefp) JsonVariant *m = NULL;
1616 int r;
1617
1618 assert_return(v, -EINVAL);
1619 assert_return(method, -EINVAL);
1620
1621 if (v->state == VARLINK_DISCONNECTED)
1622 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1623
1624 /* We allow enqueuing multiple method calls at once! */
1625 if (!IN_SET(v->state, VARLINK_IDLE_CLIENT, VARLINK_AWAITING_REPLY))
1626 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
1627
1628 r = varlink_sanitize_parameters(&parameters);
1629 if (r < 0)
1630 return varlink_log_errno(v, r, "Failed to sanitize parameters: %m");
1631
1632 r = json_build(&m, JSON_BUILD_OBJECT(
1633 JSON_BUILD_PAIR("method", JSON_BUILD_STRING(method)),
1634 JSON_BUILD_PAIR("parameters", JSON_BUILD_VARIANT(parameters))));
1635 if (r < 0)
1636 return varlink_log_errno(v, r, "Failed to build json message: %m");
1637
1638 r = varlink_enqueue_json(v, m);
1639 if (r < 0)
1640 return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
1641
1642 varlink_set_state(v, VARLINK_AWAITING_REPLY);
1643 v->n_pending++;
1644 v->timestamp = now(CLOCK_MONOTONIC);
1645
1646 return 0;
1647 }
1648
1649 int varlink_invokeb(Varlink *v, const char *method, ...) {
1650 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
1651 va_list ap;
1652 int r;
1653
1654 assert_return(v, -EINVAL);
1655
1656 va_start(ap, method);
1657 r = json_buildv(&parameters, ap);
1658 va_end(ap);
1659
1660 if (r < 0)
1661 return varlink_log_errno(v, r, "Failed to build json message: %m");
1662
1663 return varlink_invoke(v, method, parameters);
1664 }
1665
1666 int varlink_observe(Varlink *v, const char *method, JsonVariant *parameters) {
1667 _cleanup_(json_variant_unrefp) JsonVariant *m = NULL;
1668 int r;
1669
1670 assert_return(v, -EINVAL);
1671 assert_return(method, -EINVAL);
1672
1673 if (v->state == VARLINK_DISCONNECTED)
1674 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1675
1676 /* Note that we don't allow enqueuing multiple method calls when we are in more/continues mode! We
1677 * thus insist on an idle client here. */
1678 if (v->state != VARLINK_IDLE_CLIENT)
1679 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
1680
1681 r = varlink_sanitize_parameters(&parameters);
1682 if (r < 0)
1683 return varlink_log_errno(v, r, "Failed to sanitize parameters: %m");
1684
1685 r = json_build(&m, JSON_BUILD_OBJECT(
1686 JSON_BUILD_PAIR("method", JSON_BUILD_STRING(method)),
1687 JSON_BUILD_PAIR("parameters", JSON_BUILD_VARIANT(parameters)),
1688 JSON_BUILD_PAIR("more", JSON_BUILD_BOOLEAN(true))));
1689 if (r < 0)
1690 return varlink_log_errno(v, r, "Failed to build json message: %m");
1691
1692 r = varlink_enqueue_json(v, m);
1693 if (r < 0)
1694 return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
1695
1696 varlink_set_state(v, VARLINK_AWAITING_REPLY_MORE);
1697 v->n_pending++;
1698 v->timestamp = now(CLOCK_MONOTONIC);
1699
1700 return 0;
1701 }
1702
1703 int varlink_observeb(Varlink *v, const char *method, ...) {
1704 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
1705 va_list ap;
1706 int r;
1707
1708 assert_return(v, -EINVAL);
1709
1710 va_start(ap, method);
1711 r = json_buildv(&parameters, ap);
1712 va_end(ap);
1713
1714 if (r < 0)
1715 return varlink_log_errno(v, r, "Failed to build json message: %m");
1716
1717 return varlink_observe(v, method, parameters);
1718 }
1719
1720 int varlink_call(
1721 Varlink *v,
1722 const char *method,
1723 JsonVariant *parameters,
1724 JsonVariant **ret_parameters,
1725 const char **ret_error_id,
1726 VarlinkReplyFlags *ret_flags) {
1727
1728 _cleanup_(json_variant_unrefp) JsonVariant *m = NULL;
1729 int r;
1730
1731 assert_return(v, -EINVAL);
1732 assert_return(method, -EINVAL);
1733
1734 if (v->state == VARLINK_DISCONNECTED)
1735 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1736 if (v->state != VARLINK_IDLE_CLIENT)
1737 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
1738
1739 assert(v->n_pending == 0); /* n_pending can't be > 0 if we are in VARLINK_IDLE_CLIENT state */
1740
1741 /* If there was still a reply pinned from a previous call, now it's the time to get rid of it, so
1742 * that we can assign a new reply shortly. */
1743 varlink_clear_current(v);
1744
1745 r = varlink_sanitize_parameters(&parameters);
1746 if (r < 0)
1747 return varlink_log_errno(v, r, "Failed to sanitize parameters: %m");
1748
1749 r = json_build(&m, JSON_BUILD_OBJECT(
1750 JSON_BUILD_PAIR("method", JSON_BUILD_STRING(method)),
1751 JSON_BUILD_PAIR("parameters", JSON_BUILD_VARIANT(parameters))));
1752 if (r < 0)
1753 return varlink_log_errno(v, r, "Failed to build json message: %m");
1754
1755 r = varlink_enqueue_json(v, m);
1756 if (r < 0)
1757 return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
1758
1759 varlink_set_state(v, VARLINK_CALLING);
1760 v->n_pending++;
1761 v->timestamp = now(CLOCK_MONOTONIC);
1762
1763 while (v->state == VARLINK_CALLING) {
1764
1765 r = varlink_process(v);
1766 if (r < 0)
1767 return r;
1768 if (r > 0)
1769 continue;
1770
1771 r = varlink_wait(v, USEC_INFINITY);
1772 if (r < 0)
1773 return r;
1774 }
1775
1776 switch (v->state) {
1777
1778 case VARLINK_CALLED:
1779 assert(v->current);
1780
1781 varlink_set_state(v, VARLINK_IDLE_CLIENT);
1782 assert(v->n_pending == 1);
1783 v->n_pending--;
1784
1785 if (ret_parameters)
1786 *ret_parameters = json_variant_by_key(v->current, "parameters");
1787 if (ret_error_id)
1788 *ret_error_id = json_variant_string(json_variant_by_key(v->current, "error"));
1789 if (ret_flags)
1790 *ret_flags = 0;
1791
1792 return 1;
1793
1794 case VARLINK_PENDING_DISCONNECT:
1795 case VARLINK_DISCONNECTED:
1796 return varlink_log_errno(v, SYNTHETIC_ERRNO(ECONNRESET), "Connection was closed.");
1797
1798 case VARLINK_PENDING_TIMEOUT:
1799 return varlink_log_errno(v, SYNTHETIC_ERRNO(ETIME), "Connection timed out.");
1800
1801 default:
1802 assert_not_reached();
1803 }
1804 }
1805
1806 int varlink_callb(
1807 Varlink *v,
1808 const char *method,
1809 JsonVariant **ret_parameters,
1810 const char **ret_error_id,
1811 VarlinkReplyFlags *ret_flags, ...) {
1812
1813 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
1814 va_list ap;
1815 int r;
1816
1817 assert_return(v, -EINVAL);
1818
1819 va_start(ap, ret_flags);
1820 r = json_buildv(&parameters, ap);
1821 va_end(ap);
1822
1823 if (r < 0)
1824 return varlink_log_errno(v, r, "Failed to build json message: %m");
1825
1826 return varlink_call(v, method, parameters, ret_parameters, ret_error_id, ret_flags);
1827 }
1828
1829 int varlink_reply(Varlink *v, JsonVariant *parameters) {
1830 _cleanup_(json_variant_unrefp) JsonVariant *m = NULL;
1831 int r;
1832
1833 assert_return(v, -EINVAL);
1834
1835 if (v->state == VARLINK_DISCONNECTED)
1836 return -ENOTCONN;
1837 if (!IN_SET(v->state,
1838 VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE,
1839 VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
1840 return -EBUSY;
1841
1842 r = varlink_sanitize_parameters(&parameters);
1843 if (r < 0)
1844 return varlink_log_errno(v, r, "Failed to sanitize parameters: %m");
1845
1846 r = json_build(&m, JSON_BUILD_OBJECT(JSON_BUILD_PAIR("parameters", JSON_BUILD_VARIANT(parameters))));
1847 if (r < 0)
1848 return varlink_log_errno(v, r, "Failed to build json message: %m");
1849
1850 r = varlink_enqueue_json(v, m);
1851 if (r < 0)
1852 return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
1853
1854 if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) {
1855 /* We just replied to a method call that was let hanging for a while (i.e. we were outside of
1856 * the varlink_dispatch_method() stack frame), which means with this reply we are ready to
1857 * process further messages. */
1858 varlink_clear_current(v);
1859 varlink_set_state(v, VARLINK_IDLE_SERVER);
1860 } else
1861 /* We replied to a method call from within the varlink_dispatch_method() stack frame), which
1862 * means we should it handle the rest of the state engine. */
1863 varlink_set_state(v, VARLINK_PROCESSED_METHOD);
1864
1865 return 1;
1866 }
1867
1868 int varlink_replyb(Varlink *v, ...) {
1869 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
1870 va_list ap;
1871 int r;
1872
1873 assert_return(v, -EINVAL);
1874
1875 va_start(ap, v);
1876 r = json_buildv(&parameters, ap);
1877 va_end(ap);
1878
1879 if (r < 0)
1880 return r;
1881
1882 return varlink_reply(v, parameters);
1883 }
1884
1885 int varlink_error(Varlink *v, const char *error_id, JsonVariant *parameters) {
1886 _cleanup_(json_variant_unrefp) JsonVariant *m = NULL;
1887 int r;
1888
1889 assert_return(v, -EINVAL);
1890 assert_return(error_id, -EINVAL);
1891
1892 if (v->state == VARLINK_DISCONNECTED)
1893 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
1894 if (!IN_SET(v->state,
1895 VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE,
1896 VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
1897 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
1898
1899 /* Reset the list of pushed file descriptors before sending an error reply. We do this here to
1900 * simplify code that puts together a complex reply message with fds, and half-way something
1901 * fails. In that case the pushed fds need to be flushed out again. Under the assumption that it
1902 * never makes sense to send fds along with errors we simply flush them out here beforehand, so that
1903 * the callers don't need to do this explicitly. */
1904 varlink_reset_fds(v);
1905
1906 r = varlink_sanitize_parameters(&parameters);
1907 if (r < 0)
1908 return varlink_log_errno(v, r, "Failed to sanitize parameters: %m");
1909
1910 r = json_build(&m, JSON_BUILD_OBJECT(
1911 JSON_BUILD_PAIR("error", JSON_BUILD_STRING(error_id)),
1912 JSON_BUILD_PAIR("parameters", JSON_BUILD_VARIANT(parameters))));
1913 if (r < 0)
1914 return varlink_log_errno(v, r, "Failed to build json message: %m");
1915
1916 r = varlink_enqueue_json(v, m);
1917 if (r < 0)
1918 return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
1919
1920 if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) {
1921 varlink_clear_current(v);
1922 varlink_set_state(v, VARLINK_IDLE_SERVER);
1923 } else
1924 varlink_set_state(v, VARLINK_PROCESSED_METHOD);
1925
1926 return 1;
1927 }
1928
1929 int varlink_errorb(Varlink *v, const char *error_id, ...) {
1930 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
1931 va_list ap;
1932 int r;
1933
1934 assert_return(v, -EINVAL);
1935 assert_return(error_id, -EINVAL);
1936
1937 va_start(ap, error_id);
1938 r = json_buildv(&parameters, ap);
1939 va_end(ap);
1940
1941 if (r < 0)
1942 return varlink_log_errno(v, r, "Failed to build json message: %m");
1943
1944 return varlink_error(v, error_id, parameters);
1945 }
1946
1947 int varlink_error_invalid_parameter(Varlink *v, JsonVariant *parameters) {
1948 int r;
1949
1950 assert_return(v, -EINVAL);
1951 assert_return(parameters, -EINVAL);
1952
1953 /* We expect to be called in one of two ways: the 'parameters' argument is a string variant in which
1954 * case it is the parameter key name that is invalid. Or the 'parameters' argument is an object
1955 * variant in which case we'll pull out the first key. The latter mode is useful in functions that
1956 * don't expect any arguments. */
1957
1958 /* varlink_error(...) expects a json object as the third parameter. Passing a string variant causes
1959 * parameter sanitization to fail, and it returns -EINVAL. */
1960
1961 if (json_variant_is_string(parameters)) {
1962 _cleanup_(json_variant_unrefp) JsonVariant *parameters_obj = NULL;
1963
1964 r = json_build(&parameters_obj,
1965 JSON_BUILD_OBJECT(
1966 JSON_BUILD_PAIR("parameter", JSON_BUILD_VARIANT(parameters))));
1967 if (r < 0)
1968 return r;
1969
1970 return varlink_error(v, VARLINK_ERROR_INVALID_PARAMETER, parameters_obj);
1971 }
1972
1973 if (json_variant_is_object(parameters) &&
1974 json_variant_elements(parameters) > 0) {
1975 _cleanup_(json_variant_unrefp) JsonVariant *parameters_obj = NULL;
1976
1977 r = json_build(&parameters_obj,
1978 JSON_BUILD_OBJECT(
1979 JSON_BUILD_PAIR("parameter", JSON_BUILD_VARIANT(json_variant_by_index(parameters, 0)))));
1980 if (r < 0)
1981 return r;
1982
1983 return varlink_error(v, VARLINK_ERROR_INVALID_PARAMETER, parameters_obj);
1984 }
1985
1986 return -EINVAL;
1987 }
1988
1989 int varlink_error_errno(Varlink *v, int error) {
1990 return varlink_errorb(
1991 v,
1992 VARLINK_ERROR_SYSTEM,
1993 JSON_BUILD_OBJECT(JSON_BUILD_PAIR("errno", JSON_BUILD_INTEGER(abs(error)))));
1994 }
1995
1996 int varlink_notify(Varlink *v, JsonVariant *parameters) {
1997 _cleanup_(json_variant_unrefp) JsonVariant *m = NULL;
1998 int r;
1999
2000 assert_return(v, -EINVAL);
2001
2002 if (v->state == VARLINK_DISCONNECTED)
2003 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
2004 if (!IN_SET(v->state, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PENDING_METHOD_MORE))
2005 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
2006
2007 r = varlink_sanitize_parameters(&parameters);
2008 if (r < 0)
2009 return varlink_log_errno(v, r, "Failed to sanitize parameters: %m");
2010
2011 r = json_build(&m, JSON_BUILD_OBJECT(
2012 JSON_BUILD_PAIR("parameters", JSON_BUILD_VARIANT(parameters)),
2013 JSON_BUILD_PAIR("continues", JSON_BUILD_BOOLEAN(true))));
2014 if (r < 0)
2015 return varlink_log_errno(v, r, "Failed to build json message: %m");
2016
2017 r = varlink_enqueue_json(v, m);
2018 if (r < 0)
2019 return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
2020
2021 /* No state change, as more is coming */
2022 return 1;
2023 }
2024
2025 int varlink_notifyb(Varlink *v, ...) {
2026 _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
2027 va_list ap;
2028 int r;
2029
2030 assert_return(v, -EINVAL);
2031
2032 va_start(ap, v);
2033 r = json_buildv(&parameters, ap);
2034 va_end(ap);
2035
2036 if (r < 0)
2037 return varlink_log_errno(v, r, "Failed to build json message: %m");
2038
2039 return varlink_notify(v, parameters);
2040 }
2041
2042 int varlink_bind_reply(Varlink *v, VarlinkReply callback) {
2043 assert_return(v, -EINVAL);
2044
2045 if (callback && v->reply_callback && callback != v->reply_callback)
2046 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "A different callback was already set.");
2047
2048 v->reply_callback = callback;
2049
2050 return 0;
2051 }
2052
2053 void* varlink_set_userdata(Varlink *v, void *userdata) {
2054 void *old;
2055
2056 assert_return(v, NULL);
2057
2058 old = v->userdata;
2059 v->userdata = userdata;
2060
2061 return old;
2062 }
2063
2064 void* varlink_get_userdata(Varlink *v) {
2065 assert_return(v, NULL);
2066
2067 return v->userdata;
2068 }
2069
2070 static int varlink_acquire_ucred(Varlink *v) {
2071 int r;
2072
2073 assert(v);
2074
2075 if (v->ucred_acquired)
2076 return 0;
2077
2078 r = getpeercred(v->fd, &v->ucred);
2079 if (r < 0)
2080 return r;
2081
2082 v->ucred_acquired = true;
2083 return 0;
2084 }
2085
2086 int varlink_get_peer_uid(Varlink *v, uid_t *ret) {
2087 int r;
2088
2089 assert_return(v, -EINVAL);
2090 assert_return(ret, -EINVAL);
2091
2092 r = varlink_acquire_ucred(v);
2093 if (r < 0)
2094 return varlink_log_errno(v, r, "Failed to acquire credentials: %m");
2095
2096 if (!uid_is_valid(v->ucred.uid))
2097 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENODATA), "Peer uid is invalid.");
2098
2099 *ret = v->ucred.uid;
2100 return 0;
2101 }
2102
2103 int varlink_get_peer_pid(Varlink *v, pid_t *ret) {
2104 int r;
2105
2106 assert_return(v, -EINVAL);
2107 assert_return(ret, -EINVAL);
2108
2109 r = varlink_acquire_ucred(v);
2110 if (r < 0)
2111 return varlink_log_errno(v, r, "Failed to acquire credentials: %m");
2112
2113 if (!pid_is_valid(v->ucred.pid))
2114 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENODATA), "Peer uid is invalid.");
2115
2116 *ret = v->ucred.pid;
2117 return 0;
2118 }
2119
2120 int varlink_set_relative_timeout(Varlink *v, usec_t timeout) {
2121 assert_return(v, -EINVAL);
2122 assert_return(timeout > 0, -EINVAL);
2123
2124 v->timeout = timeout;
2125 return 0;
2126 }
2127
2128 VarlinkServer *varlink_get_server(Varlink *v) {
2129 assert_return(v, NULL);
2130
2131 return v->server;
2132 }
2133
2134 int varlink_set_description(Varlink *v, const char *description) {
2135 assert_return(v, -EINVAL);
2136
2137 return free_and_strdup(&v->description, description);
2138 }
2139
2140 static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
2141 Varlink *v = ASSERT_PTR(userdata);
2142
2143 assert(s);
2144
2145 handle_revents(v, revents);
2146 (void) varlink_process(v);
2147
2148 return 1;
2149 }
2150
2151 static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
2152 Varlink *v = ASSERT_PTR(userdata);
2153
2154 assert(s);
2155
2156 (void) varlink_process(v);
2157 return 1;
2158 }
2159
2160 static int defer_callback(sd_event_source *s, void *userdata) {
2161 Varlink *v = ASSERT_PTR(userdata);
2162
2163 assert(s);
2164
2165 (void) varlink_process(v);
2166 return 1;
2167 }
2168
2169 static int prepare_callback(sd_event_source *s, void *userdata) {
2170 Varlink *v = ASSERT_PTR(userdata);
2171 int r, e;
2172 usec_t until;
2173 bool have_timeout;
2174
2175 assert(s);
2176
2177 e = varlink_get_events(v);
2178 if (e < 0)
2179 return e;
2180
2181 r = sd_event_source_set_io_events(v->io_event_source, e);
2182 if (r < 0)
2183 return varlink_log_errno(v, r, "Failed to set source events: %m");
2184
2185 r = varlink_get_timeout(v, &until);
2186 if (r < 0)
2187 return r;
2188 have_timeout = r > 0;
2189
2190 if (have_timeout) {
2191 r = sd_event_source_set_time(v->time_event_source, until);
2192 if (r < 0)
2193 return varlink_log_errno(v, r, "Failed to set source time: %m");
2194 }
2195
2196 r = sd_event_source_set_enabled(v->time_event_source, have_timeout ? SD_EVENT_ON : SD_EVENT_OFF);
2197 if (r < 0)
2198 return varlink_log_errno(v, r, "Failed to enable event source: %m");
2199
2200 return 1;
2201 }
2202
2203 static int quit_callback(sd_event_source *event, void *userdata) {
2204 Varlink *v = ASSERT_PTR(userdata);
2205
2206 assert(event);
2207
2208 varlink_flush(v);
2209 varlink_close(v);
2210
2211 return 1;
2212 }
2213
2214 int varlink_attach_event(Varlink *v, sd_event *e, int64_t priority) {
2215 int r;
2216
2217 assert_return(v, -EINVAL);
2218 assert_return(!v->event, -EBUSY);
2219
2220 if (e)
2221 v->event = sd_event_ref(e);
2222 else {
2223 r = sd_event_default(&v->event);
2224 if (r < 0)
2225 return varlink_log_errno(v, r, "Failed to create event source: %m");
2226 }
2227
2228 r = sd_event_add_time(v->event, &v->time_event_source, CLOCK_MONOTONIC, 0, 0, time_callback, v);
2229 if (r < 0)
2230 goto fail;
2231
2232 r = sd_event_source_set_priority(v->time_event_source, priority);
2233 if (r < 0)
2234 goto fail;
2235
2236 (void) sd_event_source_set_description(v->time_event_source, "varlink-time");
2237
2238 r = sd_event_add_exit(v->event, &v->quit_event_source, quit_callback, v);
2239 if (r < 0)
2240 goto fail;
2241
2242 r = sd_event_source_set_priority(v->quit_event_source, priority);
2243 if (r < 0)
2244 goto fail;
2245
2246 (void) sd_event_source_set_description(v->quit_event_source, "varlink-quit");
2247
2248 r = sd_event_add_io(v->event, &v->io_event_source, v->fd, 0, io_callback, v);
2249 if (r < 0)
2250 goto fail;
2251
2252 r = sd_event_source_set_prepare(v->io_event_source, prepare_callback);
2253 if (r < 0)
2254 goto fail;
2255
2256 r = sd_event_source_set_priority(v->io_event_source, priority);
2257 if (r < 0)
2258 goto fail;
2259
2260 (void) sd_event_source_set_description(v->io_event_source, "varlink-io");
2261
2262 r = sd_event_add_defer(v->event, &v->defer_event_source, defer_callback, v);
2263 if (r < 0)
2264 goto fail;
2265
2266 r = sd_event_source_set_priority(v->defer_event_source, priority);
2267 if (r < 0)
2268 goto fail;
2269
2270 (void) sd_event_source_set_description(v->defer_event_source, "varlink-defer");
2271
2272 return 0;
2273
2274 fail:
2275 varlink_log_errno(v, r, "Failed to setup event source: %m");
2276 varlink_detach_event(v);
2277 return r;
2278 }
2279
2280 void varlink_detach_event(Varlink *v) {
2281 if (!v)
2282 return;
2283
2284 varlink_detach_event_sources(v);
2285
2286 v->event = sd_event_unref(v->event);
2287 }
2288
2289 sd_event *varlink_get_event(Varlink *v) {
2290 assert_return(v, NULL);
2291
2292 return v->event;
2293 }
2294
2295 int varlink_push_fd(Varlink *v, int fd) {
2296 int i;
2297
2298 assert_return(v, -EINVAL);
2299 assert_return(fd >= 0, -EBADF);
2300
2301 /* Takes an fd to send along with the *next* varlink message sent via this varlink connection. This
2302 * takes ownership of the specified fd. Use varlink_dup_fd() below to duplicate the fd first. */
2303
2304 if (!v->allow_fd_passing_output)
2305 return -EPERM;
2306
2307 if (v->n_pushed_fds >= INT_MAX)
2308 return -ENOMEM;
2309
2310 if (!GREEDY_REALLOC(v->pushed_fds, v->n_pushed_fds + 1))
2311 return -ENOMEM;
2312
2313 i = (int) v->n_pushed_fds;
2314 v->pushed_fds[v->n_pushed_fds++] = fd;
2315 return i;
2316 }
2317
2318 int varlink_dup_fd(Varlink *v, int fd) {
2319 _cleanup_close_ int dp = -1;
2320 int r;
2321
2322 assert_return(v, -EINVAL);
2323 assert_return(fd >= 0, -EBADF);
2324
2325 /* Like varlink_push_fd() but duplicates the specified fd instead of taking possession of it */
2326
2327 dp = fcntl(fd, F_DUPFD_CLOEXEC, 3);
2328 if (dp < 0)
2329 return -errno;
2330
2331 r = varlink_push_fd(v, dp);
2332 if (r < 0)
2333 return r;
2334
2335 TAKE_FD(dp);
2336 return r;
2337 }
2338
2339 int varlink_reset_fds(Varlink *v) {
2340 assert_return(v, -EINVAL);
2341
2342 /* Closes all currently pending fds to send. This may be used whenever the caller is in the process
2343 * of putting together a message with fds, and then eventually something fails and they need to
2344 * rollback the fds. Note that this is implicitly called whenever an error reply is sent, see above. */
2345
2346 close_many(v->output_fds, v->n_output_fds);
2347 v->n_output_fds = 0;
2348 return 0;
2349 }
2350
2351 int varlink_peek_fd(Varlink *v, size_t i) {
2352 assert_return(v, -EINVAL);
2353
2354 /* Returns one of the file descriptors that were received along with the current message. This does
2355 * not duplicate the fd nor invalidate it, it hence remains in our possession. */
2356
2357 if (!v->allow_fd_passing_input)
2358 return -EPERM;
2359
2360 if (i >= v->n_input_fds)
2361 return -ENXIO;
2362
2363 return v->input_fds[i];
2364 }
2365
2366 int varlink_take_fd(Varlink *v, size_t i) {
2367 assert_return(v, -EINVAL);
2368
2369 /* Similar to varlink_peek_fd() but the file descriptor's ownership is passed to the caller, and
2370 * we'll invalidate the reference to it under our possession. If called twice in a row will return
2371 * -EBADF */
2372
2373 if (!v->allow_fd_passing_input)
2374 return -EPERM;
2375
2376 if (i >= v->n_input_fds)
2377 return -ENXIO;
2378
2379 return TAKE_FD(v->input_fds[i]);
2380 }
2381
2382 static int verify_unix_socket(Varlink *v) {
2383 assert(v);
2384
2385 if (v->af < 0) {
2386 struct stat st;
2387
2388 if (fstat(v->fd, &st) < 0)
2389 return -errno;
2390 if (!S_ISSOCK(st.st_mode)) {
2391 v->af = AF_UNSPEC;
2392 return -ENOTSOCK;
2393 }
2394
2395 v->af = socket_get_family(v->fd);
2396 if (v->af < 0)
2397 return v->af;
2398 }
2399
2400 return v->af == AF_UNIX ? 0 : -ENOMEDIUM;
2401 }
2402
2403 int varlink_set_allow_fd_passing_input(Varlink *v, bool b) {
2404 int r;
2405
2406 assert_return(v, -EINVAL);
2407
2408 if (v->allow_fd_passing_input == b)
2409 return 0;
2410
2411 if (!b) {
2412 v->allow_fd_passing_input = false;
2413 return 1;
2414 }
2415
2416 r = verify_unix_socket(v);
2417 if (r < 0)
2418 return r;
2419
2420 v->allow_fd_passing_input = true;
2421 return 0;
2422 }
2423
2424 int varlink_set_allow_fd_passing_output(Varlink *v, bool b) {
2425 int r;
2426
2427 assert_return(v, -EINVAL);
2428
2429 if (v->allow_fd_passing_output == b)
2430 return 0;
2431
2432 if (!b) {
2433 v->allow_fd_passing_output = false;
2434 return 1;
2435 }
2436
2437 r = verify_unix_socket(v);
2438 if (r < 0)
2439 return r;
2440
2441 v->allow_fd_passing_output = true;
2442 return 0;
2443 }
2444
2445 int varlink_server_new(VarlinkServer **ret, VarlinkServerFlags flags) {
2446 VarlinkServer *s;
2447
2448 assert_return(ret, -EINVAL);
2449 assert_return((flags & ~_VARLINK_SERVER_FLAGS_ALL) == 0, -EINVAL);
2450
2451 s = new(VarlinkServer, 1);
2452 if (!s)
2453 return log_oom_debug();
2454
2455 *s = (VarlinkServer) {
2456 .n_ref = 1,
2457 .flags = flags,
2458 .connections_max = varlink_server_connections_max(NULL),
2459 .connections_per_uid_max = varlink_server_connections_per_uid_max(NULL),
2460 };
2461
2462 *ret = s;
2463 return 0;
2464 }
2465
2466 static VarlinkServer* varlink_server_destroy(VarlinkServer *s) {
2467 char *m;
2468
2469 if (!s)
2470 return NULL;
2471
2472 varlink_server_shutdown(s);
2473
2474 while ((m = hashmap_steal_first_key(s->methods)))
2475 free(m);
2476
2477 hashmap_free(s->methods);
2478 hashmap_free(s->by_uid);
2479
2480 sd_event_unref(s->event);
2481
2482 free(s->description);
2483
2484 return mfree(s);
2485 }
2486
2487 DEFINE_TRIVIAL_REF_UNREF_FUNC(VarlinkServer, varlink_server, varlink_server_destroy);
2488
2489 static int validate_connection(VarlinkServer *server, const struct ucred *ucred) {
2490 int allowed = -1;
2491
2492 assert(server);
2493 assert(ucred);
2494
2495 if (FLAGS_SET(server->flags, VARLINK_SERVER_ROOT_ONLY))
2496 allowed = ucred->uid == 0;
2497
2498 if (FLAGS_SET(server->flags, VARLINK_SERVER_MYSELF_ONLY))
2499 allowed = allowed > 0 || ucred->uid == getuid();
2500
2501 if (allowed == 0) { /* Allow access when it is explicitly allowed or when neither
2502 * VARLINK_SERVER_ROOT_ONLY nor VARLINK_SERVER_MYSELF_ONLY are specified. */
2503 varlink_server_log(server, "Unprivileged client attempted connection, refusing.");
2504 return 0;
2505 }
2506
2507 if (server->n_connections >= server->connections_max) {
2508 varlink_server_log(server, "Connection limit of %u reached, refusing.", server->connections_max);
2509 return 0;
2510 }
2511
2512 if (FLAGS_SET(server->flags, VARLINK_SERVER_ACCOUNT_UID)) {
2513 unsigned c;
2514
2515 if (!uid_is_valid(ucred->uid)) {
2516 varlink_server_log(server, "Client with invalid UID attempted connection, refusing.");
2517 return 0;
2518 }
2519
2520 c = PTR_TO_UINT(hashmap_get(server->by_uid, UID_TO_PTR(ucred->uid)));
2521 if (c >= server->connections_per_uid_max) {
2522 varlink_server_log(server, "Per-UID connection limit of %u reached, refusing.",
2523 server->connections_per_uid_max);
2524 return 0;
2525 }
2526 }
2527
2528 return 1;
2529 }
2530
2531 static int count_connection(VarlinkServer *server, const struct ucred *ucred) {
2532 unsigned c;
2533 int r;
2534
2535 assert(server);
2536 assert(ucred);
2537
2538 server->n_connections++;
2539
2540 if (FLAGS_SET(server->flags, VARLINK_SERVER_ACCOUNT_UID)) {
2541 r = hashmap_ensure_allocated(&server->by_uid, NULL);
2542 if (r < 0)
2543 return log_debug_errno(r, "Failed to allocate UID hash table: %m");
2544
2545 c = PTR_TO_UINT(hashmap_get(server->by_uid, UID_TO_PTR(ucred->uid)));
2546
2547 varlink_server_log(server, "Connections of user " UID_FMT ": %u (of %u max)",
2548 ucred->uid, c, server->connections_per_uid_max);
2549
2550 r = hashmap_replace(server->by_uid, UID_TO_PTR(ucred->uid), UINT_TO_PTR(c + 1));
2551 if (r < 0)
2552 return log_debug_errno(r, "Failed to increment counter in UID hash table: %m");
2553 }
2554
2555 return 0;
2556 }
2557
2558 int varlink_server_add_connection(VarlinkServer *server, int fd, Varlink **ret) {
2559 _cleanup_(varlink_unrefp) Varlink *v = NULL;
2560 struct ucred ucred = UCRED_INVALID;
2561 bool ucred_acquired;
2562 int r;
2563
2564 assert_return(server, -EINVAL);
2565 assert_return(fd >= 0, -EBADF);
2566
2567 if ((server->flags & (VARLINK_SERVER_ROOT_ONLY|VARLINK_SERVER_ACCOUNT_UID)) != 0) {
2568 r = getpeercred(fd, &ucred);
2569 if (r < 0)
2570 return varlink_server_log_errno(server, r, "Failed to acquire peer credentials of incoming socket, refusing: %m");
2571
2572 ucred_acquired = true;
2573
2574 r = validate_connection(server, &ucred);
2575 if (r < 0)
2576 return r;
2577 if (r == 0)
2578 return -EPERM;
2579 } else
2580 ucred_acquired = false;
2581
2582 r = varlink_new(&v);
2583 if (r < 0)
2584 return varlink_server_log_errno(server, r, "Failed to allocate connection object: %m");
2585
2586 r = count_connection(server, &ucred);
2587 if (r < 0)
2588 return r;
2589
2590 v->fd = fd;
2591 if (server->flags & VARLINK_SERVER_INHERIT_USERDATA)
2592 v->userdata = server->userdata;
2593
2594 if (ucred_acquired) {
2595 v->ucred = ucred;
2596 v->ucred_acquired = true;
2597 }
2598
2599 _cleanup_free_ char *desc = NULL;
2600 if (asprintf(&desc, "%s-%i", server->description ?: "varlink", v->fd) >= 0)
2601 v->description = TAKE_PTR(desc);
2602
2603 /* Link up the server and the connection, and take reference in both directions. Note that the
2604 * reference on the connection is left dangling. It will be dropped when the connection is closed,
2605 * which happens in varlink_close(), including in the event loop quit callback. */
2606 v->server = varlink_server_ref(server);
2607 varlink_ref(v);
2608
2609 varlink_set_state(v, VARLINK_IDLE_SERVER);
2610
2611 if (server->event) {
2612 r = varlink_attach_event(v, server->event, server->event_priority);
2613 if (r < 0) {
2614 varlink_log_errno(v, r, "Failed to attach new connection: %m");
2615 v->fd = -EBADF; /* take the fd out of the connection again */
2616 varlink_close(v);
2617 return r;
2618 }
2619 }
2620
2621 if (ret)
2622 *ret = v;
2623
2624 return 0;
2625 }
2626
2627 static VarlinkServerSocket *varlink_server_socket_free(VarlinkServerSocket *ss) {
2628 if (!ss)
2629 return NULL;
2630
2631 free(ss->address);
2632 return mfree(ss);
2633 }
2634
2635 DEFINE_TRIVIAL_CLEANUP_FUNC(VarlinkServerSocket *, varlink_server_socket_free);
2636
2637 static int connect_callback(sd_event_source *source, int fd, uint32_t revents, void *userdata) {
2638 VarlinkServerSocket *ss = ASSERT_PTR(userdata);
2639 _cleanup_close_ int cfd = -EBADF;
2640 Varlink *v = NULL;
2641 int r;
2642
2643 assert(source);
2644
2645 varlink_server_log(ss->server, "New incoming connection.");
2646
2647 cfd = accept4(fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
2648 if (cfd < 0) {
2649 if (ERRNO_IS_ACCEPT_AGAIN(errno))
2650 return 0;
2651
2652 return varlink_server_log_errno(ss->server, errno, "Failed to accept incoming socket: %m");
2653 }
2654
2655 r = varlink_server_add_connection(ss->server, cfd, &v);
2656 if (r < 0)
2657 return 0;
2658
2659 TAKE_FD(cfd);
2660
2661 if (ss->server->connect_callback) {
2662 r = ss->server->connect_callback(ss->server, v, ss->server->userdata);
2663 if (r < 0) {
2664 varlink_log_errno(v, r, "Connection callback returned error, disconnecting client: %m");
2665 varlink_close(v);
2666 return 0;
2667 }
2668 }
2669
2670 return 0;
2671 }
2672
2673 static int varlink_server_create_listen_fd_socket(VarlinkServer *s, int fd, VarlinkServerSocket **ret_ss) {
2674 _cleanup_(varlink_server_socket_freep) VarlinkServerSocket *ss = NULL;
2675 int r;
2676
2677 assert(s);
2678 assert(fd >= 0);
2679 assert(ret_ss);
2680
2681 r = fd_nonblock(fd, true);
2682 if (r < 0)
2683 return r;
2684
2685 ss = new(VarlinkServerSocket, 1);
2686 if (!ss)
2687 return log_oom_debug();
2688
2689 *ss = (VarlinkServerSocket) {
2690 .server = s,
2691 .fd = fd,
2692 };
2693
2694 if (s->event) {
2695 r = sd_event_add_io(s->event, &ss->event_source, fd, EPOLLIN, connect_callback, ss);
2696 if (r < 0)
2697 return r;
2698
2699 r = sd_event_source_set_priority(ss->event_source, s->event_priority);
2700 if (r < 0)
2701 return r;
2702 }
2703
2704 *ret_ss = TAKE_PTR(ss);
2705 return 0;
2706 }
2707
2708 int varlink_server_listen_fd(VarlinkServer *s, int fd) {
2709 _cleanup_(varlink_server_socket_freep) VarlinkServerSocket *ss = NULL;
2710 int r;
2711
2712 assert_return(s, -EINVAL);
2713 assert_return(fd >= 0, -EBADF);
2714
2715 r = varlink_server_create_listen_fd_socket(s, fd, &ss);
2716 if (r < 0)
2717 return r;
2718
2719 LIST_PREPEND(sockets, s->sockets, TAKE_PTR(ss));
2720 return 0;
2721 }
2722
2723 int varlink_server_listen_address(VarlinkServer *s, const char *address, mode_t m) {
2724 _cleanup_(varlink_server_socket_freep) VarlinkServerSocket *ss = NULL;
2725 union sockaddr_union sockaddr;
2726 socklen_t sockaddr_len;
2727 _cleanup_close_ int fd = -EBADF;
2728 int r;
2729
2730 assert_return(s, -EINVAL);
2731 assert_return(address, -EINVAL);
2732 assert_return((m & ~0777) == 0, -EINVAL);
2733
2734 r = sockaddr_un_set_path(&sockaddr.un, address);
2735 if (r < 0)
2736 return r;
2737 sockaddr_len = r;
2738
2739 fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
2740 if (fd < 0)
2741 return -errno;
2742
2743 fd = fd_move_above_stdio(fd);
2744
2745 (void) sockaddr_un_unlink(&sockaddr.un);
2746
2747 WITH_UMASK(~m & 0777) {
2748 r = mac_selinux_bind(fd, &sockaddr.sa, sockaddr_len);
2749 if (r < 0)
2750 return r;
2751 }
2752
2753 if (listen(fd, SOMAXCONN) < 0)
2754 return -errno;
2755
2756 r = varlink_server_create_listen_fd_socket(s, fd, &ss);
2757 if (r < 0)
2758 return r;
2759
2760 r = free_and_strdup(&ss->address, address);
2761 if (r < 0)
2762 return r;
2763
2764 LIST_PREPEND(sockets, s->sockets, TAKE_PTR(ss));
2765 TAKE_FD(fd);
2766 return 0;
2767 }
2768
2769 void* varlink_server_set_userdata(VarlinkServer *s, void *userdata) {
2770 void *ret;
2771
2772 assert_return(s, NULL);
2773
2774 ret = s->userdata;
2775 s->userdata = userdata;
2776
2777 return ret;
2778 }
2779
2780 void* varlink_server_get_userdata(VarlinkServer *s) {
2781 assert_return(s, NULL);
2782
2783 return s->userdata;
2784 }
2785
2786 static VarlinkServerSocket* varlink_server_socket_destroy(VarlinkServerSocket *ss) {
2787 if (!ss)
2788 return NULL;
2789
2790 if (ss->server)
2791 LIST_REMOVE(sockets, ss->server->sockets, ss);
2792
2793 sd_event_source_disable_unref(ss->event_source);
2794
2795 free(ss->address);
2796 safe_close(ss->fd);
2797
2798 return mfree(ss);
2799 }
2800
2801 int varlink_server_shutdown(VarlinkServer *s) {
2802 assert_return(s, -EINVAL);
2803
2804 while (s->sockets)
2805 varlink_server_socket_destroy(s->sockets);
2806
2807 return 0;
2808 }
2809
2810 static int varlink_server_add_socket_event_source(VarlinkServer *s, VarlinkServerSocket *ss, int64_t priority) {
2811 _cleanup_(sd_event_source_unrefp) sd_event_source *es = NULL;
2812
2813 int r;
2814
2815 assert(s);
2816 assert(s->event);
2817 assert(ss);
2818 assert(ss->fd >= 0);
2819 assert(!ss->event_source);
2820
2821 r = sd_event_add_io(s->event, &es, ss->fd, EPOLLIN, connect_callback, ss);
2822 if (r < 0)
2823 return r;
2824
2825 r = sd_event_source_set_priority(es, priority);
2826 if (r < 0)
2827 return r;
2828
2829 ss->event_source = TAKE_PTR(es);
2830 return 0;
2831 }
2832
2833 int varlink_server_attach_event(VarlinkServer *s, sd_event *e, int64_t priority) {
2834 int r;
2835
2836 assert_return(s, -EINVAL);
2837 assert_return(!s->event, -EBUSY);
2838
2839 if (e)
2840 s->event = sd_event_ref(e);
2841 else {
2842 r = sd_event_default(&s->event);
2843 if (r < 0)
2844 return r;
2845 }
2846
2847 LIST_FOREACH(sockets, ss, s->sockets) {
2848 r = varlink_server_add_socket_event_source(s, ss, priority);
2849 if (r < 0)
2850 goto fail;
2851 }
2852
2853 s->event_priority = priority;
2854 return 0;
2855
2856 fail:
2857 varlink_server_detach_event(s);
2858 return r;
2859 }
2860
2861 int varlink_server_detach_event(VarlinkServer *s) {
2862 assert_return(s, -EINVAL);
2863
2864 LIST_FOREACH(sockets, ss, s->sockets)
2865 ss->event_source = sd_event_source_disable_unref(ss->event_source);
2866
2867 sd_event_unref(s->event);
2868 return 0;
2869 }
2870
2871 sd_event *varlink_server_get_event(VarlinkServer *s) {
2872 assert_return(s, NULL);
2873
2874 return s->event;
2875 }
2876
2877 int varlink_server_bind_method(VarlinkServer *s, const char *method, VarlinkMethod callback) {
2878 _cleanup_free_ char *m = NULL;
2879 int r;
2880
2881 assert_return(s, -EINVAL);
2882 assert_return(method, -EINVAL);
2883 assert_return(callback, -EINVAL);
2884
2885 if (startswith(method, "org.varlink.service."))
2886 return log_debug_errno(SYNTHETIC_ERRNO(EEXIST), "Cannot bind server to '%s'.", method);
2887
2888 m = strdup(method);
2889 if (!m)
2890 return log_oom_debug();
2891
2892 r = hashmap_ensure_put(&s->methods, &string_hash_ops, m, callback);
2893 if (r == -ENOMEM)
2894 return log_oom_debug();
2895 if (r < 0)
2896 return log_debug_errno(r, "Failed to register callback: %m");
2897 if (r > 0)
2898 TAKE_PTR(m);
2899
2900 return 0;
2901 }
2902
2903 int varlink_server_bind_method_many_internal(VarlinkServer *s, ...) {
2904 va_list ap;
2905 int r = 0;
2906
2907 assert_return(s, -EINVAL);
2908
2909 va_start(ap, s);
2910 for (;;) {
2911 VarlinkMethod callback;
2912 const char *method;
2913
2914 method = va_arg(ap, const char *);
2915 if (!method)
2916 break;
2917
2918 callback = va_arg(ap, VarlinkMethod);
2919
2920 r = varlink_server_bind_method(s, method, callback);
2921 if (r < 0)
2922 break;
2923 }
2924 va_end(ap);
2925
2926 return r;
2927 }
2928
2929 int varlink_server_bind_connect(VarlinkServer *s, VarlinkConnect callback) {
2930 assert_return(s, -EINVAL);
2931
2932 if (callback && s->connect_callback && callback != s->connect_callback)
2933 return log_debug_errno(SYNTHETIC_ERRNO(EBUSY), "A different callback was already set.");
2934
2935 s->connect_callback = callback;
2936 return 0;
2937 }
2938
2939 int varlink_server_bind_disconnect(VarlinkServer *s, VarlinkDisconnect callback) {
2940 assert_return(s, -EINVAL);
2941
2942 if (callback && s->disconnect_callback && callback != s->disconnect_callback)
2943 return log_debug_errno(SYNTHETIC_ERRNO(EBUSY), "A different callback was already set.");
2944
2945 s->disconnect_callback = callback;
2946 return 0;
2947 }
2948
2949 unsigned varlink_server_connections_max(VarlinkServer *s) {
2950 int dts;
2951
2952 /* If a server is specified, return the setting for that server, otherwise the default value */
2953 if (s)
2954 return s->connections_max;
2955
2956 dts = getdtablesize();
2957 assert_se(dts > 0);
2958
2959 /* Make sure we never use up more than ¾th of RLIMIT_NOFILE for IPC */
2960 if (VARLINK_DEFAULT_CONNECTIONS_MAX > (unsigned) dts / 4 * 3)
2961 return dts / 4 * 3;
2962
2963 return VARLINK_DEFAULT_CONNECTIONS_MAX;
2964 }
2965
2966 unsigned varlink_server_connections_per_uid_max(VarlinkServer *s) {
2967 unsigned m;
2968
2969 if (s)
2970 return s->connections_per_uid_max;
2971
2972 /* Make sure to never use up more than ¾th of available connections for a single user */
2973 m = varlink_server_connections_max(NULL);
2974 if (VARLINK_DEFAULT_CONNECTIONS_PER_UID_MAX > m)
2975 return m / 4 * 3;
2976
2977 return VARLINK_DEFAULT_CONNECTIONS_PER_UID_MAX;
2978 }
2979
2980 int varlink_server_set_connections_per_uid_max(VarlinkServer *s, unsigned m) {
2981 assert_return(s, -EINVAL);
2982 assert_return(m > 0, -EINVAL);
2983
2984 s->connections_per_uid_max = m;
2985 return 0;
2986 }
2987
2988 int varlink_server_set_connections_max(VarlinkServer *s, unsigned m) {
2989 assert_return(s, -EINVAL);
2990 assert_return(m > 0, -EINVAL);
2991
2992 s->connections_max = m;
2993 return 0;
2994 }
2995
2996 unsigned varlink_server_current_connections(VarlinkServer *s) {
2997 assert_return(s, UINT_MAX);
2998
2999 return s->n_connections;
3000 }
3001
3002 int varlink_server_set_description(VarlinkServer *s, const char *description) {
3003 assert_return(s, -EINVAL);
3004
3005 return free_and_strdup(&s->description, description);
3006 }
3007
3008 int varlink_server_serialize(VarlinkServer *s, FILE *f, FDSet *fds) {
3009 assert(f);
3010 assert(fds);
3011
3012 if (!s)
3013 return 0;
3014
3015 LIST_FOREACH(sockets, ss, s->sockets) {
3016 int copy;
3017
3018 assert(ss->address);
3019 assert(ss->fd >= 0);
3020
3021 fprintf(f, "varlink-server-socket-address=%s", ss->address);
3022
3023 /* If we fail to serialize the fd, it will be considered an error during deserialization */
3024 copy = fdset_put_dup(fds, ss->fd);
3025 if (copy < 0)
3026 return copy;
3027
3028 fprintf(f, " varlink-server-socket-fd=%i", copy);
3029
3030 fputc('\n', f);
3031 }
3032
3033 return 0;
3034 }
3035
3036 int varlink_server_deserialize_one(VarlinkServer *s, const char *value, FDSet *fds) {
3037 _cleanup_(varlink_server_socket_freep) VarlinkServerSocket *ss = NULL;
3038 _cleanup_free_ char *address = NULL;
3039 const char *v = ASSERT_PTR(value);
3040 int r, fd = -EBADF;
3041 char *buf;
3042 size_t n;
3043
3044 assert(s);
3045 assert(fds);
3046
3047 n = strcspn(v, " ");
3048 address = strndup(v, n);
3049 if (!address)
3050 return log_oom_debug();
3051
3052 if (v[n] != ' ')
3053 return log_debug_errno(SYNTHETIC_ERRNO(EINVAL),
3054 "Failed to deserialize VarlinkServerSocket: %s: %m", value);
3055 v = startswith(v + n + 1, "varlink-server-socket-fd=");
3056 if (!v)
3057 return log_debug_errno(SYNTHETIC_ERRNO(EINVAL),
3058 "Failed to deserialize VarlinkServerSocket fd %s: %m", value);
3059
3060 n = strcspn(v, " ");
3061 buf = strndupa_safe(v, n);
3062
3063 r = safe_atoi(buf, &fd);
3064 if (r < 0)
3065 return log_debug_errno(r, "Unable to parse VarlinkServerSocket varlink-server-socket-fd=%s: %m", buf);
3066 if (fd < 0)
3067 return log_debug_errno(SYNTHETIC_ERRNO(EINVAL),
3068 "VarlinkServerSocket varlink-server-socket-fd= has an invalid value: %d", fd);
3069 if (!fdset_contains(fds, fd))
3070 return log_debug_errno(SYNTHETIC_ERRNO(EBADF),
3071 "VarlinkServerSocket varlink-server-socket-fd= has unknown fd %d: %m", fd);
3072
3073 ss = new(VarlinkServerSocket, 1);
3074 if (!ss)
3075 return log_oom_debug();
3076
3077 *ss = (VarlinkServerSocket) {
3078 .server = s,
3079 .address = TAKE_PTR(address),
3080 .fd = fdset_remove(fds, fd),
3081 };
3082
3083 r = varlink_server_add_socket_event_source(s, ss, SD_EVENT_PRIORITY_NORMAL);
3084 if (r < 0)
3085 return log_debug_errno(r, "Failed to add VarlinkServerSocket event source to the event loop: %m");
3086
3087 LIST_PREPEND(sockets, s->sockets, TAKE_PTR(ss));
3088 return 0;
3089 }
Unnamed repository; edit this file 'description' to name the repository.